Merge pull request #77 from seqeralabs/fix_env_variable_detection

Fix handling of complex environment variables

seqerakit/seqeraplatform.py

@@ -73,12 +73,13 @@ def _construct_command(self, cmd, *args, **kwargs):
     def _check_env_vars(self, command):
         full_cmd_parts = []
         for arg in command:
-            if arg.startswith("$"):
-                env_var = arg[1:]
-                if env_var not in os.environ:
-                    logging.error(f" Environment variable '{env_var}' is not set.")
-                    return None  # handle as desired
-                full_cmd_parts.append(os.environ[env_var])
+            if "$" in arg:
+                for env_var in re.findall(r"\$\{?[\w]+\}?", arg):
+                    if re.sub(r"[${}]", "", env_var) not in os.environ:
+                        raise EnvironmentError(
+                            f" Environment variable {env_var} not found!"
+                        )
+                full_cmd_parts.append(arg)
             else:
                 full_cmd_parts.append(shlex.quote(arg))
         return " ".join(full_cmd_parts)

tests/unit/test_seqeraplatform.py

@@ -3,6 +3,7 @@
 from seqerakit import seqeraplatform
 import json
 import subprocess
+import os
 
 
 class TestSeqeraPlatform(unittest.TestCase):
@@ -179,5 +180,47 @@ def test_dryrun_call(self, mock_subprocess):
         mock_subprocess.assert_not_called()
 
 
+class TestCheckEnvVars(unittest.TestCase):
+    def setUp(self):
+        self.sp = seqeraplatform.SeqeraPlatform()
+        self.original_environ = dict(os.environ)
+
+    def tearDown(self):
+        # Restore the original environment after each test
+        os.environ.clear()
+        os.environ.update(self.original_environ)
+
+    def test_with_set_env_vars(self):
+        # Set environment variables for the test
+        os.environ["VAR1"] = "value1"
+
+        command = ["tw", "pipelines", "list", "-w", "$VAR1"]
+        expected = "tw pipelines list -w $VAR1"
+        result = self.sp._check_env_vars(command)
+        self.assertEqual(result, expected)
+
+    def test_without_env_vars(self):
+        # Test case where there are no environment variables in the command
+        command = ["tw", "info"]
+        expected = "tw info"  # shlex.quote() will not alter these
+        result = self.sp._check_env_vars(command)
+        self.assertEqual(result, expected)
+
+    def test_error_raised_for_unset_env_vars(self):
+        # Unset environment variables for this test
+        unset_var = "{UNSET_VAR}"
+        if "UNSET_VAR" in os.environ:
+            del os.environ["UNSET_VAR"]
+
+        command = ["tw", "pipelines", "list", "-w", "${UNSET_VAR}"]
+
+        # Assert that EnvironmentError is raised
+        with self.assertRaises(EnvironmentError) as context:
+            self.sp._check_env_vars(command)
+        self.assertEqual(
+            str(context.exception), f" Environment variable ${unset_var} not found!"
+        )
+
+
 if __name__ == "__main__":
     unittest.main()