chore: add test to improve coverage
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / test-with-beamline-plugins (pxi_bec) (push) Skipped
CI / test-with-beamline-plugins (pxii_bec) (push) Skipped
CI / test-with-beamline-plugins (pxiii_bec) (push) Skipped
CI / test (3.12) (pull_request) Successful in 1m27s
CI / lint (pull_request) Successful in 1m37s
CI / test-with-beamline-plugins (pxii_bec) (pull_request) Successful in 1m25s
CI / test-with-beamline-plugins (pxi_bec) (pull_request) Successful in 1m29s
CI / test (3.13) (pull_request) Successful in 1m59s
CI / test-with-coverage (pull_request) Successful in 2m0s
CI / test (3.11) (pull_request) Successful in 2m21s
CI / coverage-analysis (pull_request) Successful in 4s
CI / test-with-beamline-plugins (pxiii_bec) (pull_request) Canceled after 2m23s

This commit is contained in:
2026-08-11 16:00:58 +02:00
parent 17bb65420f
commit 9906937525
@@ -0,0 +1,45 @@
"""Covers the recovery-response payload handling in DAQWorker.
The server may return a bare JSON string instead of a dict during recovery
(see fix "allow str has no get when in recovery"), so the handler must not
assume ``payload.get`` exists.
"""
import json
from unittest.mock import MagicMock
import pytest
from aare.gui.threads.daq_worker import DAQWorker
@pytest.fixture
def worker(qapp):
w = DAQWorker(base_url=None, token="test-token")
w._timer.stop() # no background polling during tests
w.send_status_request = MagicMock()
return w
def run_recovery(worker, response_data, default="default recovery message"):
worker.handle_response = MagicMock(return_value=response_data)
slot = MagicMock()
worker.recovery_action_completed.connect(slot)
worker._handle_recovery_action_response(MagicMock(), default)
return slot
def test_recovery_response_dict_payload(worker):
slot = run_recovery(worker, json.dumps({"message": "TELL recovered"}))
slot.assert_called_once_with("TELL recovered")
def test_recovery_response_dict_without_message_uses_default(worker):
slot = run_recovery(worker, json.dumps({"status": "ok"}), default="fallback")
slot.assert_called_once_with("fallback")
def test_recovery_response_str_payload(worker):
# A bare JSON string has no .get(); must be passed through as-is.
slot = run_recovery(worker, json.dumps("plain string status"))
slot.assert_called_once_with("plain string status")