130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
import types
|
|
|
|
from aare.common.coordinate import AerotechCoordinate, Coordinate
|
|
from aare.common.exception_handler import CriticalTellException, MountingFailed
|
|
from aare.common.models import DewarAddress, SampleShortInfo
|
|
from aare.devices.tell_client import TellEventValueEnum
|
|
from aare.daq.operations.mounting.models import MountingContext, MountingResult
|
|
from aare.daq.operations.mounting.service import MountingService
|
|
|
|
|
|
def _make_sample(sample_id: int, name: str) -> SampleShortInfo:
|
|
return SampleShortInfo(
|
|
db_id=sample_id,
|
|
puck_name="puck1",
|
|
dewar_name="dew1",
|
|
sample_name=name,
|
|
run_number=1,
|
|
user="group1",
|
|
pin=sample_id,
|
|
location=DewarAddress(segment="A", pos=1),
|
|
)
|
|
|
|
|
|
def _make_context(previous_sample=None):
|
|
tell = types.SimpleNamespace(
|
|
mount=lambda **kwargs: TellEventValueEnum.SUCCESS,
|
|
unmount=lambda **kwargs: None,
|
|
dry=lambda **kwargs: None,
|
|
check_enable_motion=lambda: None,
|
|
wait_not_busy=lambda timeout=360.0: None,
|
|
set_in_mount_position=lambda value: None,
|
|
)
|
|
|
|
devs = types.SimpleNamespace(
|
|
smargon_move_home=lambda: None,
|
|
aerotech_pos=None,
|
|
tell=tell,
|
|
magnet_position_sensor=types.SimpleNamespace(value=0),
|
|
)
|
|
|
|
cfg = types.SimpleNamespace(
|
|
current_sample=previous_sample,
|
|
get_mount_fail_count=lambda: 0,
|
|
record_mount_success=lambda: None,
|
|
record_mount_failure=lambda: 1,
|
|
)
|
|
|
|
return MountingContext(
|
|
cfg=cfg,
|
|
devs=devs,
|
|
mount_position=AerotechCoordinate(at_mm=Coordinate(x=0, y=0, z=0), omega_deg=0),
|
|
)
|
|
|
|
|
|
def test_execute_mount_success(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert isinstance(result, MountingResult)
|
|
assert result.success is True
|
|
assert result.mounted_sample == target_sample
|
|
assert result.previous_sample == previous_sample
|
|
assert result.did_unmount_previous is True
|
|
assert ctx.cfg.current_sample == target_sample
|
|
|
|
|
|
def test_execute_unmount_success(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=None)
|
|
|
|
assert result.success is True
|
|
assert result.mounted_sample is None
|
|
assert result.previous_sample == previous_sample
|
|
assert result.did_unmount_previous is True
|
|
assert ctx.cfg.current_sample is None
|
|
|
|
|
|
def test_execute_mount_returns_failed_result_for_no_pin_in_gripper(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.NO_PIN_IN_GRIPPER
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, MountingFailed)
|
|
assert result.is_error is True
|
|
|
|
|
|
def test_execute_mount_returns_failed_result_for_unhandled_tell_response(mock_logger):
|
|
target_sample = _make_sample(2, "new")
|
|
ctx = _make_context()
|
|
ctx.devs.tell.mount = lambda **kwargs: TellEventValueEnum.UNKNOWN
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
|
|
result = service.execute(target=target_sample)
|
|
|
|
assert result.success is False
|
|
assert isinstance(result.error, CriticalTellException)
|
|
assert result.is_error is True
|
|
|
|
|
|
def test_dry_unmounts_current_sample_before_drying(mock_logger):
|
|
previous_sample = _make_sample(1, "old")
|
|
ctx = _make_context(previous_sample=previous_sample)
|
|
unmount_calls = []
|
|
dry_calls = []
|
|
|
|
ctx.devs.tell.unmount = lambda **kwargs: unmount_calls.append(kwargs)
|
|
ctx.devs.tell.dry = lambda **kwargs: dry_calls.append(kwargs)
|
|
|
|
service = MountingService(context=ctx, logger=mock_logger)
|
|
service.dry(park=True)
|
|
|
|
assert len(unmount_calls) == 1
|
|
assert ctx.cfg.current_sample is None
|
|
assert dry_calls == [{"wait_cold": -1, "wait": True}]
|