diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index 010d41b6..daeeee3e 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -165,7 +165,8 @@ def fetch_spool(self, gate: int, spool_id: int) -> None: return def _on_result(result: dict | None) -> None: - if result is not None: + # An error reply parses as an empty spool and would blank the gate. + if result and "error" not in result: self.spool_fetched.emit(gate, result) self._ws.api.get_spool(spool_id, _on_result) diff --git a/BlocksScreen/lib/files.py b/BlocksScreen/lib/files.py index 916b7afa..392fcf8f 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -458,21 +458,25 @@ def request_print_duration(self, filename: str) -> None: job_id = str(metadata.job_id) if job_id in self._job_durations: return - # Mark before sending: error replies skip the callback. + # Mark before sending so an in-flight lookup is not re-requested. self._job_durations[job_id] = None - self.ws.api.history_get_job( + sent = self.ws.api.history_get_job( job_id, lambda result, name=filename, uid=job_id: self._history_job.emit( name, uid, result or {} ), ) + if not sent: + self._job_durations.pop(job_id, None) @QtCore.pyqtSlot(str, str, dict, name="on_history_job") def _on_history_job(self, filename: str, job_id: str, result: dict) -> None: """Cache a completed job's duration and re-emit fileinfo.""" job = result.get("job") or {} status = job.get("status") - if status == "in_progress": + lost = (result.get("error") or {}).get("code") == -1 + # Re-ask after a disconnect or a running job; a deleted job stays final. + if lost or status == "in_progress": self._job_durations.pop(job_id, None) return duration = job.get("print_duration") diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index c5960362..fe505152 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -42,6 +42,7 @@ class MoonWebSocket(QtCore.QObject, threading.Thread): klippy_state_signal = QtCore.pyqtSignal(str, name="klippy_state") query_server_info_signal = QtCore.pyqtSignal(name="query_server_information") server_components_signal = QtCore.pyqtSignal(list, name="server_components") + request_reply_signal = QtCore.pyqtSignal(object, object, name="request_reply") _KLIPPY_NOTIFY_METHODS: typing.ClassVar[frozenset[str]] = frozenset( { @@ -84,6 +85,7 @@ def __init__(self, parent: QtCore.QObject) -> None: ) self.klippy_state_signal.connect(self.api.request_printer_info) + self.request_reply_signal.connect(self._run_callback) logger.info("Websocket object initialized") @property @@ -201,6 +203,13 @@ def on_close(self, *args) -> None: _close_message = args[2] if len(args) == 3 else None with self._state_lock: self.connected = False + with self._request_lock: + callbacks = [e[2] for e in self.request_table.values() if e[2] is not None] + self.request_table.clear() + for callback in callbacks: + # Same shape as a Moonraker JSON-RPC error; -1 marks it client-side. + error = {"code": -1, "message": "disconnected"} + self.request_reply_signal.emit(callback, {"error": error}) self.ws.keep_running = False self.connection_lost[str].emit( f"code: {_close_status_code} | message {_close_message}" @@ -302,12 +311,14 @@ def on_message(self, *args) -> None: else: _callback = _entry[2] if len(_entry) > 2 else None if _callback is not None: - if "error" not in response: - _callback(response.get("result")) - else: + if "error" in response: logger.error( - "WS request %s error: %s", _entry[0], response.get("error") + "WS request %s error: %s", _entry[0], response["error"] ) + _reply = {"error": response["error"]} + else: + _reply = response.get("result") + self.request_reply_signal.emit(_callback, _reply) return if "error" in response: message_event = WebSocketMessageReceived( @@ -349,6 +360,14 @@ def on_message(self, *args) -> None: "Unexpected error while creating websocket message event: %s", e ) + @QtCore.pyqtSlot(object, object, name="run_callback") + def _run_callback(self, callback, reply) -> None: + """Run a request callback on the GUI thread; a raise must not kill the app.""" + try: + callback(reply) + except Exception: + logger.exception("WS request callback %r failed", callback) + def send_request( self, method: str, params: dict | None = None, callback=None ) -> bool: @@ -357,11 +376,14 @@ def send_request( Args: method (str): Websocket method name params (dict, optional): parameters for the websocket method. Defaults to None. - callback (callable, optional): Called with ``response["result"]`` when the - response arrives. If None, the response is routed as a + callback (callable, optional): Called once on the GUI thread with + ``response["result"]``, or with ``{"error": {"code": int, "message": str}}`` + on a JSON-RPC error or a disconnect (code -1); check for ``"error"`` + before reading the result. If None, the response is routed as a ``WebSocketMessageReceived`` event to the parent widget. Defaults to None. Returns: - bool: Whether the method finished and a request was sent + bool: True if sent; the callback then runs exactly once. False if not + sent; the callback never runs, so the caller handles the failure. """ if params is None: params = {} @@ -382,7 +404,16 @@ def send_request( "params": params, "id": _rid, } - _ws.send(json.dumps(packet)) + try: + _ws.send(json.dumps(packet)) + except (websocket.WebSocketException, OSError) as e: + logger.warning("WS request %s not sent: %s", method, e) + with self._request_lock: + queued = self.request_table.pop(_rid, None) is not None + if queued or callback is None: + return False + # on_close already failed it; False would make the caller answer it twice. + return True return True @@ -923,10 +954,11 @@ def set_spool_id(self, spool_id: int | None, callback=None) -> bool: ) def get_spool(self, spool_id: int, callback) -> bool: - """Request spool data from Moonraker's Spoolman proxy""" - return self._ws.send_request( - method="server.spoolman.get_spool", - params={"spool_id": spool_id}, + """Request one spool through the Spoolman proxy; there is no get_spool RPC.""" + return self.spoolman_proxy( + request_method="GET", + path=f"/v1/spool/{spool_id}", + use_v2_response=False, callback=callback, ) diff --git a/BlocksScreen/lib/moonrest.py b/BlocksScreen/lib/moonrest.py index 437d98e5..36601d93 100644 --- a/BlocksScreen/lib/moonrest.py +++ b/BlocksScreen/lib/moonrest.py @@ -77,19 +77,19 @@ def get_server_info(self): return self.get_request(method="server/info") def get_spool(self, spool_id: int) -> dict | None: - """Spoolman spool via Moonraker, None on error.""" - response = self.get_request(f"server/spoolman/spool/{spool_id}") + """Spoolman spool via the Moonraker proxy, None on error.""" + response = self.post_request( + "server/spoolman/proxy", + json={ + "request_method": "GET", + "path": f"/v1/spool/{spool_id}", + "use_v2_response": False, + }, + ) if not isinstance(response, dict): return None return response.get("result") - def set_spool_used_weight(self, spool_id: int, weight: float) -> bool: - """Set a spool's used_weight; True on success.""" - response = self.post_request( - f"server/spoolman/spool/{spool_id}", json={"used_weight": weight} - ) - return response is not None - def firmware_restart(self): """POST firmware_restart to Moonraker.""" return self.post_request(method="printer/firmware_restart") diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index fe462544..aa8d3714 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -381,6 +381,24 @@ def test_no_emit_on_ws_failure(self, manager, qtbot) -> None: with qtbot.assertNotEmitted(manager.spool_fetched): callback(None) + @pytest.mark.parametrize( + "error", + [ + {"code": 404, "message": "spool not found"}, + {"code": -1, "message": "disconnected"}, + ], + ) + def test_error_reply_leaves_the_gate_alone(self, manager, qtbot, error) -> None: + """An error read as spool data would send MATERIAL= COLOR= SPOOLID=-1.""" + manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) + manager.fetch_spool(0, 42) + callback = manager._ws.api.get_spool.call_args.args[1] + with ( + qtbot.assertNotEmitted(manager.spool_fetched), + qtbot.assertNotEmitted(manager.run_gcode_signal), + ): + callback({"error": error}) + class TestLoadCellUpdate: def test_noop_when_mmu_state_none(self, manager) -> None: diff --git a/tests/amu/test_moonrest_unit.py b/tests/amu/test_moonrest_unit.py index 2291aad3..64264cf7 100644 --- a/tests/amu/test_moonrest_unit.py +++ b/tests/amu/test_moonrest_unit.py @@ -19,35 +19,26 @@ def test_return_spool_dict_on_sucess(self, rest) -> None: "used_weight": 50.0, "remaining_weight": 200.0, } - with patch.object(rest, "get_request", return_value={"result": spool_data}): + with patch.object(rest, "post_request", return_value={"result": spool_data}): assert rest.get_spool(42) == spool_data def test_returns_non_on_http_error(self, rest) -> None: - with patch.object(rest, "get_request", return_value=None): + with patch.object(rest, "post_request", return_value=None): assert rest.get_spool(42) is None def test_returns_none_on_missing_result_key(self, rest): - with patch.object(rest, "get_request", return_value={"something": "else"}): + with patch.object(rest, "post_request", return_value={"something": "else"}): assert rest.get_spool(42) is None - def test_calls_correct_endpoint(self, rest) -> None: - with patch.object(rest, "get_request", return_value=None) as mock_get: - rest.get_spool(7) - mock_get.assert_called_once_with("server/spoolman/spool/7") - - -class TestSetSpoolUsedWeight: - def test_returns_true_on_sucess(self, rest) -> None: - with patch.object(rest, "post_request", return_value={"result": "ok"}): - assert rest.set_spool_used_weight(42, 75.5) - - def test_returns_false_on_error(self, rest) -> None: - with patch.object(rest, "post_request", return_value={"result": "ok"}): - assert rest.set_spool_used_weight(42, 75.5) - - def test_calls_correct_endpoint(self, rest) -> None: + # /server/spoolman/spool/{id} is a 404; only the proxy route exists. + def test_goes_through_the_proxy(self, rest) -> None: with patch.object(rest, "post_request", return_value=None) as mock_post: - rest.set_spool_used_weight(5, 100.0) + rest.get_spool(7) mock_post.assert_called_once_with( - "server/spoolman/spool/5", json={"used_weight": 100.0} + "server/spoolman/proxy", + json={ + "request_method": "GET", + "path": "/v1/spool/7", + "use_v2_response": False, + }, ) diff --git a/tests/lib/test_moonraker_comm_unit.py b/tests/lib/test_moonraker_comm_unit.py index 77e9608c..044c022b 100644 --- a/tests/lib/test_moonraker_comm_unit.py +++ b/tests/lib/test_moonraker_comm_unit.py @@ -1,6 +1,7 @@ """Unit tests for moonrakerComm.""" import json +import threading from types import SimpleNamespace from unittest.mock import MagicMock @@ -8,7 +9,7 @@ from PyQt6 import QtCore from BlocksScreen.lib import moonrakerComm -from BlocksScreen.lib.moonrakerComm import OneShotTokenError +from BlocksScreen.lib.moonrakerComm import MoonAPI, OneShotTokenError class TestOneShotTokenError: @@ -120,6 +121,42 @@ def test_reconnect_reemits_unchanged_state(self, ws): assert states == ["ready", "ready"] +class TestRequestCallbacks: + def _reply(self, sock, callback, payload: dict) -> None: + sock.request_table[9] = ["server.spoolman.proxy", {}, callback] + sock.on_message(None, json.dumps({"jsonrpc": "2.0", "id": 9, **payload})) + + def test_result_reaches_callback(self, ws): + got = [] + self._reply(ws, got.append, {"result": {"response": [1], "error": None}}) + assert got == [{"response": [1], "error": None}] + + def test_jsonrpc_error_reaches_callback(self, ws): + got = [] + err = {"code": 503, "message": "Spoolman server not available"} + self._reply(ws, got.append, {"error": err}) + assert got == [{"error": err}] + + def test_raising_callback_is_logged_not_raised(self, ws, caplog): + def boom(_result): + raise RuntimeError("boom") + + self._reply(ws, boom, {"result": {}}) + assert "callback" in caplog.text + + def test_callback_runs_on_gui_thread(self, ws, qtbot): + threads = [] + + def record(_result): + threads.append(threading.current_thread()) + + worker = threading.Thread(target=self._reply, args=(ws, record, {"result": {}})) + worker.start() + worker.join() + qtbot.waitUntil(lambda: bool(threads)) + assert threads == [threading.main_thread()] + + class TestReconnectCycle: def test_retries_forever_until_connected(self, ws, monkeypatch): connect = MagicMock(return_value=False) @@ -192,3 +229,73 @@ def test_close_leaves_reconnect_to_watchdog(self, ws, monkeypatch, qtbot): ws.on_close(None, 1006, "gone") qtbot.wait(50) connect.assert_not_called() + + def test_close_fails_pending_requests(self, ws, qtbot): + ws.ws = MagicMock() + got = [] + ws.request_table[3] = ["server.info", {}, None] + ws.request_table[4] = ["server.spoolman.proxy", {}, got.append] + ws.on_close(None, 1006, "gone") + assert ws.request_table == {} + qtbot.waitUntil(lambda: bool(got)) + assert got == [{"error": {"code": -1, "message": "disconnected"}}] + + def test_failed_send_returns_false_and_forgets_request(self, ws, monkeypatch): + # conftest stubs websocket; an except clause needs a real class + monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError) + ws.connected = True + ws.ws = MagicMock() + ws.ws.send.side_effect = OSError("broken pipe") + assert ws.send_request("server.info", callback=MagicMock()) is False + assert ws.request_table == {} + + def test_send_racing_close_answers_the_callback_once(self, ws, qtbot, monkeypatch): + """on_close already failed the callback, so True keeps the caller from a 2nd.""" + monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError) + ws.connected = True + ws.ws = MagicMock() + + def close_then_fail(_packet): + ws.on_close(None, 1006, "gone") + raise OSError("broken pipe") + + ws.ws.send.side_effect = close_then_fail + got = [] + assert ws.send_request("server.info", callback=got.append) is True + qtbot.waitUntil(lambda: bool(got)) + qtbot.wait(20) + assert got == [{"error": {"code": -1, "message": "disconnected"}}] + + def test_send_while_disconnected_never_calls_back(self, ws, qtbot): + callback = MagicMock() + assert ws.send_request("server.info", callback=callback) is False + assert ws.request_table == {} + qtbot.wait(20) + callback.assert_not_called() + + +@pytest.fixture +def api(): + """MoonAPI over a stub socket; it parents itself to ws, so ws must be a QObject.""" + sock = QtCore.QObject() + sock.send_request = MagicMock(return_value=True) + return MoonAPI(sock) + + +class TestGetSpool: + def test_goes_through_the_proxy(self, api) -> None: + api.get_spool(42, None) + params = api._ws.send_request.call_args.kwargs + assert params["method"] == "server.spoolman.proxy" + assert params["params"] == { + "use_v2_response": False, + "request_method": "GET", + "path": "/v1/spool/42", + } + + def test_passes_the_callback_through(self, api) -> None: + def cb(_result): + return None + + api.get_spool(1, cb) + assert api._ws.send_request.call_args.kwargs["callback"] is cb diff --git a/tests/widgets/test_files_data_unit.py b/tests/widgets/test_files_data_unit.py index fd01fea4..8448dc43 100644 --- a/tests/widgets/test_files_data_unit.py +++ b/tests/widgets/test_files_data_unit.py @@ -186,6 +186,28 @@ def test_in_progress_job_is_asked_again(self, files): files.request_print_duration("a.gcode") assert files.ws.api.history_get_job.call_count == 2 + def test_disconnect_reply_is_asked_again(self, files): + """A reply lost to a disconnect must not hide the duration until restart.""" + files.request_print_duration("a.gcode") + error = {"code": -1, "message": "disconnected"} + files._on_history_job("a.gcode", "000023", {"error": error}) + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 2 + + def test_deleted_job_is_not_asked_again(self, files): + """A deleted job (-32601) is final; re-asking costs a request per view.""" + files.request_print_duration("a.gcode") + error = {"code": -32601, "message": "Invalid job uid: 000023"} + files._on_history_job("a.gcode", "000023", {"error": error}) + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 1 + + def test_unsent_lookup_is_asked_again(self, files): + files.ws.api.history_get_job.return_value = False + files.request_print_duration("a.gcode") + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 2 + class TestFilelistNotifications: def test_every_batched_entry_is_applied(self, files):