Flexible working directory

constrain/api/workflow.py

@@ -5,7 +5,7 @@
 """
 
 import glob
-import sys, logging, json, os, datetime
+import sys, logging, json, os, datetime, platform
 from typing import Union
 
 sys.path.append("./constrain")
@@ -37,7 +37,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)
@@ -47,6 +46,46 @@ def __init__(self, workflow, run_workflow_now=False):
 
         self.load_states()
 
+        # First, check if the workflow_dict has "working_dir"
+        if "working_dir" not in self.workflow_dict:
+            logging.info("No working_dic 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:
+                    logging.error("working directory specified does not exist.")
+
+        # run workflow now
         if run_workflow_now:
             self.run_workflow()
 
@@ -301,6 +340,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,11 @@
+{
+    "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,160 @@
+import unittest, sys, logging, json, os
+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)
+                del 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],
+                "INFO:root:No working_dic 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.",
+            )
+
+    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.",
+            )
+
+    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"] = "./not_existing_path/test"
+
+            with open(json_case_path, "w") as f:
+                json.dump(workflow_dict, f)
+
+            workflow = Workflow(workflow=json_case_path)
+            self.assertEqual(
+                logobs.output[1],
+                "ERROR:root:working directory specified does not exist.",
+            )
+
+    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.",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()