perl_d
0135129c89
CI / lint (pull_request) Failing after 33s
CI / test (3.11) (pull_request) Skipped
CI / test (3.12) (pull_request) Skipped
CI / test (3.13) (pull_request) Skipped
CI / test-with-beamline-plugins (pxi_bec) (pull_request) Skipped
CI / test-with-beamline-plugins (pxii_bec) (pull_request) Skipped
CI / test-with-beamline-plugins (pxiii_bec) (pull_request) Skipped
CI / test-with-coverage (pull_request) Skipped
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
import types
|
|
|
|
from aarecommon.models.models import DAQOperation
|
|
|
|
from aare.daq.operations.face_detection.models import FaceDetectionResult
|
|
|
|
|
|
def test_execute_face_detection_reports_failure(monkeypatch):
|
|
from aare.daq.daq import AareDAQ
|
|
|
|
daq = object.__new__(AareDAQ)
|
|
|
|
calls = {"set_state": [], "handle_error": []}
|
|
|
|
monkeypatch.setattr(daq, "_set_state", lambda state: calls["set_state"].append(state))
|
|
monkeypatch.setattr(
|
|
daq,
|
|
"_create_face_detection_service",
|
|
lambda: types.SimpleNamespace(
|
|
run=lambda **kwargs: FaceDetectionResult(
|
|
success=False,
|
|
payload={"running": False, "samples": [], "height_fit": {}, "area_fit": {}},
|
|
error=RuntimeError("fd failed"),
|
|
comment="face detection sequence failed",
|
|
)
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
daq, "_handle_operation_error", lambda **kwargs: calls["handle_error"].append(kwargs)
|
|
)
|
|
monkeypatch.setattr(
|
|
type(daq),
|
|
"sample",
|
|
property(lambda self: types.SimpleNamespace(sample_name="sample-1", db_id=1)),
|
|
)
|
|
|
|
result = daq._execute_face_detection(steps=7, step_size=30, report_error=True)
|
|
|
|
assert result.success is False
|
|
assert len(calls["set_state"]) == 1
|
|
assert len(calls["handle_error"]) == 1
|
|
assert calls["handle_error"][0]["operation"] == DAQOperation.FACE_CENTERING
|
|
|
|
|
|
def test_execute_face_detection_can_skip_error_reporting(monkeypatch):
|
|
from aare.daq.daq import AareDAQ
|
|
|
|
daq = object.__new__(AareDAQ)
|
|
|
|
handle_error_calls = []
|
|
|
|
monkeypatch.setattr(daq, "_set_state", lambda state: None)
|
|
monkeypatch.setattr(
|
|
daq,
|
|
"_create_face_detection_service",
|
|
lambda: types.SimpleNamespace(
|
|
run=lambda **kwargs: FaceDetectionResult(
|
|
success=False,
|
|
payload={"running": False, "samples": [], "height_fit": {}, "area_fit": {}},
|
|
error=RuntimeError("fd failed"),
|
|
comment="face detection sequence failed",
|
|
)
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
daq, "_handle_operation_error", lambda **kwargs: handle_error_calls.append(kwargs)
|
|
)
|
|
|
|
result = daq._execute_face_detection(report_error=False)
|
|
|
|
assert result.success is False
|
|
assert handle_error_calls == []
|
|
|
|
|
|
def test_public_face_detection_uses_execute_face_detection(monkeypatch):
|
|
from aare.daq.daq import AareDAQ
|
|
|
|
daq = object.__new__(AareDAQ)
|
|
cfg = types.SimpleNamespace(try_set_busy=lambda timeout=360: None, state_busy=False)
|
|
daq._cfg = cfg
|
|
|
|
daq._execute_face_detection = lambda **kwargs: FaceDetectionResult(
|
|
success=True,
|
|
payload={"running": False, "samples": [{"angle": 45}], "height_fit": {}, "area_fit": {}},
|
|
)
|
|
|
|
result = daq.face_detection(steps=7, step_size=30)
|
|
|
|
assert result["running"] is False
|
|
assert result["samples"] == [{"angle": 45}]
|
|
assert cfg.state_busy is False
|