Get unit tests passing with Scan Workflows

This commit is contained in:
Julian Stirling 2026-01-16 00:08:33 +00:00
parent c82523dd5b
commit 2a3a54b936
5 changed files with 149 additions and 135 deletions

View file

@ -225,6 +225,11 @@ class FinalStitcher(BaseStitcher):
HistoricScanData for this scan as a dictionary. HistoricScanData for this scan as a dictionary.
:param stitch_tiff: Whether to stitch a pyramidal TIFF. :param stitch_tiff: Whether to stitch a pyramidal TIFF.
""" """
if not isinstance(stitching_settings, StitchingSettings):
raise StitcherValidationError(
"Final stitcher requires settings to be set as a StitchingSettings "
"model"
)
self.logger = logger self.logger = logger
overlap = stitching_settings.overlap overlap = stitching_settings.overlap
correlation_resize = stitching_settings.correlation_resize correlation_resize = stitching_settings.correlation_resize

View file

@ -146,7 +146,7 @@ class SmartStackParams(BaseModel):
@dataclass @dataclass
class CaptureInfo: class CaptureInfo:
"""The information from a capture in a z_stack.""" """The information from a capture in a smart_z_stack."""
buffer_id: int buffer_id: int
position: Mapping[str, int] position: Mapping[str, int]

View file

@ -22,11 +22,13 @@ from typing import Callable, Optional
import pytest import pytest
from fastapi import HTTPException from fastapi import HTTPException
from pydantic import BaseModel
from labthings_fastapi.exceptions import InvocationCancelledError from labthings_fastapi.exceptions import InvocationCancelledError
from labthings_fastapi.testing import create_thing_without_server from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import NotEnoughFreeSpaceError from openflexure_microscope_server.scan_directories import NotEnoughFreeSpaceError
from openflexure_microscope_server.stitching import StitchingSettings
from openflexure_microscope_server.things.smart_scan import ( from openflexure_microscope_server.things.smart_scan import (
ActiveScanData, ActiveScanData,
ScanNotRunningError, ScanNotRunningError,
@ -221,21 +223,28 @@ MOCK_SCAN_DIR = "scans/test_name_0001/images/"
MOCK_START_POS = {"x": 123, "y": 456, "z": 789} MOCK_START_POS = {"x": 123, "y": 456, "z": 789}
class MockWorkflowSettingModel(BaseModel):
"""A mock model to check that ActiveScanData can hold arbitrary models."""
foo: str = "bar"
bar: str = "foo"
dx: int = 123
dy: int = 456
def _expected_scan_data(): def _expected_scan_data():
"""Return the expected ActiveScanData object for a SmartScan with default properties.""" """Return the expected ActiveScanData object for a SmartScan with default properties."""
expected_dict = { expected_dict = {
"scan_name": MOCK_SCAN_NAME, "scan_name": MOCK_SCAN_NAME,
"starting_position": MOCK_START_POS, "starting_position": MOCK_START_POS,
"overlap": 0.45,
"max_dist": 45000,
"dx": 100,
"dy": 100,
"autofocus_dz": 1000,
"autofocus_on": True,
"skip_background": True,
"stitch_automatically": True,
"correlation_resize": 0.5,
"save_resolution": (1640, 1232), "save_resolution": (1640, 1232),
"stitch_automatically": True,
"stitching_settings": {
"overlap": 0.45,
"correlation_resize": 0.5,
},
"workflow": "Mock",
"workflow_settings": MockWorkflowSettingModel(),
} }
return ActiveScanData(start_time=datetime.now(), **expected_dict) return ActiveScanData(start_time=datetime.now(), **expected_dict)
@ -245,14 +254,14 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker):
"""Return a scan thing that is mocked so that _collect_scan_data will run.""" """Return a scan thing that is mocked so that _collect_scan_data will run."""
# Set the lock so it thinks the scan is running # Set the lock so it thinks the scan is running
with smart_scan_thing._scan_lock: with smart_scan_thing._scan_lock:
mocker.patch.object(
smart_scan_thing,
"_calc_displacement_from_test_image",
return_value=[100, 100],
)
smart_scan_thing._stage.position = MOCK_START_POS smart_scan_thing._stage.position = MOCK_START_POS
smart_scan_thing._workflow.all_settings.return_value = (
MockWorkflowSettingModel(),
StitchingSettings(correlation_resize=0.5, overlap=0.45),
)
smart_scan_thing._workflow.save_resolution = (1640, 1232)
mock_ongoing_scan = mocker.Mock() mock_ongoing_scan = mocker.Mock()
mock_ongoing_scan.name = MOCK_SCAN_NAME mock_ongoing_scan.name = MOCK_SCAN_NAME
mock_ongoing_scan.images_dir = MOCK_SCAN_DIR mock_ongoing_scan.images_dir = MOCK_SCAN_DIR
@ -265,7 +274,7 @@ def test_collect_scan_data(scan_thing_mocked_for_scan_data):
"""Run _collect_scan_data, and check the ActiveScanData 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 scan_thing = scan_thing_mocked_for_scan_data
data = scan_thing._collect_scan_data() data = scan_thing._collect_scan_data(scan_thing._workflow)
expected_data = _expected_scan_data() expected_data = _expected_scan_data()
time_diff = expected_data.start_time - data.start_time time_diff = expected_data.start_time - data.start_time
assert abs(time_diff.total_seconds()) < 1 assert abs(time_diff.total_seconds()) < 1
@ -278,7 +287,7 @@ 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 ActiveScanData.""" """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_thing_mocked_for_scan_data
scan_thing._scan_data = scan_thing._collect_scan_data() scan_thing._scan_data = scan_thing._collect_scan_data(scan_thing._workflow)
scan_thing._scan_data.image_count = 44 scan_thing._scan_data.image_count = 44
scan_thing._save_final_scan_data("Mocked!") scan_thing._save_final_scan_data("Mocked!")
# _ongoing_scan is a mock so we can check that save_scan data was called and get # _ongoing_scan is a mock so we can check that save_scan data was called and get
@ -320,10 +329,10 @@ def check_run_scan(scan_thing, caplog, expected_exception=None):
""" """
if expected_exception is None: if expected_exception is None:
with caplog.at_level(logging.WARNING): with caplog.at_level(logging.WARNING):
scan_thing._scan_data = scan_thing._run_scan() scan_thing._scan_data = scan_thing._run_scan(scan_thing._workflow)
else: else:
with pytest.raises(expected_exception), caplog.at_level(logging.WARNING): with pytest.raises(expected_exception), caplog.at_level(logging.WARNING):
scan_thing._scan_data = scan_thing._run_scan() scan_thing._scan_data = scan_thing._run_scan(scan_thing._workflow)
# The preview stitcher object should still exist. And images dir should be set. # The preview stitcher object should still exist. And images dir should be set.
assert scan_thing._preview_stitcher.images_dir == MOCK_SCAN_DIR assert scan_thing._preview_stitcher.images_dir == MOCK_SCAN_DIR

View file

@ -25,6 +25,7 @@ from openflexure_microscope_server.things.autofocus import (
_get_capture_index_by_id, _get_capture_index_by_id,
_get_peak_turning_point, _get_peak_turning_point,
) )
from openflexure_microscope_server.things.scan_workflows import HistoScanWorkflow
RANDOM_GENERATOR = np.random.default_rng() RANDOM_GENERATOR = np.random.default_rng()
@ -267,20 +268,28 @@ def autofocus_thing():
return create_thing_without_server(AutofocusThing, mock_all_slots=True) return create_thing_without_server(AutofocusThing, mock_all_slots=True)
def test_create_stack(autofocus_thing, caplog): @pytest.fixture
def histo_scan_workflow():
"""Return an autofocus thing connected to a server."""
return create_thing_without_server(HistoScanWorkflow, mock_all_slots=True)
def test_create_stack(histo_scan_workflow, caplog):
"""Run create stack with default values and check there is no coercion or logging.""" """Run create stack with default values and check there is no coercion or logging."""
initial_min_images_to_test = autofocus_thing.stack_min_images_to_test initial_min_images_to_test = histo_scan_workflow.stack_min_images_to_test
initial_images_to_save = autofocus_thing.stack_images_to_save initial_images_to_save = histo_scan_workflow.stack_images_to_save
with caplog.at_level(logging.INFO): with caplog.at_level(logging.INFO):
stack_params = autofocus_thing.create_stack_params( stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
) )
assert len(caplog.records) == 0 assert len(caplog.records) == 0
assert autofocus_thing.stack_min_images_to_test == initial_min_images_to_test assert histo_scan_workflow.stack_min_images_to_test == initial_min_images_to_test
assert autofocus_thing.stack_images_to_save == initial_images_to_save assert histo_scan_workflow.stack_images_to_save == initial_images_to_save
assert stack_params.min_images_to_test == autofocus_thing.stack_min_images_to_test assert (
assert stack_params.images_to_save == autofocus_thing.stack_images_to_save stack_params.min_images_to_test == histo_scan_workflow.stack_min_images_to_test
)
assert stack_params.images_to_save == histo_scan_workflow.stack_images_to_save
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -293,13 +302,13 @@ def test_create_stack(autofocus_thing, caplog):
], ],
) )
def test_coercing_stack_test_ims( def test_coercing_stack_test_ims(
initial_test_ims, coerced_test_ims, expected_log_start, autofocus_thing, caplog initial_test_ims, coerced_test_ims, expected_log_start, histo_scan_workflow, caplog
): ):
"""Run create stack with images to test set to values requiring coercion, and check result.""" """Run create stack with images to test set to values requiring coercion, and check result."""
autofocus_thing.stack_min_images_to_test = initial_test_ims histo_scan_workflow.stack_min_images_to_test = initial_test_ims
with caplog.at_level(logging.WARNING): with caplog.at_level(logging.WARNING):
stack_params = autofocus_thing.create_stack_params( stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
) )
@ -308,7 +317,9 @@ def test_coercing_stack_test_ims(
# Check the value is coerced in the stack_params # Check the value is coerced in the stack_params
assert stack_params.min_images_to_test == coerced_test_ims assert stack_params.min_images_to_test == coerced_test_ims
# Check that the setting in the Thing was updated to the coerced value # Check that the setting in the Thing was updated to the coerced value
assert stack_params.min_images_to_test == autofocus_thing.stack_min_images_to_test assert (
stack_params.min_images_to_test == histo_scan_workflow.stack_min_images_to_test
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -321,13 +332,13 @@ def test_coercing_stack_test_ims(
], ],
) )
def test_coercing_stack_save_ims( def test_coercing_stack_save_ims(
initial_save_ims, coerced_save_ims, expected_log_start, autofocus_thing, caplog initial_save_ims, coerced_save_ims, expected_log_start, histo_scan_workflow, caplog
): ):
"""Run create stack with images to save set to values requiring coercion, and check result.""" """Run create stack with images to save set to values requiring coercion, and check result."""
autofocus_thing.stack_images_to_save = initial_save_ims histo_scan_workflow.stack_images_to_save = initial_save_ims
with caplog.at_level(logging.WARNING): with caplog.at_level(logging.WARNING):
stack_params = autofocus_thing.create_stack_params( stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
) )
@ -336,13 +347,13 @@ def test_coercing_stack_save_ims(
# Check the value is coerced in the stack_params # Check the value is coerced in the stack_params
assert stack_params.images_to_save == coerced_save_ims assert stack_params.images_to_save == coerced_save_ims
# Check that the setting in the Thing was updated to the coerced value # Check that the setting in the Thing was updated to the coerced value
assert stack_params.images_to_save == autofocus_thing.stack_images_to_save assert stack_params.images_to_save == histo_scan_workflow.stack_images_to_save
@pytest.mark.parametrize("pass_on", [1, 2, 3, 4]) @pytest.mark.parametrize("pass_on", [1, 2, 3, 4])
def test_run_smart_stack(pass_on, autofocus_thing, mocker): def test_run_smart_stack(pass_on, histo_scan_workflow, autofocus_thing, mocker):
"""Test Running smart stack with the stack passing on different attempts.""" """Test Running smart stack with the stack passing on different attempts."""
stack_params = autofocus_thing.create_stack_params( stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
) )
assert stack_params.max_attempts == 3 assert stack_params.max_attempts == 3
@ -364,8 +375,8 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker):
failed_return = (False, fake_captures, "pick_me") failed_return = (False, fake_captures, "pick_me")
return_list = [failed_return] * (pass_on - 1) + [successful_return] return_list = [failed_return] * (pass_on - 1) + [successful_return]
# Mock z_stack and looping_autofocus # Mock smart_z_stack and looping_autofocus
autofocus_thing.z_stack = mocker.Mock(side_effect=return_list) autofocus_thing.smart_z_stack = mocker.Mock(side_effect=return_list)
autofocus_thing.looping_autofocus = mocker.Mock() autofocus_thing.looping_autofocus = mocker.Mock()
# Run it # Run it
@ -380,9 +391,10 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker):
# Final z is the one from the id returned by the stack "pick_me" # Final z is the one from the id returned by the stack "pick_me"
assert final_z == 555 assert final_z == 555
# z_stack should run up until the time it passes. Running no more than max_attempts # smart_z_stack should run up until the time it passes. Running no more than
# max_attempts
n_stacks = min(pass_on, stack_params.max_attempts) n_stacks = min(pass_on, stack_params.max_attempts)
assert autofocus_thing.z_stack.call_count == n_stacks assert autofocus_thing.smart_z_stack.call_count == n_stacks
# Move absolute should be 1 less time that the number of times z_stack_run # Move absolute should be 1 less time that the number of times z_stack_run
assert autofocus_thing._stage.move_absolute.call_count == n_stacks - 1 assert autofocus_thing._stage.move_absolute.call_count == n_stacks - 1
# As should looping autofocus # As should looping autofocus
@ -396,14 +408,16 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker):
assert autofocus_thing._cam.save_from_memory.call_count == (1 if success else 0) assert autofocus_thing._cam.save_from_memory.call_count == (1 if success else 0)
def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, mocker): def setup_and_run_smart_z_stack(
"""Set up a z_stack, run it, and return the result. check_returns, check_turning_points, histo_scan_workflow, autofocus_thing, mocker
):
"""Set up a smart_z_stack, run it, and return the result.
:param check_returns: The return values from check_stack_result. Note that if this :param check_returns: The return values from check_stack_result. Note that if this
is a list, it will be set as a side effect (and should be a list of tuples of is a list, it will be set as a side effect (and should be a list of tuples of
results). If it a tuple (or anything else), it is set as a return value. results). If it a tuple (or anything else), it is set as a return value.
""" """
stack_params = autofocus_thing.create_stack_params( stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
) )
stack_params.settling_time = 0 # Don't settle or tests take forever. stack_params.settling_time = 0 # Don't settle or tests take forever.
@ -413,58 +427,70 @@ def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing,
autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns) autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns)
else: else:
autofocus_thing.check_stack_result = mocker.Mock(return_value=check_returns) autofocus_thing.check_stack_result = mocker.Mock(return_value=check_returns)
return autofocus_thing.z_stack( return autofocus_thing.smart_z_stack(
stack_parameters=stack_params, stack_parameters=stack_params,
check_turning_points=check_turning_points, check_turning_points=check_turning_points,
) )
def test_z_stack_turning_toggle_passed(autofocus_thing, mocker): def test_z_stack_turning_toggle_passed(histo_scan_workflow, autofocus_thing, mocker):
"""Check that the toggling of turning points is passed to the check.""" """Check that the toggling of turning points is passed to the check."""
check_returns = ("success", "mock_id") check_returns = ("success", "mock_id")
for check_turning in [True, False]: for check_turning in [True, False]:
setup_and_run_z_stack(check_returns, check_turning, autofocus_thing, mocker) setup_and_run_smart_z_stack(
check_returns, check_turning, histo_scan_workflow, autofocus_thing, mocker
)
check_kwargs = autofocus_thing.check_stack_result.call_args.kwargs check_kwargs = autofocus_thing.check_stack_result.call_args.kwargs
assert check_kwargs["check_turning_points"] == check_turning assert check_kwargs["check_turning_points"] == check_turning
def test_z_stack_returns_on_success_and_restart(autofocus_thing, mocker): def test_z_stack_returns_on_success_and_restart(
histo_scan_workflow, autofocus_thing, mocker
):
"""Check that if the check returns success or restart then the stack exits with correct return value.""" """Check that if the check returns success or restart then the stack exits with correct return value."""
for result in ["success", "restart"]: for result in ["success", "restart"]:
check_returns = (result, "mock_id") check_returns = (result, "mock_id")
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) ret = setup_and_run_smart_z_stack(
check_returns, True, histo_scan_workflow, autofocus_thing, mocker
)
assert autofocus_thing.check_stack_result.call_count == 1 assert autofocus_thing.check_stack_result.call_count == 1
# Check the number of images taken is exactly the call count. # Check the number of images taken is exactly the call count.
ims_taken = autofocus_thing.capture_stack_image.call_count ims_taken = autofocus_thing.capture_stack_image.call_count
assert ims_taken == autofocus_thing.stack_min_images_to_test assert ims_taken == histo_scan_workflow.stack_min_images_to_test
# And the result is as expected. # And the result is as expected.
assert ret[0] == (result == "success") assert ret[0] == (result == "success")
def test_z_stack_exits_if_focus_never_found(autofocus_thing, mocker): def test_z_stack_exits_if_focus_never_found(
histo_scan_workflow, autofocus_thing, mocker
):
"""Check that if the check returns continue the stack exits eventually with a failure.""" """Check that if the check returns continue the stack exits eventually with a failure."""
check_returns = ("continue", "mock_id") check_returns = ("continue", "mock_id")
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) ret = setup_and_run_smart_z_stack(
check_returns, True, histo_scan_workflow, autofocus_thing, mocker
)
assert autofocus_thing.check_stack_result.call_count == EXTRA_STACK_CAPTURES + 1 assert autofocus_thing.check_stack_result.call_count == EXTRA_STACK_CAPTURES + 1
# Check the number of images taken is the maximum possible, set by the min images to # Check the number of images taken is the maximum possible, set by the min images to
# test and the number of extra images that can be taken # test and the number of extra images that can be taken
ims_taken = autofocus_thing.capture_stack_image.call_count ims_taken = autofocus_thing.capture_stack_image.call_count
max_ims = autofocus_thing.stack_min_images_to_test + EXTRA_STACK_CAPTURES max_ims = histo_scan_workflow.stack_min_images_to_test + EXTRA_STACK_CAPTURES
assert ims_taken == max_ims assert ims_taken == max_ims
# And the result is as expected. # And the result is as expected.
assert not ret[0] assert not ret[0]
def test_z_stack_return(autofocus_thing, mocker): def test_z_stack_return(histo_scan_workflow, autofocus_thing, mocker):
"""Check z-stack returns as expected for more complex cases the fixed results above.""" """Check z-stack returns as expected for more complex cases the fixed results above."""
for i in range(2, EXTRA_STACK_CAPTURES): for i in range(2, EXTRA_STACK_CAPTURES):
check_returns = [ check_returns = [
("restart" if j == i - 1 else "continue", f"id_{j}") for j in range(i) ("restart" if j == i - 1 else "continue", f"id_{j}") for j in range(i)
] ]
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) ret = setup_and_run_smart_z_stack(
check_returns, True, histo_scan_workflow, autofocus_thing, mocker
)
# Calculate images taken # Calculate images taken
images_taken = autofocus_thing.stack_min_images_to_test + i - 1 images_taken = histo_scan_workflow.stack_min_images_to_test + i - 1
assert autofocus_thing.capture_stack_image.call_count == images_taken assert autofocus_thing.capture_stack_image.call_count == images_taken
# Check it reports a failure # Check it reports a failure
assert not ret[0] assert not ret[0]
@ -473,7 +499,9 @@ def test_z_stack_return(autofocus_thing, mocker):
check_returns = [ check_returns = [
("success" if j == i - 1 else "continue", f"id_{j}") for j in range(i) ("success" if j == i - 1 else "continue", f"id_{j}") for j in range(i)
] ]
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) ret = setup_and_run_smart_z_stack(
check_returns, True, histo_scan_workflow, autofocus_thing, mocker
)
# Calculate images taken # Calculate images taken
assert autofocus_thing.capture_stack_image.call_count == images_taken assert autofocus_thing.capture_stack_image.call_count == images_taken
# Check it reports a success # Check it reports a success

View file

@ -11,15 +11,16 @@ import time
from copy import copy from copy import copy
import pytest import pytest
from pydantic import BaseModel
import labthings_fastapi as lt import labthings_fastapi as lt
from openflexure_microscope_server.stitching import ( from openflexure_microscope_server.stitching import (
STITCHING_RESOLUTION,
BaseStitcher, BaseStitcher,
FinalStitcher, FinalStitcher,
PreviewStitcher, PreviewStitcher,
StitcherValidationError, StitcherValidationError,
StitchingSettings,
) )
from ..shared_utils.lt_test_utils import LabThingsTestEnv from ..shared_utils.lt_test_utils import LabThingsTestEnv
@ -88,87 +89,41 @@ FINAL_EXPECTED_COMMAND = [
FAKE_DIR, FAKE_DIR,
] ]
DEFAULT_SETTINGS = StitchingSettings(correlation_resize=0.5, overlap=0.1)
def test_final_stitcher_command_defaults(caplog):
"""Check the FinalStitcher stitches with expected default values.
It should warn when default values are used as they are a fallback.
"""
n_logs = 0
# Test with no dictionary data and with irrelevant dictionary data.
with caplog.at_level(logging.WARNING):
for data_dict in [None, {"irrelevant": "data"}]:
stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER, scan_data_dict=data_dict)
# Should log for overlap being None and correlation_resize being None
n_logs += 2
assert len(caplog.records) == n_logs
assert stitcher.command == FINAL_EXPECTED_COMMAND
def test_final_stitcher_command_tiff(caplog): def test_final_stitcher_command_tiff():
"""Check that the tiff can be requested.""" """Check that the tiff can be requested."""
# Modify defaults # Modify defaults
expected_command = copy(FINAL_EXPECTED_COMMAND) expected_command = copy(FINAL_EXPECTED_COMMAND)
expected_command[4] = "--stitch_tiff" expected_command[4] = "--stitch_tiff"
stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER, stitch_tiff=True) stitcher = FinalStitcher(
# Should log for overlap being None and correlation_resize being None FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS, stitch_tiff=True
assert len(caplog.records) == 2 )
assert stitcher.command == expected_command assert stitcher.command == expected_command
def test_final_stitcher_command_set_val_directly(): def test_final_stitcher_command_with_settings():
"""Check that values are set as expected when directly input."""
# Modify defaults
expected_command = copy(FINAL_EXPECTED_COMMAND)
expected_command[8] = "0.36"
expected_command[10] = "0.25"
# Test with no data dictionary, irrelevant data, and also the wrong data
# When wrong data is submitted, it should take the directly input data.
dict_vals = [
None,
{"irrelevant": "data"},
{"overlap": 0.2, "save_resolution": [5, 5]},
]
for data_dict in dict_vals:
stitcher = FinalStitcher(
FAKE_DIR,
logger=LOGGER,
overlap=0.4,
correlation_resize=0.25,
scan_data_dict=data_dict,
)
assert stitcher.command == expected_command
def test_final_stitcher_command_set_with_dict():
"""Check that values are set as expected when set from a ScanData dictionary.""" """Check that values are set as expected when set from a ScanData dictionary."""
# Modify defaults # Modify defaults
expected_command = copy(FINAL_EXPECTED_COMMAND) expected_command = copy(FINAL_EXPECTED_COMMAND)
expected_command[8] = "0.36" expected_command[8] = "0.36"
expected_command[10] = "0.25" expected_command[10] = "0.25"
# Check same thing works with a dictionary, resize is calculated from the saved image
# resolution. Make 4x bigger than STITCHING_RESOLUTION to get 0.25 stitcher = FinalStitcher(
resolution = [dim * 4 for dim in STITCHING_RESOLUTION] FAKE_DIR,
# Check legacy key as well as current one: logger=LOGGER,
for resolution_key in ["save_resolution", "capture resolution"]: stitching_settings=StitchingSettings(correlation_resize=0.25, overlap=0.4),
stitcher = FinalStitcher( )
FAKE_DIR, assert stitcher.command == expected_command
logger=LOGGER,
scan_data_dict={"overlap": 0.4, resolution_key: resolution},
)
assert stitcher.command == expected_command
def _validation_error_tester(scan_path, **kwargs): def _validation_error_tester(scan_path, **kwargs):
"""Check each type of stitcher throws a validation error for the given init args.""" """Check stitcher throws a validation error for the given init args."""
# If scan_data_dict is in the kwargs only test the scan_data_dict
if "scan_data_dict" not in kwargs:
with pytest.raises(StitcherValidationError):
BaseStitcher(scan_path, **kwargs).command
with pytest.raises(StitcherValidationError):
PreviewStitcher(scan_path, **kwargs).command
with pytest.raises(StitcherValidationError): with pytest.raises(StitcherValidationError):
FinalStitcher(scan_path, logger=LOGGER, **kwargs).command BaseStitcher(scan_path, **kwargs).command
with pytest.raises(StitcherValidationError):
PreviewStitcher(scan_path, **kwargs).command
def test_validation_error(): def test_validation_error():
@ -176,10 +131,23 @@ def test_validation_error():
The stitcher should throw a validation error each attempt. The stitcher should throw a validation error each attempt.
""" """
# Tests for preview (and base) stitcher
_validation_error_tester("/dir;rm -rf /;", overlap=".2", correlation_resize=".25") _validation_error_tester("/dir;rm -rf /;", overlap=".2", correlation_resize=".25")
_validation_error_tester(FAKE_DIR, overlap=".2", correlation_resize=".25;rm -rf /;") _validation_error_tester(FAKE_DIR, overlap=".2", correlation_resize=".25;rm -rf /;")
_validation_error_tester(FAKE_DIR, overlap=".2;rm -rf /;", correlation_resize=".25") _validation_error_tester(FAKE_DIR, overlap=".2;rm -rf /;", correlation_resize=".25")
_validation_error_tester(FAKE_DIR, scan_data_dict={"overlap": ".2;rm -rf /;"})
class EvilModel(BaseModel):
overlap: str
correlation_resize: str
with pytest.raises(StitcherValidationError):
FinalStitcher(
FAKE_DIR,
logger=LOGGER,
stitching_settings=EvilModel(
overlap=".2;rm -rf /;", correlation_resize=".25"
),
)
def test_extra_arg_validation(): def test_extra_arg_validation():
@ -188,7 +156,9 @@ def test_extra_arg_validation():
Currently extra args do not come from user input. But this makes checks more Currently extra args do not come from user input. But this makes checks more
future-proof. future-proof.
""" """
stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER) stitcher = FinalStitcher(
FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS
)
stitcher._extra_args = ["&&rm -rf /&&"] stitcher._extra_args = ["&&rm -rf /&&"]
with pytest.raises(StitcherValidationError): with pytest.raises(StitcherValidationError):
stitcher.command stitcher.command
@ -233,7 +203,9 @@ class StitchingTestThing(lt.Thing):
@lt.action @lt.action
def run_final(self): def run_final(self):
"""Run the final stitcher.""" """Run the final stitcher."""
stitcher = FinalStitcher(FAKE_DIR, logger=self.logger) stitcher = FinalStitcher(
FAKE_DIR, logger=self.logger, stitching_settings=DEFAULT_SETTINGS
)
# Send in the argument HANG to mock-stitch and it just hang for 10s # Send in the argument HANG to mock-stitch and it just hang for 10s
stitcher._extra_args = ["HANG"] stitcher._extra_args = ["HANG"]
stitcher.run() stitcher.run()
@ -277,9 +249,8 @@ def test_final_stitching_command(caplog, mocker):
mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd) mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd)
with caplog.at_level(logging.INFO): with caplog.at_level(logging.INFO):
# Input values to prevent logging
stitcher = FinalStitcher( stitcher = FinalStitcher(
FAKE_DIR, logger=LOGGER, overlap=0.1, correlation_resize=0.5 FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS
) )
# For the final stitcher it will always complete before returning. # For the final stitcher it will always complete before returning.
stitcher.run() stitcher.run()
@ -315,16 +286,17 @@ def test_final_stitching_command_cancelled(stitching_test_env, mocker):
assert re.match(r"^Invocation [0-9a-f-]+ was cancelled", logs[-1]["message"]) assert re.match(r"^Invocation [0-9a-f-]+ was cancelled", logs[-1]["message"])
def test_final_stitching_command_error(caplog, mocker): def test_final_stitching_command_error(mocker):
"""Check that ChildProcessError is raised if the final stitch errors.""" """Check that ChildProcessError is raised if the final stitch errors."""
mock_cmd = f"python {MOCK_STITCHER}" mock_cmd = f"python {MOCK_STITCHER}"
mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd) mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd)
with caplog.at_level(logging.INFO): stitcher = FinalStitcher(
stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER) FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS
# Send in the argument ERROR to mock-stitch and it will raise an error rather )
# than echo. # Send in the argument ERROR to mock-stitch and it will raise an error rather
stitcher._extra_args = ["ERROR"] # than echo.
with pytest.raises(ChildProcessError): stitcher._extra_args = ["ERROR"]
stitcher.run() with pytest.raises(ChildProcessError):
stitcher.run()