From fae3e9f941757e5e96ab55439229beed51f8a5b1 Mon Sep 17 00:00:00 2001 From: "Ware, Joseph (DLSLtd,RAL,LSCI)" Date: Fri, 18 Oct 2024 18:07:13 +0100 Subject: [PATCH] Test infrastructure --- tests/plans/test_compliance.py | 68 +++++++++++++++++++++++++++ tests/plans/test_scanspec.py | 85 ++++++++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 tests/plans/test_compliance.py create mode 100644 tests/plans/test_scanspec.py diff --git a/tests/plans/test_compliance.py b/tests/plans/test_compliance.py new file mode 100644 index 0000000000..9a9a810009 --- /dev/null +++ b/tests/plans/test_compliance.py @@ -0,0 +1,68 @@ +import inspect +from types import ModuleType +from typing import Any, get_type_hints + +from dls_bluesky_core.core import MsgGenerator, PlanGenerator + +from dodal import plan_stubs, plans + + +def is_bluesky_plan_generator(func: Any) -> bool: + try: + return callable(func) and get_type_hints(func).get("return") == MsgGenerator + except TypeError: + # get_type_hints fails on some objects (such as Union or Optional) + return False + + +def get_all_available_generators(mod: ModuleType): + def get_named_subset(names: list[str]): + for name in names: + yield getattr(mod, name) + + if "__export__" in mod.__dict__: + yield from get_named_subset(mod.get("__export__")) + elif "__all__" in mod.__dict__: + yield from get_named_subset(mod.get("__all__")) + else: + for name, value in mod.__dict__.items(): + if not name.startswith("_"): + yield value + + +def assert_hard_requirements(plan: PlanGenerator, signature: inspect.Signature): + assert plan.__doc__ is not None, f"'{plan.__name__}' has no docstring" + for parameter in signature.parameters.values(): + assert ( + parameter.kind is not parameter.VAR_POSITIONAL + and parameter.kind is not parameter.VAR_KEYWORD + ), f"'{plan.__name__}' has variadic arguments" + + +def assert_metadata_requirements(plan: PlanGenerator, signature: inspect.Signature): + assert ( + "metadata" in signature.parameters + ), f"'{plan.__name__}' does not allow metadata" + metadata = signature.parameters["metadata"] + assert ( + metadata.annotation == dict[str, Any] | None + and metadata.default is not inspect.Parameter.empty + ), f"'{plan.__name__}' metadata is not optional" + assert metadata.default is None, f"'{plan.__name__}' metadata default is mutable" + + +def test_plans_comply(): + for plan in get_all_available_generators(plans): + if is_bluesky_plan_generator(plan): + signature = inspect.Signature.from_callable(plan) + assert_hard_requirements(plan, signature) + assert_metadata_requirements(plan, signature) + + +def test_stubs_comply(): + for plan in get_all_available_generators(plan_stubs): + if is_bluesky_plan_generator(plan): + signature = inspect.Signature.from_callable(plan) + assert_hard_requirements(plan, signature) + if "metadata" in signature.parameters: + assert_metadata_requirements(plan, signature) diff --git a/tests/plans/test_scanspec.py b/tests/plans/test_scanspec.py new file mode 100644 index 0000000000..b7eb1437cd --- /dev/null +++ b/tests/plans/test_scanspec.py @@ -0,0 +1,85 @@ +from pathlib import Path + +import pytest +from bluesky.run_engine import RunEngine +from event_model.documents import ( + DocumentType, +) +from ophyd_async.core import ( + DeviceCollector, + PathProvider, + callback_on_mock_put, + set_mock_value, +) +from ophyd_async.epics.adaravis import AravisDetector +from ophyd_async.epics.motor import Motor +from scanspec.specs import Line, Spiral + +from dodal.common.beamlines.beamline_utils import set_path_provider +from dodal.common.visit import StaticVisitPathProvider +from dodal.plans import spec_scan + + +@pytest.fixture +def x_axis(RE: RunEngine) -> Motor: + with DeviceCollector(mock=True): + x_axis = Motor("DUMMY:X:") + set_mock_value(x_axis.velocity, 1) + return x_axis + + +@pytest.fixture +def y_axis(RE: RunEngine) -> Motor: + with DeviceCollector(mock=True): + y_axis = Motor("DUMMY:X:") + set_mock_value(y_axis.velocity, 1) + return y_axis + + +@pytest.fixture +def path_provider(static_path_provider: PathProvider): + assert isinstance(static_path_provider, StaticVisitPathProvider) + set_path_provider(static_path_provider) + yield static_path_provider + set_path_provider(None) # type: ignore + + +@pytest.fixture +def det(RE: RunEngine, path_provider: PathProvider, tmp_path: Path) -> AravisDetector: + with DeviceCollector(mock=True): + det = AravisDetector("DUMMY:DET", path_provider=path_provider) + + def ready_to_write(file_name: str, *_, **__): + set_mock_value(det.hdf.file_path_exists, True) + set_mock_value(det.hdf.full_file_name, str(tmp_path / f"{file_name}.h5")) + + callback_on_mock_put(det.hdf.file_path, ready_to_write) + set_mock_value(det.hdf.capture, True) + + return det + + +def test_metadata_of_simple_spec(RE: RunEngine, x_axis: Motor, det: AravisDetector): + spec = Line(axis=x_axis, start=1, stop=2, num=3) + + docs: list[tuple[str, DocumentType]] = [] + + def capture_doc(name: str, doc: DocumentType): + docs.append((name, doc)) + + RE(spec_scan({det}, spec), capture_doc) + + # Start, Descriptor, StreamResource, StreamDatum, Event * 3, Stop + assert len(docs) == 8 + + +def test_metadata_of_spiral_spec( + RE: RunEngine, x_axis: Motor, y_axis: Motor, det: AravisDetector +): + spec = Spiral.spaced(x_axis, y_axis, 0, 0, 5, 1) + docs: list[tuple[str, DocumentType]] = [] + + def capture_doc(name: str, doc: DocumentType): + docs.append((name, doc)) + + RE(spec_scan({det}, spec), capture_doc)