test: drop racy printer_state_to_dict patch; assert structural shape

The patch on printer_state_to_dict raced against the broadcast coroutine
  under pytest-xdist's parallel workers. Mostly won locally, lost
  occasionally on CI — surfaced first as AttributeError on .kprofiles,
  then (after _fake_state was hardened) as a dict-content mismatch
  between the patched return value and the real 36-key dict.

  Fix: stop patching printer_state_to_dict; let it run for real against
  the complete _fake_state stub. Assertions now check the broadcast fired
  with the right printer_id and a dict containing awaiting_plate_clear,
  not the exact dict shape — that decouples the test from
  printer_state_to_dict's evolving body.
This commit is contained in:
maziggy
2026-05-11 14:08:06 +02:00
parent 96a0c60c60
commit e4ff84e94a
@@ -168,7 +168,17 @@ class TestBroadcastStatusChange:
@pytest.mark.asyncio
async def test_emits_ws_update_when_state_present(self, manager):
"""Happy path: printer has a known status, broadcast goes out
with the dict produced by ``printer_state_to_dict``."""
with the dict produced by ``printer_state_to_dict``.
Note: we deliberately don't patch ``printer_state_to_dict`` here.
The patch was observed to race on parallel xdist runners — when it
didn't catch the call the real function ran, leaving the test
comparing the patched return value against the real dict shape.
Letting the real function run (against a complete ``_fake_state``)
makes the test deterministic; we assert structural shape, not the
exact ~36 keys, because pinning those couples the test to the
evolving ``printer_state_to_dict`` body and adds zero value over
what ``test_printer_manager.py`` already covers."""
state = _fake_state()
with (
patch.object(manager, "get_status", return_value=state),
@@ -177,20 +187,17 @@ class TestBroadcastStatusChange:
"backend.app.core.websocket.ws_manager.send_printer_status",
new_callable=AsyncMock,
) as send_status,
patch(
"backend.app.services.printer_manager.printer_state_to_dict",
return_value={"id": 7, "awaiting_plate_clear": False},
) as to_dict,
):
await manager._broadcast_status_change(7)
send_status.assert_awaited_once()
# First positional arg is the printer ID, second is the status dict.
printer_id_arg, payload_arg = send_status.await_args.args
assert printer_id_arg == 7
assert payload_arg == {"id": 7, "awaiting_plate_clear": False}
# Verify the dict was built from the right inputs (state + id + model).
to_dict.assert_called_once_with(state, 7, "P1S")
assert isinstance(payload_arg, dict)
# The ``awaiting_plate_clear`` key is the whole point of this broadcast
# path (#1128). Any future restructuring that drops it from the dict
# would silently break the UI; pin its presence.
assert "awaiting_plate_clear" in payload_arg
@pytest.mark.asyncio
async def test_skips_when_status_unknown(self, manager):
@@ -221,10 +228,6 @@ class TestBroadcastStatusChange:
with (
patch.object(manager, "get_status", return_value=_fake_state()),
patch.object(manager, "get_model", return_value="P1S"),
patch(
"backend.app.services.printer_manager.printer_state_to_dict",
return_value={"id": 7},
),
patch(
"backend.app.core.websocket.ws_manager.send_printer_status",
new_callable=AsyncMock,
@@ -253,13 +256,14 @@ class TestEndToEndUnderRunningLoop:
# the broadcast short-circuits before reaching ws_manager.
manager._awaiting_plate_clear.add(7)
# _fake_state defaults awaiting_plate_clear=False via printer_state_to_dict's
# is_awaiting_plate_clear(printer_id) lookup, which reads from
# manager._awaiting_plate_clear (the in-memory set). Since we just
# removed 7 from that set by calling set_awaiting_plate_clear(7, False),
# the broadcast payload's awaiting_plate_clear field will be False.
with (
patch.object(manager, "get_status", return_value=_fake_state()),
patch.object(manager, "get_model", return_value="P1S"),
patch(
"backend.app.services.printer_manager.printer_state_to_dict",
return_value={"id": 7, "awaiting_plate_clear": False},
),
patch(
"backend.app.core.websocket.ws_manager.send_printer_status",
new_callable=AsyncMock,