style: Removed get workflow by name (and API fixes)

Author: Alan Christie

tests/test_test_api_adapter.py

@@ -54,19 +54,6 @@ def test_get_workflow():
     assert wf["workflow"]["name"] == "blah"
 
 
-def test_get_workflow_by_name():
-    # Arrange
-    utaa = UnitTestWorkflowAPIAdapter()
-    _ = 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 = UnitTestWorkflowAPIAdapter()
@@ -256,35 +243,6 @@ def test_get_running_workflow_step():
     assert rwfs["running_workflow"] == rwfid
 
 
-def test_get_running_workflow_steps():
-    # Arrange
-    utaa = UnitTestWorkflowAPIAdapter()
-    response = utaa.create_workflow(workflow_definition={"name": "blah"})
-    wfid = response["id"]
-    response = utaa.create_running_workflow(
-        user_id="dlister",
-        workflow_id=wfid,
-        project_id=TEST_PROJECT_ID,
-        variables={},
-    )
-    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
-
-
 def test_create_instance():
     # Arrange
     utaa = UnitTestWorkflowAPIAdapter()

tests/wapi_adapter.py

@@ -18,7 +18,7 @@
 import os
 from multiprocessing import Lock
 from pickle import Pickler, Unpickler
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import yaml
 
@@ -29,7 +29,7 @@
     os.path.dirname(__file__), "job-definitions", "job-definitions.yaml"
 )
 with open(_JOB_DEFINITION_FILE, "r", encoding="utf8") as jd_file:
-    _JOB_DEFINITIONS: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
+    _JOB_DEFINITIONS: dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
 assert _JOB_DEFINITIONS
 
 # Table UUID formats
@@ -80,7 +80,7 @@ def __init__(self):
                 Pickler(pickle_file).dump({})
         UnitTestWorkflowAPIAdapter.lock.release()
 
-    def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
+    def create_workflow(self, *, workflow_definition: dict[str, Any]) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
@@ -95,34 +95,22 @@ def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
 
         return {"id": workflow_definition_id}
 
-    def get_workflow(self, *, workflow_id: str) -> Dict[str, Any]:
+    def get_workflow(self, *, workflow_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
         UnitTestWorkflowAPIAdapter.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]:
-        UnitTestWorkflowAPIAdapter.lock.acquire()
-        with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
-            workflow = Unpickler(pickle_file).load()
-        UnitTestWorkflowAPIAdapter.lock.release()
-
-        item = {}
-        for wfid, value in workflow.items():
-            if value["name"] == name:
-                item = {"id": wfid, "workflow": value}
-        return item
-
     def create_running_workflow(
         self,
         *,
         user_id: str,
         workflow_id: str,
         project_id: str,
-        variables: Dict[str, Any],
-    ) -> str:
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
         assert user_id
         assert isinstance(variables, dict)
 
@@ -149,13 +137,23 @@ def create_running_workflow(
 
         return {"id": running_workflow_id}
 
+    def get_running_workflow(self, *, running_workflow_id: str) -> dict[str, Any]:
+        UnitTestWorkflowAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow = Unpickler(pickle_file).load()
+        UnitTestWorkflowAPIAdapter.lock.release()
+
+        if running_workflow_id not in running_workflow:
+            return {}
+        return {"running_workflow": running_workflow[running_workflow_id]}
+
     def set_running_workflow_done(
         self,
         *,
         running_workflow_id: str,
         success: bool,
-        error: Optional[int] = None,
-        error_msg: Optional[str] = None,
+        error: int | None = None,
+        error_msg: str | None = None,
     ) -> None:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
@@ -171,19 +169,9 @@ def set_running_workflow_done(
             Pickler(pickle_file).dump(running_workflow)
         UnitTestWorkflowAPIAdapter.lock.release()
 
-    def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
-        UnitTestWorkflowAPIAdapter.lock.acquire()
-        with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
-            running_workflow = Unpickler(pickle_file).load()
-        UnitTestWorkflowAPIAdapter.lock.release()
-
-        if running_workflow_id not in running_workflow:
-            return {}
-        return {"running_workflow": running_workflow[running_workflow_id]}
-
     def create_running_workflow_step(
         self, *, running_workflow_id: str, step: str
-    ) -> str:
+    ) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
@@ -208,7 +196,7 @@ def create_running_workflow_step(
 
     def get_running_workflow_step(
         self, *, running_workflow_step_id: str
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
@@ -225,8 +213,8 @@ def set_running_workflow_step_done(
         *,
         running_workflow_step_id: str,
         success: bool,
-        error: Optional[int] = None,
-        error_msg: Optional[str] = None,
+        error: int | None = None,
+        error_msg: str | None = None,
     ) -> None:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
@@ -242,9 +230,7 @@ def set_running_workflow_step_done(
             Pickler(pickle_file).dump(running_workflow_step)
         UnitTestWorkflowAPIAdapter.lock.release()
 
-    def get_running_workflow_steps(
-        self, *, running_workflow_id: str
-    ) -> List[Dict[str, Any]]:
+    def get_running_workflow_steps(self, *, running_workflow_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
@@ -257,7 +243,7 @@ def get_running_workflow_steps(
                 steps.append(item)
         return {"count": len(steps), "running_workflow_steps": steps}
 
-    def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
+    def create_instance(self, *, running_workflow_step_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
             instances = Unpickler(pickle_file).load()
@@ -275,15 +261,15 @@ def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
 
         return {"id": instance_id}
 
-    def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
+    def get_instance(self, *, instance_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
             instances = Unpickler(pickle_file).load()
         UnitTestWorkflowAPIAdapter.lock.release()
 
         return {} if instance_id not in instances else instances[instance_id]
 
-    def create_task(self, *, instance_id: str) -> Dict[str, Any]:
+    def create_task(self, *, instance_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
             tasks = Unpickler(pickle_file).load()
@@ -302,17 +288,15 @@ def create_task(self, *, instance_id: str) -> Dict[str, Any]:
 
         return {"id": task_id}
 
-    def get_task(self, *, task_id: str) -> Dict[str, Any]:
+    def get_task(self, *, task_id: str) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
             tasks = Unpickler(pickle_file).load()
         UnitTestWorkflowAPIAdapter.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]]:
+    def get_job(self, *, collection: str, job: str, version: str) -> dict[str, Any]:
         assert collection == _JOB_DEFINITIONS["collection"]
         assert job in _JOB_DEFINITIONS["jobs"]
         assert version

workflow/workflow_abc.py

@@ -97,20 +97,7 @@ def get_workflow(
         # {
         #    "workflow": <workflow>,
         # }
-
-    @abstractmethod
-    def get_workflow_by_name(
-        self,
-        *,
-        name: str,
-        version: str,
-    ) -> dict[str, Any]:
-        """Get a Workflow Record by name"""
-        # If present this should return:
-        # {
-        #    "id": "workflow-00000000-0000-0000-0000-000000000001",
-        #    "workflow": <workflow>,
-        # }
+        # If not present an empty dictionary should be returned.
 
     @abstractmethod
     def create_running_workflow(
@@ -127,18 +114,6 @@ def create_running_workflow(
         #    "id": "r-workflow-00000000-0000-0000-0000-000000000001",
         # }
 
-    @abstractmethod
-    def set_running_workflow_done(
-        self,
-        *,
-        running_workflow_id: str,
-        success: bool,
-        error: int | None = None,
-        error_msg: str | None = None,
-    ) -> None:
-        """Set the success value for a RunningWorkflow Record.
-        If not successful an error code and message should be provided."""
-
     @abstractmethod
     def get_running_workflow(self, *, running_workflow_id: str) -> dict[str, Any]:
         """Get a RunningWorkflow Record"""
@@ -156,6 +131,19 @@ def get_running_workflow(self, *, running_workflow_id: str) -> dict[str, Any]:
         #       "variables": {"x": 1, "y": 2},
         #    }
         # }
+        # If not present an empty dictionary should be returned.
+
+    @abstractmethod
+    def set_running_workflow_done(
+        self,
+        *,
+        running_workflow_id: str,
+        success: bool,
+        error: int | None = None,
+        error_msg: str | None = None,
+    ) -> None:
+        """Set the success value for a RunningWorkflow Record.
+        If not successful an error code and message should be provided."""
 
     @abstractmethod
     def create_running_workflow_step(
@@ -186,6 +174,7 @@ def get_running_workflow_step(
         #       "running_workflow": "r-workflow-00000000-0000-0000-0000-000000000001",
         #    },
         # }
+        # If not present an empty dictionary should be returned.
 
     @abstractmethod
     def set_running_workflow_step_done(
@@ -200,9 +189,7 @@ def set_running_workflow_step_done(
         If not successful an error code and message should be provided."""
 
     @abstractmethod
-    def get_running_workflow_steps(
-        self, *, running_workflow_id: str
-    ) -> list[dict[str, Any]]:
+    def get_running_workflow_steps(self, *, running_workflow_id: str) -> dict[str, Any]:
         """Gets all the RunningWorkflowStep Records (for a RunningWorkflow)"""
         # Should return:
         # {
@@ -221,6 +208,7 @@ def get_running_workflow_steps(
         #       ...
         #    ]
         # }
+        # If there are not steps an empty dictionary should be returned and a count of 0
 
     @abstractmethod
     def create_instance(self, running_workflow_step_id: str) -> dict[str, Any]:
@@ -239,6 +227,7 @@ def get_instance(self, *, instance_id: str) -> dict[str, Any]:
         #    "running_workflow_step": "r-workflow-step-00000000-0000-0000-0000-000000000001",
         #    [...],
         # }
+        # If not present an empty dictionary should be returned.
 
     @abstractmethod
     def create_task(self, instance_id: str) -> dict[str, Any]:
@@ -257,6 +246,7 @@ def get_task(self, *, task_id: str) -> dict[str, Any]:
         #    "exit_code": 0,
         #    [...],
         # }
+        # If not present an empty dictionary should be returned.
 
     @abstractmethod
     def get_job(
@@ -265,8 +255,9 @@ def get_job(
         collection: str,
         job: str,
         version: str,
-    ) -> dict[str, Any] | None:
+    ) -> dict[str, Any]:
         """Get a Job"""
+        # If not present an empty dictionary should be returned.
 
 
 class MessageDispatcher(ABC):