feat: Add test for WorkflowValidator and minimal now a file

Author: Alan Christie

tests/test_decoder_minimal.py

@@ -1,40 +1,35 @@
 # Tests for the decoder package.
+import os
 from typing import Any, Dict
 
 import pytest
+import yaml
 
 pytestmark = pytest.mark.unit
 
 from workflow import decoder
 
-# A minimal Workflow Definition.
-# Tests can use this and adjust accordingly.
-_MINIMAL: Dict[str, Any] = {
-    "kind": "DataManagerWorkflow",
-    "kind-version": "2024.1",
-    "name": "workflow-minimal",
-    "steps": [
-        {
-            "name": "step-1",
-            "specification": "{}",
-        }
-    ],
-}
+_MINIMAL_WORKFLOW_FILE: str = os.path.join(
+    os.path.dirname(__file__), "workflow-definitions", "minimal.yaml"
+)
+with open(_MINIMAL_WORKFLOW_FILE, "r", encoding="utf8") as workflow_file:
+    _MINIMAL_WORKFLOW: Dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+assert _MINIMAL_WORKFLOW
 
 
 def test_validate_minimal():
     # Arrange
 
     # Act
-    error = decoder.validate_schema(_MINIMAL)
+    error = decoder.validate_schema(_MINIMAL_WORKFLOW)
 
     # Assert
     assert error is None
 
 
 def test_validate_without_name():
     # Arrange
-    workflow = _MINIMAL.copy()
+    workflow = _MINIMAL_WORKFLOW.copy()
     _ = workflow.pop("name", None)
 
     # Act
@@ -46,7 +41,7 @@ def test_validate_without_name():
 
 def test_validate_name_with_spaces():
     # Arrange
-    workflow = _MINIMAL.copy()
+    workflow = _MINIMAL_WORKFLOW.copy()
     workflow["name"] = "workflow with spaces"
 
     # Act

tests/test_worflow_validator.py

@@ -0,0 +1,42 @@
+# Tests for the decoder package.
+from typing import Any, Dict
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+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 {}
+
+
+def test_validate_minimal_for_create():
+    # Arrange
+    db_adapter = DummyDatabaseAdapter()
+    validator = WorkflowValidator(db_adapter=db_adapter)
+
+    # Act
+    error = validator.validate(
+        level=ValidationLevel.CREATE,
+        workflow_definition=_MINIMAL_WORKFLOW,
+    )
+
+    # Assert
+    assert error.error == 0
+    assert error.error_msg is None

tests/workflow-definitions/minimal.yaml

@@ -0,0 +1,7 @@
+---
+kind: DataManagerWorkflow
+kind-version: "2024.1"
+name: workflow-minimal
+steps:
+- name: step-1
+  specification: "{}"

workflow/worklfow_validator.py

@@ -48,7 +48,7 @@ def validate(
         *,
         level: ValidationLevel,
         workflow_definition: Dict[str, Any],
-        workflow_inputs: Optional[Dict[str, Any]],
+        workflow_inputs: Optional[Dict[str, Any]] = None,
     ) -> ValidationResult:
         """Validates the workflow definition (and inputs) based
         on the provided 'level'."""