Add test of logic for selecting worflow.

src/openflexure_microscope_server/things/smart_scan.py

@@ -12,6 +12,7 @@ import threading
 import time
 from datetime import datetime
 from subprocess import SubprocessError
+from types import TracebackType
 from typing import (
     Annotated,
     Any,
@@ -164,6 +165,17 @@ class SmartScanThing(lt.Thing):
         self._workflow_name = valid_name
         return self
 
+    def __exit__(
+        self,
+        _exc_type: type[BaseException],
+        _exc_value: Optional[BaseException],
+        _traceback: Optional[TracebackType],
+    ) -> None:
+        """Clean up after context manager is closed.
+
+        In this case it doesn't need to do anything.
+        """
+
     # Note that the default detector name is set at init. This is over written if
     # setting is loaded from disk.
     @lt.setting

tests/conftest.py

@@ -0,0 +1,63 @@
+"""Fixtures for all test suites."""
+
+import logging
+import re
+from collections.abc import Iterable
+from contextlib import contextmanager
+from typing import Optional
+
+import pytest
+
+
+@pytest.fixture
+def check_side_effect(caplog):
+    """Supply a context manager for checking code either logs, raises an error, or doesn't."""
+
+    @contextmanager
+    def _checker(
+        side_effect: Optional[type | int | Iterable[int]],
+        match: Optional[str | Iterable[str]] = None,
+    ) -> None:
+        """Check the code within this context has the expected side effect.
+
+        :param side_effect: The side effect of the code within this context. An int
+            should be a logging level number, or a list of level numbers if multiple
+            logs are expected.
+        :param match: Optionally a match regex for raises or the logger, can be a list
+            if side effect is a list of levels.
+        """
+        # If the side effect is an exception
+        if isinstance(side_effect, type) and issubclass(side_effect, BaseException):
+            with pytest.raises(side_effect, match=match):
+                yield
+            return
+
+        # If the side effect is None or logging
+        caplog.clear()
+
+        with caplog.at_level(logging.INFO):
+            yield
+
+        if side_effect is None:
+            # No side effect
+            assert caplog.records == []
+        elif isinstance(side_effect, Iterable):
+            # Multiple logs
+            if match is not None:
+                assert isinstance(match, Iterable)
+                assert len(match) == len(side_effect)
+            assert len(caplog.records) == len(side_effect)
+            for i, level in enumerate(side_effect):
+                record = caplog.records[i]
+                assert record.levelno == level
+                if match is not None:
+                    assert re.match(match[i], record.message) is not None
+        else:
+            # Single log
+            assert len(caplog.records) == 1
+            record = caplog.records[0]
+            assert record.levelno == side_effect
+            if match is not None:
+                assert re.match(match, record.message) is not None
+
+    return _checker

tests/unit_tests/test_smart_scan.py

@@ -17,6 +17,8 @@ import logging
 import os
 import shutil
 import tempfile
+from collections.abc import Iterable
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Callable, Optional
 
@@ -67,6 +69,89 @@ def test_initial_properties(smart_scan_thing):
     assert smart_scan_thing.latest_scan_name is None
 
 
+@dataclass
+class WorkflowSelectorTestCase:
+    """The information from a capture in a smart_z_stack."""
+
+    workflows: dict
+    default_wf: str
+    expected_wf: Optional[str]
+    loaded_wf: Optional[str] = None
+    side_effect: Optional[type | int | Iterable[int]] = None
+    """The side effect of entering the Thing (None, Logger level number, Error type)"""
+    match: Optional[str | Iterable[str]] = None
+
+
+SELECTOR_CASES = [
+    WorkflowSelectorTestCase(
+        workflows={},
+        default_wf="foo",
+        expected_wf=None,
+        side_effect=RuntimeError,
+        match="Could not set Scan Workflow",
+    ),
+    WorkflowSelectorTestCase(
+        workflows={"foo": "foo_workflow", "bar": "bar_workflow"},
+        default_wf="foo",
+        expected_wf="foo",
+        side_effect=None,
+    ),
+    WorkflowSelectorTestCase(
+        workflows={"foo": "foo_workflow", "bar": "bar_workflow"},
+        default_wf="bar",
+        expected_wf="bar",
+        side_effect=None,
+    ),
+    WorkflowSelectorTestCase(
+        workflows={"foo": "foo_workflow", "bar": "bar_workflow"},
+        default_wf="wrong",
+        expected_wf="foo",
+        side_effect=logging.WARNING,
+        match="Could not select default key 'wrong'",
+    ),
+    WorkflowSelectorTestCase(
+        workflows={"foo": "foo_workflow", "bar": "bar_workflow"},
+        default_wf="wrong",
+        loaded_wf="bar",
+        expected_wf="bar",
+        side_effect=None,
+    ),
+    WorkflowSelectorTestCase(
+        workflows={"foo": "foo_workflow", "bar": "bar_workflow"},
+        default_wf="wrong",
+        loaded_wf="wrong",
+        expected_wf="foo",
+        side_effect=[logging.WARNING, logging.WARNING],
+        match=[
+            "Could not select 'wrong' from Thing mapping",
+            "Could not select default key 'wrong'",
+        ],
+    ),
+]
+
+
+@pytest.mark.parametrize("case", SELECTOR_CASES)
+def test_workflow_set_on_enter(case, check_side_effect, mocker):
+    """Check workflow is set on enter."""
+    with check_side_effect(case.side_effect, match=case.match):
+        # Patch the slot before creation so we don't get caught up in LabThings errors
+        mocker.patch.object(
+            SmartScanThing, "_all_workflows", return_value=mocker.PropertyMock
+        )
+        smart_scan_thing = create_thing_without_server(
+            SmartScanThing,
+            scans_folder=SCAN_DIR,
+            default_workflow=case.default_wf,
+            mock_all_slots=False,
+        )
+        # Set the workflows
+        smart_scan_thing._all_workflows = case.workflows
+        # Load in "loaded" as the sever would
+        smart_scan_thing._workflow_name = case.loaded_wf
+        with smart_scan_thing:
+            assert smart_scan_thing._workflow_name == case.expected_wf
+
+
 def test_inaccessible_scan_methods(smart_scan_thing):
     """Test that method with @_scan_running decorator is inaccessible.