fix: orgs and teams overwrite behaviour when resolving env vars

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

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)

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()