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

Start updating unit tests after refactoring into ScanWorkflows

Browse source
This commit is contained in:
Julian Stirling 2026-01-15 22:59:58 +00:00
parent 11ef1217e0
commit c82523dd5b
6 changed files with 119 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

100
tests/unit_tests/test_scan_data.py
View file

@ -5,7 +5,11 @@ from copy import copy
from datetime import datetime, timedelta
from math import floor
from openflexure_microscope_server.scan_directories import ScanData
from pydantic import BaseModel
from openflexure_microscope_server.scan_directories import HistoricScanData
from openflexure_microscope_server.stitching import StitchingSettings
from openflexure_microscope_server.things.smart_scan import ActiveScanData
MOCK_START_TIME = datetime(
year=2024,
@ -28,8 +32,8 @@ MOCK_END_TIME = datetime(
)
def _fake_legacy_scan_data(**kwargs) -> ScanData:
"""Make fake legacy scan data, the start time is now. Final properties are not added.
def _fake_legacy_scan_data(**kwargs) -> HistoricScanData:
"""Make fake legacy scan data.
:param **kwargs: Key word arguments can be used to override other values.
"""
@ -50,12 +54,73 @@ def _fake_legacy_scan_data(**kwargs) -> ScanData:
}
for key, value in kwargs.items():
data_dict[key] = value
return ScanData(**data_dict)
return HistoricScanData(**data_dict)
class MockWorkflowSettingModel(BaseModel):
"""A mock model to check that ActiveScanData can hold arbitrary models."""
setting_1: int
setting_2: int
setting_3: str
def fake_active_scan_data():
"""Fake scan data for and active scan.
The start time is now. Final properties are not added.
"""
return ActiveScanData(
schema_version=2,
scan_name="fake_scan_0001",
starting_position={"x": 123, "y": 456, "z": 789},
start_time=copy(MOCK_START_TIME),
stitch_automatically=True,
save_resolution=(1000, 1000),
stitching_settings=StitchingSettings(correlation_resize=0.25, overlap=0.1),
workflow="MockWorkflow",
workflow_settings=MockWorkflowSettingModel(
setting_1=1,
setting_2=2,
setting_3="three",
),
)
def assert_active_and_historic_data_equivalent(active_data, historic_data):
"""Raise and error if active and historic scan data is not equivalent."""
assert isinstance(active_data, ActiveScanData)
assert isinstance(historic_data, HistoricScanData)
# For the round trip to be equal we must remove microseconds from the start
# time as they are not saved
active_data.start_time = active_data.start_time.replace(microsecond=0)
for key in active_data.model_fields:
if key == "workflow_settings":
# For workflow_settings check the base model serialises to the historic
# data.
active_wf_setting_dict = active_data.workflow_settings.model_dump()
assert historic_data.workflow_settings == active_wf_setting_dict
continue
assert getattr(active_data, key) == getattr(historic_data, key)
def test_legacy_data_validates():
"""Check that legacy scan data validates."""
scan_data = _fake_legacy_scan_data()
assert isinstance(scan_data, HistoricScanData)
assert scan_data.image_count == 0
assert scan_data.duration is None
assert scan_data.scan_result is None
# Most importantly legacy stitching data should now be in the StitchingSettings
# model
assert scan_data.stitching_settings.correlation_resize == 0.25
assert scan_data.stitching_settings.overlap == 0.1
def test_set_final_data():
"""Check that adding final data to a ScanData object works as expected."""
scan_data = _fake_legacy_scan_data()
"""Check that adding final data to a ActiveScanData object works as expected."""
scan_data = fake_active_scan_data()
assert scan_data.image_count == 0
assert scan_data.duration is None
@ -75,8 +140,8 @@ def test_set_final_data():
def test_custom_serialisation():
"""Check that the custom serialisation in ScanData works as expected."""
scan_data = _fake_legacy_scan_data()
"""Check that the custom serialisation in ActiveScanData works as expected."""
scan_data = fake_active_scan_data()
# Serialise to string then load directly as json
scan_data_dict = json.loads(scan_data.model_dump_json())
assert scan_data_dict["start_time"] == "2024-12-25_11:00:00"
@ -95,26 +160,23 @@ def test_custom_serialisation():
def test_round_trip_not_finalised():
"""Check that ScanData without final data can be serialised and deserialised."""
scan_data = _fake_legacy_scan_data()
"""Check that ActiveScanData without final data can be serialised and deserialised."""
scan_data = fake_active_scan_data()
scan_data_dict = json.loads(scan_data.model_dump_json())
scan_data_reloaded = ScanData(**scan_data_dict)
scan_data_reloaded = HistoricScanData(**scan_data_dict)
# For the round trip to be equal we must remove microseconds from the start
# time as they are not saved
scan_data.start_time = scan_data.start_time.replace(microsecond=0)
assert scan_data == scan_data_reloaded
assert_active_and_historic_data_equivalent(scan_data, scan_data_reloaded)
def test_round_trip_finalised():
"""Check that finalised ScanData can be serialised and deserialised."""
scan_data = _fake_legacy_scan_data()
"""Check that finalised HistoricScanData can be serialised and deserialised."""
scan_data = fake_active_scan_data()
# Finalise the data.
scan_data.image_count += 123
scan_data.set_final_data(result="Success")
scan_data_dict = json.loads(scan_data.model_dump_json())
scan_data_reloaded = ScanData(**scan_data_dict)
scan_data_reloaded = HistoricScanData(**scan_data_dict)
# For the round trip to be equal we must remove microseconds from the start
# time and duration as they are not saved
@ -122,4 +184,4 @@ def test_round_trip_finalised():
scan_data.start_time = scan_data.start_time.replace(microsecond=0)
scan_data.duration = timedelta(seconds=floor(scan_data.duration.total_seconds()))
assert scan_data == scan_data_reloaded
assert_active_and_historic_data_equivalent(scan_data, scan_data_reloaded)

53
tests/unit_tests/test_scan_directories.py
View file

@ -21,7 +21,10 @@ from openflexure_microscope_server.scan_directories import (
get_files_in_zip,
)
from .test_scan_data import _fake_scan_data
from .test_scan_data import (
assert_active_and_historic_data_equivalent,
fake_active_scan_data,
)
from .utilities import assert_unique_of_length
# Use our own dir in the root temp dir not a dynamically generated one so we
@ -173,7 +176,7 @@ def test_scan_sequence_and_listing(caplog):
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
# Create some scan data and mark it as successful to get an end date.
scan_data = _fake_scan_data()
scan_data = fake_active_scan_data()
scan_data.set_final_data(result="Success")
# Make 4 scans
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
@ -362,26 +365,33 @@ def test_get_scan_data_path():
assert scan_dir_manager.get_scan_data_path(scan_name) is None
def test_get_scan_data_dict():
"""Check that the dictionary for the scan data is returned, or None if doesn't exist."""
def test_get_scan_data():
"""Check that the scan data is returned, or None if doesn't exist."""
_clear_scan_dir()
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
scan_name = scan_dir.name
# Doesn't yet exist
assert scan_dir_manager.get_scan_data_dict(scan_name) is None
assert scan_dir_manager.get_scan_data(scan_name) is None
fake_data = {"foo": 1, "bar": "foobar"}
fake_active_data = fake_active_scan_data()
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
json.dump(fake_data, json_file)
json.dump(fake_active_data.model_dump(), json_file)
# Should now be able to load this fake data from disk
assert scan_dir_manager.get_scan_data_dict(scan_name) == fake_data
fake_historic_data = scan_dir_manager.get_scan_data(scan_name)
assert_active_and_historic_data_equivalent(fake_active_data, fake_historic_data)
# Check None is returned if the data cannot be read.
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
json_file.write("this is not json")
assert scan_dir_manager.get_scan_data_dict(scan_name) is None
assert scan_dir_manager.get_scan_data(scan_name) is None
# Check None is returned if the data cannot or is json but cannot be serialised to
# the data model
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
json_file.write(json.dumps({"foo": "bar"}))
assert scan_dir_manager.get_scan_data(scan_name) is None
def test_empty_scan_info():
@ -442,29 +452,6 @@ def test_zipping_scan_data():
assert not file.endswith(".dzi")
def test_saving_and_loading_scan_data():
"""Test that scan data is saved and loaded as expected."""
_clear_scan_dir()
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
scan_name = scan_dir.name
# Should start without a scan data file.
assert not os.path.isfile(scan_dir.scan_data_path)
# Create
scan_data_obj = _fake_scan_data()
scan_dir.save_scan_data(scan_data_obj)
# File should now exist
assert os.path.isfile(scan_dir.scan_data_path)
# Dump the scan json to a string an reload it
# Note that more detailed checking of the dumping and loading of ScanData is in
# tests/test_scan_data.py
scan_data_dict = json.loads(scan_data_obj.model_dump_json())
# What is loaded from file should be the same as from dumping and loading.
assert scan_dir_manager.get_scan_data_dict(scan_name) == scan_data_dict
def test_saving_scan_data_error():
"""Test that saving scan data if there is no images directory raises FileNotFoundError."""
_clear_scan_dir()
@ -475,7 +462,7 @@ def test_saving_scan_data_error():
shutil.rmtree(scan_dir.images_dir)
# Should raise FileNotFoundError.
with pytest.raises(FileNotFoundError):
scan_dir.save_scan_data(_fake_scan_data())
scan_dir.save_scan_data(fake_active_scan_data())
def test_all_files():

18
tests/unit_tests/test_smart_scan.py
View file

@ -26,11 +26,9 @@ 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,
ScanData,
)
from openflexure_microscope_server.scan_directories import NotEnoughFreeSpaceError
from openflexure_microscope_server.things.smart_scan import (
ActiveScanData,
ScanNotRunningError,
SmartScanThing,
)
@ -224,7 +222,7 @@ MOCK_START_POS = {"x": 123, "y": 456, "z": 789}
def _expected_scan_data():
"""Return the expected ScanData object for a SmartScan with default properties."""
"""Return the expected ActiveScanData object for a SmartScan with default properties."""
expected_dict = {
"scan_name": MOCK_SCAN_NAME,
"starting_position": MOCK_START_POS,
@ -239,7 +237,7 @@ def _expected_scan_data():
"correlation_resize": 0.5,
"save_resolution": (1640, 1232),
}
return ScanData(start_time=datetime.now(), **expected_dict)
return ActiveScanData(start_time=datetime.now(), **expected_dict)
@pytest.fixture
@ -264,7 +262,7 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker):
def test_collect_scan_data(scan_thing_mocked_for_scan_data):
"""Run _collect_scan_data, and check the ScanData object has the expected values."""
"""Run _collect_scan_data, and check the ActiveScanData object has the expected values."""
scan_thing = scan_thing_mocked_for_scan_data
data = scan_thing._collect_scan_data()
@ -277,7 +275,7 @@ def test_collect_scan_data(scan_thing_mocked_for_scan_data):
def test_save_final_scan_data(scan_thing_mocked_for_scan_data):
"""Run _save_final_scan_data, check save is called with final results in ScanData."""
"""Run _save_final_scan_data, check save is called with final results in ActiveScanData."""
scan_thing = scan_thing_mocked_for_scan_data
scan_thing._scan_data = scan_thing._collect_scan_data()
@ -287,7 +285,7 @@ def test_save_final_scan_data(scan_thing_mocked_for_scan_data):
# the value
scan_thing._ongoing_scan.save_scan_data.assert_called()
final_data = scan_thing._ongoing_scan.save_scan_data.call_args[0][0]
assert isinstance(final_data, ScanData)
assert isinstance(final_data, ActiveScanData)
assert final_data.scan_result == "Mocked!"
assert final_data.image_count == 44
assert final_data.duration.total_seconds() < 1
@ -341,7 +339,7 @@ def check_run_scan(scan_thing, caplog, expected_exception=None):
def test_run_scan(scan_thing_mocked_for_run_scan, caplog):
"""Run _save_final_scan_data, check save is called with final results in ScanData."""
"""Run _save_final_scan_data, check save is called with final results in ActiveScanData."""
result, logs, calls = check_run_scan(scan_thing_mocked_for_run_scan, caplog)
assert result == "success"

16
tests/unit_tests/test_stack.py
View file

@ -19,7 +19,7 @@ from openflexure_microscope_server.things.autofocus import (
AutofocusThing,
CaptureInfo,
NotAPeakError,
StackParams,
SmartStackParams,
_count_turning_points,
_get_capture_by_id,
_get_capture_index_by_id,
@ -64,7 +64,7 @@ def test_stack_params_validation(save_ims, extra_ims):
# Coerce min_images_to_test as the max extra ims depends on save_ims so is hard
# to do automatically in hypothesis. This clamps the number between 3 and 9.
min_images_to_test = max(min(save_ims + extra_ims, 9), 3)
StackParams(
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=min_images_to_test,
@ -91,7 +91,7 @@ def test_stack_params_not_enough_test_images(save_ims, extra_ims):
"Can't save more images than the minimum number tested)"
)
with pytest.raises(ValueError, match=match):
StackParams(
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
@ -116,7 +116,7 @@ def test_stack_params_negative_images_to_save(save_ims, extra_ims):
"Images to save must be positive and odd)"
)
with pytest.raises(ValueError, match=match):
StackParams(
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
@ -142,7 +142,7 @@ def test_even_min_images_to_test(save_ims, extra_ims):
"Minimum number of images to test should be positive and odd)"
)
with pytest.raises(ValueError, match=match):
StackParams(
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
@ -166,7 +166,7 @@ def test_even_images_to_save(save_ims, extra_ims):
"Images to save must be positive and odd)"
)
with pytest.raises(ValueError, match=match):
StackParams(
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
@ -177,11 +177,11 @@ def test_even_images_to_save(save_ims, extra_ims):
def test_computed_stack_params():
"""Test StackParams computed properties are as expected.
"""Test SmartStackParams computed properties are as expected.
Not using hypothesis or we will just copy in the same formulas.
"""
stack_parameters = StackParams(
stack_parameters = SmartStackParams(
stack_dz=50,
images_to_save=5,
min_images_to_test=9,