feat: Add simple test DB adapter (reads test JD file)

Author: Alan Christie

tests/database_adapter.py

@@ -0,0 +1,41 @@
+import os
+from typing import Any, Dict, Optional
+
+import yaml
+
+from workflow.workflow_abc import DatabaseAdapter
+
+# Load the Unit test Job Definitions file now.
+_JOB_DEFINITION_FILE: str = os.path.join(
+    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)
+assert _JOB_DEFINITIONS
+
+
+class UnitTestDatabaseAdapter(DatabaseAdapter):
+    """A minimal Database adapter simply to serve-up Job Definitions."""
+
+    def get_workflow(self, *, workflow_definition_id: str) -> Optional[Dict[str, Any]]:
+        return {}
+
+    def get_workflow_by_name(
+        self, *, name: str, version: str
+    ) -> Optional[Dict[str, Any]]:
+        return {}
+
+    def get_job(
+        self, *, collection: str, job: str, version: str
+    ) -> Optional[Dict[str, Any]]:
+        if collection != _JOB_DEFINITIONS["collection"]:
+            return 1
+        if job not in _JOB_DEFINITIONS["jobs"]:
+            return 2
+        if version != _JOB_DEFINITIONS["jobs"][job]["version"]:
+            return 3
+        jd = _JOB_DEFINITIONS["jobs"][job]
+        response = {"command": jd["command"]}
+        if "variables" in jd:
+            response["variables"] = jd["variables"]
+        return response

tests/job-definitions/job-definitions.yaml

@@ -0,0 +1,23 @@
+---
+# Job definitions used exclusively for unit testing.
+# They provide sufficient information to 'masquerade' as real jobs
+# and are served up by the UnitTestDatabaseAdapter (found in the tests directory).
+#
+# This offers bare-essential job definitions that are used to test
+# the workflow engine. It does not fully comply with the Job definition schema -
+# only those that we need for testing the Workflow Engine.
+#
+# All jobs used for unit testing must be defined in this file.
+#
+# For each job you must provide: -
+# - Job version (and only one version for each Job)
+# - Job command
+
+collection: workflow-engine-unit-test-jobs
+
+jobs:
+
+  nop:
+    version: "1.0.0"
+    command: >-
+      python --version

tests/test_database_adapter.py

@@ -0,0 +1,23 @@
+# Tests for the decoder package.
+import os
+from typing import Any, Dict
+
+import pytest
+import yaml
+
+pytestmark = pytest.mark.unit
+
+from tests.database_adapter import UnitTestDatabaseAdapter
+
+
+def test_get_nop_job():
+    # Arrange
+    utda = UnitTestDatabaseAdapter()
+
+    # Act
+    jd = utda.get_job(
+        collection="workflow-engine-unit-test-jobs", job="nop", version="1.0.0"
+    )
+
+    # Assert
+    assert jd["command"] == "python --version"

tests/test_worflow_validator.py

@@ -1,34 +1,17 @@
-# Tests for the decoder package.
-from typing import Any, Dict
+# Tests for the WorkflowValidator
 
 import pytest
 
 pytestmark = pytest.mark.unit
 
+from tests.database_adapter import UnitTestDatabaseAdapter
 from tests.test_decoder_minimal import _MINIMAL_WORKFLOW
-from workflow.worklfow_validator import (
-    DatabaseAdapter,
-    ValidationLevel,
-    WorkflowValidator,
-)
-
-
-class DummyDatabaseAdapter(DatabaseAdapter):
-    """A minimal Database adapter simply to satisfy the tests in this file."""
-
-    def get_workflow(self, *, workflow_definition_id: str) -> Dict[str, Any]:
-        return _MINIMAL_WORKFLOW
-
-    def get_workflow_by_name(self, *, name: str, version: str) -> Dict[str, Any]:
-        return _MINIMAL_WORKFLOW
-
-    def get_job(self, *, collection: str, job: str, version: str) -> Dict[str, Any]:
-        return {}
+from workflow.worklfow_validator import ValidationLevel, WorkflowValidator
 
 
 def test_validate_minimal_for_create():
     # Arrange
-    db_adapter = DummyDatabaseAdapter()
+    db_adapter = UnitTestDatabaseAdapter()
     validator = WorkflowValidator(db_adapter=db_adapter)
 
     # Act