feat: Work toward use of project directory

Author: Alan Christie

tests/api_adapter.py

@@ -98,14 +98,19 @@ def get_workflow_by_name(self, *, name: str, version: str) -> Dict[str, Any]:
                 item = {"id": wfid, "workflow": value}
         return item
 
-    def create_running_workflow(self, *, workflow_id: str) -> str:
+    def create_running_workflow(self, *, workflow_id: str, project_id: str) -> str:
         UnitTestAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             running_workflow = Unpickler(pickle_file).load()
 
         next_id: int = len(running_workflow) + 1
         running_workflow_id: str = _RUNNING_WORKFLOW_ID_FORMAT.format(id=next_id)
-        record = {"done": False, "success": False, "workflow": workflow_id}
+        record = {
+            "done": False,
+            "success": False,
+            "workflow": workflow_id,
+            "project_id": project_id,
+        }
         running_workflow[running_workflow_id] = record
 
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:

tests/instance_launcher.py

@@ -52,6 +52,9 @@ def launch(
         assert workflow_id
         assert step_specification
 
+        # Project must be `project-00000000-0000-0000-0000-000000000001`
+        assert project_id == "project-00000000-0000-0000-0000-000000000001"
+
         # We're passed a RunningWorkflowStep ID but a record is expected to have been
         # created bt the caller, we simply create instance records.
         response = self._api_adapter.get_running_workflow_step(
@@ -66,15 +69,22 @@ def launch(
         response = self._api_adapter.create_task(instance_id=instance_id)
         task_id = response["id"]
 
+        # Where to run the job (i.e. in the project directory)
+        execution_directory = f"project-root/{project_id}"
+        os.makedirs(execution_directory, exist_ok=True)
+
         # Just run the Python module that matched the 'job' in the step specification.
-        # Don't care about 'version' or 'collection'.
+        # Don't care about 'version' or 'collection'. It will be relative to the
+        # execution directory.
         job: str = step_specification["job"]
         job_module = f"{_JOB_DIRECTORY}/{job}.py"
         assert os.path.isfile(job_module)
 
         job_cmd: List[str] = ["python", job_module]
-        print(f"Running job command: {job_cmd}")
-        completed_process: CompletedProcess = subprocess.run(job_cmd, check=False)
+        print(f"Running job command: {job_module}")
+        completed_process: CompletedProcess = subprocess.run(
+            job_cmd, check=False, cwd=execution_directory
+        )
 
         # Simulate a PodMessage (that will contain the instance ID),
         # filling-in only the fields that are of use to the Engine.

tests/project-root/README.txt

@@ -0,0 +1,4 @@
+This is the directory used as the working directory for any workflow Job executions.
+Any and all files generated by Jobs will be placed in a project directory here.
+So expect to find a temporary project directory called
+`project-00000000-0000-0000-0000-000000000001` with files in it here.

tests/test_test_api_adapter.py

@@ -17,7 +17,7 @@ def test_get_nop_job():
     )
 
     # Assert
-    assert jd["command"] == "python --version"
+    assert jd["command"] == "nop.py"
 
 
 def test_get_unknown_workflow():
@@ -74,7 +74,10 @@ def test_create_running_workflow():
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
 
     # Act
-    response = utaa.create_running_workflow(workflow_id=response["id"])
+    response = utaa.create_running_workflow(
+        workflow_id=response["id"],
+        project_id="project-00000000-0000-0000-0000-000000000001",
+    )
 
     # Assert
     assert response["id"] == "r-workflow-00000000-0000-0000-0000-000000000001"
@@ -85,7 +88,9 @@ def test_get_running_workflow():
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
     wfid = response["id"]
-    response = utaa.create_running_workflow(workflow_id=wfid)
+    response = utaa.create_running_workflow(
+        workflow_id=wfid, project_id="project-00000000-0000-0000-0000-000000000001"
+    )
     rwfid = response["id"]
 
     # Act
@@ -101,7 +106,10 @@ def test_set_running_workflow_done_when_success():
     # Arrange
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
-    response = utaa.create_running_workflow(workflow_id=response["id"])
+    response = utaa.create_running_workflow(
+        workflow_id=response["id"],
+        project_id="project-00000000-0000-0000-0000-000000000001",
+    )
     rwfid = response["id"]
 
     # Act
@@ -117,7 +125,10 @@ def test_set_running_workflow_done_when_failed():
     # Arrange
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
-    response = utaa.create_running_workflow(workflow_id=response["id"])
+    response = utaa.create_running_workflow(
+        workflow_id=response["id"],
+        project_id="project-00000000-0000-0000-0000-000000000001",
+    )
     rwfid = response["id"]
 
     # Act
@@ -133,7 +144,10 @@ def test_create_running_workflow_step():
     # Arrange
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
-    response = utaa.create_running_workflow(workflow_id=response["id"])
+    response = utaa.create_running_workflow(
+        workflow_id=response["id"],
+        project_id="project-00000000-0000-0000-0000-000000000001",
+    )
 
     # Act
     response = utaa.create_running_workflow_step(
@@ -149,7 +163,9 @@ def test_get_running_workflow_step():
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
     wfid = response["id"]
-    response = utaa.create_running_workflow(workflow_id=wfid)
+    response = utaa.create_running_workflow(
+        workflow_id=wfid, project_id="project-00000000-0000-0000-0000-000000000001"
+    )
     rwfid = response["id"]
     response = utaa.create_running_workflow_step(
         running_workflow_id=rwfid, step="step-1"
@@ -171,7 +187,9 @@ def test_get_running_workflow_steps():
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
     wfid = response["id"]
-    response = utaa.create_running_workflow(workflow_id=wfid)
+    response = utaa.create_running_workflow(
+        workflow_id=wfid, project_id="project-00000000-0000-0000-0000-000000000001"
+    )
     rwfid = response["id"]
     response = utaa.create_running_workflow_step(
         running_workflow_id=rwfid, step="step-1"

tests/test_test_instance_launcher.py

@@ -26,16 +26,19 @@ def test_get_nop_job(basic_launcher):
     utaa = basic_launcher[0]
     launcher = basic_launcher[1]
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
-    response = utaa.create_running_workflow(workflow_id=response["id"])
+    response = utaa.create_running_workflow(
+        workflow_id=response["id"],
+        project_id="project-00000000-0000-0000-0000-000000000001",
+    )
     response = utaa.create_running_workflow_step(
         running_workflow_id=response["id"], step="step-1"
     )
     rwfsid = response["id"]
 
     # Act
     result = launcher.launch(
-        project_id="project-00000000-0000-0000-0000-000000000000",
-        workflow_id="workflow-00000000-0000-0000-0000-000000000000",
+        project_id="project-00000000-0000-0000-0000-000000000001",
+        workflow_id="workflow-00000000-0000-0000-0000-000000000001",
         running_workflow_step_id=rwfsid,
         workflow_definition={},
         step_specification={"job": "nop", "variables": {"x": 1}},

tests/test_workflow_engine_examples.py

@@ -71,7 +71,9 @@ def start_workflow(md, da, workflow_file_name) -> str:
     assert wfid
     print(f"Created workflow definition {wfid}")
     # 2.
-    response = da.create_running_workflow(workflow_id=wfid)
+    response = da.create_running_workflow(
+        workflow_id=wfid, project_id="project-00000000-0000-0000-0000-000000000001"
+    )
     r_wfid = response["id"]
     assert r_wfid
     print(f"Created running workflow {r_wfid}")

workflow/workflow_abc.py

@@ -96,6 +96,7 @@ def create_running_workflow(
         self,
         *,
         workflow_id: str,
+        project_id: str,
     ) -> Dict[str, Any]:
         """Create a RunningWorkflow Record (from a Workflow)"""
         # Should return:
@@ -117,7 +118,8 @@ def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
         #    "running_workflow": {
         #       "done": False,
         #       "success": false,
-        #       "workflow": "workflow-000",
+        #       "workflow": {"id": "workflow-000"},
+        #       "project_id": "project-000",
         #    }
         # }
 

workflow/workflow_engine.py

@@ -77,6 +77,7 @@ def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
             assert "running_workflow" in response
             running_workflow = response["running_workflow"]
             _LOGGER.info("RunningWorkflow: %s", running_workflow)
+            project_id = running_workflow["project_id"]
             workflow_id = running_workflow["workflow"]["id"]
             response = self._api_adapter.get_workflow(workflow_id=workflow_id)
             assert "workflow" in response
@@ -94,7 +95,7 @@ def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
             step = workflow["steps"][0]
             step_specification: Dict[str, Any] = ast.literal_eval(step["specification"])
             self._instance_launcher.launch(
-                project_id="project-000",
+                project_id=project_id,
                 workflow_id=workflow_id,
                 running_workflow_step_id=running_workflow_step_id,
                 workflow_definition=workflow,
@@ -162,6 +163,7 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
         response = self._api_adapter.get_running_workflow(
             running_workflow_id=running_workflow_id
         )
+        project_id = response["running_workflow"]["project_id"]
         workflow_id = response["running_workflow"]["workflow"]["id"]
         assert workflow_id
         response = self._api_adapter.get_workflow(workflow_id=workflow_id)
@@ -189,7 +191,7 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
                             next_step["specification"]
                         )
                         self._instance_launcher.launch(
-                            project_id="project-000",
+                            project_id=project_id,
                             workflow_id=workflow_id,
                             running_workflow_step_id=running_workflow_step_id,
                             workflow_definition=workflow,