Merge branch 'main' into simple-linear

Author: kaliif

.gitignore

@@ -2,7 +2,9 @@
 dist/
 **/__pycache__/
 
-
 *~
 \#*#
 \#*
+
+**/*.pickle
+tests/project-root/project-*/

poetry.lock

@@ -14,6 +14,7 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.12"
 im-protobuf = "^7.1.0"
+im-data-manager-job-decoder = "^2.1.0"
 jsonschema = "^4.21.1"
 pyyaml = ">= 5.3.1, < 7.0"
 

pyproject.toml

@@ -1,4 +1,23 @@
+"""The UnitTest API Adapter.
+
+This 'simulates' the sort of responses you can expect from the DM API/Model.
+It stimulates a Database by using dictionaries that are pickled to (and unpickled from)
+the file system after acquiring a lock object. Pickling is required to
+(because it's a simple built-in mechanism in Python) to persist data
+between processes in the multi-processing framework we run in because we ultimately need
+to simulate the multi-pod messaging-based framework of the DM.
+
+A separate pickle file is used for each 'simulated' model table and the object
+initialiser resets all the pickle files (located in 'tests/pickle-files').
+
+Job definitions are loaded (statically) from the content of the
+'tests/job-definitions/job-definitions.yaml' file and yielded through the 'get_job()'
+method.
+"""
+
 import os
+from multiprocessing import Lock
+from pickle import Pickler, Unpickler
 from typing import Any, Dict, List, Optional
 
 import yaml
@@ -13,62 +32,162 @@
     _JOB_DEFINITIONS: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
 assert _JOB_DEFINITIONS
 
+# Table UUID formats
 _INSTANCE_ID_FORMAT: str = "instance-00000000-0000-0000-0000-{id:012d}"
+_TASK_ID_FORMAT: str = "task-00000000-0000-0000-0000-{id:012d}"
 _WORKFLOW_DEFINITION_ID_FORMAT: str = "workflow-00000000-0000-0000-0000-{id:012d}"
 _RUNNING_WORKFLOW_ID_FORMAT: str = "r-workflow-00000000-0000-0000-0000-{id:012d}"
 _RUNNING_WORKFLOW_STEP_ID_FORMAT: str = (
     "r-workflow-step-00000000-0000-0000-0000-{id:012d}"
 )
 
+# Pickle files (for each 'Table')
+_PICKLE_DIRECTORY: str = "tests/pickle-files"
+_WORKFLOW_PICKLE_FILE: str = f"{_PICKLE_DIRECTORY}/workflow.pickle"
+_RUNNING_WORKFLOW_PICKLE_FILE: str = f"{_PICKLE_DIRECTORY}/running-workflow.pickle"
+_RUNNING_WORKFLOW_STEP_PICKLE_FILE: str = (
+    f"{_PICKLE_DIRECTORY}/running-workflow-step.pickle"
+)
+_INSTANCE_PICKLE_FILE: str = f"{_PICKLE_DIRECTORY}/instance.pickle"
+_TASK_PICKLE_FILE: str = f"{_PICKLE_DIRECTORY}/task.pickle"
+
 
 class UnitTestAPIAdapter(APIAdapter):
     """A minimal API adapter. It serves-up Job Definitions
     from the job-definitions/job-definitions.yaml file and provides basic
-    (in-memory) storage for Workflow Definitions and related tables."""
+    storage for Workflow Definitions and related tables.
+
+    Because the adapter is used by the multi-processing test suite, it uses both a lock
+    and pickle files to store data, so that data can be shared between processes.
+    """
+
+    lock = Lock()
 
     def __init__(self):
         super().__init__()
-        # A map of workflow definitions, keyed by workflow definition ID.
-        self._workflow_definitions: Dict[str, Dict[str, Any]] = {}
-        self._running_workflow: Dict[str, Dict[str, Any]] = {}
-        self._running_workflow_steps: Dict[str, Dict[str, Any]] = {}
-        self._instances: Dict[str, Dict[str, Any]] = {}
-        self._tasks: Dict[str, Dict[str, Any]] = {}
+        # Safely initialise the pickle files
+        UnitTestAPIAdapter.lock.acquire()
+        if not os.path.exists(_PICKLE_DIRECTORY):
+            os.makedirs(_PICKLE_DIRECTORY)
+        for file in [
+            _WORKFLOW_PICKLE_FILE,
+            _RUNNING_WORKFLOW_PICKLE_FILE,
+            _RUNNING_WORKFLOW_STEP_PICKLE_FILE,
+            _INSTANCE_PICKLE_FILE,
+            _TASK_PICKLE_FILE,
+        ]:
+            with open(file, "wb") as pickle_file:
+                Pickler(pickle_file).dump({})
+        UnitTestAPIAdapter.lock.release()
 
     def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
-        next_id: int = len(self._workflow_definitions) + 1
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            workflow = Unpickler(pickle_file).load()
+
+        next_id: int = len(workflow) + 1
         workflow_definition_id: str = _WORKFLOW_DEFINITION_ID_FORMAT.format(id=next_id)
-        self._workflow_definitions[workflow_definition_id] = workflow_definition
+        workflow[workflow_definition_id] = workflow_definition
+
+        with open(_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(workflow)
+        UnitTestAPIAdapter.lock.release()
+
         return {"id": workflow_definition_id}
 
-    def get_workflow(self, *, workflow_definition_id: str) -> Dict[str, Any]:
-        if workflow_definition_id not in self._workflow_definitions:
-            return {}
-        return {"workflow": self._workflow_definitions[workflow_definition_id]}
+    def get_workflow(self, *, workflow_id: str) -> Dict[str, Any]:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            workflow = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
+        return {"workflow": workflow[workflow_id]} if workflow_id in workflow else {}
 
     def get_workflow_by_name(self, *, name: str, version: str) -> Dict[str, Any]:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            workflow = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
         item = {}
-        for wfid, value in self._workflow_definitions.items():
+        for wfid, value in workflow.items():
             if value["name"] == name:
                 item = {"id": wfid, "workflow": value}
         return item
 
-    def create_running_workflow(self, *, workflow_definition_id: str) -> str:
-        next_id: int = len(self._running_workflow) + 1
+    def create_running_workflow(
+        self,
+        *,
+        user_id: str,
+        workflow_id: str,
+        project_id: str,
+        variables: Dict[str, Any],
+    ) -> str:
+        assert user_id
+        assert isinstance(variables, dict)
+
+        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_definition_id}
-        self._running_workflow[running_workflow_id] = record
+        record = {
+            "user_id": user_id,
+            "done": False,
+            "success": False,
+            "workflow": workflow_id,
+            "project_id": project_id,
+            "variables": variables,
+        }
+        running_workflow[running_workflow_id] = record
+
+        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow)
+        UnitTestAPIAdapter.lock.release()
+
         return {"id": running_workflow_id}
 
+    def set_running_workflow_done(
+        self,
+        *,
+        running_workflow_id: str,
+        success: bool,
+        error: Optional[int] = None,
+        error_msg: Optional[str] = None,
+    ) -> None:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow = Unpickler(pickle_file).load()
+
+        assert running_workflow_id in running_workflow
+        running_workflow[running_workflow_id]["done"] = True
+        running_workflow[running_workflow_id]["success"] = success
+        running_workflow[running_workflow_id]["error"] = error
+        running_workflow[running_workflow_id]["error_msg"] = error_msg
+
+        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow)
+        UnitTestAPIAdapter.lock.release()
+
     def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
-        if running_workflow_id not in self._running_workflow:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
+        if running_workflow_id not in running_workflow:
             return {}
-        return {"running_workflow": self._running_workflow[running_workflow_id]}
+        return {"running_workflow": running_workflow[running_workflow_id]}
 
     def create_running_workflow_step(
         self, *, running_workflow_id: str, step: str
     ) -> str:
-        next_id: int = len(self._running_workflow_steps) + 1
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+
+        next_id: int = len(running_workflow_step) + 1
         running_workflow_step_id: str = _RUNNING_WORKFLOW_STEP_ID_FORMAT.format(
             id=next_id
         )
@@ -78,64 +197,124 @@ def create_running_workflow_step(
             "success": False,
             "running_workflow": running_workflow_id,
         }
-        self._running_workflow_steps[running_workflow_step_id] = record
+        running_workflow_step[running_workflow_step_id] = record
+
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow_step)
+        UnitTestAPIAdapter.lock.release()
+
         return {"id": running_workflow_step_id}
 
     def get_running_workflow_step(
         self, *, running_workflow_step_id: str
     ) -> Dict[str, Any]:
-        if running_workflow_step_id not in self._running_workflow_steps:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
+        if running_workflow_step_id not in running_workflow_step:
             return {}
         return {
-            "running_workflow_step": self._running_workflow_steps[
-                running_workflow_step_id
-            ]
+            "running_workflow_step": running_workflow_step[running_workflow_step_id]
         }
 
+    def set_running_workflow_step_done(
+        self,
+        *,
+        running_workflow_step_id: str,
+        success: bool,
+        error: Optional[int] = None,
+        error_msg: Optional[str] = None,
+    ) -> None:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+
+        assert running_workflow_step_id in running_workflow_step
+        running_workflow_step[running_workflow_step_id]["done"] = True
+        running_workflow_step[running_workflow_step_id]["success"] = success
+        running_workflow_step[running_workflow_step_id]["error"] = error
+        running_workflow_step[running_workflow_step_id]["error_msg"] = error_msg
+
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow_step)
+        UnitTestAPIAdapter.lock.release()
+
     def get_running_workflow_steps(
         self, *, running_workflow_id: str
     ) -> List[Dict[str, Any]]:
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
         steps = []
-        for key, value in self._running_workflow_steps.items():
+        for key, value in running_workflow_step.items():
             if value["running_workflow"] == running_workflow_id:
                 item = {"running_workflow_step": value, "id": key}
                 steps.append(item)
         return {"count": len(steps), "running_workflow_steps": steps}
 
     def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
-        next_id: int = len(self._instances) + 1
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
+            instances = Unpickler(pickle_file).load()
+
+        next_id: int = len(instances) + 1
         instance_id: str = _INSTANCE_ID_FORMAT.format(id=next_id)
         record = {
             "running_workflow_step": running_workflow_step_id,
         }
-        self._instances[instance_id] = record
+        instances[instance_id] = record
+
+        with open(_INSTANCE_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(instances)
+        UnitTestAPIAdapter.lock.release()
+
         return {"id": instance_id}
 
     def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
-        if instance_id not in self._instances:
-            return {}
-        return self._instances[instance_id]
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
+            instances = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
+        return {} if instance_id not in instances else instances[instance_id]
 
     def create_task(self, *, instance_id: str) -> Dict[str, Any]:
-        next_id: int = len(self._instances) + 1
-        task_id: str = _INSTANCE_ID_FORMAT.format(id=next_id)
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
+            tasks = Unpickler(pickle_file).load()
+
+        next_id: int = len(tasks) + 1
+        task_id: str = _TASK_ID_FORMAT.format(id=next_id)
         record = {
             "done": False,
             "exit_code": 0,
         }
-        self._tasks[task_id] = record
+        tasks[task_id] = record
+
+        with open(_TASK_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(tasks)
+        UnitTestAPIAdapter.lock.release()
+
         return {"id": task_id}
 
     def get_task(self, *, task_id: str) -> Dict[str, Any]:
-        if task_id not in self._tasks:
-            return {}
-        return self._tasks[task_id]
+        UnitTestAPIAdapter.lock.acquire()
+        with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
+            tasks = Unpickler(pickle_file).load()
+        UnitTestAPIAdapter.lock.release()
+
+        return {} if task_id not in tasks else tasks[task_id]
 
     def get_job(
         self, *, collection: str, job: str, version: str
     ) -> Optional[Dict[str, Any]]:
         assert collection == _JOB_DEFINITIONS["collection"]
         assert job in _JOB_DEFINITIONS["jobs"]
+        assert version
 
         jd = _JOB_DEFINITIONS["jobs"][job]
         response = {"command": jd["command"]}

tests/api_adapter.py

@@ -0,0 +1,5 @@
+"""An emulation of the Data Manager config module.
+A module of common constants."""
+
+# The project ID used in the tests (there is only one atm)
+TEST_PROJECT_ID: str = "project-00000000-0000-0000-0000-000000000001"