diff --git a/seqerakit/cli.py b/seqerakit/cli.py index 40859f4..d7207df 100644 --- a/seqerakit/cli.py +++ b/seqerakit/cli.py @@ -106,6 +106,11 @@ def parse_args(args=None): type=str, help="Path to a YAML file containing environment variables for configuration.", ) + yaml_processing.add_argument( + "--overwrite", + action="store_true", + help="Globally enable overwrite for all resources defined in YAML input(s).", + ) return parser.parse_args(args) @@ -157,8 +162,10 @@ def handle_block(self, block, args, destroy=False, dryrun=False): ), } - # Check if overwrite is set to True, and call overwrite handler - overwrite_option = args.get("overwrite", False) + # Check if overwrite is set to True or globally, and call overwrite handler + overwrite_option = args.get("overwrite", False) or getattr( + self.sp, "overwrite", False + ) if overwrite_option and not dryrun: logging.debug(f" Overwrite is set to 'True' for {block}\n") self.overwrite_method.handle_overwrite( @@ -239,6 +246,7 @@ def main(args=None): sp = seqeraplatform.SeqeraPlatform( cli_args=cli_args_list, dryrun=options.dryrun, json=options.json ) + sp.overwrite = options.overwrite # If global overwrite is set # If the info flag is set, run 'tw info' try: diff --git a/seqerakit/overwrite.py b/seqerakit/overwrite.py index d98d916..3fe9069 100644 --- a/seqerakit/overwrite.py +++ b/seqerakit/overwrite.py @@ -155,7 +155,7 @@ def _get_team_args(self, args): # Get the teamId from the json data team_id = utils.find_key_value_in_dict( - json.loads(json_out), "name", args["name"], "teamId" + json.loads(json_out), "name", utils.resolve_env_var(args["name"]), "teamId" ) return ("delete", "--id", str(team_id), "--organization", args["organization"]) @@ -277,7 +277,9 @@ def check_resource_exists(self, name_key, sp_args): Check if a resource exists in Seqera Platform by looking for the name and value in the json data generated from the list() method. """ - return utils.check_if_exists(self.cached_jsondata, name_key, sp_args["name"]) + return utils.check_if_exists( + self.cached_jsondata, name_key, utils.resolve_env_var(sp_args["name"]) + ) def delete_resource(self, block, operation, sp_args): """ @@ -313,12 +315,15 @@ def _find_workspace_id(self, organization, workspace_name): """ jsondata = json.loads(self.cached_jsondata) workspaces = jsondata["workspaces"] + for workspace in workspaces: - if ( - workspace.get("orgName") == organization - and workspace.get("workspaceName") == workspace_name + if workspace.get("orgName") == utils.resolve_env_var( + organization + ) and workspace.get("workspaceName") == utils.resolve_env_var( + workspace_name ): - return workspace.get("workspaceId") + workspace_id = workspace.get("workspaceId") + return workspace_id return None def _find_label_id(self, label_name, label_value): @@ -329,6 +334,8 @@ def _find_label_id(self, label_name, label_value): jsondata = json.loads(self.cached_jsondata) labels = jsondata["labels"] for label in labels: - if label.get("name") == label_name and label.get("value") == label_value: + if label.get("name") == utils.resolve_env_var(label_name) and label.get( + "value" + ) == utils.resolve_env_var(label_value): return label.get("id") return None diff --git a/seqerakit/utils.py b/seqerakit/utils.py index bfa0203..11b6326 100644 --- a/seqerakit/utils.py +++ b/seqerakit/utils.py @@ -67,11 +67,9 @@ def check_if_exists(json_data, namekey, namevalue): if not json_data: return False logging.info(f" Checking if {namekey} {namevalue} exists in Seqera Platform...") - # Regex pattern to match environment variables in the string - env_var_pattern = re.compile(r"\$\{?[\w]+\}?") # Substitute environment variables in namevalue - resolved_value = env_var_pattern.sub(replace_env_var, namevalue) + resolved_value = resolve_env_var(namevalue) data = json.loads(json_data) if find_key_value_in_dict(data, namekey, resolved_value, return_key=None): @@ -145,8 +143,8 @@ def read_file(file_path): for key, value in combined_params.items(): if isinstance(value, str): - expanded_value = re.sub(r"\$\{?\w+\}?", replace_env_var, value) - combined_params[key] = quoted_str(expanded_value) + resolved_value = resolve_env_var(value) + combined_params[key] = quoted_str(resolved_value) with tempfile.NamedTemporaryFile( mode="w", delete=False, suffix=".yaml" @@ -155,9 +153,29 @@ def read_file(file_path): return temp_file.name -def replace_env_var(match): - var_name = match.group().lstrip("$").strip("{}") - var_value = os.getenv(var_name) - if var_value is None: - raise EnvironmentError(f"Environment variable {var_name} not found") - return var_value +def resolve_env_var(value): + """ + Resolves environment variables in a string value. + Handles both $VAR and ${VAR} formats. + + Args: + value (str): The value that might contain environment variables + (e.g. "$MYVAR" or "${MYVAR}") + + Returns: + str: The resolved value with environment variables replaced + + Raises: + EnvironmentError: If an environment variable is not found + """ + if not isinstance(value, str) or "$" not in value: + return value + + def _replace(match): + var_name = match.group().lstrip("$").strip("{}") + var_value = os.getenv(var_name) + if var_value is None: + raise EnvironmentError(f"Environment variable {var_name} not found") + return var_value + + return re.sub(r"\$\{?\w+\}?", _replace, value) diff --git a/setup.py b/setup.py index 6bbd75e..01f188d 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import find_packages, setup -VERSION = "0.5.1" +VERSION = "0.5.2" with open("README.md") as f: readme = f.read() diff --git a/tests/unit/test_overwrite.py b/tests/unit/test_overwrite.py new file mode 100644 index 0000000..53da0e3 --- /dev/null +++ b/tests/unit/test_overwrite.py @@ -0,0 +1,173 @@ +import unittest +from unittest.mock import Mock, patch +import json +from seqerakit.overwrite import Overwrite +from seqerakit.seqeraplatform import ResourceExistsError + + +class TestOverwrite(unittest.TestCase): + def setUp(self): + self.mock_sp = Mock() + + # Mock context manager for suppress_output + self.mock_sp.suppress_output.return_value.__enter__ = Mock() + self.mock_sp.suppress_output.return_value.__exit__ = Mock() + + self.overwrite = Overwrite(self.mock_sp) + + self.sample_teams_json = json.dumps( + {"teams": [{"name": "test-team", "teamId": "123", "orgName": "test-org"}]} + ) + + self.sample_workspace_json = json.dumps( + { + "workspaces": [ + { + "workspaceId": "456", + "workspaceName": "test-workspace", + "orgName": "test-org", + } + ] + } + ) + + self.sample_labels_json = json.dumps( + {"labels": [{"id": "789", "name": "test-label", "value": "test-value"}]} + ) + + def test_handle_overwrite_generic_deletion(self): + # Test for credentials, secrets, compute-envs, datasets, actions, pipelines + args = ["--name", "test-resource", "--workspace", "test-workspace"] + + self.mock_sp.__getattr__("-o json").return_value = json.dumps( + {"name": "test-resource"} + ) + + self.overwrite.handle_overwrite("credentials", args, overwrite=True) + + self.mock_sp.credentials.assert_called_with( + "delete", "--name", "test-resource", "--workspace", "test-workspace" + ) + + def test_handle_overwrite_resource_exists_no_overwrite(self): + args = ["--name", "test-resource", "--workspace", "test-workspace"] + + # Mock JSON response indicating resource exists + self.mock_sp.__getattr__("-o json").return_value = json.dumps( + {"name": "test-resource"} + ) + + with self.assertRaises(ResourceExistsError): + self.overwrite.handle_overwrite("credentials", args, overwrite=False) + + def test_team_deletion(self): + args = {"name": "test-team", "organization": "test-org"} + + json_method_mock = Mock(side_effect=lambda *args: self.sample_teams_json) + + self.mock_sp.configure_mock(**{"-o json": json_method_mock}) + + team_args = self.overwrite._get_team_args(args) + + self.assertEqual( + team_args, ("delete", "--id", "123", "--organization", "test-org") + ) + + json_method_mock.assert_called_with("teams", "list", "-o", "test-org") + + # Test caching behavior + # Second call should use cached data + team_args_cached = self.overwrite._get_team_args(args) + + # Should return same result + self.assertEqual( + team_args_cached, ("delete", "--id", "123", "--organization", "test-org") + ) + + # But json_mock should only have been called once + self.assertEqual(json_method_mock.call_count, 1) + + def test_workspace_deletion(self): + args = ["--name", "test-workspace", "--organization", "test-org"] + + self.mock_sp.__getattr__("-o json").return_value = self.sample_workspace_json + + self.overwrite.handle_overwrite("workspaces", args, overwrite=True) + + self.mock_sp.workspaces.assert_called_with("delete", "--id", "456") + + def test_label_deletion(self): + args = [ + "--name", + "test-label", + "--value", + "test-value", + "--workspace", + "test-workspace", + ] + + self.mock_sp.__getattr__("-o json").return_value = self.sample_labels_json + + self.overwrite.handle_overwrite("labels", args, overwrite=True) + + self.mock_sp.labels.assert_called_with( + "delete", "--id", "789", "-w", "test-workspace" + ) + + def test_participant_deletion(self): + team_args = [ + "--name", + "test-team", + "--type", + "TEAM", + "--workspace", + "test-workspace", + ] + + self.mock_sp.__getattr__("-o json").return_value = json.dumps( + {"teamName": "test-team"} + ) + + self.overwrite.handle_overwrite("participants", team_args, overwrite=True) + + self.mock_sp.participants.assert_called_with( + "delete", + "--name", + "test-team", + "--type", + "TEAM", + "--workspace", + "test-workspace", + ) + + @patch("seqerakit.utils.resolve_env_var") + def test_organization_deletion_with_env_var(self, mock_resolve_env_var): + + args = ["--name", "${ORG_NAME}"] + + # Setup environment variable mock + mock_resolve_env_var.side_effect = lambda x: ( + "resolved-org-name" if x == "${ORG_NAME}" else x + ) + + # Create a mock for the json method that returns our JSON data + # The JSON response needs to match what check_if_exists method looks for + json_method_mock = Mock( + side_effect=lambda *args: json.dumps( + {"organizations": [{"orgName": "resolved-org-name"}]} + ) + ) + + self.mock_sp.configure_mock(**{"-o json": json_method_mock}) + + self.overwrite.handle_overwrite("organizations", args, overwrite=True) + + mock_resolve_env_var.assert_any_call("${ORG_NAME}") + + self.mock_sp.organizations.assert_called_with("delete", "--name", "${ORG_NAME}") + + +# TODO: tests for destroy and JSON caching + +if __name__ == "__main__": + unittest.main()