rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'v3' into labthings-debug-logging-config

Browse source
This commit is contained in:
Beth Probert 2026-03-05 10:27:05 +00:00
commit c3ad127f1c
19 changed files with 183 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/integration_tests/test_actions.py
View file

@ -33,7 +33,10 @@ def test_env():
"""Yield a server with a very basic configuration."""
with open(SIM_CONFIG, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
with LabThingsTestEnv(things=config_dict["things"]) as env:
with LabThingsTestEnv(
things=config_dict["things"],
application_config=config_dict["application_config"],
) as env:
yield env

6
tests/shared_utils/lt_test_utils.py
View file

@ -34,6 +34,7 @@ class LabThingsTestEnv:
self,
things: Mapping[str, lt.Thing | str],
settings_folder: Optional[str] = None,
application_config: Optional[Mapping[str, Any]] = None,
) -> None:
"""Initialise the test environment.
@ -47,6 +48,7 @@ class LabThingsTestEnv:
self._test_client: Optional[TestClient]
self._things_config = things
self._settings_folder = settings_folder
self._application_config = application_config
self._tmp_dir_obj: Optional[tempfile.TemporaryDirectory] = None
def __enter__(self) -> Self:
@ -55,7 +57,9 @@ class LabThingsTestEnv:
self._tmp_dir_obj = tempfile.TemporaryDirectory()
self._settings_folder = self._tmp_dir_obj.name
self._server = lt.ThingServer(
things=self._things_config, settings_folder=self._settings_folder
things=self._things_config,
settings_folder=self._settings_folder,
application_config=self._application_config,
)
self._test_client = TestClient(self._server.app)
self._test_client.__enter__()

27
tests/unit_tests/test_serve_static_files.py
View file

@ -81,7 +81,7 @@ def test_add_static_with_no_static_dir(mocker):
)
# Should raise as the mock static dir does not exist
with pytest.raises(FileNotFoundError):
serve_static_files.add_static_files(mock_app, scans_folder=None)
serve_static_files.add_static_files(mock_app, data_folder=None)
assert mock_app.get.call_count == 0
@ -126,7 +126,8 @@ def test_add_static_files(mock_static_dir, mocker):
)
# Get the wrapper function from the mocked decorator
wrapper = mock_app.get.return_value
serve_static_files.add_static_files(mock_app, scans_folder=None)
with tempfile.TemporaryDirectory() as datadir:
serve_static_files.add_static_files(mock_app, data_folder=datadir)
# Get should have been called twice to create a route for index
assert mock_app.get.call_count == 2
@ -151,23 +152,11 @@ def test_add_static_files(mock_static_dir, mocker):
assert second_wrapped().headers["Pragma"] == "no-cache"
# Also should have mounted both dirs
assert mock_app.mount.call_count == 1
mounted_path = mock_app.mount.call_args.args[0]
mounted_dir = mock_app.mount.call_args.args[1].directory
assert mock_app.mount.call_count == 2
mounted_path = mock_app.mount.call_args_list[0].args[0]
mounted_dir = mock_app.mount.call_args_list[0].args[1].directory
assert "/assets/" in mounted_path
assert os.path.join(mock_static_dir, "assets") in mounted_dir
def test_add_static_files_with_scan_dir(mock_static_dir, mocker):
"""Check the scan dir mounts if supplied."""
mock_app = mocker.Mock()
mocker.patch(
"openflexure_microscope_server.server.serve_static_files.STATIC_PATH",
mock_static_dir,
)
with tempfile.TemporaryDirectory() as scandir:
serve_static_files.add_static_files(mock_app, scans_folder=scandir)
# The scan dir should be the last to mount so can use call args. It should mount
# at /scans/
assert mock_app.mount.call_args.args[0] == "/scans/"
assert mock_app.mount.call_args.args[1].directory == scandir
assert mock_app.mount.call_args_list[1].args[0] == "/data/"
assert mock_app.mount.call_args_list[1].args[1].directory == datadir

9
tests/unit_tests/test_server_cli.py
View file

@ -15,7 +15,8 @@ from .test_server_config import SIM_CONFIG
def test_no_config():
"""Check that an error is thrown if no configuration is set for the microspe via CLI."""
with pytest.raises(RuntimeError, match="No configuration"):
msg = "OpenFlexure Microscope Server must have a configuration file specified."
with pytest.raises(RuntimeError, match=msg):
ofm_server.serve_from_cli([])
@ -35,6 +36,8 @@ def test_successful_start(mocker, caplog):
"openflexure_microscope_server.server.lt.ThingServer.from_config",
return_value=mock_server,
)
# Also mock configure_logging or it will touch a load of code and log.
mock_log_configure = mocker.patch.object(ofm_server, "configure_logging")
# Also mock customisation or it will try to access the hard drive and make files.
mock_customise = mocker.patch.object(ofm_server, "customise_server")
# And mock uvicorn.run as we don't want to start a server process
@ -43,7 +46,9 @@ def test_successful_start(mocker, caplog):
# Run the mock CLI
ofm_server.serve_from_cli(["-c", SIM_CONFIG])
# Check that the server was customised and the run
# Check that the logging was configures, the server was customised, and uvicorn was
# run
assert mock_log_configure.call_count == 1
assert mock_customise.call_count == 1
assert mock_uvicorn_run.call_count == 1
# Read the log config that was set

91
tests/unit_tests/test_server_config.py
View file

@ -1,9 +1,7 @@
"""Test server booting and shut down."""
import json
import os
from argparse import Namespace
from copy import deepcopy
import pytest
@ -12,11 +10,13 @@ from labthings_fastapi.server.config_model import ThingServerConfig
# Import as ofm server to attempt to minimise confusion with server as a var in other
# functions and also FastAPI `Server`.
from openflexure_microscope_server import server as ofm_server
from openflexure_microscope_server.server import OFMApplicationData
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(os.path.dirname(THIS_DIR))
FULL_CONFIG = os.path.join(REPO_ROOT, "ofm_config_full.json")
SIM_CONFIG = os.path.join(REPO_ROOT, "ofm_config_simulation.json")
MANUAL_CONFIG = os.path.join(REPO_ROOT, "ofm_config_manual.json")
@pytest.mark.parametrize("side_effect", [None, RuntimeError("Mock")])
@ -60,7 +60,6 @@ def test_customise_server(mocker):
# but also to limit excess coverage reporting.
mocked_add_static = mocker.patch.object(ofm_server, "add_static_files")
mocked_v2_endpoints = mocker.patch.object(ofm_server, "add_v2_endpoints")
mocked_configure_logging = mocker.patch.object(ofm_server, "configure_logging")
mocked_retrieve_log = mocker.patch.object(ofm_server, "retrieve_log")
mocked_retrieve_log_file = mocker.patch.object(ofm_server, "retrieve_log_from_file")
@ -68,13 +67,15 @@ def test_customise_server(mocker):
# The wrapper returned for app.get so we can see what functions are decorated.
wrapper = mock_app.get.return_value
application_data = OFMApplicationData(
log_folder="mock_log_folder", data_folder="mock_data_folder"
)
# Finally we can run it!
ofm_server.customise_server(mock_server, "mock_log_folder", "mock_scan_folder")
ofm_server.customise_server(mock_server, application_data)
# Check each internal customisation function is called
assert mocked_add_static.call_count == 1
assert mocked_v2_endpoints.call_count == 1
assert mocked_configure_logging.call_count == 1
# Check app.get adds two routes
assert mock_app.get.call_count == 2
assert wrapper.call_count == 2
@ -88,67 +89,26 @@ def test_customise_server(mocker):
assert mocked_retrieve_log_file
@pytest.mark.parametrize(
("config_file", "expected_scan_dir"),
[(FULL_CONFIG, "/var/openflexure/scans/"), (SIM_CONFIG, "./openflexure/scans/")],
)
def test_get_scans_dir_ok(config_file, expected_scan_dir):
"""Test the _get_scans_dir function and also check the standard config files."""
with open(config_file, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
assert ofm_server._get_scans_dir(config_dict) == expected_scan_dir
def test_get_scans_dir_no_smart_scan():
"""Test _get_scans_dir with no SmartScanThing."""
# Load the standard config dict
with open(FULL_CONFIG, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
# Delete smart scan
del config_dict["things"]["smart_scan"]
# No SmartScanThing, should return None
assert ofm_server._get_scans_dir(config_dict) is None
def test_get_scans_dir_bad_smart_scan_config():
"""Test _get_scans_dir with a SmartScanThing that doesn't set a config dir."""
# Load the standard config dict
with open(FULL_CONFIG, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
# Copy the dictionary
broken_config = deepcopy(config_dict)
# Delete all the smart scan kwargs
del broken_config["things"]["smart_scan"]["kwargs"]
# Creates an Error
with pytest.raises(RuntimeError):
ofm_server._get_scans_dir(broken_config)
# Same thing should happen if just the scans_folder key is deleted
broken_config = deepcopy(config_dict)
del broken_config["things"]["smart_scan"]["kwargs"]
# Creates an Error
with pytest.raises(RuntimeError):
ofm_server._get_scans_dir(broken_config)
def test_full_config_from_args_json():
"""Check that _full_config_from_args returns as expected if json is supplied."""
"""Check that _full_config_from_args errors if json config is supplied."""
msg = "OpenFlexure Microscope Server must have a configuration file specified."
args = Namespace(
config=None,
json='{"things":{"example": "labthings_fastapi.example_things:MyThing"}}',
)
lt_conf, ofm_conf = ofm_server._full_config_from_args(args)
# If json is supplied OFM config is default. As the json is passed dircectly to
# The Pydantic Mocel in LabThings. No custom data allowed.
assert isinstance(lt_conf, ThingServerConfig)
assert "log_folder" in ofm_conf
assert ofm_conf["log_folder"] == "./openflexure/logs"
assert "scans_folder" in ofm_conf
assert ofm_conf["scans_folder"] is None
with pytest.raises(RuntimeError, match=msg):
ofm_server._full_config_from_args(args)
def test_full_config_from_args_full(mocker):
@pytest.mark.parametrize(
("config", "log_dir", "data_dir"),
[
(FULL_CONFIG, "/var/openflexure/logs/", "/var/openflexure/data/"),
(SIM_CONFIG, "./openflexure/logs/", "./openflexure/data/"),
(MANUAL_CONFIG, "./openflexure/logs/", "./openflexure/data/"),
],
)
def test_full_config_from_args(config, log_dir, data_dir, mocker):
"""Check that _full_config_from_args returns as expected if json is supplied."""
# Mock picamera library so LabThings can create its model.
mocker.patch.dict(
@ -159,12 +119,13 @@ def test_full_config_from_args_full(mocker):
"picamera2.outputs": mocker.Mock(),
},
)
args = Namespace(config=FULL_CONFIG, json=None)
lt_conf, ofm_conf = ofm_server._full_config_from_args(args)
args = Namespace(config=config, json=None)
lt_conf = ofm_server._full_config_from_args(args)
# If json is supplied OFM config is default. As the json is passed dircectly to
# The Pydantic Mocel in LabThings. No custom data allowed.
assert isinstance(lt_conf, ThingServerConfig)
assert "log_folder" in ofm_conf
assert ofm_conf["log_folder"] == "/var/openflexure/logs/"
assert "scans_folder" in ofm_conf
assert ofm_conf["scans_folder"] == "/var/openflexure/scans/"
# Check the configuration validates
application_config = OFMApplicationData(**lt_conf.application_config)
# And has the correct values
assert application_config.log_folder == log_dir
assert application_config.data_folder == data_dir

66
tests/unit_tests/test_smart_scan.py
View file

@ -41,7 +41,7 @@ from openflexure_microscope_server.things.smart_scan import (
# Use our own dir in the root temp dir not a dynamically generated one so we
# have some control of when it is deleted
SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans")
SCAN_DIR = os.path.join(tempfile.gettempdir(), "smartscanthing")
def _clear_scan_dir() -> None:
@ -51,17 +51,31 @@ def _clear_scan_dir() -> None:
@pytest.fixture
def smart_scan_thing():
def smart_scan_thing(mocker):
"""Return a smart scan thing as a fixture."""
return create_thing_without_server(
thing = create_thing_without_server(
SmartScanThing,
scans_folder=SCAN_DIR,
default_workflow="mock-_all_workflows",
mock_all_slots=True,
)
type(thing._thing_server_interface).application_config = mocker.PropertyMock(
return_value={"data_folder": tempfile.gettempdir()}
)
return thing
def custom_smart_scan_thing(default_workflow, all_workflows):
@pytest.fixture
def entered_smart_scan_thing(smart_scan_thing):
"""Yield a smart scan thing as a fixture that has been entred.
This will make a scan directory manager. This fixture also clears the scan dir
"""
_clear_scan_dir()
with smart_scan_thing:
yield smart_scan_thing
def custom_smart_scan_thing(default_workflow, all_workflows, mocker):
"""Set up a custom smart scan thing with workflows adjusted.
This allows setting a default workflow and to adjust all workflows from simple
@ -69,10 +83,13 @@ def custom_smart_scan_thing(default_workflow, all_workflows):
"""
smart_scan_thing = create_thing_without_server(
SmartScanThing,
scans_folder=SCAN_DIR,
default_workflow=default_workflow,
mock_all_slots=True,
)
interface_type = type(smart_scan_thing._thing_server_interface)
interface_type.application_config = mocker.PropertyMock(
return_value={"data_folder": tempfile.gettempdir()}
)
# Pop the existing mock workflow and add specified ones (if any)
smart_scan_thing._all_workflows.pop("mock-_all_workflows")
for key, thing in all_workflows.items():
@ -81,12 +98,13 @@ def custom_smart_scan_thing(default_workflow, all_workflows):
return smart_scan_thing
def test_initial_properties(smart_scan_thing):
def test_initial_properties(entered_smart_scan_thing):
"""Check the initial values of properties.
Test properties of SmartScanThing are available without a ThingServer
and return expected default values.
"""
smart_scan_thing = entered_smart_scan_thing
assert smart_scan_thing._scan_dir_manager.base_dir == SCAN_DIR
assert smart_scan_thing.latest_scan_name is None
@ -168,9 +186,9 @@ SELECTOR_CASES = [
@pytest.mark.parametrize("case", SELECTOR_CASES)
def test_workflow_set_on_enter(case, check_side_effect):
def test_workflow_set_on_enter(case, check_side_effect, mocker):
"""Check workflow is set on enter."""
smart_scan_thing = custom_smart_scan_thing(case.default_wf, case.workflows)
smart_scan_thing = custom_smart_scan_thing(case.default_wf, case.workflows, mocker)
with check_side_effect(case.side_effect, match=case.match):
# Load in "loaded" as the sever would
smart_scan_thing._workflow_name = case.loaded_wf
@ -178,14 +196,13 @@ def test_workflow_set_on_enter(case, check_side_effect):
assert smart_scan_thing._workflow_name == case.expected_wf
def test_setting_workflows(caplog):
def test_setting_workflows(caplog, mocker):
"""Check that setting workflow works, or warns if incorrect."""
workflows = {
"foo": mock.MagicMock(spec=ScanWorkflow),
"bar": mock.MagicMock(spec=ScanWorkflow),
}
smart_scan_thing = custom_smart_scan_thing("foo", workflows)
smart_scan_thing = custom_smart_scan_thing("foo", workflows, mocker)
with caplog.at_level(logging.WARNING), smart_scan_thing:
assert smart_scan_thing._workflow_name == "foo"
assert smart_scan_thing._workflow is workflows["foo"]
@ -225,15 +242,13 @@ def test_inaccessible_scan_methods(smart_scan_thing):
smart_scan_thing.ongoing_scan
def test_private_delete_scan(smart_scan_thing, caplog):
def test_private_delete_scan(entered_smart_scan_thing, caplog):
"""Test the private _delete_scan method deletes directories or warns if it can't."""
_clear_scan_dir()
smart_scan_thing = entered_smart_scan_thing
with caplog.at_level(logging.INFO):
fake_scan_name = "fake_scan_0001"
fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name)
# Make the outer scan dir, but not the one to delete
os.makedirs(SCAN_DIR)
# Attempt to delete the fake scan. Expect it to fail and provide a warning
deleted = smart_scan_thing._delete_scan(fake_scan_name)
assert not deleted
@ -251,16 +266,13 @@ def test_private_delete_scan(smart_scan_thing, caplog):
assert len(caplog.records) == 1
def test_public_delete_scan(smart_scan_thing, caplog):
def test_public_delete_scan(entered_smart_scan_thing, caplog):
"""Test the delete_scan API call deletes directories or warns if it can't."""
_clear_scan_dir()
smart_scan_thing = entered_smart_scan_thing
with caplog.at_level(logging.INFO):
fake_scan_name = "fake_scan_0001"
fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name)
# Make the outer scan dir, but not the one to delete
os.makedirs(SCAN_DIR)
# Attempt to delete the fake scan. Expect it to fail
with pytest.raises(HTTPException) as exc_info:
smart_scan_thing.delete_scan(fake_scan_name)
@ -280,9 +292,9 @@ def test_public_delete_scan(smart_scan_thing, caplog):
assert len(caplog.records) == 1
def test_delete_all_scans(smart_scan_thing, caplog):
def test_delete_all_scans(entered_smart_scan_thing, caplog):
"""Check the delete_all_scan API really does delete all the scans."""
_clear_scan_dir()
smart_scan_thing = entered_smart_scan_thing
with caplog.at_level(logging.INFO):
fake_scan_names = [
"fake_scan_0001",
@ -346,15 +358,15 @@ def _run_only_outer_scan(
return smart_scan_thing, exec_info
def test_outer_scan(smart_scan_thing, mocker):
def test_outer_scan(entered_smart_scan_thing, mocker):
"""Test setup and teardown of the scan."""
mock_ss_thing, exec_info = _run_only_outer_scan(smart_scan_thing, mocker)
mock_ss_thing, exec_info = _run_only_outer_scan(entered_smart_scan_thing, mocker)
assert exec_info is None
# Checked the mocked _run_scan was run exactly once
assert mock_ss_thing._run_scan.call_count == 1
def test_outer_scan_wo_sample_skip(smart_scan_thing, mocker):
def test_outer_scan_wo_sample_skip(entered_smart_scan_thing, mocker):
"""Test setup and teardown of the scan."""
def _set_skip_background(mock_ss_thing):
@ -363,7 +375,7 @@ def test_outer_scan_wo_sample_skip(smart_scan_thing, mocker):
mock_ss_thing.__dict__["skip_background"] = False
mock_ss_thing, exec_info = _run_only_outer_scan(
smart_scan_thing, mocker, _set_skip_background
entered_smart_scan_thing, mocker, _set_skip_background
)
assert exec_info is None