fix: Better logic

Author: Alan Christie

tests/api_adapter.py

@@ -94,14 +94,14 @@ 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_definition_id: str) -> str:
+    def create_running_workflow(self, *, workflow_id: str) -> str:
         UnitTestAPIAdapter.mp_lock.acquire()
         with open(f"tests/{_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}
+        record = {"done": False, "success": False, "workflow": workflow_id}
         running_workflow[running_workflow_id] = record
 
         with open(f"tests/{_RUNNING_WORKFLOW_PICKLE_FILE}", "wb") as pickle_file:

tests/test_test_api_adapter.py

@@ -74,7 +74,7 @@ def test_create_running_workflow():
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
 
     # Act
-    response = utaa.create_running_workflow(workflow_definition_id=response["id"])
+    response = utaa.create_running_workflow(workflow_id=response["id"])
 
     # Assert
     assert response["id"] == "r-workflow-00000000-0000-0000-0000-000000000001"
@@ -85,7 +85,7 @@ def test_get_running_workflow():
     utaa = UnitTestAPIAdapter()
     response = utaa.create_workflow(workflow_definition={"name": "blah"})
     wfid = response["id"]
-    response = utaa.create_running_workflow(workflow_definition_id=wfid)
+    response = utaa.create_running_workflow(workflow_id=wfid)
     rwfid = response["id"]
 
     # Act
@@ -97,11 +97,43 @@ def test_get_running_workflow():
     assert rwf["workflow"] == wfid
 
 
+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"])
+    rwfid = response["id"]
+
+    # Act
+    utaa.set_running_workflow_done(running_workflow_id=rwfid, success=True)
+
+    # Assert
+    response = utaa.get_running_workflow(running_workflow_id=rwfid)
+    assert response["running_workflow"]["done"]
+    assert response["running_workflow"]["success"]
+
+
+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"])
+    rwfid = response["id"]
+
+    # Act
+    utaa.set_running_workflow_done(running_workflow_id=rwfid, success=False)
+
+    # Assert
+    response = utaa.get_running_workflow(running_workflow_id=rwfid)
+    assert response["running_workflow"]["done"]
+    assert not response["running_workflow"]["success"]
+
+
 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"])
+    response = utaa.create_running_workflow(workflow_id=response["id"])
 
     # Act
     response = utaa.create_running_workflow_step(
@@ -117,7 +149,7 @@ 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_definition_id=wfid)
+    response = utaa.create_running_workflow(workflow_id=wfid)
     rwfid = response["id"]
     response = utaa.create_running_workflow_step(
         running_workflow_id=rwfid, step="step-1"
@@ -139,7 +171,7 @@ 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_definition_id=wfid)
+    response = utaa.create_running_workflow(workflow_id=wfid)
     rwfid = response["id"]
     response = utaa.create_running_workflow_step(
         running_workflow_id=rwfid, step="step-1"

tests/test_test_instance_launcher.py

@@ -26,7 +26,7 @@ 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_definition_id=response["id"])
+    response = utaa.create_running_workflow(workflow_id=response["id"])
     response = utaa.create_running_workflow_step(
         running_workflow_id=response["id"], step="step-1"
     )

tests/test_workflow_engine_examples.py

@@ -46,7 +46,14 @@ def basic_engine():
 
 
 def start_workflow(md, da, workflow_file_name) -> str:
-    """A convenience function to handle all the 'START' logic for a workflow."""
+    """A convenience function to handle all the 'START' logic for a workflow.
+    It is given the message dispatcher, data adapter, and the base-name of the
+    workflow definition - i.e. the filename without the '.yaml' extension
+    (expected to be in the workflow-definitions directory).
+
+    It loads the workflow definition into the API adapter, creates a running workflow
+    from it and then sends a 'START' message which should cause the workflow engine to
+    start the workflow."""
 
     # To start a workflow we need to:
     # 1. Load and create a Workflow Definition
@@ -64,7 +71,7 @@ def start_workflow(md, da, workflow_file_name) -> str:
     assert wfid
     print(f"Created workflow definition {wfid}")
     # 2.
-    response = da.create_running_workflow(workflow_definition_id=wfid)
+    response = da.create_running_workflow(workflow_id=wfid)
     r_wfid = response["id"]
     assert r_wfid
     print(f"Created running workflow {r_wfid}")
@@ -81,7 +88,10 @@ def start_workflow(md, da, workflow_file_name) -> str:
 
 def wait_for_workflow(da, mq, r_wfid, expect_success=True):
     """A convenience function to wait for and check a workflow execution
-    (by inspecting the anticipated DB/API records)."""
+    (by inspecting the anticipated DB/API records). The workflow is expected
+    to start (because start_workflow() has been called), this function
+    waits for the running workflow to complete (by polling the API).
+    """
 
     # We wait for the workflow to complete by polling the API and checking
     # the running workflow's 'done' status. The user can specify whether
@@ -106,7 +116,7 @@ def wait_for_workflow(da, mq, r_wfid, expect_success=True):
     assert r_wf["success"] == expect_success
 
 
-def test_workflow_engine_with_two_step_nop(basic_engine):
+def test_workflow_engine_example_two_step_nop(basic_engine):
     # Arrange
     da, mq, md, _ = basic_engine
 
@@ -125,7 +135,7 @@ def test_workflow_engine_with_two_step_nop(basic_engine):
         assert step["running_workflow_step"]["success"]
 
 
-def test_workflow_engine_with_nop_fail(basic_engine):
+def test_workflow_engine_example_nop_fail(basic_engine):
     # Arrange
     da, mq, md, _ = basic_engine
 

workflow/workflow_abc.py

@@ -95,7 +95,7 @@ def get_workflow_by_name(
     def create_running_workflow(
         self,
         *,
-        workflow_definition_id: str,
+        workflow_id: str,
     ) -> Dict[str, Any]:
         """Create a RunningWorkflow Record (from a Workflow)"""
         # Should return: