feat: Launcher now uses API adapter (and better API adapter & tests)

Alan Christie · Alan Christie · commit b9fe02ebac47 · 2024-10-28T14:23:29.000+01:00
diff --git a/tests/api_adapter.py b/tests/api_adapter.py
@@ -13,6 +13,7 @@
     _JOB_DEFINITIONS: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
 assert _JOB_DEFINITIONS
 
+_INSTANCE_ID_FORMAT: str = "instance-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 = (
@@ -31,8 +32,9 @@ def __init__(self):
         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]] = {}
 
-    def save_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
+    def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
         next_id: int = len(self._workflow_definitions) + 1
         workflow_definition_id: str = _WORKFLOW_DEFINITION_ID_FORMAT.format(id=next_id)
         self._workflow_definitions[workflow_definition_id] = workflow_definition
@@ -60,17 +62,20 @@ def create_running_workflow(self, *, workflow_definition_id: str) -> str:
     def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
         if running_workflow_id not in self._running_workflow:
             return {}
-        return {"running-workflow": self._running_workflow[running_workflow_id]}
+        return {"running_workflow": self._running_workflow[running_workflow_id]}
 
-    def create_running_workflow_step(self, *, running_workflow_id: str) -> str:
+    def create_running_workflow_step(
+        self, *, running_workflow_id: str, step: str
+    ) -> str:
         next_id: int = len(self._running_workflow_steps) + 1
         running_workflow_step_id: str = _RUNNING_WORKFLOW_STEP_ID_FORMAT.format(
             id=next_id
         )
         record = {
+            "step": step,
             "done": False,
             "success": False,
-            "running-workflow": running_workflow_id,
+            "running_workflow": running_workflow_id,
         }
         self._running_workflow_steps[running_workflow_step_id] = record
         return {"id": running_workflow_step_id}
@@ -80,17 +85,35 @@ def get_running_workflow_step(
     ) -> Dict[str, Any]:
         if running_workflow_step_id not in self._running_workflow_steps:
             return {}
-        return {"id": self._running_workflow_steps[running_workflow_step_id]}
+        return {
+            "running_workflow_step": self._running_workflow_steps[
+                running_workflow_step_id
+            ]
+        }
 
     def get_running_workflow_steps(
         self, *, running_workflow_id: str
     ) -> List[Dict[str, Any]]:
         steps = []
         for key, value in self._running_workflow_steps.items():
-            if value["running-workflow"] == running_workflow_id:
-                item = {"running-workflow-step": value, "id": key}
+            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}
+        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
+        instance_id: str = _INSTANCE_ID_FORMAT.format(id=next_id)
+        record = {
+            "running_workflow_step": running_workflow_step_id,
+        }
+        self._instances[instance_id] = record
+        return {"instance_id": running_workflow_step_id}
+
+    def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
+        if instance_id not in self._instances:
+            return {}
+        return {self._instances[instance_id]}
 
     def get_job(
         self, *, collection: str, job: str, version: str
diff --git a/tests/instance_launcher.py b/tests/instance_launcher.py
@@ -5,6 +5,7 @@
 
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
+from tests.api_adapter import UnitTestAPIAdapter
 from tests.message_dispatcher import UnitTestMessageDispatcher
 from workflow.workflow_abc import InstanceLauncher, LaunchResult
 
@@ -18,9 +19,12 @@ class UnitTestInstanceLauncher(InstanceLauncher):
     'end of instance' PodMessage (to the WorkflowEngine).
     """
 
-    def __init__(self, msg_dispatcher: UnitTestMessageDispatcher):
+    def __init__(
+        self, api_adapter: UnitTestAPIAdapter, msg_dispatcher: UnitTestMessageDispatcher
+    ):
         super().__init__()
 
+        self._api_adapter = api_adapter
         self._msg_dispatcher = msg_dispatcher
 
     def launch(
diff --git a/tests/test_test_api_adapter.py b/tests/test_test_api_adapter.py
@@ -33,21 +33,21 @@ def test_get_unknown_workflow():
     assert wfd == {}
 
 
-def test_save_workflow():
+def test_create_workflow():
     # Arrange
     utaa = UnitTestAPIAdapter()
 
     # Act
-    wfid = utaa.save_workflow(workflow_definition={"name": "blah"})
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
 
     # Assert
-    assert wfid == {"id": "workflow-00000000-0000-0000-0000-000000000001"}
+    assert response["id"] == "workflow-00000000-0000-0000-0000-000000000001"
 
 
 def test_get_workflow():
     # Arrange
     utaa = UnitTestAPIAdapter()
-    response = utaa.save_workflow(workflow_definition={"name": "blah"})
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
     wfid = response["id"]
 
     # Act
@@ -60,11 +60,101 @@ def test_get_workflow():
 def test_get_workflow_by_name():
     # Arrange
     utaa = UnitTestAPIAdapter()
-    _ = utaa.save_workflow(workflow_definition={"name": "blah"})
+    _ = utaa.create_workflow(workflow_definition={"name": "blah"})
 
     # Act
     response = utaa.get_workflow_by_name(name="blah", version="1.0.0")
 
     # Assert
     assert response["workflow"]["name"] == "blah"
     assert "id" in response
+
+
+def test_create_running_workflow():
+    # Arrange
+    utaa = UnitTestAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+
+    # Act
+    response = utaa.create_running_workflow(workflow_definition_id=response["id"])
+
+    # Assert
+    assert response["id"] == "r-workflow-00000000-0000-0000-0000-000000000001"
+
+
+def test_get_running_workflow():
+    # Arrange
+    utaa = UnitTestAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    wfid = response["id"]
+    response = utaa.create_running_workflow(workflow_definition_id=wfid)
+    rwfid = response["id"]
+
+    # Act
+    response = utaa.get_running_workflow(running_workflow_id=rwfid)
+
+    # Assert
+    rwf = response["running_workflow"]
+    assert not rwf["done"]
+    assert rwf["workflow"] == wfid
+
+
+def test_create_running_workflow_step():
+    # Arrange
+    utaa = UnitTestAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    response = utaa.create_running_workflow(workflow_definition_id=response["id"])
+
+    # Act
+    response = utaa.create_running_workflow_step(
+        running_workflow_id=response["id"], step="step-1"
+    )
+
+    # Assert
+    assert response["id"] == "r-workflow-step-00000000-0000-0000-0000-000000000001"
+
+
+def test_get_running_workflow_step():
+    # Arrange
+    utaa = UnitTestAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    wfid = response["id"]
+    response = utaa.create_running_workflow(workflow_definition_id=wfid)
+    rwfid = response["id"]
+    response = utaa.create_running_workflow_step(
+        running_workflow_id=rwfid, step="step-1"
+    )
+    rwfsid = response["id"]
+
+    # Act
+    response = utaa.get_running_workflow_step(running_workflow_step_id=rwfsid)
+
+    # Assert
+    rwfs = response["running_workflow_step"]
+    assert rwfs["step"] == "step-1"
+    assert not rwfs["done"]
+    assert rwfs["running_workflow"] == rwfid
+
+
+def test_get_running_workflow_steps():
+    # Arrange
+    utaa = UnitTestAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    wfid = response["id"]
+    response = utaa.create_running_workflow(workflow_definition_id=wfid)
+    rwfid = response["id"]
+    response = utaa.create_running_workflow_step(
+        running_workflow_id=rwfid, step="step-1"
+    )
+    rwfsid = response["id"]
+
+    # Act
+    response = utaa.get_running_workflow_steps(running_workflow_id=rwfid)
+
+    # Assert
+    assert response["count"] == 1
+    rwfs = response["running_workflow_steps"][0]
+    assert rwfs["id"] == rwfsid
+    assert rwfs["running_workflow_step"]["step"] == "step-1"
+    assert not rwfs["running_workflow_step"]["done"]
+    assert rwfs["running_workflow_step"]["running_workflow"] == rwfid
diff --git a/tests/test_test_instance_launcher.py b/tests/test_test_instance_launcher.py
@@ -4,16 +4,20 @@
 
 pytestmark = pytest.mark.unit
 
+from tests.api_adapter import UnitTestAPIAdapter
 from tests.instance_launcher import UnitTestInstanceLauncher
 from tests.message_dispatcher import UnitTestMessageDispatcher
 from tests.message_queue import UnitTestMessageQueue
 
 
 @pytest.fixture
 def basic_launcher():
-    utmq = UnitTestMessageQueue()
-    utmd = UnitTestMessageDispatcher(msg_queue=utmq)
-    return UnitTestInstanceLauncher(msg_dispatcher=utmd)
+    api_adapter = UnitTestAPIAdapter()
+    message_queue = UnitTestMessageQueue()
+    message_dispatcher = UnitTestMessageDispatcher(msg_queue=message_queue)
+    return UnitTestInstanceLauncher(
+        api_adapter=api_adapter, msg_dispatcher=message_dispatcher
+    )
 
 
 def test_get_nop_job(basic_launcher):
diff --git a/tests/test_workflow_engine_with_examples.py b/tests/test_workflow_engine_with_examples.py
@@ -20,7 +20,9 @@ def basic_engine():
     api_adapter = UnitTestAPIAdapter()
     message_queue = UnitTestMessageQueue()
     message_dispatcher = UnitTestMessageDispatcher(msg_queue=message_queue)
-    instance_launcher = UnitTestInstanceLauncher(msg_dispatcher=message_dispatcher)
+    instance_launcher = UnitTestInstanceLauncher(
+        api_adapter=api_adapter, msg_dispatcher=message_dispatcher
+    )
     return [
         api_adapter,
         message_dispatcher,
@@ -34,7 +36,7 @@ def test_workflow_engine_with_example_1(basic_engine):
     # LOAD THE EXAMPLE-1 WORKFLOW DEFINITION INTO THE DATABASE
     # TODO
     # SIMULATE THE API CREATION OF A RUNNING WORKFLOW FROM THE WORKFLOW
-    wfid = da.save_workflow(workflow_definition={"name": "example-1"})
+    wfid = da.create_workflow(workflow_definition={"name": "example-1"})
     assert wfid
     response = da.create_running_workflow(workflow_definition_id=wfid)
     r_wfid = response["id"]
@@ -57,7 +59,7 @@ def test_workflow_engine_with_example_1(basic_engine):
     r_wf = None
     while not done:
         response = da.get_running_workflow(running_workflow_id=r_wfid)
-        r_wf = response["running-workflow"]
+        r_wf = response["running_workflow"]
         if r_wf["done"]:
             done = True
         else:
@@ -73,6 +75,6 @@ def test_workflow_engine_with_example_1(basic_engine):
     response = da.get_running_workflow_steps(running_workflow_id=r_wfid)
     # TODO - The following should not be zero but the implementation does not set it yet
     assert response["count"] == 0
-    for step in response["running-workflow-steps"]:
-        assert step["running-workflow-step"]["done"]
-        assert step["running-workflow-step"]["success"]
+    for step in response["running_workflow_steps"]:
+        assert step["running_workflow_step"]["done"]
+        assert step["running_workflow_step"]["success"]
diff --git a/workflow/workflow_abc.py b/workflow/workflow_abc.py
@@ -61,12 +61,12 @@ class APIAdapter(ABC):
     records returning dictionary (API-like) responses."""
 
     @abstractmethod
-    def save_workflow(
+    def create_workflow(
         self,
         *,
         workflow_definition: Dict[str, Any],
-    ) -> str:
-        """Save a Workflow, getting an ID in return"""
+    ) -> Dict[str, Any]:
+        """Create a Workflow, getting an ID in return"""
         # Should return:
         # {
         #    "id": "workflow-00000000-0000-0000-0000-000000000001",
@@ -103,7 +103,7 @@ def create_running_workflow(
         self,
         *,
         workflow_definition_id: str,
-    ) -> str:
+    ) -> Dict[str, Any]:
         """Create a RunningWorkflow Record (from a Workflow)"""
         # Should return:
         # {
@@ -115,7 +115,7 @@ def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
         """Get a RunningWorkflow Record"""
         # Should return:
         # {
-        #    "running-workflow": {
+        #    "running_workflow": {
         #       "done": False,
         #       "success": false,
         #       "workflow": "workflow-000",
@@ -127,7 +127,7 @@ def create_running_workflow_step(
         self,
         *,
         running_workflow_id: str,
-    ) -> str:
+    ) -> Dict[str, Any]:
         """Create a RunningWorkflowStep Record (from a RunningWorkflow)"""
         # Should return:
         # {
@@ -141,10 +141,11 @@ def get_running_workflow_step(
         """Get a RunningWorkflowStep Record"""
         # Should return:
         # {
-        #    "running-workflow-step": {
+        #    "running_workflow_step": {
+        #       "step:": "step-1234",
         #       "done": False,
         #       "success": false,
-        #       "running-workflow": "r-workflow-000",
+        #       "running_workflow": "r-workflow-000",
         #    },
         # }
 
@@ -156,10 +157,11 @@ def get_running_workflow_steps(
         # Should return:
         # {
         #    "count": 1,
-        #    "running-workflow-steps": [
+        #    "running_workflow_steps": [
         #       {
         #           "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
-        #           "running-workflow-step": {
+        #           "running_workflow_step": {
+        #               "step:": "step-1234",
         #               "done": False,
         #               "success": false,
         #               "workflow": "workflow-000",
@@ -168,6 +170,24 @@ def get_running_workflow_steps(
         #    ]
         # }
 
+    @abstractmethod
+    def create_instance(self, running_workflow_step_id: str) -> Dict[str, Any]:
+        """Create an Instance Record (for a RunningWorkflowStep)"""
+        # Should return:
+        # {
+        #    "instance_id": "instance-00000000-0000-0000-0000-000000000001",
+        #    "task_id": "task-00000000-0000-0000-0000-000000000001",
+        # }
+
+    @abstractmethod
+    def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
+        """Get an Instance Record"""
+        # Should return:
+        # {
+        #    "running_workflow_step": "r-workflow-step-00000000-0000-0000-0000-000000000001",
+        #    [...],
+        # }
+
     @abstractmethod
     def get_job(
         self,
