From be539f3ca98725af158ae865d78e8042a020f192 Mon Sep 17 00:00:00 2001 From: appel_c Date: Tue, 21 Oct 2025 10:25:18 +0200 Subject: [PATCH] fix(psi-detector-base): add test for interface --- tests/test_base_classes.py | 214 ++++++++----------------------------- 1 file changed, 43 insertions(+), 171 deletions(-) diff --git a/tests/test_base_classes.py b/tests/test_base_classes.py index 02726bd..d04783f 100644 --- a/tests/test_base_classes.py +++ b/tests/test_base_classes.py @@ -39,182 +39,54 @@ def test_stage(detector_base): assert mock_on_stage.call_count == 2 -# def test_stage(detector_base): -# detector_base._staged = Staged.yes -# with pytest.raises(RedundantStaging): -# detector_base.stage() -# assert detector_base.stopped is False -# detector_base._staged = Staged.no -# with ( -# mock.patch.object(detector_base.custom_prepare, "on_stage") as mock_on_stage, -# mock.patch.object(detector_base.scaninfo, "load_scan_metadata") as mock_load_metadata, -# ): -# rtr = detector_base.stage() -# assert isinstance(rtr, list) -# mock_on_stage.assert_called_once() -# mock_load_metadata.assert_called_once() -# assert detector_base.stopped is False +def test_pre_scan(detector_base): + with mock.patch.object(detector_base, "on_pre_scan") as mock_on_pre_scan: + detector_base.pre_scan() + mock_on_pre_scan.assert_called_once() -# def test_pre_scan(detector_base): -# with mock.patch.object(detector_base.custom_prepare, "on_pre_scan") as mock_on_pre_scan: -# detector_base.pre_scan() -# mock_on_pre_scan.assert_called_once() +def test_trigger(detector_base): + status = DeviceStatus(detector_base) + with mock.patch.object( + detector_base, "on_trigger", side_effect=[None, status] + ) as mock_on_trigger: + st = detector_base.trigger() + assert isinstance(st, DeviceStatus) + st.wait(timeout=1) + assert st.done is True + st = detector_base.trigger() + assert st.done is False + assert id(st) == id(status) -# def test_trigger(detector_base): -# status = DeviceStatus(detector_base) -# with mock.patch.object( -# detector_base.custom_prepare, "on_trigger", side_effect=[None, status] -# ) as mock_on_trigger: -# st = detector_base.trigger() -# assert isinstance(st, DeviceStatus) -# time.sleep(0.1) -# assert st.done is True -# st = detector_base.trigger() -# assert st.done is False -# assert id(st) == id(status) +def test_unstage(detector_base): + detector_base.stopped = True + with (mock.patch.object(detector_base, "on_unstage") as mock_on_unstage,): + rtr = detector_base.unstage() + assert isinstance(rtr, list) + assert mock_on_unstage.call_count == 1 + detector_base.stopped = False + rtr = detector_base.unstage() + assert isinstance(rtr, list) + assert mock_on_unstage.call_count == 2 -# def test_unstage(detector_base): -# detector_base.stopped = True -# with ( -# mock.patch.object(detector_base.custom_prepare, "on_unstage") as mock_on_unstage, -# mock.patch.object(detector_base, "check_scan_id") as mock_check_scan_id, -# ): -# rtr = detector_base.unstage() -# assert isinstance(rtr, list) -# assert mock_check_scan_id.call_count == 1 -# assert mock_on_unstage.call_count == 1 -# detector_base.stopped = False -# rtr = detector_base.unstage() -# assert isinstance(rtr, list) -# assert mock_check_scan_id.call_count == 2 -# assert mock_on_unstage.call_count == 2 +def test_complete(detector_base): + status = DeviceStatus(detector_base) + with mock.patch.object( + detector_base, "on_complete", side_effect=[None, status] + ) as mock_on_complete: + st = detector_base.complete() + assert isinstance(st, DeviceStatus) + time.sleep(0.1) + assert st.done is True + st = detector_base.complete() + assert st.done is False + assert id(st) == id(status) -# def test_complete(detector_base): -# status = DeviceStatus(detector_base) -# with mock.patch.object( -# detector_base.custom_prepare, "on_complete", side_effect=[None, status] -# ) as mock_on_complete: -# st = detector_base.complete() -# assert isinstance(st, DeviceStatus) -# time.sleep(0.1) -# assert st.done is True -# st = detector_base.complete() -# assert st.done is False -# assert id(st) == id(status) - - -# def test_stop(detector_base): -# with mock.patch.object(detector_base.custom_prepare, "on_stop") as mock_on_stop: -# detector_base.stop() -# mock_on_stop.assert_called_once() -# assert detector_base.stopped is True - - -# def test_check_scan_id(detector_base): -# detector_base.scaninfo.scan_id = "abcde" -# detector_base.stopped = False -# detector_base.check_scan_id() -# assert detector_base.stopped is True -# detector_base.stopped = False -# detector_base.check_scan_id() -# assert detector_base.stopped is False - - -# def test_wait_for_signal(detector_base): -# my_value = False - -# def my_callback(): -# return my_value - -# detector_base -# status = detector_base.custom_prepare.wait_with_status( -# [(my_callback, True)], -# check_stopped=True, -# timeout=5, -# interval=0.01, -# exception_on_timeout=None, -# ) -# time.sleep(0.1) -# assert status.done is False -# # Check first that it is stopped when detector_base.stop() is called -# detector_base.stop() -# # some delay to allow the stop to take effect -# time.sleep(0.15) -# assert status.done is True -# assert status.exception().args == DeviceStopError(f"{detector_base.name} was stopped").args -# detector_base.stopped = False -# status = detector_base.custom_prepare.wait_with_status( -# [(my_callback, True)], -# check_stopped=True, -# timeout=5, -# interval=0.01, -# exception_on_timeout=None, -# ) -# # Check that thread resolves when expected value is set -# my_value = True -# # some delay to allow the stop to take effect -# time.sleep(0.15) -# assert status.done is True -# assert status.success is True -# assert status.exception() is None - -# detector_base.stopped = False -# # Check that wait for status runs into timeout with expectd exception -# my_value = "random_value" -# exception = TimeoutError("Timeout") -# status = detector_base.custom_prepare.wait_with_status( -# [(my_callback, True)], -# check_stopped=True, -# timeout=0.01, -# interval=0.01, -# exception_on_timeout=exception, -# ) -# time.sleep(0.2) -# assert status.done is True -# assert id(status.exception()) == id(exception) -# assert status.success is False - - -# def test_wait_for_signal_returns_exception(detector_base): -# my_value = False - -# def my_callback(): -# return my_value - -# # Check that wait for status runs into timeout with expectd exception - -# exception = TimeoutError("Timeout") -# status = detector_base.custom_prepare.wait_with_status( -# [(my_callback, True)], -# check_stopped=True, -# timeout=0.01, -# interval=0.01, -# exception_on_timeout=exception, -# ) -# time.sleep(0.2) -# assert status.done is True -# assert id(status.exception()) == id(exception) -# assert status.success is False - -# detector_base.stopped = False -# # Check that standard exception is thrown -# status = detector_base.custom_prepare.wait_with_status( -# [(my_callback, True)], -# check_stopped=True, -# timeout=0.01, -# interval=0.01, -# exception_on_timeout=None, -# ) -# time.sleep(0.2) -# assert status.done is True -# assert ( -# status.exception().args -# == DeviceTimeoutError( -# f"Timeout error for {detector_base.name} while waiting for signals {[(my_callback, True)]}" -# ).args -# ) -# assert status.success is False +def test_stop(detector_base): + with mock.patch.object(detector_base, "on_stop") as mock_on_stop: + detector_base.stop() + mock_on_stop.assert_called_once() + assert detector_base.stopped is True