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

Final updates for main test suite for labthings-fastapi 0.0.12 compatibility

Browse source
This commit is contained in:
Julian Stirling 2025-12-14 22:11:08 +00:00
parent dec90f5b6b
commit f79505546c
5 changed files with 94 additions and 93 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/test_smart_scan.py
View file

@ -24,6 +24,7 @@ import pytest
from fastapi import HTTPException
from labthings_fastapi.exceptions import InvocationCancelledError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import (
NotEnoughFreeSpaceError,
@ -56,7 +57,7 @@ def _clear_scan_dir() -> None:
@pytest.fixture
def smart_scan_thing():
"""Return a smart scan thing as a fixture."""
return SmartScanThing(SCAN_DIR)
return create_thing_without_server(SmartScanThing, scans_folder=SCAN_DIR)
def test_initial_properties(smart_scan_thing):
@ -198,7 +199,9 @@ def _run_only_outer_scan(adjust_initial_state: Optional[Callable] = None):
assert self._csm is csm_mock
# mock smart scan thing
mock_ss_thing = MockedSmartScanThing(SCAN_DIR)
mock_ss_thing = create_thing_without_server(
MockedSmartScanThing, scans_folder=SCAN_DIR
)
if adjust_initial_state is not None:
adjust_initial_state(mock_ss_thing)

13
tests/test_stack.py
View file

@ -1,17 +1,15 @@
"""Tests for the smart and fast stacking."""
import logging
import tempfile
from random import randint
from typing import Optional
import numpy as np
import pytest
from fastapi.testclient import TestClient
from hypothesis import given
from hypothesis import strategies as st
import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import IMAGE_REGEX
from openflexure_microscope_server.things.autofocus import (
@ -266,13 +264,8 @@ def test_retrieval_of_captures(start):
@pytest.fixture
def autofocus_thing():
"""Yield an autofocus thing connected to a server."""
autofocus_thing = AutofocusThing()
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(autofocus_thing, "autofocus")
with TestClient(server.app):
yield autofocus_thing
"""Return an autofocus thing connected to a server."""
return create_thing_without_server(AutofocusThing)
def test_create_stack(autofocus_thing, caplog):

117
tests/test_stage.py
View file

@ -11,6 +11,7 @@ from hypothesis import strategies as st
import labthings_fastapi as lt
from labthings_fastapi.exceptions import NotConnectedToServerError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.stage import (
BaseStage,
@ -33,25 +34,21 @@ path3d = st.lists(point3d, min_size=5, max_size=10)
@pytest.fixture
def dummy_stage():
"""Return a dummy stage with a very low step time."""
return DummyStage(step_time=0.000001)
return create_thing_without_server(DummyStage, step_time=0.000001)
@pytest.fixture
def thing_server(dummy_stage):
def stage_server():
"""Yield a server with a very basic configuration."""
thing_conf = {
"camera": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"stage": "openflexure_microscope_server.things.stage.dummy:DummyStage",
}
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(dummy_stage, "stage")
server = lt.ThingServer(things=thing_conf, settings_folder=tmpdir)
yield server
@pytest.fixture
def stage_client(thing_server):
"""Yield a labthings ThingClient for the stage."""
with TestClient(thing_server.app) as test_client:
yield lt.ThingClient.from_url("stage", client=test_client)
def test_override_base_movement():
"""Child classes of stage should implement functions in the hardware reference frame.
@ -73,7 +70,7 @@ def test_override_base_movement():
pass
with pytest.raises(RedefinedBaseMovementError):
BadStage1()
create_thing_without_server(BadStage1)
class BadStage2(BaseStage):
@lt.action
@ -86,7 +83,7 @@ def test_override_base_movement():
pass
with pytest.raises(RedefinedBaseMovementError):
BadStage2()
create_thing_without_server(BadStage2)
def _set_axis_direction(dummy_stage, direction):
@ -149,48 +146,61 @@ def test_apply_axis_errors(dummy_stage):
dummy_stage._apply_axis_direction({"x": -1, "y": 2, "up": -3})
def test_default_values(stage_client, dummy_stage):
def test_default_values(dummy_stage):
"""Check the default values for the dummy stage."""
# axes are x, y, z (note that going through the thing, client the tuple is
# converted to a list.
assert stage_client.axis_names == ["x", "y", "z"]
assert dummy_stage.axis_names == ("x", "y", "z")
# position starts at 0, 0, 0
assert stage_client.position == {"x": 0, "y": 0, "z": 0}
assert dummy_stage.position == {"x": 0, "y": 0, "z": 0}
# axis direction starts is -1, 1, 1 for the dummy stage
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# And check the thing state
assert dummy_stage.thing_state == {"position": {"x": 0, "y": 0, "z": 0}}
def test_direction_inversion(stage_client, dummy_stage):
def test_direction_inversion(dummy_stage):
"""Check axes invert as expected when called."""
# Can't set an arbitrary value via a client as read only:
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.axis_inverted = {"x": 2, "y": 1, "z": 1}
# Read only should set a 405 error code
assert excinfo.value.response.status_code == 405
# And not modify th initial value
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="x")
assert stage_client.axis_inverted == {"x": False, "y": False, "z": False}
stage_client.invert_axis_direction(axis="x")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="y")
assert stage_client.axis_inverted == {"x": True, "y": True, "z": False}
stage_client.invert_axis_direction(axis="y")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="z")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": True}
stage_client.invert_axis_direction(axis="z")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Check initial value
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# Start inverting
dummy_stage.invert_axis_direction(axis="x")
assert dummy_stage.axis_inverted == {"x": False, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="x")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="y")
assert dummy_stage.axis_inverted == {"x": True, "y": True, "z": False}
dummy_stage.invert_axis_direction(axis="y")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="z")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": True}
dummy_stage.invert_axis_direction(axis="z")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# Should error if axis doesn't exist, this is a KeyError in the server
with pytest.raises(KeyError):
dummy_stage.invert_axis_direction(axis="theta")
# But a 422 over HTTP
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.invert_axis_direction(axis="theta")
assert excinfo.value.response.status_code == 422
def test_direction_errors_local_and_http(stage_server):
"""Check for expected errors both locally and over http."""
dummy_stage = stage_server.things["stage"]
with TestClient(stage_server.app) as test_client:
stage_client = lt.ThingClient.from_url("/stage/", client=test_client)
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Can't set an arbitrary value via a client as read only:
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.axis_inverted = {"x": 2, "y": 1, "z": 1}
# Read only should set a 405 error code ...
assert excinfo.value.response.status_code == 405
# ... and should not modify the initial value
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Should error if axis doesn't exist, this is a KeyError in the server
with pytest.raises(KeyError):
dummy_stage.invert_axis_direction(axis="theta")
# But a 422 over HTTP
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.invert_axis_direction(axis="theta")
assert excinfo.value.response.status_code == 422
def _test_move_relative(dummy_stage, axis_inverted, path):
@ -225,22 +235,17 @@ def _test_move_relative(dummy_stage, axis_inverted, path):
suppress_health_check=[HealthCheck.function_scoped_fixture],
deadline=10000,
)
def test_move_relative(stage_client, dummy_stage, path):
def test_move_relative(dummy_stage, path):
"""Loop over different inversion options and check that the stage moves as expected.
3 paths are tried for each case of axis inversion. This checks both that the
reported position changes as is input in path, and that the hardware position is
inverted when appropriate.
NOTE: it is essential that `stage_client` is imported even if any time it is used
dummy stage could be used. This is because the fixture that creates stage client
handles adding the dummy_stage to a server. It needs to have been added to a
server for it not to throw errors about not being connected to a server.
"""
# Note that the fixture is not reset. This is fine, because the stage_should work
# no matter the starting position.
axis_names = stage_client.axis_names
axis_names = dummy_stage.axis_names
# Create every combination of True/False for x,y,z.
inversion_combinations = [
dict(zip(axis_names, inverted, strict=True))
@ -286,14 +291,14 @@ def _test_move_absolute(dummy_stage, axis_inverted, path):
suppress_health_check=[HealthCheck.function_scoped_fixture],
deadline=10000,
)
def test_move_absolute(stage_client, dummy_stage, path):
def test_move_absolute(dummy_stage, path):
"""Loop over different inversion options and check that the stage moves as expected.
3 paths are tried for each case of axis inversion. This checks both that the
reported position changes as is input in path, and that the hardware position is
inverted when appropriate.
NOTE: it is essential that `stage_client` is imported even if any time it is used
NOTE: it is essential that `dummy_stage` is imported even if any time it is used
dummy stage could be used. This is because the fixture that creates stage client
handles adding the dummy_stage to a server. It needs to have been added to a
server for it not to throw errors about not being connected to a server.
@ -301,7 +306,7 @@ def test_move_absolute(stage_client, dummy_stage, path):
# Note that the fixture is not reset. This is fine, because the stage_should work
# no matter the starting position.
axis_names = stage_client.axis_names
axis_names = dummy_stage.axis_names
# Create every combination of True/False for x,y,z.
inversion_combinations = [
dict(zip(axis_names, inverted, strict=True))
@ -328,7 +333,7 @@ def test_thing_description_equivalence(dummy_stage, mocker):
# Flash LED isn't a standard stage action, most stages do not control illumination.
extra_sanga_actions = ["flash_led"]
base_td = BaseStage().thing_description()
base_td = create_thing_without_server(BaseStage).thing_description()
base_actions = set(base_td.actions.keys())
base_properties = set(base_td.properties.keys())
@ -336,7 +341,7 @@ def test_thing_description_equivalence(dummy_stage, mocker):
dummy_actions = set(dummy_td.actions.keys())
dummy_properties = set(dummy_td.properties.keys())
sanga_td = SangaboardThing().thing_description()
sanga_td = create_thing_without_server(SangaboardThing).thing_description()
# Remove known extra actions
sanga_actions = list(sanga_td.actions.keys())
for action in extra_sanga_actions:

16
tests/test_stage_measure.py
View file

@ -2,14 +2,12 @@
import dataclasses
import logging
import tempfile
from copy import copy
import numpy as np
import pytest
from fastapi.testclient import TestClient
import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things import stage_measure
from openflexure_microscope_server.things.camera_stage_mapping import (
@ -145,22 +143,18 @@ def test_parasitic_detect(par_fraction, too_high):
def test_error_if_no_stream_res_set_when_requesting_img_coords():
"""Check a RuntimeError thrown when requesting image coordinates if resolution unset."""
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing)
with pytest.raises(RuntimeError, match="Stream resolution must be set"):
stage_measure.RangeofMotionThing()._img_percentage_to_img_coords(20, "x")
rom_thing._img_percentage_to_img_coords(20, "x")
@pytest.fixture
def rom_thing(example_rom_data) -> stage_measure.RangeofMotionThing:
"""Yield a RangeofMotionThing already populated with some example rom_data."""
rom_thing = stage_measure.RangeofMotionThing()
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing)
rom_thing._stream_resolution = [800, 600]
rom_thing._rom_data = example_rom_data
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(rom_thing, "rom_thing")
with TestClient(server.app):
yield rom_thing
return rom_thing
@pytest.fixture

34
tests/test_system_thing.py
View file

@ -4,10 +4,20 @@ import os
from signal import SIGTERM
from uuid import UUID
import pytest
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things import system
from openflexure_microscope_server.utilities import VersionData, robust_version_strings
@pytest.fixture
def system_thing():
"""Return a OpenFlexureSystem with a mocked server interface."""
return create_thing_without_server(system.OpenFlexureSystem)
def _is_raspberrypi() -> bool:
"""Check if the machine running the test is a Raspberry Pi.
@ -22,17 +32,16 @@ def _is_raspberrypi() -> bool:
return "Raspberry Pi" in model_name
def test_is_raspberry():
def test_is_raspberry(system_thing):
"""Check the thing property reports whether this is a Raspberry Pi correctly."""
assert system.OpenFlexureSystem().is_raspberrypi == _is_raspberrypi()
assert system_thing.is_raspberrypi == _is_raspberrypi()
def test_version_data():
def test_version_data(system_thing):
"""Check VersionData is returned.
The content of robust_version_strings() is tested elsewhere.
"""
system_thing = system.OpenFlexureSystem()
data = system_thing.version_data
assert isinstance(data, VersionData)
assert data == robust_version_strings()
@ -43,27 +52,24 @@ def test_version_data():
assert data == fake_data
def test_hostname(mocker):
def test_hostname(system_thing, mocker):
"""Check the hostname matches what socket.gethostname() returns."""
mock_gethostname = mocker.patch("socket.gethostname", return_value="foobar")
system_thing = system.OpenFlexureSystem()
assert system_thing.hostname == "foobar"
mock_gethostname.assert_called_once_with()
def test_microscope_id():
def test_microscope_id(system_thing):
"""Check the microscope UUID is a valid UUID and doesn't change when read again."""
system_thing = system.OpenFlexureSystem()
microscope_id = system_thing.microscope_id
assert isinstance(microscope_id, UUID)
assert microscope_id == system_thing.microscope_id
def test_thing_state(mocker):
def test_thing_state(system_thing, mocker):
"""Check the thing state contains version data, hostname, and a UUID string."""
mocker.patch("socket.gethostname", return_value="foobar")
version_data = robust_version_strings()
system_thing = system.OpenFlexureSystem()
state_dict = system_thing.thing_state
assert state_dict["hostname"] == "foobar"
# Check the UUID in the dictionary is a string
@ -93,7 +99,7 @@ def test_pi_shutdown(mocker):
# Mock the shutdown command as we don't want to shutdown when running tests.
mocker.patch.object(system, "SHUTDOWN_CMD", new=["echo", "shutdown"])
# Call shutdown on a MockPiSystem
system_thing = MockPiSystem()
system_thing = create_thing_without_server(MockPiSystem)
result = system_thing.shutdown()
# Check the result of the echo mock command was returned
@ -108,7 +114,7 @@ def test_pi_reboot(mocker):
# Mock the reboot command as we don't want to reboot when running tests.
mocker.patch.object(system, "REBOOT_CMD", new=["echo", "restart"])
# Call reboot on a MockPiSystem
system_thing = MockPiSystem()
system_thing = create_thing_without_server(MockPiSystem)
result = system_thing.reboot()
# Check the result of the echo mock command was returned
@ -125,7 +131,7 @@ def test_non_pi_shutdown(mocker):
mock_kill = mocker.patch("os.kill")
# Get this subprocess
pid = os.getpid()
system_thing = MockNonPiSystem()
system_thing = create_thing_without_server(MockNonPiSystem)
result = system_thing.shutdown()
# Check it tried to kill this process with SIGTERM
@ -137,7 +143,7 @@ def test_non_pi_shutdown(mocker):
def test_non_pi_reboot():
"""Check that a server not on a pi refuses to restart."""
system_thing = MockNonPiSystem()
system_thing = create_thing_without_server(MockNonPiSystem)
result = system_thing.reboot()
# Check output is an appropriate error message