"""Test the stage without creating a full HTTP server and socket connection.""" from collections.abc import Mapping import tempfile import itertools from fastapi.testclient import TestClient import pytest from httpx import HTTPStatusError from hypothesis import given, settings, HealthCheck, strategies as st import labthings_fastapi as lt from labthings_fastapi.exceptions import NotConnectedToServerError from openflexure_microscope_server.things.stage import ( BaseStage, RedefinedBaseMovementError, ) from openflexure_microscope_server.things.stage.dummy import DummyStage # Keep the size and number of moves fairly small or the tests can take forever point3d = st.tuples( st.integers(min_value=-100, max_value=100), st.integers(min_value=-100, max_value=100), st.integers(min_value=-100, max_value=100), ) 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) @pytest.fixture def thing_server(dummy_stage): """Yield a server with a very basic configuration.""" with tempfile.TemporaryDirectory() as tmpdir: server = lt.ThingServer(settings_folder=tmpdir) server.add_thing(dummy_stage, "/stage/") 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. ``move_absolute`` and ``move_relative`` are in the program reference frame and convert to the hardware frame. It is recommended to override ``_hardware_move_relative`` and ``_hardware_move_absolute`` instead. Check that the expected error is raised if the methods are overridden. """ class BadStage1(BaseStage): @lt.thing_action def move_relative( self, cancel: lt.deps.CancelHook, block_cancellation: bool = False, **kwargs: Mapping[str, int], ): pass with pytest.raises(RedefinedBaseMovementError): BadStage1() class BadStage2(BaseStage): @lt.thing_action def move_absolute( self, cancel: lt.deps.CancelHook, block_cancellation: bool = False, **kwargs: Mapping[str, int], ): pass with pytest.raises(RedefinedBaseMovementError): BadStage2() def _set_axis_direction(dummy_stage, direction): """Set the axis direction directly even if not connected to server.""" try: dummy_stage.axis_inverted = direction except NotConnectedToServerError: pass def test_apply_axis_direction_all_pos(dummy_stage): """Test the apply axis direction function behaves as expected when axis +v3.""" # Directly create a stage not through a ThingServer to access private methods _set_axis_direction(dummy_stage, {"x": False, "y": False, "z": False}) # A list of positions to try positions = [ [1, 2, 3], # list {"x": 1, "y": 2, "z": 3}, # mapping {"x": 1, "z": 2, "y": 3}, # mapping out of order {"x": 3}, # Mapping with only 1 value {"x": 1, "z": 2}, # Mapping with only 2 values ] for pos in positions: assert dummy_stage._apply_axis_direction(pos) == pos # Check tuple separately as it gets converted to list assert dummy_stage._apply_axis_direction((1, 2, 0)) == [1, 2, 0] def test_apply_axis_direction_mixed(dummy_stage): """Test the apply axis direction function behaves as expected when axis dirs are mixed.""" # Make x and z negative _set_axis_direction(dummy_stage, {"x": True, "y": False, "z": True}) # A list of (input position, output position) to try position_pairs = [ ([1, 2, 3], [-1, 2, -3]), # list ([1, "2", "3"], [-1, 2, -3]), # list with strings ((1, 2, 0), [-1, 2, 0]), # tuple (gets converted to list) ({"x": 1, "y": 2, "z": 3}, {"x": -1, "y": 2, "z": -3}), # mapping ({"x": 1, "z": 2, "y": 3}, {"x": -1, "z": -2, "y": 3}), # mapping out of order ({"x": 1, "y": "2", "z": "3"}, {"x": -1, "y": 2, "z": -3}), # mapping w strings ({"x": 3}, {"x": -3}), # Mapping with only 1 value ({"x": 1, "z": 2}, {"x": -1, "z": -2}), # Mapping with only 2 values ] for pos, expected_pos in position_pairs: assert dummy_stage._apply_axis_direction(pos) == expected_pos def test_apply_axis_errors(dummy_stage): """Test the apply axis direction returns appropriate errors.""" with pytest.raises(TypeError): dummy_stage._apply_axis_direction(None) with pytest.raises(TypeError): dummy_stage._apply_axis_direction("Onwards!") with pytest.raises(KeyError): dummy_stage._apply_axis_direction({"x": -1, "y": 2, "z": 4, "up": -3}) with pytest.raises(KeyError): dummy_stage._apply_axis_direction({"x": -1, "y": 2, "up": -3}) def test_default_values(stage_client, 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"] # position starts at 0, 0, 0 assert stage_client.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} # 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): """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} # 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(stage_client, dummy_stage, axis_inverted, path): """Test moving relative, ensuring position and hardware position behave as expected. :param axis_inverted: Is used to set the inversion. :param path: The 3d path to move over, generated by hypothesis. """ _set_axis_direction(dummy_stage, axis_inverted) # Explicitly do axes calculation here to check logic in main code. x_dir = -1 if axis_inverted["x"] else 1 y_dir = -1 if axis_inverted["y"] else 1 z_dir = -1 if axis_inverted["z"] else 1 position = list(stage_client.position.values()) for movement in path: stage_client.move_relative(x=movement[0], y=movement[1], z=movement[2]) position = [pos + move for pos, move in zip(position, movement)] stage_pos = stage_client.get_xyz_position() hw_pos = dummy_stage._hardware_position assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir @given(path=path3d) @settings( max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=10000, ) def test_move_relative(stage_client, 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 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 # Create every combination of True/False for x,y,z. inversion_combinations = [ dict(zip(axis_names, inverted)) for inverted in itertools.product([True, False], repeat=len(axis_names)) ] print(path) for axis_inverted in inversion_combinations: _test_move_relative( stage_client=stage_client, dummy_stage=dummy_stage, axis_inverted=axis_inverted, path=path, ) def _test_move_absolute(stage_client, dummy_stage, axis_inverted, path): """Test moving relative, ensuring position and hardware position behave as expected. :param axis_inverted: Is used to set the inversion. :param path: The 3d path to move over, generated by hypothesis. """ _set_axis_direction(dummy_stage, axis_inverted) # Explicitly do axes calculation here to check logic in main code. x_dir = -1 if axis_inverted["x"] else 1 y_dir = -1 if axis_inverted["y"] else 1 z_dir = -1 if axis_inverted["z"] else 1 position = list(stage_client.position.values()) for move_to in path: stage_client.move_absolute(x=move_to[0], y=move_to[1], z=move_to[2]) position = move_to stage_pos = stage_client.get_xyz_position() hw_pos = dummy_stage._hardware_position assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir @given(path=path3d) @settings( max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=10000, ) def test_move_absolute(stage_client, 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 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 # Create every combination of True/False for x,y,z. inversion_combinations = [ dict(zip(axis_names, inverted)) for inverted in itertools.product([True, False], repeat=len(axis_names)) ] print(path) for axis_inverted in inversion_combinations: _test_move_absolute( stage_client=stage_client, dummy_stage=dummy_stage, axis_inverted=axis_inverted, path=path, ) def test_thing_description_equivalence(dummy_stage, mocker): """Stop extra actions getting added to child classes without explicit approval. To add an extra action to a stage this test needs to be updated, highlighting it at review. This tests should explain why the action isn't on the general base class. """ mock_sangaboard = mocker.Mock() mocker.patch.dict("sys.modules", {"sangaboard": mock_sangaboard}) from openflexure_microscope_server.things.stage.sangaboard import SangaboardThing # 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_actions = set(base_td.actions.keys()) base_properties = set(base_td.properties.keys()) dummy_td = dummy_stage.thing_description() dummy_actions = set(dummy_td.actions.keys()) dummy_properties = set(dummy_td.properties.keys()) sanga_td = SangaboardThing().thing_description() # Remove known extra actions sanga_actions = list(sanga_td.actions.keys()) for action in extra_sanga_actions: index = sanga_actions.index(action) sanga_actions.pop(index) sanga_actions = set(sanga_actions) sanga_properties = set(sanga_td.properties.keys()) assert sanga_actions == dummy_actions == base_actions assert sanga_properties == dummy_properties == base_properties