|
|
@@ -0,0 +1,860 @@
|
|
|
+import asyncio
|
|
|
+import logging
|
|
|
+
|
|
|
+import httpx
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services.bambu_mqtt import HMSError, PrinterState
|
|
|
+from backend.app.services.wled import WLEDManager, WLEDResponseError, effective_wled_status
|
|
|
+
|
|
|
+
|
|
|
+def _state(state: str, **values) -> PrinterState:
|
|
|
+ return PrinterState(connected=values.pop("connected", True), state=state, **values)
|
|
|
+
|
|
|
+
|
|
|
+def _manager(handler) -> tuple[WLEDManager, httpx.AsyncClient]:
|
|
|
+ client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
|
+ return WLEDManager(client), client
|
|
|
+
|
|
|
+
|
|
|
+def test_effective_status_uses_existing_bambuddy_states_and_priorities():
|
|
|
+ assert effective_wled_status(_state("IDLE")) == "idle"
|
|
|
+ assert effective_wled_status(_state("RUNNING")) == "printing"
|
|
|
+ assert effective_wled_status(_state("PRINTING")) == "printing"
|
|
|
+ assert effective_wled_status(_state("PREPARE")) == "prepare"
|
|
|
+ assert effective_wled_status(_state("SLICING")) == "prepare"
|
|
|
+ assert effective_wled_status(_state("PAUSE")) == "paused"
|
|
|
+ assert effective_wled_status(_state("FINISH")) == "finished"
|
|
|
+ assert effective_wled_status(_state("FAILED")) == "error"
|
|
|
+ assert effective_wled_status(_state("IDLE"), awaiting_plate_clear=True, require_plate_clear=True) == "queue_waiting"
|
|
|
+ assert effective_wled_status(_state("PAUSE", stg_cur=6)) == "filament_problem"
|
|
|
+ assert effective_wled_status(_state("PAUSE", stg_cur=16)) == "paused"
|
|
|
+ assert effective_wled_status(_state("PAUSE", stg_cur=16, ams_status_main=1, mc_print_sub_stage=3)) == "paused"
|
|
|
+ assert (
|
|
|
+ effective_wled_status(_state("PAUSE", stg_cur=6, ams_status_main=0, mc_print_sub_stage=0)) == "filament_problem"
|
|
|
+ )
|
|
|
+ fault = HMSError(code="0300_0008", attr=0x03000200, module=3, severity=1, description="Nozzle temperature abnormal")
|
|
|
+ assert effective_wled_status(_state("IDLE", hms_errors=[fault])) == "hms_error"
|
|
|
+ assert effective_wled_status(_state("RUNNING", connected=False)) == "offline"
|
|
|
+ assert effective_wled_status(_state("UNKNOWN")) is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ ("state", "awaiting_plate_clear", "require_plate_clear", "expected"),
|
|
|
+ [
|
|
|
+ ("FINISH", True, False, "finished"),
|
|
|
+ ("FAILED", True, False, "error"),
|
|
|
+ ("IDLE", True, False, "idle"),
|
|
|
+ ("FINISH", True, True, "queue_waiting"),
|
|
|
+ ("FAILED", True, True, "queue_waiting"),
|
|
|
+ ("IDLE", True, True, "queue_waiting"),
|
|
|
+ ("FINISH", False, True, "finished"),
|
|
|
+ ("FAILED", False, True, "error"),
|
|
|
+ ("RUNNING", True, True, "printing"),
|
|
|
+ ("PRINTING", True, True, "printing"),
|
|
|
+ ("PREPARE", True, True, "prepare"),
|
|
|
+ ("SLICING", True, True, "prepare"),
|
|
|
+ ("PAUSE", True, True, "paused"),
|
|
|
+ ("UNKNOWN", True, True, None),
|
|
|
+ ],
|
|
|
+)
|
|
|
+def test_plate_gate_precedes_only_idle_finished_and_error(state, awaiting_plate_clear, require_plate_clear, expected):
|
|
|
+ assert (
|
|
|
+ effective_wled_status(
|
|
|
+ _state(state), awaiting_plate_clear=awaiting_plate_clear, require_plate_clear=require_plate_clear
|
|
|
+ )
|
|
|
+ == expected
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ ("severity", "description", "actions", "full_code", "expected"),
|
|
|
+ [
|
|
|
+ (0, "Invalid level", ["OK_BUTTON"], "0500000000004038", "printing"),
|
|
|
+ (2, None, [], "0C0001000002001B", "printing"),
|
|
|
+ (3, "The top cover is open", [], "0300970000030001", "printing"),
|
|
|
+ (1, "Nozzle temperature abnormal", [], "0300020000010008", "hms_error"),
|
|
|
+ (2, "Filament fault", [], "0500060000020005", "hms_error"),
|
|
|
+ (3, None, ["OK_BUTTON"], "0300970000030001", "hms_error"),
|
|
|
+ (3, "Unable to start drying", [], "1880C003", "hms_error"),
|
|
|
+ ],
|
|
|
+)
|
|
|
+def test_hms_filter_matches_notifications(severity, description, actions, full_code, expected):
|
|
|
+ fault = HMSError(
|
|
|
+ code="fault", attr=0, module=0, severity=severity, description=description, actions=actions, full_code=full_code
|
|
|
+ )
|
|
|
+ assert effective_wled_status(_state("RUNNING", hms_errors=[fault])) == expected
|
|
|
+
|
|
|
+
|
|
|
+def test_specific_fault_priorities_are_preserved_with_plate_gate():
|
|
|
+ ignored = HMSError(code="ignored", attr=0, module=0, severity=0)
|
|
|
+ fault = HMSError(code="fault", attr=0, module=0, severity=2, description="Filament fault")
|
|
|
+ for state in ("RUNNING", "FINISH", "FAILED", "IDLE", "PAUSE"):
|
|
|
+ assert (
|
|
|
+ effective_wled_status(
|
|
|
+ _state(state, hms_errors=[ignored, fault]), awaiting_plate_clear=True, require_plate_clear=True
|
|
|
+ )
|
|
|
+ == "hms_error"
|
|
|
+ )
|
|
|
+ assert (
|
|
|
+ effective_wled_status(
|
|
|
+ _state("PAUSE", stg_cur=6, hms_errors=[fault]), awaiting_plate_clear=True, require_plate_clear=True
|
|
|
+ )
|
|
|
+ == "filament_problem"
|
|
|
+ )
|
|
|
+ assert effective_wled_status(_state("PAUSE", stg_cur=16, hms_errors=[fault])) == "hms_error"
|
|
|
+ assert (
|
|
|
+ effective_wled_status(
|
|
|
+ _state("FINISH", connected=False, hms_errors=[fault]), awaiting_plate_clear=True, require_plate_clear=True
|
|
|
+ )
|
|
|
+ == "offline"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_list_presets_returns_only_valid_named_slots():
|
|
|
+ manager, client = _manager(
|
|
|
+ lambda request: httpx.Response(
|
|
|
+ 200,
|
|
|
+ json={"0": {}, "3": {"n": "Printing Blue"}, "7": {"n": ""}, "251": {"n": "No"}},
|
|
|
+ )
|
|
|
+ )
|
|
|
+ assert await manager.list_presets("http://wled.local") == [
|
|
|
+ {"id": 3, "name": "Printing Blue"},
|
|
|
+ {"id": 7, "name": "7"},
|
|
|
+ ]
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ "response",
|
|
|
+ [httpx.Response(200, text="not-json"), httpx.Response(200, json=[])],
|
|
|
+)
|
|
|
+async def test_list_presets_rejects_invalid_responses(response):
|
|
|
+ manager, client = _manager(lambda request: response)
|
|
|
+ with pytest.raises(WLEDResponseError):
|
|
|
+ await manager.list_presets("http://wled.local")
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_get_info_returns_optional_wled_identity_and_rejects_invalid_response():
|
|
|
+ manager, client = _manager(lambda request: httpx.Response(200, json={"name": " Kitchen LEDs ", "ver": "0.15.0"}))
|
|
|
+ assert await manager.get_info("http://wled.local") == {"name": "Kitchen LEDs", "version": "0.15.0"}
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+ invalid, invalid_client = _manager(lambda request: httpx.Response(200, json=[]))
|
|
|
+ with pytest.raises(WLEDResponseError):
|
|
|
+ await invalid.get_info("http://wled.local")
|
|
|
+ await invalid_client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_send_preset_posts_expected_payload_and_handles_failures(caplog):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+ requests = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ requests.append(request)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ assert await manager.send_preset(1, "http://wled.local", 3) is True
|
|
|
+ assert requests[0].url == httpx.URL("http://wled.local/json/state")
|
|
|
+ assert requests[0].content == b'{"ps":3}'
|
|
|
+ assert (logging.INFO, "[WLED] Printer 1 state=manual -> preset=3") in [
|
|
|
+ (record.levelno, record.message) for record in caplog.records
|
|
|
+ ]
|
|
|
+ caplog.clear()
|
|
|
+
|
|
|
+ for response in (httpx.Response(500), httpx.Response(200, text="bad")):
|
|
|
+ failing, failing_client = _manager(lambda request, result=response: result)
|
|
|
+ assert await failing.send_preset(1, "http://wled.local", 3) is False
|
|
|
+ await failing_client.aclose()
|
|
|
+
|
|
|
+ offline, offline_client = _manager(lambda request: (_ for _ in ()).throw(httpx.ConnectError("offline")))
|
|
|
+ assert await offline.send_preset(1, "http://wled.local", 3) is False
|
|
|
+ records = [record for record in caplog.records if record.name == "backend.app.services.wled"]
|
|
|
+ assert len(records) == 3
|
|
|
+ assert all(record.levelno == logging.WARNING and "preset=3 failed" in record.message for record in records)
|
|
|
+ assert all("wled.local" not in record.message for record in records)
|
|
|
+ await offline_client.aclose()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_status_logging_follows_successful_sends_without_poll_duplicates(caplog):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+ manager, client = _manager(lambda request: httpx.Response(200, json={"success": True}))
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {
|
|
|
+ "enabled": True,
|
|
|
+ "base_url": "http://wled.local",
|
|
|
+ "presets": {"printing": 32, "finished": 34, "queue_waiting": 36},
|
|
|
+ },
|
|
|
+ )
|
|
|
+ manager.require_plate_clear = True
|
|
|
+ for state, awaiting in (("RUNNING", False), ("FINISH", False), ("FINISH", True)):
|
|
|
+ manager.handle_status(1, _state(state), awaiting_plate_clear=awaiting)
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ manager.handle_status(1, _state(state), awaiting_plate_clear=awaiting)
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert [
|
|
|
+ (record.levelno, record.message) for record in caplog.records if record.name == "backend.app.services.wled"
|
|
|
+ ] == [
|
|
|
+ (logging.INFO, "[WLED] Printer 1 state=printing -> preset=32"),
|
|
|
+ (logging.INFO, "[WLED] Printer 1 state=finished -> preset=34"),
|
|
|
+ (logging.INFO, "[WLED] Printer 1 state=queue_waiting -> preset=36"),
|
|
|
+ ]
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_cancelled_preset_send_does_not_log_a_switch_or_failure(caplog):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+
|
|
|
+ async def handler(request):
|
|
|
+ raise asyncio.CancelledError
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ with pytest.raises(asyncio.CancelledError):
|
|
|
+ await manager.send_preset(1, "http://wled.local", 32, status="printing")
|
|
|
+ assert not [record for record in caplog.records if record.name == "backend.app.services.wled"]
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_status_changes_are_deduplicated_and_missing_or_disabled_mappings_do_nothing():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "paused": 7}},
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ manager.handle_status(1, _state("RUNNING", progress=50))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 1
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 2
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("IDLE"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 2
|
|
|
+
|
|
|
+ manager.configure_printer(2, {"enabled": False, "base_url": "http://wled.local", "presets": {"idle": 1}})
|
|
|
+ manager.handle_status(2, _state("IDLE"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 2
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_failed_status_send_retries_after_the_interval_then_deduplicates_success(caplog, monkeypatch):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+ now = [1000.0]
|
|
|
+ monkeypatch.setattr("backend.app.services.wled.time.monotonic", lambda: now[0])
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(503 if len(sent) <= 2 else 200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert len(sent) == 1
|
|
|
+ assert manager._runtime[1].status is None
|
|
|
+ assert manager._runtime[1].preset_id is None
|
|
|
+
|
|
|
+ # Status updates arrive every second or so: none of them retries before
|
|
|
+ # the interval is over.
|
|
|
+ for _ in range(5):
|
|
|
+ now[0] += 1
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 1
|
|
|
+
|
|
|
+ now[0] += 30
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert len(sent) == 2
|
|
|
+
|
|
|
+ now[0] += 30
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert len(sent) == 3
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 3
|
|
|
+ messages = [
|
|
|
+ (record.levelno, record.message) for record in caplog.records if record.name == "backend.app.services.wled"
|
|
|
+ ]
|
|
|
+ # One warning for the whole outage, not one per retry.
|
|
|
+ assert messages == [
|
|
|
+ (logging.WARNING, "[WLED] Printer 1 state=printing -> preset=3 failed (HTTP 503)"),
|
|
|
+ (logging.INFO, "[WLED] Printer 1: retrying every 30s until WLED answers"),
|
|
|
+ (logging.INFO, "[WLED] Printer 1 state=printing -> preset=3"),
|
|
|
+ (logging.INFO, "[WLED] Printer 1: WLED answers again"),
|
|
|
+ ]
|
|
|
+ assert all("wled.local" not in message for _, message in messages)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_a_new_state_is_sent_at_once_during_an_outage_without_a_new_warning(caplog, monkeypatch):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+ monkeypatch.setattr("backend.app.services.wled.time.monotonic", lambda: 1000.0)
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request.content)
|
|
|
+ return httpx.Response(503)
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "paused": 7}}
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert sent == [b'{"ps":3}', b'{"ps":7}']
|
|
|
+ warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
|
|
|
+ assert len(warnings) == 1
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_identical_status_does_not_start_parallel_sends():
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ async def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ started.set()
|
|
|
+ await release.wait()
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ task = manager._runtime[1].send_task
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING", progress=20))
|
|
|
+ manager.handle_status(1, _state("RUNNING", progress=30))
|
|
|
+ assert manager._runtime[1].send_task is task
|
|
|
+ assert manager._runtime[1].status is None
|
|
|
+ assert len(sent) == 1
|
|
|
+
|
|
|
+ release.set()
|
|
|
+ await task
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_failed_return_to_previously_confirmed_state_still_retries():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(503 if len(sent) == 2 else 200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.retry_seconds = 0
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "paused": 7}},
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert manager._runtime[1].status == "printing"
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert len(sent) == 2
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+ assert manager._runtime[1].confirmed_generation != manager._runtime[1].generation
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert len(sent) == 3
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_manual_preset_restores_automatic_mapping_once_on_next_status():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request.content)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+
|
|
|
+ assert await manager.test_preset(1, "http://wled.local", 11) is True
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert sent == [b'{"ps":3}', b'{"ps":11}', b'{"ps":3}']
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_failed_manual_preset_keeps_automatic_mapping_retryable():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request.content)
|
|
|
+ return httpx.Response(503 if request.content == b'{"ps":11}' else 200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+
|
|
|
+ assert await manager.test_preset(1, "http://wled.local", 11) is False
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert sent == [b'{"ps":3}', b'{"ps":11}', b'{"ps":3}']
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("printing", 3)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_manual_preset_invalidates_slow_automatic_send(monkeypatch):
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ async def delayed_send(printer_id, base_url, preset_id, *, status=None, quiet=False):
|
|
|
+ sent.append(preset_id)
|
|
|
+ if preset_id == 3 and sent.count(3) == 1:
|
|
|
+ started.set()
|
|
|
+ try:
|
|
|
+ await release.wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ await release.wait()
|
|
|
+ return True
|
|
|
+
|
|
|
+ manager = WLEDManager()
|
|
|
+ monkeypatch.setattr(manager, "send_preset", delayed_send)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ old_task = manager._runtime[1].send_task
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+
|
|
|
+ assert await manager.test_preset(1, "http://wled.local", 11) is True
|
|
|
+ release.set()
|
|
|
+ await old_task
|
|
|
+ assert manager._runtime[1].status is None
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert sent == [3, 11, 3]
|
|
|
+ assert manager._runtime[1].status == "printing"
|
|
|
+ await manager.shutdown()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_status_change_during_manual_send_restores_new_mapping(monkeypatch):
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ async def delayed_send(printer_id, base_url, preset_id, *, status=None, quiet=False):
|
|
|
+ sent.append(preset_id)
|
|
|
+ if preset_id == 11:
|
|
|
+ started.set()
|
|
|
+ await release.wait()
|
|
|
+ return True
|
|
|
+
|
|
|
+ manager = WLEDManager()
|
|
|
+ monkeypatch.setattr(manager, "send_preset", delayed_send)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "paused": 7}},
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+
|
|
|
+ manual_task = asyncio.create_task(manager.test_preset(1, "http://wled.local", 11))
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert manager._runtime[1].status == "paused"
|
|
|
+
|
|
|
+ release.set()
|
|
|
+ assert await manual_task is True
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert sent == [3, 11, 7, 7]
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("paused", 7)
|
|
|
+ await manager.shutdown()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_offline_lifecycle_send_and_reconnect_restore_mapping():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request.content)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "offline": 9}}
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ manager.handle_offline(1)
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("offline", 9)
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert sent == [b'{"ps":3}', b'{"ps":9}', b'{"ps":3}']
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_offline_lifecycle_skips_disabled_or_missing_preset():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request.content)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.handle_offline(1)
|
|
|
+ manager.configure_printer(1, {"enabled": False, "base_url": "http://wled.local", "presets": {"offline": 9}})
|
|
|
+ manager.handle_offline(1)
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3}})
|
|
|
+ manager.handle_offline(1)
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert sent == []
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_offline_lifecycle_http_failure_does_not_block_and_is_retryable(caplog):
|
|
|
+ manager, client = _manager(lambda request: httpx.Response(503, json={"error": "offline"}))
|
|
|
+ manager.retry_seconds = 0
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": {"offline": 9}})
|
|
|
+ manager.handle_offline(1)
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert manager._runtime[1].confirmed_generation is None
|
|
|
+ assert "state=offline -> preset=9 failed (HTTP 503)" in caplog.text
|
|
|
+ manager.handle_offline(1)
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_late_automatic_send_cannot_confirm_over_offline(monkeypatch):
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ async def delayed_send(printer_id, base_url, preset_id, *, status=None, quiet=False):
|
|
|
+ sent.append(preset_id)
|
|
|
+ if preset_id == 3:
|
|
|
+ started.set()
|
|
|
+ try:
|
|
|
+ await release.wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ await release.wait()
|
|
|
+ return True
|
|
|
+
|
|
|
+ manager = WLEDManager()
|
|
|
+ monkeypatch.setattr(manager, "send_preset", delayed_send)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1, {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "offline": 9}}
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ old_task = manager._runtime[1].send_task
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+ manager.handle_offline(1)
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ release.set()
|
|
|
+ await old_task
|
|
|
+ assert sent == [3, 9]
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("offline", 9)
|
|
|
+ await manager.shutdown()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_late_old_send_cannot_confirm_over_new_state(monkeypatch):
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+
|
|
|
+ async def delayed_send(printer_id, base_url, preset_id, *, status=None, quiet=False):
|
|
|
+ if preset_id == 3:
|
|
|
+ started.set()
|
|
|
+ try:
|
|
|
+ await release.wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ await release.wait()
|
|
|
+ return True
|
|
|
+
|
|
|
+ manager = WLEDManager()
|
|
|
+ monkeypatch.setattr(manager, "send_preset", delayed_send)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {"enabled": True, "base_url": "http://wled.local", "presets": {"printing": 3, "paused": 7}},
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ old_task = manager._runtime[1].send_task
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("PAUSE"))
|
|
|
+ assert old_task.cancelled() or old_task.cancelling()
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("paused", 7)
|
|
|
+
|
|
|
+ release.set()
|
|
|
+ await old_task
|
|
|
+ assert (manager._runtime[1].status, manager._runtime[1].preset_id) == ("paused", 7)
|
|
|
+ await manager.shutdown()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ ("raw_state", "stage", "hms", "gate", "required", "presets", "expected_state", "expected_preset"),
|
|
|
+ [
|
|
|
+ ("PAUSE", 6, False, False, True, {"paused": 7}, "paused", 7),
|
|
|
+ ("PAUSE", 6, True, False, True, {"hms_error": 8, "paused": 7}, "hms_error", 8),
|
|
|
+ ("PAUSE", 16, False, False, True, {"filament_problem": 6, "paused": 7}, "paused", 7),
|
|
|
+ ("RUNNING", -1, True, False, True, {"printing": 3}, "printing", 3),
|
|
|
+ ("FINISH", -1, False, True, False, {"queue_waiting": 5, "finished": 9}, "finished", 9),
|
|
|
+ ("FINISH", -1, False, True, True, {"queue_waiting": 5, "finished": 9}, "queue_waiting", 5),
|
|
|
+ ("FINISH", -1, False, True, True, {"finished": 9}, "finished", 9),
|
|
|
+ ("FAILED", -1, False, True, True, {"error": 10}, "error", 10),
|
|
|
+ ("IDLE", -1, False, True, True, {"idle": 1}, "idle", 1),
|
|
|
+ ("FINISH", -1, False, True, True, {"idle": 1}, None, None),
|
|
|
+ ("RUNNING", -1, False, False, True, {"idle": 1}, None, None),
|
|
|
+ ],
|
|
|
+)
|
|
|
+async def test_missing_preset_falls_back_only_to_applicable_states(
|
|
|
+ raw_state, stage, hms, gate, required, presets, expected_state, expected_preset
|
|
|
+):
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.require_plate_clear = required
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": "http://wled.local", "presets": presets})
|
|
|
+ fault = HMSError(code="fault", attr=0, module=0, severity=1, description="Nozzle fault")
|
|
|
+ manager.handle_status(
|
|
|
+ 1,
|
|
|
+ _state(raw_state, stg_cur=stage, hms_errors=[fault] if hms else []),
|
|
|
+ awaiting_plate_clear=gate,
|
|
|
+ )
|
|
|
+ await asyncio.sleep(0)
|
|
|
+
|
|
|
+ assert [request.content for request in sent] == (
|
|
|
+ [f'{{"ps":{expected_preset}}}'.encode()] if expected_preset else []
|
|
|
+ )
|
|
|
+ assert manager._runtime[1].status == expected_state
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+def test_invalid_persisted_config_is_treated_as_disabled():
|
|
|
+ manager = WLEDManager()
|
|
|
+ manager.configure_printer(1, {"enabled": True, "base_url": None})
|
|
|
+ assert manager._configs[1].enabled is False
|
|
|
+
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {"enabled": True, "base_url": "http://169.254.169.254/latest/meta-data", "presets": {"printing": 3}},
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ assert manager._configs[1].enabled is False
|
|
|
+ assert manager._runtime == {}
|
|
|
+ assert manager._client is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_finished_timeout_confirms_idle_until_new_print_cycle(caplog):
|
|
|
+ caplog.set_level(logging.INFO, logger="backend.app.services.wled")
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {
|
|
|
+ "enabled": True,
|
|
|
+ "base_url": "http://wled.local",
|
|
|
+ "presets": {"finished": 9, "idle": 1, "printing": 3},
|
|
|
+ "finished_timeout_seconds": 10,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ runtime = manager._runtime[1]
|
|
|
+ await manager._finish_timeout(1, runtime.generation, "http://wled.local", 1, 0)
|
|
|
+ assert [request.content for request in sent] == [b'{"ps":9}', b'{"ps":1}']
|
|
|
+ assert "[WLED] Printer 1 state=idle -> preset=1" in caplog.messages
|
|
|
+ assert (runtime.status, runtime.preset_id) == ("idle", 1)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("FINISH", progress=100))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert len(sent) == 2
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("IDLE"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert [request.content for request in sent].count(b'{"ps":9}') == 1
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert sent[-1].content == b'{"ps":3}'
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ assert [request.content for request in sent].count(b'{"ps":9}') == 2
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_finished_timeout_cancels_when_printing_resumes():
|
|
|
+ manager, client = _manager(lambda request: httpx.Response(200, json={"success": True}))
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {
|
|
|
+ "enabled": True,
|
|
|
+ "base_url": "http://wled.local",
|
|
|
+ "presets": {"finished": 9, "idle": 1, "printing": 3},
|
|
|
+ "finished_timeout_seconds": 10,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ timer = manager._runtime[1].finished_task
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ assert timer is not None and timer.cancelling()
|
|
|
+ await manager._runtime[1].send_task
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_failed_finished_timeout_idle_send_is_retryable():
|
|
|
+ sent = []
|
|
|
+
|
|
|
+ def handler(request: httpx.Request):
|
|
|
+ sent.append(request)
|
|
|
+ return httpx.Response(503 if request.content == b'{"ps":1}' and len(sent) == 2 else 200, json={"success": True})
|
|
|
+
|
|
|
+ manager, client = _manager(handler)
|
|
|
+ manager.retry_seconds = 0
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {
|
|
|
+ "enabled": True,
|
|
|
+ "base_url": "http://wled.local",
|
|
|
+ "presets": {"finished": 9, "idle": 1},
|
|
|
+ "finished_timeout_seconds": 10,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ runtime = manager._runtime[1]
|
|
|
+ await runtime.send_task
|
|
|
+ runtime.finished_task.cancel()
|
|
|
+ runtime.finished_task = asyncio.create_task(
|
|
|
+ manager._finish_timeout(1, runtime.generation, "http://wled.local", 1, 0)
|
|
|
+ )
|
|
|
+ await runtime.finished_task
|
|
|
+ assert (runtime.status, runtime.preset_id) == ("finished", 9)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ await runtime.finished_task
|
|
|
+ assert [request.content for request in sent] == [b'{"ps":9}', b'{"ps":1}', b'{"ps":1}']
|
|
|
+ assert (runtime.status, runtime.preset_id) == ("idle", 1)
|
|
|
+ await manager.shutdown()
|
|
|
+ await client.aclose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_late_finished_timeout_idle_cannot_confirm_over_new_printing(monkeypatch):
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+
|
|
|
+ async def delayed_send(printer_id, base_url, preset_id, *, status=None, quiet=False):
|
|
|
+ if preset_id == 1:
|
|
|
+ started.set()
|
|
|
+ try:
|
|
|
+ await release.wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ await release.wait()
|
|
|
+ return True
|
|
|
+
|
|
|
+ manager = WLEDManager()
|
|
|
+ monkeypatch.setattr(manager, "send_preset", delayed_send)
|
|
|
+ manager.configure_printer(
|
|
|
+ 1,
|
|
|
+ {
|
|
|
+ "enabled": True,
|
|
|
+ "base_url": "http://wled.local",
|
|
|
+ "presets": {"finished": 9, "idle": 1, "printing": 3},
|
|
|
+ "finished_timeout_seconds": 10,
|
|
|
+ },
|
|
|
+ )
|
|
|
+ manager.handle_status(1, _state("FINISH"))
|
|
|
+ runtime = manager._runtime[1]
|
|
|
+ await runtime.send_task
|
|
|
+ runtime.finished_task.cancel()
|
|
|
+ old_timeout = asyncio.create_task(manager._finish_timeout(1, runtime.generation, "http://wled.local", 1, 0))
|
|
|
+ runtime.finished_task = old_timeout
|
|
|
+ await asyncio.wait_for(started.wait(), 1)
|
|
|
+
|
|
|
+ manager.handle_status(1, _state("RUNNING"))
|
|
|
+ await runtime.send_task
|
|
|
+ assert (runtime.status, runtime.preset_id) == ("printing", 3)
|
|
|
+ release.set()
|
|
|
+ await old_timeout
|
|
|
+ assert (runtime.status, runtime.preset_id) == ("printing", 3)
|
|
|
+ await manager.shutdown()
|