Flexible working directory

constrain/api/workflow.py

@@ -5,7 +5,8 @@
 """
 
 import glob
-import sys, logging, json, os, datetime
+import sys, logging, json, os, datetime, platform
+from pathlib import Path
 from typing import Union
 
 sys.path.append("./constrain")
@@ -37,7 +38,6 @@ def __init__(self, workflow, run_workflow_now=False):
         self.states = {}
         self.workflow_dict = {}
         self.running_sequence = []
-
         # checking workflow validity is handed over to the API method.
         if isinstance(workflow, str):
             self.load_workflow_json(workflow)
@@ -46,10 +46,68 @@ def __init__(self, workflow, run_workflow_now=False):
             self.workflow_dict = workflow
 
         self.load_states()
+        # change working dir
+        self.change_work_dir()
 
+        # run workflow now
         if run_workflow_now:
             self.run_workflow()
 
+    def change_work_dir(self) -> None:
+        # First, check if the workflow_dict has "working_dir"
+        if "working_dir" not in self.workflow_dict:
+            logging.info("No working_dir is specified")
+        else:
+            # Change the working path to the working_dir value in workflow_dict.
+            if not isinstance(self.workflow_dict["working_dir"], str):
+                # First, detect if the working dir is a valid string
+                logging.error("working directory specified is not a valid string.")
+            else:
+                # Then detect if the working dir provided is in Linux format or Windows Format.
+                if ("/" in self.workflow_dict["working_dir"]) and (
+                    "\\" not in self.workflow_dict["working_dir"]
+                ):
+                    # in Linux Format
+                    if (
+                        platform.system() == "Windows"
+                    ):  # convert it to the working platform if it is windows
+                        self.workflow_dict["working_dir"] = self.workflow_dict[
+                            "working_dir"
+                        ].replace("/", "\\")
+                    logging.info("the working dir provided is in Linux format.")
+                elif ("/" not in self.workflow_dict["working_dir"]) and (
+                    "\\" in self.workflow_dict["working_dir"]
+                ):
+                    # in windows format
+                    if (
+                        platform.system() == "Linux"
+                    ):  # convert it to the working platform if it is Linux
+                        self.workflow_dict["working_dir"] = self.workflow_dict[
+                            "working_dir"
+                        ].replace("\\", "/")
+                    logging.info("the working dir provided is in Win format.")
+
+                # change the working directory if it exists
+                if os.path.exists(self.workflow_dict["working_dir"]):
+                    os.chdir(self.workflow_dict["working_dir"])
+                    logging.info("Change current working path to the specified path.")
+                else:
+                    try:
+                        Path(self.workflow_dict["working_dir"]).mkdir(
+                            parents=True, exist_ok=True
+                        )
+                        logging.info(
+                            "working directory specified does not exist and create a new director."
+                        )
+                        os.chdir(self.workflow_dict["working_dir"])
+                    except Exception as e:
+                        # If an invalid escapse sequence string is specified. E.g. ".\tests\api\test"
+                        if e.winerror == 123:
+                            # The error is : OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect.
+                            logging.error(
+                                "The working directory specified is an invalid escape sequence string."
+                            )
+
     def validate(self, verbose: bool = False) -> bool:
         """function to be implemented to check for high level validity of the workflow definition"""
         workflow_schema = {
@@ -301,6 +359,7 @@ def create_workflow_engine(
         Returns:
             Union[None, WorkflowEngine]: Instantiated WorkflowEngine object if provided workflow is valid; None otherwise.
         """
+
         if (isinstance(workflow, str) and os.path.isfile(workflow)) or isinstance(
             workflow, dict
         ):

tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json

@@ -0,0 +1 @@
+{"workflow_name": "A minimal case", "meta": {"author": "ConStrain Team", "date": "03/23/2024", "version": "1.0", "description": "A minimal case to test if flexible call feature works"}, "imports": [], "states": {}}

tests/api/test_flexible_calling.py

@@ -0,0 +1,294 @@
+import unittest, sys, logging, json, os
+import platform
+from unittest.mock import patch
+
+import constrain
+from constrain.api import Workflow
+
+sys.path.append("./constrain")
+
+
+class TestFlexibleCalling(unittest.TestCase):
+    def test_no_dir_provided(self):
+        """This test checks when no working directory is provided,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Delete working_dir value in the json file
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict.pop("working_dir", None)
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            self.assertEqual(
+                logobs.output[0],
+                "INFO:root:No working_dir is specified",
+            )
+
+    def test_invalid_str(self):
+        """This test checks when working directory is not a valid string,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a invalid string
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = []
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            self.assertEqual(
+                logobs.output[0],
+                "ERROR:root:working directory specified is not a valid string.",
+            )
+
+    def test_Linux_path(self):
+        """This test check if the program can detect the working path provided is in Linux format."""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path in Linux format
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = "./tests/api/result"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            # change current working directory back
+            os.chdir("../../..")
+
+            self.assertEqual(
+                logobs.output[0],
+                "INFO:root:the working dir provided is in Linux format.",
+            )
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_Win_path(self):
+        """This test check if the program can detect the working path provided is in WIN format."""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path in Win format
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = ".\\tests\\api\\result"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            # change current working directory back
+            os.chdir("../../..")
+
+            self.assertEqual(
+                logobs.output[0],
+                "INFO:root:the working dir provided is in Win format.",
+            )
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_valid_dir(self):
+        """This test checks when a working directory is provided and it also points to the correct path,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = "./tests/api/result"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # change current working directory back
+            os.chdir("../../..")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+        with self.assertLogs() as logobs:
+            # Change working_dir value in the json file to a valid path in Win format
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = ".\\tests\\api\\result"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # Change current working directory back
+            os.chdir("../../..")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_dir_with_space(self):
+        """This test checks when a working directory with space is provided and it also points to the correct path,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = "./tests/api/result/dir space"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # change current working directory back
+            os.chdir("../../../../")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+        with self.assertLogs() as logobs:
+            # Change working_dir value in the json file to a valid path in Win format
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = ".\\tests\\api\\result\\dir space"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # Change current working directory back
+            os.chdir("../../../../")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_dir_simple(self):
+        """This test checks when a simple working directory without any "\\" or "/" is provided and it also points to the correct path,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = "./tests"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # change current working directory back
+            os.chdir("../")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+        with self.assertLogs() as logobs:
+            # Change working_dir value in the json file to a valid path in Win format
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = ".\\tests"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # Change current working directory back
+            os.chdir("../")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_valid_absolute_dir(self):
+        """This test checks when a absolute working directory is provided and it also points to the correct path,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a valid path
+            print(os.getcwd())
+
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                if platform.system() == "Windows":
+                    workflow_dict["working_dir"] = os.getcwd() + "\\tests\\api\\result"
+                else:
+                    workflow_dict["working_dir"] = os.getcwd() + "/tests/api/result"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+
+            # change current working directory back
+            os.chdir("../../..")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:Change current working path to the specified path.",
+            )
+
+    def test_dir_not_exist(self):
+        """This test checks when a valid wd is provided but it doesn't exist,
+        if the program will behave correctly"""
+        with self.assertLogs() as logobs:
+            json_case_path = "./tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json"
+
+            # Change working_dir value in the json file to a path that does not exist
+            with open(json_case_path, "r") as f:
+                workflow_dict = json.load(f)
+                workflow_dict["working_dir"] = "./tests/api/result/not_existing_path"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            # change current working directory back
+            os.chdir("../../../../")
+
+            # then delete this path.
+            # print(os.getcwd())
+
+            # But I always get error deleting this directory:
+            # os.remove("./tests/api/result/not_existing_path/test")
+
+            self.assertEqual(
+                logobs.output[1],
+                "INFO:root:working directory specified does not exist and create a new director.",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()