diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index c415a9c5..333f6875 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..a7d77bfe --- /dev/null +++ b/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 diff --git a/tests/unit_tests/test_smart_scan.py b/tests/unit_tests/test_smart_scan.py index 49d9223d..38cf2466 100644 --- a/tests/unit_tests/test_smart_scan.py +++ b/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.