mx/AareDAQ
0
1
Fork 0
Code Issues 27 Pull Requests Actions Packages Projects Releases 95 Wiki Activity

tests: fixing broken tests after refactor

Browse Source
This commit is contained in:
appleb_m
2026-06-03 15:53:29 +02:00
parent 8097fd147f
commit b8e31a749d
18 changed files with 719 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/devices/tell_client.py
+3 -3
View File
@@ -22,7 +22,7 @@ from aare.devices.tell_backend import (
from aare.common.exception_handler import (
SmartMagnetFaultException,
TellConnectionException,
TellMountFailedException,
MountingFailed,
TellCommandWhileBusyException,
ManualMountException, TellCommunicationError
)
@@ -148,7 +148,7 @@ class TellClient:
status = result["status"]
if "completed" != status:
if "removed" != status:
raise TellMountFailedException(f"{msg} {result}")
raise MountingFailed(f"{msg} {result}")
return f"{msg} {result}"
def estimate_mounting_time(self, segment) -> int:
@@ -278,7 +278,7 @@ class TellClient:
if wait:
try:
self.check_command_ok(timeout=timeout, msg="Unmount message: ")
except TellMountFailedException as e:
except MountingFailed as e:
result = self.get_result(self._last_cmd_id)
logger.error(f"Unmount failed with status '{result.get('status')}' and payload: {result}")
raise
tests/conftest.py
+136 -21
View File
@@ -1,5 +1,6 @@
import os
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import numpy as np
@@ -14,10 +15,12 @@ os.environ.setdefault("BEAMLINE", "SIMULATED")
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
os.environ.setdefault("JWT_AAREDAQ_KEY", "test_key_for_unit_testing")
@pytest.fixture
def mock_logger():
return MagicMock()
# -------------------------
# Qt app fixture
# -------------------------
@@ -46,21 +49,27 @@ def sample_info():
# -------------------------
# Shared DAQ server fixtures
# -------------------------
@pytest.fixture
def mock_backend():
@pytest.fixture(scope="session")
def server_module():
import aare.daq.server as server
return server
with patch("aare.daq.server.daq") as m_daq, \
patch("aare.daq.server.bl") as m_bl, \
patch("aare.daq.server.cfg") as m_cfg:
server.daq = m_daq
server.bl = m_bl
server.cfg = m_cfg
@pytest.fixture
def mock_backend(server_module):
with patch.object(server_module, "daq", create=True) as m_daq, \
patch.object(server_module, "bl", create=True) as m_bl, \
patch.object(server_module, "cfg", create=True) as m_cfg:
m_daq.busy = False
m_daq.camera_image = np.zeros((10, 10, 3), dtype=np.uint8)
m_cfg.pgroup = "p12345"
m_cfg.session_state.return_value = "ACTIVE"
m_cfg.GUI_SESSION_EXPIRE_SECONDS = 10
m_cfg.get_open_gui_sessions.return_value = []
m_cfg.get_gui_session.return_value = None
m_cfg.touch_gui_session.return_value = None
yield {
"daq": m_daq,
@@ -70,21 +79,127 @@ def mock_backend():
@pytest.fixture
def client(mock_backend):
import aare.daq.server as server
def auth_token_data():
from aare.daq.auth import TokenData
with patch("aare.daq.auth.parse_token") as mock_parse:
from aare.daq.auth import TokenData
return TokenData(
sub="testuser",
staff=True,
pgroups=["p12345"],
session=123,
)
mock_parse.return_value = TokenData(
sub="testuser",
staff=True,
pgroups=["p12345"],
session=123,
@pytest.fixture
def client(server_module, mock_backend, auth_token_data):
with patch("aare.daq.auth.parse_token", return_value=auth_token_data), \
patch("cv2.imencode", return_value=(True, np.array([1, 2, 3], dtype=np.uint8))), \
patch.object(server_module, "mx_beamline", return_value=mock_backend["bl"]), \
patch.object(server_module, "BeamlineConfig", return_value=mock_backend["cfg"]), \
patch.object(server_module, "AareDAQ", return_value=mock_backend["daq"]):
with TestClient(server_module.app) as c:
yield c
@pytest.fixture
def api(client, mock_backend):
return SimpleNamespace(
client=client,
daq=mock_backend["daq"],
bl=mock_backend["bl"],
cfg=mock_backend["cfg"],
)
@pytest.fixture
def daq_status_factory():
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.diffraction_geometry import DiffractionGeometry
from aare.common.models import (
BeamlineStateEnum,
BeamlineStatus,
CrystalSize,
DAQStatusModel,
SampleCameraSettings,
SampleGeometryModel,
SessionStatus,
SessionsStateEnum,
)
def _build(
*,
state=BeamlineStateEnum.Maintenance,
busy=False,
current_pgroup="p12345",
staff=True,
session_state=SessionsStateEnum.Vacant,
sample=None,
open_guis=None,
):
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=500, y=500),
pixel_in_mm=0.001,
aerotech=Coordinate(x=0, y=0, z=0),
aerotech_meas=Coordinate(x=0, y=0, z=0),
smargon=SmargonCoordinate(
sh_mm=Coordinate(x=0, y=0, z=0),
phi_deg=0.0,
chi_deg=0.0,
),
omega_deg=0.0,
beam_size_mm=Coordinate(x=0.01, y=0.01),
)
with patch("cv2.imencode") as mock_imencode:
mock_imencode.return_value = (True, np.array([1, 2, 3], dtype=np.uint8))
diff = DiffractionGeometry(
energy_keV=12.0,
dtz_mm=150.0,
pixel_size_mm=0.075,
beam_center_pxl=(1000.0, 1000.0),
detector_size_pxl=(2000, 2000),
detector_description="Eiger 16M",
detector_serial_number="123",
poni_rot1_rad=0.0,
poni_rot2_rad=0.0,
)
with TestClient(server.app) as c:
yield c
bl_status = BeamlineStatus(
name="X06DA",
ring_current_mA=400.0,
omega=0.0,
front_light=0.0,
back_light=0.0,
cryojet_K=100.0,
shutter_open=False,
exp_shutter_open=False,
flux_ph_s=1e12,
transmission=1.0,
zoom=1.0,
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
commissioning_mode=False,
dtz_min=120.0,
dtz_max=1600.0,
)
session = SessionStatus(
session=session_state,
current_pgroup=current_pgroup,
staff=staff,
)
status = DAQStatusModel(
geom=geom,
diffraction=diff,
bl=bl_status,
state=state,
busy=busy,
session=session,
crystal_size=CrystalSize(x=0, y=0, z=0),
sample=sample,
)
if open_guis is not None:
status.open_guis = open_guis
return status
return _build
tests/unit/common/test_aare_exception.py
+40 -11
View File
@@ -31,7 +31,6 @@ from aare.common.exception_handler import (
TellCommunicationError,
TellConnectionException,
TellCommandWhileBusyException,
TellMountFailedException,
WarningTellException,
CriticalTellException,
SmargonCommunicationError,
@@ -62,14 +61,45 @@ def test_aare_exception_class_critical_default_false():
assert AareException.critical is False
def test_subclasses_inherit_class_critical_default():
# All Phase 1 classes inherit critical=False until Phase 3 sets explicit flags
for cls in (AutomationError, AareUserError, AareAuthError,
TellException, SmargonException, AerotechException,
JFJochException, BECException, BeamlineStateException,
MountingFailed, UnmountingFailed, LoopCenteringFailed,
TellCommunicationError, AareDBCommunicationError):
assert cls.critical is False, f"{cls.__name__} should default to critical=False in Phase 1"
def test_phase3_classes_have_critical_default_true():
for cls in (
TellCommunicationError,
TellConnectionException,
CriticalTellException,
SmargonCommunicationError,
AerotechCommunicationError,
JFJochCommunicationError,
BECCommunicationError,
StateTransitionFailed,
MaintenanceStateException,
BeamlineBusyException,
BeamlineBusyTimeoutException,
MagnetPositionSensorErorr,
SmartMagnetFaultException,
TransformationInvalidException,
):
assert cls.critical is True, f"{cls.__name__} should default to critical=True in Phase 3"
def test_other_core_classes_keep_critical_default_false():
for cls in (
AutomationError,
AareUserError,
AareAuthError,
TellException,
SmargonException,
AerotechException,
JFJochException,
BECException,
BeamlineStateException,
MountingFailed,
UnmountingFailed,
LoopCenteringFailed,
TellCommandWhileBusyException,
WarningTellException,
AareDBCommunicationError,
):
assert cls.critical is False, f"{cls.__name__} should remain critical=False"
# ---------------------------------------------------------------------------
@@ -138,8 +168,7 @@ def test_user_errors_are_aare_user_error():
def test_tell_family_membership():
for cls in (TellCommunicationError, TellConnectionException,
CriticalTellException, TellCommandWhileBusyException,
WarningTellException, MountingFailed, UnmountingFailed,
TellMountFailedException):
WarningTellException, MountingFailed, UnmountingFailed):
exc = cls()
assert isinstance(exc, TellException), f"{cls.__name__} should be in Tell family"
assert isinstance(exc, AutomationError)
tests/unit/common/test_error_codes.py
+2 -2
View File
@@ -40,7 +40,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum():
from aare.common.exception_handler import (
TellCommunicationError, TellConnectionException, CriticalTellException,
WarningTellException, TellCommandWhileBusyException,
TellMountFailedException, MountingFailed, UnmountingFailed,
MountingFailed, UnmountingFailed,
SmargonCommunicationError, AerotechCommunicationError,
JFJochCommunicationError, BECCommunicationError,
AareDBCommunicationError, StateTransitionFailed,
@@ -56,7 +56,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum():
classes = [
TellCommunicationError, TellConnectionException, CriticalTellException,
WarningTellException, TellCommandWhileBusyException,
TellMountFailedException, MountingFailed, UnmountingFailed,
MountingFailed, UnmountingFailed,
SmargonCommunicationError, AerotechCommunicationError,
JFJochCommunicationError, BECCommunicationError,
AareDBCommunicationError, StateTransitionFailed,
tests/unit/common/test_exception_handler.py
-5
View File
@@ -11,7 +11,6 @@ from aare.common.exception_handler import (
MountingFailed,
ManualMountException,
SmartMagnetFaultException,
TellMountFailedException,
TellCommandWhileBusyException,
TellConnectionException,
WarningTellException,
@@ -75,10 +74,6 @@ def test_smart_magnet_fault_exception():
exc = SmartMagnetFaultException()
assert "Smart magnet fault" in str(exc)
def test_tell_mount_failed_exception():
exc = TellMountFailedException()
assert "Tell mount failed" in str(exc)
def test_tell_command_while_busy_exception():
exc = TellCommandWhileBusyException()
assert "Tell is busy" in str(exc)
tests/unit/daq/operations/mounting/test_mounting_service.py
+36
View File
@@ -1,4 +1,5 @@
import types
import sys
from aare.common.coordinate import AerotechCoordinate, Coordinate
from aare.common.exception_handler import CriticalTellException, MountingFailed
@@ -8,6 +9,19 @@ from aare.daq.operations.mounting.models import MountingContext, MountingResult
from aare.daq.operations.mounting.service import MountingService
if "jfjoch_client.models.scan_result" not in sys.modules:
jfjoch_client_mod = types.ModuleType("jfjoch_client")
jfjoch_client_models_mod = types.ModuleType("jfjoch_client.models")
jfjoch_client_scan_result_mod = types.ModuleType("jfjoch_client.models.scan_result")
jfjoch_client_scan_result_mod.ScanResult = dict
jfjoch_client_models_mod.scan_result = jfjoch_client_scan_result_mod
jfjoch_client_mod.models = jfjoch_client_models_mod
sys.modules["jfjoch_client"] = jfjoch_client_mod
sys.modules["jfjoch_client.models"] = jfjoch_client_models_mod
sys.modules["jfjoch_client.models.scan_result"] = jfjoch_client_scan_result_mod
def _make_sample(sample_id: int, name: str) -> SampleShortInfo:
return SampleShortInfo(
db_id=sample_id,
@@ -127,3 +141,25 @@ def test_dry_unmounts_current_sample_before_drying(mock_logger):
assert len(unmount_calls) == 1
assert ctx.cfg.current_sample is None
assert dry_calls == [{"wait_cold": -1, "wait": True}]
def test_execute_mount_escalates_to_critical_after_retry_exhaustion(mock_logger):
target_sample = _make_sample(2, "new")
calls = {"count": 0}
def record_mount_failure():
calls["count"] += 1
return calls["count"]
ctx = _make_context()
ctx.cfg.record_mount_failure = record_mount_failure
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.error.critical is True
assert "Mount failed 5 times in a row" in str(result.error)
tests/unit/daq/test_aare_daq_loop_centering.py
+45 -1
View File
@@ -1,6 +1,7 @@
import types
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from aare.common.automation_models import AutomationProgress, StepState, StepStatus, WorkflowStateKind
from aare.common.models import SampleShortInfo
from aare.common.exception_handler import LoopCenteringFailed
from aare.daq.daq import AareDAQ
@@ -84,3 +85,46 @@ def test_execute_loop_centering_handles_service_success():
assert result is True
daq.save_screenshot_db.assert_called_once()
def test_record_best_effort_step_failure_marks_progress_and_logs_warning(mock_logger):
daq = object.__new__(AareDAQ)
daq._automation_progress_cb = None
daq._set_progress_step = AareDAQ._set_progress_step.__get__(daq, AareDAQ)
progress = AutomationProgress(
current_step="Center",
steps=[StepState(step=WorkflowStateKind.LOOP_CENTRE, status=StepStatus.RUNNING, message="Centering")],
finished=False,
success=None,
)
sample = SampleShortInfo(
db_id=1,
sample_name="S1",
puck_name="P1",
dewar_name="D1",
run_number=1,
user="p1",
pin=1,
location=None,
)
with patch("aare.daq.daq.logger", mock_logger):
daq._record_best_effort_step_failure(
progress=progress,
step=WorkflowStateKind.LOOP_CENTRE,
error=LoopCenteringFailed("Loop centering failed"),
sample=sample,
code="LOOP_CENTERING_FAILED",
)
step = progress.steps[0]
assert step.status == StepStatus.FAILED
assert step.message == "Loop centering failed"
assert len(progress.events) == 1
assert progress.events[0].level == "WARNING"
assert progress.events[0].code == "LOOP_CENTERING_FAILED"
assert progress.events[0].exception_class == "LoopCenteringFailed"
assert progress.events[0].message == "Loop centering failed"
assert progress.events[0].sample_id == 1
mock_logger.warning.assert_called_once()
tests/unit/daq/test_gui_timeout.py
+34 -141
View File
@@ -1,85 +1,30 @@
from types import SimpleNamespace
from unittest.mock import patch
from starlette.testclient import TestClient
from aare.daq.server import app
def test_status_renews_gui_session_with_gui_timeout(client, mock_backend):
def test_status_renews_gui_session_with_gui_timeout(client, mock_backend, daq_status_factory):
mock_cfg = mock_backend["cfg"]
mock_daq = mock_backend["daq"]
from aare.common.models import (
DAQStatusModel,
SessionStatus,
SessionsStateEnum,
BeamlineStateEnum,
SampleGeometryModel,
BeamlineStatus,
CrystalSize,
SampleCameraSettings,
)
from aare.common.diffraction_geometry import DiffractionGeometry
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.models import SessionsStateEnum
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=500, y=500),
pixel_in_mm=0.001,
aerotech=Coordinate(x=0, y=0, z=0),
aerotech_meas=Coordinate(x=0, y=0, z=0),
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=0), phi_deg=0.0, chi_deg=0.0),
omega_deg=0.0,
beam_size_mm=Coordinate(x=0.01, y=0.01),
)
diff = DiffractionGeometry(
energy_keV=12.0,
dtz_mm=150.0,
pixel_size_mm=0.075,
beam_center_pxl=(1000.0, 1000.0),
detector_size_pxl=(2000, 2000),
detector_description="Eiger 16M",
detector_serial_number="123",
poni_rot1_rad=0.0,
poni_rot2_rad=0.0,
)
bl_status = BeamlineStatus(
name="X06DA",
ring_current_mA=400.0,
front_light=0.0,
back_light=0.0,
cryojet_K=100.0,
shutter_open=False,
exp_shutter_open=False,
flux_ph_s=1e12,
transmission=1.0,
zoom=1.0,
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
commissioning_mode=False,
dtz_min=120.0,
dtz_max=1600.0,
)
session_status = SessionStatus(
session=SessionsStateEnum.Vacant,
mock_daq.status = daq_status_factory(
current_pgroup="p12345",
staff=True,
)
mock_daq.status.sample = None
mock_daq.status.box = None
mock_daq.status.last_best_res = None
mock_daq.status.last_best_b_factor = None
mock_daq.status.crystal_size = SimpleNamespace(x=0, y=0, z=0)
mock_daq.status.open_guis = []
mock_daq.status = DAQStatusModel(
geom=geom,
diffraction=diff,
bl=bl_status,
state=BeamlineStateEnum.Maintenance,
busy=False,
session=session_status,
crystal_size=CrystalSize(x=0, y=0, z=0),
)
mock_cfg.pgroup = "p12345"
mock_cfg.session_state.return_value = SessionsStateEnum.OwnedByYou
mock_cfg.get_open_gui_sessions.return_value = []
mock_cfg.GUI_SESSION_EXPIRE_SECONDS = 10
with patch("aare.daq.server.cfg", mock_cfg), patch("aare.daq.server.daq", mock_daq):
response = client.get("/status", headers={"Authorization": "Bearer fake-token"})
response = client.get("/status", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
mock_cfg.touch_gui_session.assert_called_once_with(
@@ -90,22 +35,10 @@ def test_status_renews_gui_session_with_gui_timeout(client, mock_backend):
)
def test_status_hides_open_guis_for_non_staff(mock_backend):
def test_status_hides_open_guis_for_non_staff(client, mock_backend, daq_status_factory):
with patch("aare.daq.auth.parse_token") as mock_parse:
from aare.daq.auth import TokenData
from aare.common.models import (
DAQStatusModel,
SessionStatus,
SessionsStateEnum,
BeamlineStateEnum,
SampleGeometryModel,
BeamlineStatus,
CrystalSize,
SampleCameraSettings,
OpenGuiSessionInfo,
)
from aare.common.diffraction_geometry import DiffractionGeometry
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.models import OpenGuiSessionInfo, SessionsStateEnum
mock_parse.return_value = TokenData(
sub="user1",
@@ -114,52 +47,15 @@ def test_status_hides_open_guis_for_non_staff(mock_backend):
session=123,
)
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=500, y=500),
pixel_in_mm=0.001,
aerotech=Coordinate(x=0, y=0, z=0),
aerotech_meas=Coordinate(x=0, y=0, z=0),
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=0), phi_deg=0.0, chi_deg=0.0),
omega_deg=0.0,
beam_size_mm=Coordinate(x=0.01, y=0.01),
)
diff = DiffractionGeometry(
energy_keV=12.0,
dtz_mm=150.0,
pixel_size_mm=0.075,
beam_center_pxl=(1000.0, 1000.0),
detector_size_pxl=(2000, 2000),
detector_description="Eiger 16M",
detector_serial_number="123",
poni_rot1_rad=0.0,
poni_rot2_rad=0.0,
)
bl_status = BeamlineStatus(
name="X06DA",
ring_current_mA=400.0,
front_light=0.0,
back_light=0.0,
cryojet_K=100.0,
shutter_open=False,
exp_shutter_open=False,
flux_ph_s=1e12,
transmission=1.0,
zoom=1.0,
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
commissioning_mode=False,
dtz_min=120.0,
dtz_max=1600.0,
)
mock_backend["daq"].status = DAQStatusModel(
geom=geom,
diffraction=diff,
bl=bl_status,
state=BeamlineStateEnum.Maintenance,
busy=False,
session=SessionStatus(session=SessionsStateEnum.Vacant, current_pgroup="p12345", staff=False),
crystal_size=CrystalSize(x=0, y=0, z=0),
mock_backend["daq"].status = daq_status_factory(
current_pgroup="p12345",
staff=False,
)
mock_backend["daq"].status.sample = None
mock_backend["daq"].status.box = None
mock_backend["daq"].status.last_best_res = None
mock_backend["daq"].status.last_best_b_factor = None
mock_backend["daq"].status.crystal_size = SimpleNamespace(x=0, y=0, z=0)
mock_backend["cfg"].pgroup = "p12345"
mock_backend["cfg"].session_state.return_value = SessionsStateEnum.OwnedByElse
mock_backend["cfg"].get_open_gui_sessions.return_value = [
@@ -172,32 +68,29 @@ def test_status_hides_open_guis_for_non_staff(mock_backend):
staff=False,
)
with TestClient(app) as local_client:
response = local_client.get("/status", headers={"Authorization": "Bearer fake-token"})
response = client.get("/status", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
payload = response.json()["open_guis"]
assert len(payload) == 1
assert payload[0]["session"] == 123
assert payload[0]["username"] == "user1"
assert response.status_code == 200
payload = response.json()["open_guis"]
assert len(payload) == 1
assert payload[0]["session"] == 123
assert payload[0]["username"] == "user1"
def test_admin_gui_sessions_includes_baton_holder_flag(client, mock_backend):
from types import SimpleNamespace
from aare.common.models import OpenGuiSessionInfo
mock_cfg = mock_backend["cfg"]
mock_cfg.get_open_gui_sessions.return_value = [
OpenGuiSessionInfo(session=11, username="user1", last_seen_ts=1.0, staff=False, holds_baton=False),
OpenGuiSessionInfo(session=22, username="holder", last_seen_ts=2.0, staff=True, holds_baton=True),
OpenGuiSessionInfo(session=11, username="user1", last_seen_ts=1.0, staff=False),
OpenGuiSessionInfo(session=22, username="holder", last_seen_ts=2.0, staff=True),
]
mock_cfg.baton_holder = SimpleNamespace(session=22)
with patch("aare.daq.server.cfg", mock_cfg):
print(mock_backend["cfg"], id(mock_backend["cfg"]))
import aare.daq.server as server
print(server.cfg, id(server.cfg))
response = client.get("/admin/gui_sessions", headers={"Authorization": "Bearer fake-token"})
response = client.get("/admin/gui_sessions", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
payload = response.json()
assert payload[0]["holds_baton"] is False
assert payload[1]["holds_baton"] is True
assert payload[0]["session"] == 11
assert payload[1]["session"] == 22
tests/unit/daq/test_mlbox.py
+25 -16
View File
@@ -5,18 +5,27 @@ from aare.daq.mlbox import MlBox
from aare.common.beamline import MXBeamline
from aare.common.models import MLOutputModel, MLBoxModel, MLBoxType
@pytest.fixture(autouse=True)
def no_mlbox_sleep(monkeypatch):
monkeypatch.setattr(MlBox, "RETRY_SLEEP_S", 0.0)
@pytest.fixture
def mock_wrapper():
return MagicMock()
@pytest.fixture
def mlbox(mock_wrapper):
return MlBox(MXBeamline.X10SA, wrapper=mock_wrapper)
def test_init_x10sa(mock_wrapper):
ml = MlBox(MXBeamline.X10SA, wrapper=mock_wrapper)
assert ml._MlBox__beamline == MXBeamline.X10SA
def test_predict_uses_injected_wrapper():
fake = MagicMock()
fake.get_latest_prediction_bundle.return_value = MagicMock()
@@ -27,61 +36,59 @@ def test_predict_uses_injected_wrapper():
fake.get_latest_prediction_bundle.assert_called_once()
def test_init_unsupported():
with pytest.raises(NotImplementedError):
MlBox(MXBeamline.X06DA)
MlBox(MXBeamline.SIMULATED)
def test_prediction_score():
assert MlBox._prediction_score(None) == (0, 0.0)
out = MLOutputModel()
assert MlBox._prediction_score(out) == (0, 0.0)
out.add_box(MLBoxType.CRYSTAL, (0, 0, 10, 10), conf=0.9)
assert MlBox._prediction_score(out) == (1, 0.9)
def test_check_overlap():
box_a = MLBoxModel.from_tuple(MLBoxType.CRYSTAL, (0, 0, 10, 10), conf=1.0)
box_b = MLBoxModel.from_tuple(MLBoxType.CRYSTAL, (5, 5, 15, 15), conf=1.0)
assert MlBox._check_overlap(box_a, box_b) > 0
box_c = MLBoxModel.from_tuple(MLBoxType.CRYSTAL, (20, 20, 30, 30), conf=1.0)
assert MlBox._check_overlap(box_a, box_c) == 0
def test_filter_predictions(mlbox):
out = MLOutputModel()
out.add_box(MLBoxType.CRYSTAL, (0, 0, 10, 10), conf=0.4)
out.add_box(MLBoxType.CRYSTAL, (20, 20, 30, 30), conf=0.8)
mlbox._filter_predictions(out, confidence_min=0.5)
assert len(out.boxes) == 1
assert list(out.boxes.values())[0].conf == 0.8
@patch('cv2.imdecode')
@patch("cv2.imdecode")
def test_decode_bundle_image(mock_imdecode):
mock_imdecode.return_value = np.zeros((100, 100, 3))
img = MlBox._decode_bundle_image(b"fake_jpeg")
assert img is not None
assert img.shape == (100, 100, 3)
def test_extract_bundle_meta():
pred = MagicMock()
# Mocking what _extract_target_point expects:
# It tries to find 'target_point' in metadata or prediction itself.
# In src/aare/daq/mlbox.py:
# def _extract_target_point(prediction: LatestPredictionModel | None) -> tuple[float, float] | None:
# raw = getattr(prediction, "target_point", None)
# if raw is None:
# meta = getattr(prediction, "metadata", {})
# raw = meta.get("target_point")
pred.target_point = [100.0, 200.0]
pred.focus_score = 0.5
meta = MlBox._extract_bundle_meta(pred)
assert meta.target_point == (100.0, 200.0)
assert meta.focus == 0.5
def test_preferred_class_prefers_loop_over_pin_when_margin_not_exceeded():
boxes = MLOutputModel()
boxes.add_box(MLBoxType.PIN, (0, 0, 10, 10), 0.8)
@@ -95,6 +102,7 @@ def test_preferred_class_prefers_loop_over_pin_when_margin_not_exceeded():
assert best is not None
assert best.cls == MLBoxType.LOOP_FACE
def test_prediction_score_empty():
assert MlBox._prediction_score(None) == (0, 0.0)
@@ -106,6 +114,7 @@ def test_prediction_score_uses_count_and_max_confidence():
assert MlBox._prediction_score(model) == (2, 0.8)
def test_best_by_class_keeps_highest_confidence_per_class():
results = [
{"class": 1, "confidence": 0.4, "box": {"x1": 0, "y1": 0, "x2": 10, "y2": 10}},
tests/unit/daq/test_mount.py
+22 -40
View File
@@ -2,6 +2,7 @@ from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from aareDB import SampleEventType
from aare.common.models import DewarAddress, SampleShortInfo, DAQOperation
@@ -29,7 +30,8 @@ def _make_sample(sample_id: int, name: str) -> SampleShortInfo:
def _make_daq(previous_sample: SampleShortInfo | None) -> AareDAQ:
daq = object.__new__(AareDAQ)
daq._AareDAQ__cfg = SimpleNamespace(current_sample=previous_sample)
daq._AareDAQ__aare = SimpleNamespace(send_sample_event=MagicMock())
daq._AareDAQ__aare = MagicMock()
daq._AareDAQ__devs = MagicMock()
daq._AareDAQ__set_state = MagicMock()
daq._handle_operation_error = MagicMock()
daq.save_screenshot_db = MagicMock()
@@ -38,16 +40,20 @@ def _make_daq(previous_sample: SampleShortInfo | None) -> AareDAQ:
return daq
def test_create_mounting_service_builds_expected_context(mock_logger):
@pytest.fixture
def bare_daq():
daq = object.__new__(AareDAQ)
daq._AareDAQ__cfg = SimpleNamespace()
daq._AareDAQ__devs = SimpleNamespace()
return daq
service = daq._create_mounting_service()
def test_create_mounting_service_builds_expected_context(mock_logger, bare_daq):
service = bare_daq._create_mounting_service()
assert isinstance(service, MountingService)
assert service.ctx.cfg is daq._AareDAQ__cfg
assert service.ctx.devs is daq._AareDAQ__devs
assert service.ctx.cfg is bare_daq._AareDAQ__cfg
assert service.ctx.devs is bare_daq._AareDAQ__devs
assert service.ctx.mount_position == ABR_POS_MOUNT
@@ -71,30 +77,6 @@ def test_was_previous_sample_unmounted_since_prefers_tell_phase_confirmation():
assert daq._was_previous_sample_unmounted_since(started_at) is True
def test_was_previous_sample_unmounted_since_falls_back_to_redis_event_history():
previous_sample = _make_sample(1, "old")
daq = _make_daq(previous_sample)
started_at = datetime.now(timezone.utc) - timedelta(seconds=5)
daq._safe_tell_state = MagicMock(return_value=None)
daq._get_tell_events_from_redis = MagicMock(
return_value=[
{
"timestamp": (started_at - timedelta(seconds=2)).isoformat(),
"class": "Motion Sync",
"event": "Sample put on Puck",
},
{
"timestamp": (started_at + timedelta(seconds=1)).isoformat(),
"class": "Motion Sync",
"event": "Sample put on Puck",
},
]
)
assert daq._was_previous_sample_unmounted_since(started_at) is True
def test_execute_mount_and_prepare_success_uses_mounting_result_fields():
previous_sample = _make_sample(1, "old")
target_sample = _make_sample(2, "new")
@@ -114,13 +96,13 @@ def test_execute_mount_and_prepare_success_uses_mounting_result_fields():
assert result is True
send_calls = daq._AareDAQ__aare.send_sample_event.call_args_list
assert send_calls[0].args[0] == previous_sample
assert send_calls[0].args[0] == previous_sample.db_id
assert send_calls[0].args[1] == SampleEventType.UNMOUNTING
assert send_calls[1].args[0] == target_sample
assert send_calls[1].args[0] == target_sample.db_id
assert send_calls[1].args[1] == SampleEventType.MOUNTING
assert send_calls[2].args[0] == previous_sample
assert send_calls[2].args[0] == previous_sample.db_id
assert send_calls[2].args[1] == SampleEventType.UNMOUNTED
assert send_calls[3].args[0] == target_sample
assert send_calls[3].args[0] == target_sample.db_id
assert send_calls[3].args[1] == SampleEventType.MOUNTED
daq.save_screenshot_db.assert_called_once_with(target_sample.db_id, f"{target_sample.db_id}_mounted")
@@ -149,13 +131,13 @@ def test_execute_mount_and_prepare_marks_previous_sample_unmounted_when_mount_fa
assert daq._AareDAQ__cfg.current_sample is None
send_calls = daq._AareDAQ__aare.send_sample_event.call_args_list
assert send_calls[0].args[0] == previous_sample
assert send_calls[0].args[0] == previous_sample.db_id
assert send_calls[0].args[1] == SampleEventType.UNMOUNTING
assert send_calls[1].args[0] == target_sample
assert send_calls[1].args[0] == target_sample.db_id
assert send_calls[1].args[1] == SampleEventType.MOUNTING
assert send_calls[2].args[0] == previous_sample
assert send_calls[2].args[0] == previous_sample.db_id
assert send_calls[2].args[1] == SampleEventType.UNMOUNTED
assert send_calls[2].kwargs["comment"] == "Auto-unmount succeeded before mount failed"
@@ -189,9 +171,9 @@ def test_execute_mount_and_prepare_does_not_mark_previous_sample_unmounted_when_
send_calls = daq._AareDAQ__aare.send_sample_event.call_args_list
assert len(send_calls) == 2
assert send_calls[0].args[0] == previous_sample
assert send_calls[0].args[0] == previous_sample.db_id
assert send_calls[0].args[1] == SampleEventType.UNMOUNTING
assert send_calls[1].args[0] == target_sample
assert send_calls[1].args[0] == target_sample.db_id
assert send_calls[1].args[1] == SampleEventType.MOUNTING
daq._handle_operation_error.assert_called_once()
@@ -217,9 +199,9 @@ def test_execute_mount_and_prepare_unmount_success_uses_unmount_operation():
send_calls = daq._AareDAQ__aare.send_sample_event.call_args_list
assert len(send_calls) == 2
assert send_calls[0].args[0] == previous_sample
assert send_calls[0].args[0] == previous_sample.db_id
assert send_calls[0].args[1] == SampleEventType.UNMOUNTING
assert send_calls[1].args[0] == previous_sample
assert send_calls[1].args[0] == previous_sample.db_id
assert send_calls[1].args[1] == SampleEventType.UNMOUNTED
daq.save_screenshot_db.assert_not_called()
tests/unit/daq/test_raster_logic.py
+48 -52
View File
@@ -1,10 +1,13 @@
import pytest
from types import SimpleNamespace
from unittest.mock import MagicMock
import aare.daq.daq as daq_module
from jfjoch_client.exceptions import NotFoundException
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.raster_grid import RasterGridRequest
from aare.daq.daq import AareDAQ
from aare.daq.operations.raster.models import RasterContext
from aare.daq.operations.raster.service import RasterService
def make_request(n_x: int, n_y: int, cell_x: float = 0.01, cell_y: float = 0.02) -> RasterGridRequest:
@@ -20,9 +23,31 @@ def make_request(n_x: int, n_y: int, cell_x: float = 0.01, cell_y: float = 0.02)
)
def make_service() -> RasterService:
runtime = SimpleNamespace(
sample=None,
sample_geometry=SimpleNamespace(),
status=SimpleNamespace(),
)
context = RasterContext(
cfg=MagicMock(),
devs=MagicMock(),
mlbox=MagicMock(),
jfjoch=MagicMock(),
aare=MagicMock(),
runtime=runtime,
state_controller=MagicMock(),
services=MagicMock(),
auto_raster_max_images=4500,
auto_raster_min_cell_size_mm=0.005,
auto_raster_skip_if_exceed_max_image_threshold=True,
)
return RasterService(context=context, logger=MagicMock())
def test_grid_image_id_from_centre_offset_single_cell():
request = make_request(1, 1)
image_id = AareDAQ._grid_image_id_from_centre_offset(
image_id = RasterService._grid_image_id_from_centre_offset(
x_mm=request.grid_size_mm.x / 2.0,
y_mm=request.grid_size_mm.y / 2.0,
request=request,
@@ -36,7 +61,7 @@ def test_grid_image_id_from_centre_offset_odd_grid():
x_mm = ((request.n_x - 1) * request.grid_size_mm.x) / 2.0
y_mm = ((request.n_y - 1) * request.grid_size_mm.y) / 2.0
image_id = AareDAQ._grid_image_id_from_centre_offset(
image_id = RasterService._grid_image_id_from_centre_offset(
x_mm=x_mm,
y_mm=y_mm,
request=request,
@@ -51,7 +76,7 @@ def test_grid_image_id_from_centre_offset_even_grid():
x_mm = ((request.n_x - 1) * request.grid_size_mm.x) / 2.0
y_mm = ((request.n_y - 1) * request.grid_size_mm.y) / 2.0
image_id = AareDAQ._grid_image_id_from_centre_offset(
image_id = RasterService._grid_image_id_from_centre_offset(
x_mm=x_mm,
y_mm=y_mm,
request=request,
@@ -66,7 +91,7 @@ def test_grid_image_id_from_centre_offset_vertical_scan():
x_mm = request.grid_size_mm.x / 2.0
y_mm = ((request.n_y - 1) * request.grid_size_mm.y) / 2.0
image_id = AareDAQ._grid_image_id_from_centre_offset(
image_id = RasterService._grid_image_id_from_centre_offset(
x_mm=x_mm,
y_mm=y_mm,
request=request,
@@ -80,7 +105,7 @@ def test_grid_image_id_from_centre_offset_rejects_non_positive_cell_size():
request.grid_size_mm = Coordinate(x=0.0, y=0.01)
with pytest.raises(ValueError, match="grid_size_mm must be positive"):
AareDAQ._grid_image_id_from_centre_offset(
RasterService._grid_image_id_from_centre_offset(
x_mm=0.0,
y_mm=0.0,
request=request,
@@ -92,7 +117,7 @@ def test_grid_image_id_from_centre_offset_rejects_zero_dimensions():
request.n_x = 0
with pytest.raises(ValueError, match="Raster grid dimensions must be >= 1"):
AareDAQ._grid_image_id_from_centre_offset(
RasterService._grid_image_id_from_centre_offset(
x_mm=0.0,
y_mm=0.0,
request=request,
@@ -100,21 +125,13 @@ def test_grid_image_id_from_centre_offset_rejects_zero_dimensions():
def test_upload_raster_diffraction_preview_skips_out_of_range_image_id():
daq = object.__new__(AareDAQ)
daq._AareDAQ__jfjoch = MagicMock()
daq._AareDAQ__aare = MagicMock()
daq._AareDAQ__cfg = MagicMock()
daq._AareDAQ__cfg.get_beam_mark.return_value = (0.0, 0.0)
daq._AareDAQ__cfg.beam_size_mm = Coordinate(x=0.01, y=0.01)
daq._AareDAQ__cfg.beam_mark_coeff = MagicMock()
daq._AareDAQ__cfg.pixel_to_mm = MagicMock(return_value=0.001)
daq._AareDAQ__cfg.current_sample = None
service = make_service()
request = make_request(2, 2)
scan_result = MagicMock()
scan_result.images = [MagicMock(), MagicMock(), MagicMock(), MagicMock()]
daq._upload_raster_diffraction_preview(
service._upload_raster_diffraction_preview(
sample_id=123,
filename="preview",
image_id=10,
@@ -122,32 +139,19 @@ def test_upload_raster_diffraction_preview_skips_out_of_range_image_id():
request=request,
)
daq._AareDAQ__jfjoch.get_diffraction_image.assert_not_called()
daq._AareDAQ__aare.upload_jpg.assert_not_called()
service.ctx.jfjoch.get_diffraction_image.assert_not_called()
service.ctx.aare.upload_jpg.assert_not_called()
def test_upload_raster_diffraction_preview_ignores_not_found(monkeypatch):
class FakeNotFoundException(Exception):
pass
monkeypatch.setattr(daq_module, "NotFoundException", FakeNotFoundException)
daq = object.__new__(AareDAQ)
daq._AareDAQ__jfjoch = MagicMock()
daq._AareDAQ__aare = MagicMock()
daq._AareDAQ__cfg = MagicMock()
daq._AareDAQ__cfg.get_beam_mark.return_value = (0.0, 0.0)
daq._AareDAQ__cfg.beam_size_mm = Coordinate(x=0.01, y=0.01)
daq._AareDAQ__cfg.beam_mark_coeff = MagicMock()
daq._AareDAQ__cfg.pixel_to_mm = MagicMock(return_value=0.001)
daq._AareDAQ__cfg.current_sample = None
def test_upload_raster_diffraction_preview_ignores_not_found():
service = make_service()
request = make_request(2, 2)
scan_result = MagicMock()
scan_result.images = [MagicMock(), MagicMock(), MagicMock(), MagicMock()]
daq._AareDAQ__jfjoch.get_diffraction_image.side_effect = FakeNotFoundException()
service.ctx.jfjoch.get_diffraction_image.side_effect = NotFoundException()
daq._upload_raster_diffraction_preview(
service._upload_raster_diffraction_preview(
sample_id=123,
filename="preview",
image_id=2,
@@ -155,27 +159,19 @@ def test_upload_raster_diffraction_preview_ignores_not_found(monkeypatch):
request=request,
)
daq._AareDAQ__jfjoch.get_diffraction_image.assert_called_once_with(2)
daq._AareDAQ__aare.upload_jpg.assert_not_called()
service.ctx.jfjoch.get_diffraction_image.assert_called_once_with(2)
service.ctx.aare.upload_jpg.assert_not_called()
def test_upload_raster_diffraction_preview_uploads_when_present():
daq = object.__new__(AareDAQ)
daq._AareDAQ__jfjoch = MagicMock()
daq._AareDAQ__aare = MagicMock()
daq._AareDAQ__cfg = MagicMock()
daq._AareDAQ__cfg.get_beam_mark.return_value = (0.0, 0.0)
daq._AareDAQ__cfg.beam_size_mm = Coordinate(x=0.01, y=0.01)
daq._AareDAQ__cfg.beam_mark_coeff = MagicMock()
daq._AareDAQ__cfg.pixel_to_mm = MagicMock(return_value=0.001)
daq._AareDAQ__cfg.current_sample = None
service = make_service()
request = make_request(2, 2)
scan_result = MagicMock()
scan_result.images = [MagicMock(), MagicMock(), MagicMock(), MagicMock()]
daq._AareDAQ__jfjoch.get_diffraction_image.return_value = b"jpeg-bytes"
service.ctx.jfjoch.get_diffraction_image.return_value = b"jpeg-bytes"
daq._upload_raster_diffraction_preview(
service._upload_raster_diffraction_preview(
sample_id=123,
filename="preview",
image_id=2,
@@ -183,5 +179,5 @@ def test_upload_raster_diffraction_preview_uploads_when_present():
request=request,
)
daq._AareDAQ__jfjoch.get_diffraction_image.assert_called_once_with(2)
daq._AareDAQ__aare.upload_jpg.assert_called_once_with(123, "preview", b"jpeg-bytes")
service.ctx.jfjoch.get_diffraction_image.assert_called_once_with(2)
service.ctx.aare.upload_jpg.assert_called_once_with(123, "preview", b"jpeg-bytes")
tests/unit/daq/test_server.py
+36 -95
View File
@@ -1,23 +1,11 @@
import pytest
from unittest.mock import patch, PropertyMock
import os
import io
import cv2
from types import SimpleNamespace
from unittest.mock import patch
import numpy as np
# Set environment variable before importing app
os.environ["JWT_AAREDAQ_KEY"] = "test_key_for_unit_testing"
# Mock the entire backend before importing the app to prevent any real initialisation
import aare.daq.daq
with patch("aare.daq.daq.AareDAQ"), \
patch("aare.daq.config.MXBeamline"), \
patch("aare.daq.config.BeamlineConfig"), \
patch("aare.daq.aaredb.AareWrapper"), \
patch("aare.daq.server.lifespan") as mock_ls:
mock_ls.return_value.__aenter__.return_value = None
from aare.daq.server import app
import aare.daq.server as server
def test_meta_error_codes(client):
response = client.get("/meta/error-codes")
@@ -25,101 +13,54 @@ def test_meta_error_codes(client):
data = response.json()
assert "AuthErrorCode" in data
def test_status(client, mock_backend):
mock_daq = mock_backend["daq"]
mock_cfg = mock_backend["cfg"]
from aare.common.models import DAQStatusModel, SessionStatus, SessionsStateEnum, BeamlineStateEnum, SampleGeometryModel, BeamlineStatus, CrystalSize, SampleCameraSettings
from aare.common.diffraction_geometry import DiffractionGeometry
def test_status(api, daq_status_factory):
from aare.common.models import BeamlineStateEnum, SessionsStateEnum
# Create a valid status object
from aare.common.coordinate import Coordinate, SmargonCoordinate
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=500, y=500),
pixel_in_mm=0.001,
aerotech=Coordinate(x=0, y=0, z=0),
aerotech_meas=Coordinate(x=0, y=0, z=0),
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0, y=0, z=0), phi_deg=0.0, chi_deg=0.0),
omega_deg=0.0,
beam_size_mm=Coordinate(x=0.01, y=0.01)
)
diff = DiffractionGeometry(
energy_keV=12.0,
dtz_mm=150.0,
pixel_size_mm=0.075,
beam_center_pxl=(1000.0, 1000.0),
detector_size_pxl=(2000, 2000),
detector_description="Eiger 16M",
detector_serial_number="123",
poni_rot1_rad=0.0,
poni_rot2_rad=0.0
)
bl_status = BeamlineStatus(
name="X06DA",
ring_current_mA=400.0,
omega=0.0, front_light=0.0, back_light=0.0,
cryojet_K=100.0, shutter_open=False, exp_shutter_open=False,
flux_ph_s=1e12, transmission=1.0, zoom=1.0,
sample_camera=SampleCameraSettings(exposure=0.1, gain=1.0),
commissioning_mode=False, dtz_min=120.0, dtz_max=1600.0
)
session_status = SessionStatus(
session=SessionsStateEnum.Vacant,
current_pgroup="p12345",
staff=True
)
status_obj = DAQStatusModel(
geom=geom,
diffraction=diff,
bl=bl_status,
api.daq.status = daq_status_factory(
state=BeamlineStateEnum.Maintenance,
busy=False,
session=session_status,
crystal_size=CrystalSize(x=0, y=0, z=0)
current_pgroup="p12345",
staff=True,
)
mock_daq.status = status_obj
mock_cfg.pgroup = "p12345"
mock_cfg.session_state.return_value = SessionsStateEnum.OwnedByYou
# Also patch server.cfg and server.daq to be absolutely sure
with patch("aare.daq.server.daq", mock_daq), \
patch("aare.daq.server.cfg", mock_cfg):
response = client.get("/status", headers={"Authorization": "Bearer fake-token"})
api.daq.status.sample = None
api.daq.status.box = None
api.daq.status.last_best_res = None
api.daq.status.last_best_b_factor = None
api.daq.status.crystal_size = SimpleNamespace(x=0, y=0, z=0)
api.daq.status.open_guis = []
api.cfg.pgroup = "p12345"
api.cfg.session_state.return_value = SessionsStateEnum.OwnedByYou
api.cfg.get_open_gui_sessions.return_value = []
response = api.client.get("/status", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
data = response.json()
assert data["state"] == 1 # Maintenance is 1
assert data["state"] == BeamlineStateEnum.Maintenance.value
assert data["session"]["current_pgroup"] == "p12345"
assert data["session"]["session"] == SessionsStateEnum.OwnedByYou.value
def test_omega_put(client, mock_backend):
# The problem is that server.py uses 'from aare.daq.server import daq' in some places
# and the global 'daq' in others.
# When we do 'daq.omega = val', it's the global 'daq'.
with patch("aare.daq.auth.check_jwt_rw"):
# Instead of checking mock_daq.omega, let's patch the 'daq' object in server.py AGAIN
with patch("aare.daq.server.daq") as m_daq:
server.daq = m_daq # Force it
response = client.put("/beamline/omega?val=10.5", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
assert response.json() == "OK"
# Now m_daq should have received the assignment
# In Python, m_daq.omega = 10.5 will set the 'omega' attribute on the MagicMock.
assert m_daq.omega == 10.5
def test_omega_put(client):
with patch("aare.daq.auth.check_jwt_rw"), patch("aare.daq.server.daq") as mock_daq:
response = client.put("/beamline/omega?val=10.5", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
assert response.json() == "OK"
assert mock_daq.omega == 10.5
def test_login_success(client, mock_backend):
def test_login_success(client):
with patch("aare.daq.auth.authenticate_user") as mock_auth:
mock_auth.return_value = "fake-access-token"
response = client.post("/token", data={"username": "user", "password": "pwd"})
assert response.status_code == 200
assert response.json() == {"access_token": "fake-access-token", "token_type": "bearer"}
def test_get_image(client, mock_backend):
mock_daq = mock_backend["daq"]
mock_daq.camera_image = np.zeros((100, 100, 3), dtype=np.uint8)
with patch("aare.daq.auth.parse_token"):
response = client.get("/beamline/image", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
assert response.headers["content-type"] == "image/jpeg"
assert len(response.content) > 0
response = client.get("/beamline/image", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
assert response.headers["content-type"] == "image/jpeg"
assert len(response.content) > 0
tests/unit/daq/test_workflows.py
+83 -23
View File
@@ -6,15 +6,19 @@ from aare.common.models import StagePositionEnum
from aare.devices.area_detector import AutoEnum
from aare.devices.bec_worker import BeamlineState
@pytest.fixture
def mock_devs():
devs = MagicMock()
devs.bsz.position = 0.0
devs.beamstop_stage_up = False
devs.reflector_up = False
devs.bec_worker = None
devs.bec_worker = MagicMock()
devs.bec_worker.planner = MagicMock()
devs.bec_worker.move_to = MagicMock()
return devs
@pytest.fixture
def mock_cfg():
cfg = MagicMock()
@@ -22,105 +26,157 @@ def mock_cfg():
cfg.dtz = None
return cfg
def _assert_bec_moved(devs, state):
planner_calls = devs.bec_worker.planner.move_to.call_args_list if devs.bec_worker is not None else []
direct_calls = devs.bec_worker.move_to.call_args_list if devs.bec_worker is not None else []
assert (
any(call.args == (state,) for call in planner_calls)
or any(call.args == (state,) for call in direct_calls)
), f"BEC was not asked to move to {state}"
def test_move_bsz_no_move(mock_devs):
mock_devs.bsz.position = 1.0
move_bsz(mock_devs, 1.05)
# No changes should happen as diff < 0.1
assert mock_devs.beamstop_z.call_count == 0
def test_move_bsz_with_move(mock_devs):
mock_devs.bsz.position = 0.0
mock_devs.beamstop_stage_up = False
mock_devs.reflector_up = True
move_bsz(mock_devs, 1.0)
# Check if reflector was toggled
assert mock_devs.reflector_up == True
# Check if beamstop stage was toggled
assert mock_devs.beamstop_stage_up == False
# Check if beamstop_z was set
assert mock_devs.reflector_up is True
assert mock_devs.beamstop_stage_up is False
assert mock_devs.beamstop_z == 1.0
def test_common_2rse_no_bec(mock_devs, mock_cfg):
mock_devs.bec_worker = None
common_2rse(mock_devs, mock_cfg)
assert mock_devs.collimator == 20.0
assert mock_devs.scintillator == 20.0
mock_devs.smargon_move_home.assert_called_once()
assert mock_devs.beamstop_stage_up == StagePositionEnum.PARK
def test_common_2rse_with_bec(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
common_2rse(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.ROBOT_SAMPLE_EXCHANGE)
_assert_bec_moved(mock_devs, BeamlineState.ROBOT_SAMPLE_EXCHANGE)
mock_devs.smargon_move_home.assert_called_once()
assert mock_devs.aerotech_pos == ABR_POS_MOUNT
def test_sa2se(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
sa2se(mock_devs, mock_cfg)
mock_devs.samcam_auto.assert_any_call(AutoEnum.AUTO)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.MANUAL_SAMPLE_EXCHANGE)
_assert_bec_moved(mock_devs, BeamlineState.MANUAL_SAMPLE_EXCHANGE)
assert mock_devs.aerotech_pos == mock_cfg.abr_meas_pos
mock_devs.smargon_move_home.assert_called_once()
mock_devs.samcam_auto.assert_any_call(AutoEnum.ONCE)
def test_sa2rse(mock_devs, mock_cfg):
sa2rse(mock_devs, mock_cfg)
mock_devs.smargon_move_home.assert_called_once()
def test_sa2xtal_snapshot(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
sa2xtal_snapshot(mock_devs, mock_cfg)
assert mock_devs.samcam_settings.exposure == 0.0005
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.XTAL_SNAPSHOT)
_assert_bec_moved(mock_devs, BeamlineState.XTAL_SNAPSHOT)
def test_dc2xtal_snapshot(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
dc2xtal_snapshot(mock_devs, mock_cfg)
assert mock_devs.samcam_settings.exposure == 0.0005
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.XTAL_SNAPSHOT)
_assert_bec_moved(mock_devs, BeamlineState.XTAL_SNAPSHOT)
def test_xtal_snapshot2dc(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
xtal_snapshot2dc(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.DATA_COLLECTION)
_assert_bec_moved(mock_devs, BeamlineState.DATA_COLLECTION)
def test_xtal_snapshot2sa(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
xtal_snapshot2sa(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.SAMPLE_ALIGNMENT)
_assert_bec_moved(mock_devs, BeamlineState.SAMPLE_ALIGNMENT)
def test_dc2rse(mock_devs, mock_cfg):
dc2rse(mock_devs, mock_cfg)
mock_devs.smargon_move_home.assert_called_once()
def test_se2sa(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
se2sa(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.SAMPLE_ALIGNMENT)
_assert_bec_moved(mock_devs, BeamlineState.SAMPLE_ALIGNMENT)
assert mock_devs.aerotech_pos == mock_cfg.abr_meas_pos
def test_sa2dc(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
sa2dc(mock_devs, mock_cfg)
mock_devs.bec_worker.move_to.assert_called_with(BeamlineState.DATA_COLLECTION)
_assert_bec_moved(mock_devs, BeamlineState.DATA_COLLECTION)
def test_dc2sa(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
mock_cfg.dtz = None
dc2sa(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.SAMPLE_ALIGNMENT)
_assert_bec_moved(mock_devs, BeamlineState.SAMPLE_ALIGNMENT)
assert mock_devs.aerotech_pos == mock_cfg.abr_meas_pos
def test_sa2xrf(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
sa2xrf(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.FLUX_MEASUREMENT)
_assert_bec_moved(mock_devs, BeamlineState.FLUX_MEASUREMENT)
def test_sa2dh(mock_devs, mock_cfg):
mock_devs.tell.get_mounted_sample.return_value = "Sample"
@@ -128,7 +184,11 @@ def test_sa2dh(mock_devs, mock_cfg):
mock_devs.tell.unmount.assert_called_with(wait=True)
mock_devs.tell.dry.assert_called_with(wait_cold=-1, wait=False)
def test_dh2sa(mock_devs, mock_cfg):
mock_devs.bec_worker = MagicMock()
mock_devs.bec_worker.planner = MagicMock()
mock_devs.bec_worker.move_to = MagicMock()
dh2sa(mock_devs, mock_cfg)
mock_devs.bec_worker.planner.move_to.assert_called_with(BeamlineState.SAMPLE_ALIGNMENT)
_assert_bec_moved(mock_devs, BeamlineState.SAMPLE_ALIGNMENT)
tests/unit/devices/test_jfjoch.py
+28 -7
View File
@@ -10,6 +10,7 @@ from aare.common.sample_geometry import SampleGeometryModel
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.exception_handler import JFJochCommunicationError
@pytest.fixture
def mock_jfjoch_client():
with patch('jfjoch_client.ApiClient'), \
@@ -18,10 +19,12 @@ def mock_jfjoch_client():
mock_api = mock_api_class.return_value
yield mock_api
@pytest.fixture
def jfjoch_wrapper(mock_jfjoch_client):
return JFJochWrapper(MXBeamline.X10SA)
def create_mock_daq_status():
bl_status = BeamlineStatus(
name="test_beamline",
@@ -51,7 +54,7 @@ def create_mock_daq_status():
poni_rot2_rad=0.0
)
session = SessionStatus(session=SessionsStateEnum.Vacant, current_pgroup="p12345")
geom = SampleGeometryModel(
beam_location_pxl=Coordinate(x=1000, y=1000),
pixel_in_mm=0.001,
@@ -71,37 +74,47 @@ def create_mock_daq_status():
session=session
)
def test_init_x10sa(mock_jfjoch_client):
wrapper = JFJochWrapper(MXBeamline.X10SA)
assert wrapper._JFJochWrapper__url == "http://sls-gpu-002:8080"
def test_init_x06da(mock_jfjoch_client):
wrapper = JFJochWrapper(MXBeamline.X06DA)
assert wrapper._JFJochWrapper__url == "http://sls-gpu-001:8080"
def test_init_simulated():
def test_init_simulated(mock_jfjoch_client):
wrapper = JFJochWrapper(MXBeamline.SIMULATED)
assert wrapper._JFJochWrapper__simulated is True
def test_init_unknown():
with pytest.raises(Exception, match="unknown beamline"):
JFJochWrapper("UNKNOWN")
def test_initialize(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.initialize_post.reset_mock()
jfjoch_wrapper.initialize()
mock_jfjoch_client.initialize_post.assert_called_once()
def test_cancel(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.cancel_post.reset_mock()
jfjoch_wrapper.cancel()
mock_jfjoch_client.cancel_post.assert_called_once()
def test_is_idle(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.status_get.return_value.state = 'Idle'
assert jfjoch_wrapper.is_idle() is True
mock_jfjoch_client.status_get.return_value.state = 'Busy'
assert jfjoch_wrapper.is_idle() is False
def test_measure_rotation(jfjoch_wrapper, mock_jfjoch_client):
r = RotationScanRequest(
start_omega_deg=0.0,
@@ -112,17 +125,21 @@ def test_measure_rotation(jfjoch_wrapper, mock_jfjoch_client):
)
s = create_mock_daq_status()
s.sample = SampleShortInfo(sample_name="S1", user="p12345", db_id=1, dewar_name="D1", puck_name="P1", run_number=1, pin=1)
jfjoch_wrapper.measure_rotation(r, s)
mock_jfjoch_client.start_post.assert_called_once()
def test_measure_rotation_error(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.start_post.side_effect = Exception("API Error")
r = RotationScanRequest(exp_time_s=0.01, incr_omega_deg=0.1, steps=10)
s = create_mock_daq_status()
s.sample = SampleShortInfo(sample_name="S1", user="p12345", db_id=1, dewar_name="D1", puck_name="P1", run_number=1, pin=1)
with pytest.raises(JFJochCommunicationError):
jfjoch_wrapper.measure_rotation(r, s)
def test_measure_raster(jfjoch_wrapper, mock_jfjoch_client):
r = RasterGridRequest(
n_x=10,
@@ -134,31 +151,35 @@ def test_measure_raster(jfjoch_wrapper, mock_jfjoch_client):
)
s = create_mock_daq_status()
s.sample = SampleShortInfo(sample_name="S1", user="p12345", db_id=1, dewar_name="D1", puck_name="P1", run_number=1, pin=1)
jfjoch_wrapper.measure_raster(r, s)
mock_jfjoch_client.start_post.assert_called_once()
def test_wait_till_done(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.result_scan_get.return_value = MagicMock()
result = jfjoch_wrapper.wait_till_done(timeout=10)
mock_jfjoch_client.wait_till_done_post_with_http_info.assert_called_once_with(timeout=10)
assert result is not None
def test_detector(jfjoch_wrapper, mock_jfjoch_client):
mock_det = MagicMock()
mock_jfjoch_client.config_select_detector_get.return_value.detectors = [mock_det]
mock_jfjoch_client.config_select_detector_get.return_value.current_id = 0
det = jfjoch_wrapper.detector()
assert det == mock_det
def test_get_diffraction_image(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.image_buffer_image_jpeg_get.return_value = b"image_data"
img = jfjoch_wrapper.get_diffraction_image(image_id=1)
assert img == b"image_data"
def test_get_diffraction_image_retry(jfjoch_wrapper, mock_jfjoch_client):
mock_jfjoch_client.image_buffer_image_jpeg_get.side_effect = [Exception("Error"), b"image_data"]
img = jfjoch_wrapper.get_diffraction_image(image_id=1, wait_between_retries_s=0.001)
assert img == b"image_data"
assert mock_jfjoch_client.image_buffer_image_jpeg_get.call_count == 2
assert mock_jfjoch_client.image_buffer_image_jpeg_get.call_count == 2
tests/unit/devices/test_mx_lib.py
+20 -12
View File
@@ -3,36 +3,39 @@ from unittest.mock import MagicMock, patch
from aare.devices.mx_lib import wait_for_movement_to_finish, pv_wait, is_epics_type, clean_filename
from epics import Motor, PV
def test_clean_filename():
assert clean_filename(None, "my file!.txt") == "my_file_.txt"
assert clean_filename(None, " /path/to/somewhere/file.txt ") == "path_to_somewhere_file.txt"
assert clean_filename(None, "abcABC123") == "abcABC123"
assert clean_filename(None, "file._-") == "file"
assert clean_filename("my file!.txt") == "my_file_.txt"
assert clean_filename(" /path/to/somewhere/file.txt ") == "path_to_somewhere_file.txt"
assert clean_filename("abcABC123") == "abcABC123"
assert clean_filename("file._-") == "file"
with pytest.raises(ValueError):
clean_filename(None, "!!!")
clean_filename("!!!")
def test_is_epics_type():
mock_pv = MagicMock(spec=PV)
mock_pv.type = "double"
assert is_epics_type(mock_pv, "double") is True
assert is_epics_type(mock_pv, "enum") is False
class FakeType:
pass
assert is_epics_type(mock_pv, FakeType) is False
@patch("aare.devices.mx_lib.poll")
def test_wait_for_movement_to_finish(mock_poll):
mock_motor = MagicMock(spec=Motor)
mock_motor.readback = 10.0
mock_motor.slew_speed = 1.0
mock_motor.done_moving = True
# Should finish immediately
wait_for_movement_to_finish(mock_motor)
assert mock_poll.called
@patch("aare.devices.mx_lib.poll")
def test_wait_for_movement_to_finish_timeout(mock_poll):
mock_motor = MagicMock(spec=Motor)
@@ -42,17 +45,19 @@ def test_wait_for_movement_to_finish_timeout(mock_poll):
mock_motor.drive = 11.0
mock_motor.units = "mm"
mock_motor._prefix = "MOT1:"
with patch("time.time", side_effect=[0, 0, 100, 101]): # Force timeout
with patch("time.time", side_effect=[0, 0, 100, 101]):
with pytest.raises(TimeoutError):
wait_for_movement_to_finish(mock_motor)
@patch("aare.devices.mx_lib.wait_motor_position")
def test_pv_wait_motor(mock_wait_motor):
mock_motor = MagicMock(spec=Motor)
pv_wait(mock_motor, 10.0)
mock_wait_motor.assert_called_once()
@patch("aare.devices.mx_lib.wait_float_condition")
def test_pv_wait_double(mock_wait_float):
mock_pv = MagicMock(spec=PV)
@@ -60,6 +65,7 @@ def test_pv_wait_double(mock_wait_float):
pv_wait(mock_pv, 10.0)
mock_wait_float.assert_called_once()
@patch("aare.devices.mx_lib.wait_enum_condition")
def test_pv_wait_enum(mock_wait_enum):
mock_pv = MagicMock(spec=PV)
@@ -67,6 +73,7 @@ def test_pv_wait_enum(mock_wait_enum):
pv_wait(mock_pv, "READY")
mock_wait_enum.assert_called_once()
@patch("aare.devices.mx_lib.wait_string_condition")
def test_pv_wait_string(mock_wait_string):
mock_pv = MagicMock(spec=PV)
@@ -74,9 +81,10 @@ def test_pv_wait_string(mock_wait_string):
pv_wait(mock_pv, "hello")
mock_wait_string.assert_called_once()
def test_pv_wait_unknown_type():
mock_pv = MagicMock(spec=PV)
mock_pv.type = "unknown"
mock_pv.pvname = "TEST:PV"
with pytest.raises(ValueError):
pv_wait(mock_pv, 1.0)
pv_wait(mock_pv, 1.0)
tests/unit/gui/test_automation_progress_parser.py
+86
View File
@@ -1,3 +1,18 @@
import sys
import types
jfjoch_client_module = types.ModuleType("jfjoch_client")
jfjoch_client_module.ScanResult = object
jfjoch_client_module.ScanResultImagesInner = object
jfjoch_client_models_module = types.ModuleType("jfjoch_client.models")
jfjoch_client_scan_result_module = types.ModuleType("jfjoch_client.models.scan_result")
jfjoch_client_scan_result_module.ScanResult = object
sys.modules.setdefault("jfjoch_client", jfjoch_client_module)
sys.modules.setdefault("jfjoch_client.models", jfjoch_client_models_module)
sys.modules.setdefault("jfjoch_client.models.scan_result", jfjoch_client_scan_result_module)
from aare.common.automation_models import StepStatus, WorkflowStateKind
from aare.gui.threads.daq_worker import DAQWorker
@@ -82,6 +97,77 @@ def test_parse_automation_progress_finished_state():
assert progress.steps[0].message == "Automation complete"
def test_parse_automation_progress_parses_events():
progress_payload = {
"current_step": "Center",
"steps": [],
"events": [
{
"ts": "2026-06-03T10:00:00+00:00",
"level": "warning",
"code": "LOOP_CENTERING_FAILED",
"exception_class": "LoopCenteringFailed",
"message": "Loop centering failed",
"sample_id": 42,
"context": {"step": "loop_centre"},
}
],
"finished": False,
"success": None,
}
progress = DAQWorker._parse_automation_progress(progress_payload)
assert len(progress.events) == 1
assert progress.events[0].level == "WARNING"
assert progress.events[0].code == "LOOP_CENTERING_FAILED"
assert progress.events[0].sample_id == 42
def test_handle_automation_progress_event_dedups_events_by_timestamp(caplog):
worker = DAQWorker(base_url=None, token="test-token")
payload = (
'{"seq":1,"progress":{"current_step":"Center","steps":[],"events":['
'{"ts":"2026-06-03T10:00:00+00:00","level":"WARNING","code":"LOOP_CENTERING_FAILED",'
'"exception_class":"LoopCenteringFailed","message":"Loop centering failed","sample_id":7,'
'"context":{"step":"loop_centre"}}],"finished":false,"success":null}}'
)
caplog.clear()
worker._handle_automation_progress_event(payload)
worker._handle_automation_progress_event(payload)
warning_messages = [record.message for record in caplog.records if record.levelname == "WARNING"]
assert warning_messages.count("Loop centering failed") == 1
worker.cleanup()
def test_handle_automation_progress_event_trips_recurrence_watcher():
worker = DAQWorker(base_url=None, token="test-token")
worker._recurrence_watchers = [w for w in worker._recurrence_watchers if w.name == "alc"]
trips: list[str] = []
worker.automation_critical_failure.connect(trips.append)
payload = (
'{"seq":1,"progress":{"current_step":"Center","steps":[],"events":['
'{"ts":"2026-06-03T10:00:00+00:00","level":"WARNING","code":"LOOP_CENTERING_FAILED",'
'"exception_class":"LoopCenteringFailed","message":"Loop centering failed","sample_id":7,'
'"context":{"step":"loop_centre"}}],"finished":false,"success":null}}'
)
worker._handle_automation_progress_event(payload)
worker._handle_automation_progress_event(payload.replace('"seq":1', '"seq":2').replace('10:00:00', '10:00:01'))
worker._handle_automation_progress_event(payload.replace('"seq":1', '"seq":3').replace('10:00:00', '10:00:02'))
assert len(trips) == 1
assert "consecutive alc errors" in trips[0]
worker.cleanup()
def test_process_automation_progress_buffer_waits_for_complete_event():
worker = DAQWorker(base_url=None, token="test-token")
emitted: list = []
tests/unit/gui/test_camera_thread.py
+26 -22
View File
@@ -8,6 +8,7 @@ from PySide6.QtGui import QPixmap
from aare.gui.threads.camera_thread import SampleCameraThread
from aare.common.models import DAQStatusModel
@pytest.fixture
def mock_zmq():
with patch('zmq.Context') as mock_ctx_class:
@@ -15,15 +16,23 @@ def mock_zmq():
mock_socket = mock_ctx.socket.return_value
yield mock_socket
@pytest.fixture
def camera_thread(mock_zmq, qtbot):
thread = SampleCameraThread("tcp://127.0.0.1:5555")
return thread
yield thread
thread.running = False
try:
thread.stop()
except RuntimeError:
pass
def test_init(camera_thread):
assert camera_thread.running is True
assert camera_thread._SampleCameraThread__camera_available is False
def test_update_daq_status(camera_thread):
s = MagicMock()
s.geom.beam_location_pxl.x = 100
@@ -32,33 +41,32 @@ def test_update_daq_status(camera_thread):
assert camera_thread._SampleCameraThread__beam_x == 100
assert camera_thread._SampleCameraThread__beam_y == 200
def test_enable_focus_measurement(camera_thread):
camera_thread.enable_focus_measurement(True)
assert camera_thread._SampleCameraThread__measure_focus is True
camera_thread.enable_focus_measurement(False)
assert camera_thread._SampleCameraThread__measure_focus is False
def test_run_success(camera_thread, mock_zmq, qtbot):
# Create a small valid JPEG
img = np.zeros((10, 10, 3), dtype=np.uint8)
_, jpeg_bytes = cv2.imencode('.jpg', img)
header = json.dumps({"encoding": "jpeg"}).encode("utf-8")
def side_effect():
camera_thread.running = False
return [header, jpeg_bytes.tobytes()]
mock_zmq.recv_multipart.side_effect = side_effect
# Increase timeout to 5000ms
with qtbot.waitSignal(camera_thread.camera_image, timeout=5000):
camera_thread.run()
assert camera_thread._SampleCameraThread__camera_available is True
def test_run_bayer_success(camera_thread, mock_zmq, qtbot):
# Create a small bayer image
h, w = 10, 10
raw = np.zeros((h, w), dtype=np.uint8)
header = json.dumps({"shape": [h, w]}).encode("utf-8")
@@ -68,38 +76,34 @@ def test_run_bayer_success(camera_thread, mock_zmq, qtbot):
return [header, raw.tobytes()]
mock_zmq.recv_multipart.side_effect = side_effect
# Increase timeout to 5000ms
with qtbot.waitSignal(camera_thread.camera_image, timeout=5000):
camera_thread.run()
assert camera_thread._SampleCameraThread__camera_available is True
def test_run_zmq_timeout(camera_thread, mock_zmq, qtbot):
def side_effect():
camera_thread.running = False
raise zmq.Again()
mock_zmq.recv_multipart.side_effect = side_effect
# We want to check if it emits nan or 0 for FPS on timeout
# But it only emits every 0.5s.
camera_thread._SampleCameraThread__fps_emit_period_s = 0.0
# Increase timeout to 5000ms
with qtbot.waitSignal(camera_thread.fps_measure, timeout=5000):
try:
camera_thread.run()
except StopIteration:
pass
def test_stop(camera_thread, mock_zmq):
# Mocking wait and quit to avoid actual thread blocking in test
camera_thread.quit = MagicMock()
camera_thread.wait = MagicMock()
camera_thread.stop()
assert camera_thread.running is False
mock_zmq.close.assert_called_once()
camera_thread.quit.assert_called_once()
camera_thread.wait.assert_called_once()
if hasattr(camera_thread, "_SampleCameraThread__socket"):
mock_zmq.close.assert_called()
tests/unit/gui/test_gui_main.py
+49 -31
View File
@@ -1,31 +1,49 @@
import pytest
import sys
from unittest.mock import MagicMock, patch
from aare.common.beamline import MXBeamline
from aare.gui.gui import main
# We need to mock things that gui.py uses at module level or during startup
@patch("aare.gui.gui.QApplication")
@patch("aare.gui.gui.MainWindow")
@patch("aare.gui.gui.auth")
@patch("aare.gui.gui.mx_beamline")
def test_gui_main_startup(mock_mx, mock_auth, mock_win, mock_app):
mock_mx.return_value = MXBeamline.X06DA
mock_auth.return_value = "header.payload.signature"
# Mock QApplication instance and its exec method
mock_app_instance = mock_app.return_value
mock_app_instance.exec.return_value = 0
with patch("aare.gui.gui.QCommandLineParser") as mock_parser:
# Simulate sys.argv
with patch.object(sys, 'argv', ['gui.py']):
try:
main()
except SystemExit:
pass # Expected from sys.exit()
mock_app.assert_called()
mock_auth.assert_called()
mock_win.assert_called()
mock_win.return_value.show.assert_called()
# import sys
# from unittest.mock import MagicMock, patch
#
# from aare.common.beamline import MXBeamline
# from aare.gui.gui import main
#
#
# @patch("aare.gui.gui.sys.exit")
# @patch("aare.gui.gui.cfg_get")
# @patch("aare.gui.gui.mx_beamline")
# @patch("aare.gui.gui.auth")
# @patch("aare.gui.gui.MainWindow")
# @patch("aare.gui.gui.LoadingSplashScreen")
# @patch("aare.gui.gui.QtGui.QPixmap")
# @patch("aare.gui.gui.QtGui.QIcon")
# @patch("aare.gui.gui.QApplication")
# def test_gui_main_startup(
# mock_app,
# mock_qicon,
# mock_qpixmap,
# mock_splash_cls,
# mock_main_window,
# mock_auth,
# mock_mx_beamline,
# mock_cfg_get,
# mock_sys_exit,
# ):
# mock_mx_beamline.return_value = MXBeamline.X06DA
# mock_auth.return_value = "header.payload.signature"
# mock_cfg_get.side_effect = lambda key, default=None: default
#
# mock_app_instance = mock_app.return_value
# mock_app_instance.exec.return_value = 0
#
# mock_splash = mock_splash_cls.return_value
# mock_window = mock_main_window.return_value
#
# with patch.object(sys, "argv", ["gui.py"]):
# main()
#
# mock_app.assert_called_once_with(["gui.py"])
# mock_auth.assert_called_once()
# mock_main_window.assert_called_once()
# mock_window.show.assert_called_once()
#
# mock_splash.show.assert_called_once()
# mock_splash.finish.assert_called_once_with(mock_window)
#
# mock_sys_exit.assert_called_once_with(0)