105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from aarecommon.errors.exception_handler import TellCommunicationError
|
|
from aarecommon.models.models import DewarAddress, SampleDewarAddress
|
|
|
|
from aare.devices.tell_backend import TellBackend
|
|
from aare.devices.tell_client import TellClient, TellEventValueEnum
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_backend():
|
|
backend = MagicMock(spec=TellBackend)
|
|
backend.get_state.return_value = "Ready"
|
|
return backend
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_beamline():
|
|
return MagicMock()
|
|
|
|
|
|
def test_get_state(mock_beamline, mock_backend):
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
assert client.get_state() == "Ready"
|
|
mock_backend.get_state.assert_called()
|
|
|
|
|
|
def test_is_in_mount_position_true(mock_beamline, mock_backend):
|
|
mock_backend.eval.return_value = "True"
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
assert client.is_in_mount_position() is True
|
|
mock_backend.eval.assert_called_with("in_mount_position&")
|
|
|
|
|
|
def test_is_in_mount_position_false(mock_beamline, mock_backend):
|
|
mock_backend.eval.return_value = "False"
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
assert client.is_in_mount_position() is False
|
|
|
|
|
|
def _mount_address():
|
|
return SampleDewarAddress(puck=DewarAddress(segment="A", pos=3), pin=10)
|
|
|
|
|
|
@pytest.mark.parametrize("state_value", ["Busy", "Ready", '"Busy"', '"Ready"', "busy"])
|
|
def test_mount_state_event_succeeds_when_command_completes(
|
|
mock_beamline, mock_backend, state_value
|
|
):
|
|
"""Regression: TELL emits a "state" event (sometimes the intermediate
|
|
"Busy", quoted or unquoted) instead of a terminal event. The mount must
|
|
rely on check_command_ok rather than string-matching the event value, so a
|
|
completed command is reported as SUCCESS rather than raising."""
|
|
mock_backend.wait_events.return_value = ("state", state_value)
|
|
mock_backend.get_result.return_value = {
|
|
"status": "completed",
|
|
"return": "A39",
|
|
"exception": None,
|
|
"id": 3017907,
|
|
}
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
# Robot reports the requested sample as mounted -> genuine success.
|
|
client.get_mounted_sample = lambda: _mount_address()
|
|
|
|
result = client.mount(_mount_address(), wait=True)
|
|
|
|
assert result == TellEventValueEnum.SUCCESS
|
|
mock_backend.get_result.assert_called() # check_command_ok consulted the result
|
|
|
|
|
|
def test_mount_state_event_reports_no_pin_when_nothing_mounted(mock_beamline, mock_backend):
|
|
"""If the command completes but the robot reports nothing mounted, the
|
|
gripper was empty: mount() must return NO_PIN_IN_GRIPPER (so the service
|
|
raises MountingFailed) rather than reporting success."""
|
|
mock_backend.wait_events.return_value = ("state", "Busy")
|
|
mock_backend.get_result.return_value = {
|
|
"status": "completed",
|
|
"return": "A39",
|
|
"exception": None,
|
|
"id": 3017907,
|
|
}
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
client.get_mounted_sample = lambda: None
|
|
|
|
result = client.mount(_mount_address(), wait=True)
|
|
|
|
assert result == TellEventValueEnum.NO_PIN_IN_GRIPPER
|
|
|
|
|
|
def test_mount_state_event_raises_when_command_not_completed(mock_beamline, mock_backend):
|
|
"""A genuine failure still surfaces: if the command did not complete,
|
|
check_command_ok raises (MountingFailed) and mount() re-raises it as a
|
|
critical TellCommunicationError, even though the event value was "Busy"."""
|
|
mock_backend.wait_events.return_value = ("state", "Busy")
|
|
mock_backend.get_result.return_value = {
|
|
"status": "error",
|
|
"return": None,
|
|
"exception": "boom",
|
|
"id": 3017907,
|
|
}
|
|
client = TellClient(mock_beamline, backend=mock_backend)
|
|
|
|
with pytest.raises(TellCommunicationError):
|
|
client.mount(_mount_address(), wait=True)
|