Files
AareDAQ/tests/unit/devices/test_tell_client.py
appleb_mandClaude Opus 4.8 5370c0f168 tell_client: restore "No Pin in Gripper" detection on empty-gripper mount
Commit 071b4ab made mount() return SUCCESS as soon as it saw the intermediate
TELL "state"/Busy event, before the later "Gripper detection" event is read,
so an empty-gripper mount completed silently with no MountingFailed/pop-up.

check_command_ok blocks (wait_not_busy) until the mount sequence finishes, so
after it returns the robot's own record is authoritative: if get_mounted_sample()
is None the gripper was empty -> return NO_PIN_IN_GRIPPER, which the mounting
service maps to MountingFailed("No Pin in Gripper"). Covers manual and automation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-24 15:30:08 +02:00

101 lines
3.8 KiB
Python

import pytest
from unittest.mock import MagicMock
from aare.common.exception_handler import TellCommunicationError
from aare.common.models import DewarAddress, SampleDewarAddress
from aare.devices.tell_client import TellClient, TellEventValueEnum
from aare.devices.tell_backend import TellBackend
@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)