"""Tests for the VP MQTT bridge — non-proxy mirror of target printer state to slicer.""" import asyncio import json import logging import socket import time from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest from backend.app.services.virtual_printer.mqtt_bridge import ( MQTTBridge, _ip_to_uint32_le, _resolve_host_interface_for_target, _resolve_target_to_ipv4, ) from backend.app.services.virtual_printer.mqtt_server import ( _UPLOAD_SETTLE_SECONDS, SimpleMQTTServer, ) H2D_SERIAL = "0948BB540200427" VP_SERIAL = "09400A391800003" H2D_IP = "192.168.255.133" VP_IP = "192.168.255.16" def _make_server(serial: str = VP_SERIAL, bind_address: str = VP_IP) -> SimpleMQTTServer: return SimpleMQTTServer( serial=serial, access_code="deadbeef", cert_path=Path("/tmp/unused.crt"), # nosec B108 key_path=Path("/tmp/unused.key"), # nosec B108 model="O1D", bind_address=bind_address, ) def _make_paho_client( serial: str = H2D_SERIAL, ip: str = H2D_IP, *, connected: bool = True, ) -> MagicMock: """Build a mock BambuMQTTClient that satisfies MQTTBridge's interface.""" client = MagicMock() client.serial_number = serial client.ip_address = ip client.state = MagicMock() client.state.connected = connected client.publish_raw = MagicMock(return_value=True) client._raw_handlers: list = [] def _register(handler): client._raw_handlers.append(handler) def _unregister(handler): if handler in client._raw_handlers: client._raw_handlers.remove(handler) client.register_raw_message_handler.side_effect = _register client.unregister_raw_message_handler.side_effect = _unregister # No-op for _request_version / request_status_update so the post-bind nudge doesn't crash. client._request_version = MagicMock() client.request_status_update = MagicMock() return client def _make_printer_manager(client) -> MagicMock: pm = MagicMock() pm.get_client = MagicMock(return_value=client) return pm def _make_bridge(server: SimpleMQTTServer, target: MagicMock | None = None) -> MQTTBridge: target = target if target is not None else _make_paho_client() pm = _make_printer_manager(target) return MQTTBridge( vp_id=1, vp_name="vp1", vp_serial=VP_SERIAL, target_printer_id=42, mqtt_server=server, printer_manager=pm, ) # --------------------------------------------------------------------------- # Lifecycle # --------------------------------------------------------------------------- class TestBridgeLifecycle: @pytest.mark.asyncio async def test_start_registers_handler_on_target_client(self): target = _make_paho_client() bridge = _make_bridge(_make_server(), target) await bridge.start() assert len(target._raw_handlers) == 1 assert bridge.is_active is True await bridge.stop() assert len(target._raw_handlers) == 0 @pytest.mark.asyncio async def test_start_with_no_target_client_does_not_crash(self): pm = MagicMock() pm.get_client = MagicMock(return_value=None) bridge = MQTTBridge( vp_id=1, vp_name="vp1", vp_serial=VP_SERIAL, target_printer_id=42, mqtt_server=_make_server(), printer_manager=pm, ) await bridge.start() assert bridge.is_active is False await bridge.stop() @pytest.mark.asyncio async def test_resolve_rebinds_when_paho_client_replaced(self): """BambuMQTTClient is destroyed and recreated on connect_printer; bridge must rebind.""" old_client = _make_paho_client(serial="REAL_OLD") new_client = _make_paho_client(serial="REAL_NEW") pm = _make_printer_manager(old_client) bridge = MQTTBridge( vp_id=1, vp_name="vp1", vp_serial=VP_SERIAL, target_printer_id=42, mqtt_server=_make_server(), printer_manager=pm, ) await bridge.start() assert len(old_client._raw_handlers) == 1 assert bridge._target_serial == "REAL_OLD" pm.get_client.return_value = new_client bridge._resolve_client() assert len(old_client._raw_handlers) == 0 assert len(new_client._raw_handlers) == 1 assert bridge._target_serial == "REAL_NEW" await bridge.stop() @pytest.mark.asyncio async def test_post_bind_nudge_requests_version_and_status(self): target = _make_paho_client() bridge = _make_bridge(_make_server(), target) await bridge.start() target._request_version.assert_called_once() target.request_status_update.assert_called_once() await bridge.stop() @pytest.mark.asyncio async def test_post_bind_nudge_skipped_when_target_not_connected(self): """#1721: the bridge can attach before the real printer's MQTT TLS handshake completes. Calling request_status_update on a disconnected client logs WARNING (bambu_mqtt.py:3224); on A1 firmware that reconnects aggressively, every bind cycle pollutes the support bundle with a benign line. The bridge must check state.connected before nudging — the next periodic pushall picks up the cache anyway. """ target = _make_paho_client(connected=False) bridge = _make_bridge(_make_server(), target) await bridge.start() target._request_version.assert_not_called() target.request_status_update.assert_not_called() await bridge.stop() # --------------------------------------------------------------------------- # Caching: push_status # --------------------------------------------------------------------------- class TestPushStatusCache: """push_status snapshots feed `_send_status_report` via the cache, not a fan-out.""" @pytest.mark.asyncio async def test_push_status_is_cached_not_fanned_out(self): server = _make_server() server.push_raw_to_clients = AsyncMock() bridge = _make_bridge(server) await bridge.start() payload = json.dumps({"print": {"command": "push_status", "ams": {"ams": []}, "gcode_state": "IDLE"}}).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) server.push_raw_to_clients.assert_not_awaited() cached = bridge.get_latest_print_state() assert cached is not None assert cached["command"] == "push_status" assert cached["gcode_state"] == "IDLE" await bridge.stop() @pytest.mark.asyncio async def test_serial_rewritten_in_cached_push(self): server = _make_server() bridge = _make_bridge(server) await bridge.start() payload = json.dumps( { "print": { "command": "push_status", "upgrade_state": {"sn": H2D_SERIAL, "status": "IDLE"}, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["upgrade_state"]["sn"] == VP_SERIAL await bridge.stop() @pytest.mark.asyncio async def test_net_info_ip_rewritten_to_vp_ip(self): """BambuStudio reads `net.info[].ip` (LE uint32) for the FTP destination — must be rewritten to the VP's bind IP or the slicer bypasses the VP.""" server = _make_server(bind_address=VP_IP) bridge = _make_bridge(server) await bridge.start() h2d_le = _ip_to_uint32_le(H2D_IP) vp_le = _ip_to_uint32_le(VP_IP) payload = json.dumps( { "print": { "command": "push_status", "net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}, {"ip": 0, "mask": 0}]}, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["net"]["info"][0]["ip"] == vp_le assert cached["net"]["info"][1]["ip"] == 0 # untouched await bridge.stop() @pytest.mark.asyncio async def test_net_info_ip_rewritten_for_unknown_secondary_interface(self): """Regression for #1429: real printers (X1C / H2D Pro) report multiple active interfaces (WiFi + Ethernet) — only ONE matches the IP Bambuddy tracks. The rewrite must catch every non-zero entry, not just the one whose IP equals `_target_ip_uint32_le`, or the slicer's FTP fallback path leaks straight to the real printer.""" server = _make_server(bind_address=VP_IP) bridge = _make_bridge(server) await bridge.start() h2d_le = _ip_to_uint32_le(H2D_IP) # A second IP Bambuddy never saw (e.g. printer's ethernet interface # while Bambuddy talks over wifi). other_le = _ip_to_uint32_le("192.168.99.42") vp_le = _ip_to_uint32_le(VP_IP) payload = json.dumps( { "print": { "command": "push_status", "net": { "info": [ {"ip": h2d_le, "mask": 0xFFFFFF}, {"ip": other_le, "mask": 0xFFFFFF}, {"ip": 0, "mask": 0}, ] }, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["net"]["info"][0]["ip"] == vp_le assert cached["net"]["info"][1]["ip"] == vp_le # secondary interface also rewritten assert cached["net"]["info"][2]["ip"] == 0 # placeholder untouched await bridge.stop() @pytest.mark.asyncio async def test_late_arriving_printer_ip_rewrites_existing_cache(self): """Regression for #1429: if the printer's `ip_address` is empty at first bind (DB row stale, or the client object exists before the first SSDP refresh fills it in), the rewrite stays disabled and the first cached push poisons the cache with the real-printer IP. Once `ip_address` becomes valid, the next refresh tick must (a) arm the encoding and (b) sweep the cached `net.info[].ip` so the slicer sees the rewritten value on its next pull. Without the sweep the sticky-key preservation keeps the poisoned value alive across every subsequent incremental push.""" server = _make_server(bind_address=VP_IP) # Bind to a client whose ip_address is empty at start — simulates the # late-arrival path. target = _make_paho_client(ip="") bridge = _make_bridge(server, target) await bridge.start() assert bridge._target_ip_uint32_le is None # not yet armed h2d_le = _ip_to_uint32_le(H2D_IP) vp_le = _ip_to_uint32_le(VP_IP) payload = json.dumps( { "print": { "command": "push_status", "net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}]}, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) # First push landed before encoding was armed → cache holds real IP. cached = bridge.get_latest_print_state() assert cached["net"]["info"][0]["ip"] == h2d_le # Printer's IP becomes known. Next refresh tick must self-heal. target.ip_address = H2D_IP bridge._resolve_client() cached = bridge.get_latest_print_state() assert cached["net"]["info"][0]["ip"] == vp_le, ( "cache must be swept once encoding becomes valid; sticky-key " "preservation would otherwise keep the poisoned IP forever" ) assert bridge._target_ip_uint32_le == h2d_le await bridge.stop() @pytest.mark.asyncio async def test_request_topic_message_is_ignored(self): server = _make_server() bridge = _make_bridge(server) await bridge.start() payload = json.dumps({"print": {"command": "push_status"}}).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/request", payload) await asyncio.sleep(0.01) assert bridge.get_latest_print_state() is None await bridge.stop() @pytest.mark.asyncio async def test_incremental_push_preserves_ams_from_previous_cache(self): """Regression for #1371: Bambu firmware sends FULL push_status on pushall (with AMS/vt_tray/net/etc.) but typically OMITS those fields from 1 Hz incremental push_status updates. Without preserving the sticky keys across pushes, the cache forgets AMS info after the first incremental update, and BambuStudio (which reads the cache via the VP's 1 Hz status push) sees no AMS info until the user power-cycles the printer (forcing a fresh pushall). """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Initial pushall response with full state, AMS included. full_push = json.dumps( { "print": { "command": "push_status", "gcode_state": "IDLE", "wifi_signal": "-50dBm", "ams": { "ams": [ { "id": "0", "tray": [ {"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"}, {"id": "1", "tray_type": "PETG", "tray_color": "00FF00FF"}, ], } ], "tray_exist_bits": "3", }, "vt_tray": {"id": "254", "tray_type": ""}, "lights_report": [{"node": "chamber_light", "mode": "on"}], } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", full_push) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["ams"]["ams"][0]["tray"][0]["tray_type"] == "PLA" assert cached["vt_tray"]["id"] == "254" assert cached["lights_report"][0]["mode"] == "on" # 2. Incremental push with only temp/wifi changes — NO ams field. # This is what the printer sends every ~1 s between full pushalls. incremental_push = json.dumps( { "print": { "command": "push_status", "wifi_signal": "-55dBm", "chamber_temper": 26.0, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", incremental_push) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() # New fields take effect. assert cached["wifi_signal"] == "-55dBm" assert cached["chamber_temper"] == 26.0 # Sticky fields preserved from the previous cache (the #1371 fix). assert "ams" in cached, "AMS field must be preserved across incremental pushes (#1371)" assert cached["ams"]["ams"][0]["tray"][0]["tray_type"] == "PLA" assert cached["ams"]["tray_exist_bits"] == "3" assert cached["vt_tray"]["id"] == "254" assert cached["lights_report"][0]["mode"] == "on" await bridge.stop() @pytest.mark.asyncio async def test_incremental_push_preserves_non_allowlisted_capability_fields(self): """Regression for #1622: BambuStudio gates Device-tab UIs (manage calibration, AMS-slot filament dropdown, ...) on capability / lifecycle fields (cali_version, print_type, mc_print_stage, device, ...) it reads off the cached push_status. Before the fix these fields were not in the allowlist and drained out of the bridge cache on the first 1 Hz incremental tick, so the slicer's Device tab would grey out the gated UIs once the cache thinned. After the fix the cache accumulates everything the printer has ever sent, dropped only when explicitly overwritten. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() full_push = json.dumps( { "print": { "command": "push_status", "cali_version": 2, "print_type": "idle", "gcode_state": "IDLE", "mc_print_stage": "0", "mc_stage": 0, "device": {"ext_tool": {"info": []}}, "cfg": "", "home_flag": 256, "wifi_signal": "-50dBm", } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", full_push) await asyncio.sleep(0.01) # Incremental push carrying only temps + wifi — none of the # capability/lifecycle fields above are mentioned. incremental_push = json.dumps( { "print": { "command": "push_status", "wifi_signal": "-55dBm", "nozzle_temper": 24.5, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", incremental_push) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() # Incremental values applied. assert cached["wifi_signal"] == "-55dBm" assert cached["nozzle_temper"] == 24.5 # Capability / lifecycle fields preserved from the prior pushall # — the symptoms in #1622 (Device-tab UIs disabled) trace to these # exact keys missing. assert cached["cali_version"] == 2 assert cached["print_type"] == "idle" assert cached["gcode_state"] == "IDLE" assert cached["mc_print_stage"] == "0" assert cached["mc_stage"] == 0 assert cached["device"] == {"ext_tool": {"info": []}} assert cached["cfg"] == "" assert cached["home_flag"] == 256 await bridge.stop() @pytest.mark.asyncio async def test_partial_vt_tray_update_overlays_onto_cached_full_dict(self): """Regression for #1622 round 5 (reported by @shaddowlink): right after the slicer picks a filament for the external spool (vt_tray, ams_id=255), Bambu firmware pushes a partial vt_tray carrying just the changed fields — typically ``{tray_info_idx, tray_color}`` — and omits the ~18 other keys (tray_type, state, k, n, cali_idx, nozzle_temp_min/max, tray_uuid, xcam_info, ...) the slicer needs to render the slot. Before this fix the per-field accumulate replaced the cached vt_tray wholesale (it only carried over prev keys NOT present in new), so the next 1 Hz cached-as-base push handed the slicer a stripped vt_tray and BambuStudio rendered the external slot as "invalid" until a reload triggered a fresh pushall. AMS slots didn't suffer because `_merge_ams_dict` already deep-merged them. The fix overlays incoming keys onto the previous dict for every top-level dict-shaped field (excluding ams, which keeps its own deep merge). """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Pushall response with the full ~20-field vt_tray dict a real # P1S sends to bootstrap the slot. full_push = json.dumps( { "print": { "command": "push_status", "vt_tray": { "id": "254", "tray_info_idx": "Pea5f68f", "tray_type": "PLA", "tray_sub_brands": "", "tray_color": "F72323FF", "tray_weight": "0", "tray_diameter": "0.00", "tray_temp": "0", "tray_time": "0", "bed_temp_type": "0", "bed_temp": "0", "nozzle_temp_max": "240", "nozzle_temp_min": "190", "xcam_info": "000000000000000000000000", "tray_uuid": "00000000000000000000000000000000", "ctype": 0, "remain": -1, "k": 0.01999999955296, "n": 1, "cali_idx": -1, "state": 3, }, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", full_push) await asyncio.sleep(0.01) # 2. Incremental push carrying just the two fields the slicer's pick # changed — exactly the shape the P1S firmware sends after an # ams_filament_setting ack. This is what shaddowlink's wire dump # captured for the failing case. incremental_push = json.dumps( { "print": { "command": "push_status", "vt_tray": { "tray_info_idx": "Pea5f68f", "tray_color": "76D9F4FF", }, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", incremental_push) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() vt = cached["vt_tray"] # Incoming fields applied. assert vt["tray_info_idx"] == "Pea5f68f" assert vt["tray_color"] == "76D9F4FF" # All other fields preserved from the prior pushall — without these # the slicer rendered the slot as invalid. assert vt["tray_type"] == "PLA" assert vt["state"] == 3 assert vt["remain"] == -1 assert vt["k"] == 0.01999999955296 assert vt["n"] == 1 assert vt["cali_idx"] == -1 assert vt["nozzle_temp_min"] == "190" assert vt["nozzle_temp_max"] == "240" assert vt["tray_uuid"] == "00000000000000000000000000000000" assert vt["id"] == "254" await bridge.stop() @pytest.mark.asyncio async def test_partial_ams_status_update_preserves_unit_list(self): """#1387: Bambu firmware also sends `ams` updates where the key is present but the inner `ams` array is missing — e.g. just ``{ams_status: 1}`` or a humidity change. Before the deep-merge fix the bridge would overwrite the cached AMS with this stripped blob, the slicer would read it on the next 1 Hz push, and BambuStudio would drop the unit list and fall back to its "no AMS" render (only the external spool visible — the reporter's exact symptom). Now the partial update only mutates the fields it carries; the cached unit list survives. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Pushall with full AMS state. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ { "id": "0", "humidity": "1", "tray": [{"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"}], } ], "tray_exist_bits": "1", "ams_status": "0", }, } } ).encode(), ) await asyncio.sleep(0.01) # 2. Partial AMS update — only `ams_status` and `humidity` changed. # No `ams.ams` array, so prev's unit list must be preserved. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": {"ams_status": "1", "humidity": "2"}, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() # Scalar fields take the new values. assert cached["ams"]["ams_status"] == "1" assert cached["ams"]["humidity"] == "2" # Unit + tray data preserved from the pushall. assert cached["ams"]["tray_exist_bits"] == "1" assert len(cached["ams"]["ams"]) == 1 assert cached["ams"]["ams"][0]["tray"][0]["tray_type"] == "PLA" assert cached["ams"]["ams"][0]["tray"][0]["tray_color"] == "FF0000FF" await bridge.stop() @pytest.mark.asyncio async def test_partial_ams_unit_update_preserves_other_units(self): """#1387: when multiple AMS units are configured (e.g. H2D with two AMS), an incremental push during a print typically only carries the unit / tray that changed state. Naive replacement of `ams.ams` wipes the other unit. The bridge merges unit-by-unit by id, preserving units the incremental doesn't mention. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Pushall with two AMS units configured. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ {"id": "0", "tray": [{"id": "0", "tray_type": "PLA"}]}, {"id": "1", "tray": [{"id": "0", "tray_type": "PETG"}]}, ], # bit 0 (AMS 0 slot 0) + bit 4 (AMS 1 slot 0) = 0x11. # `_on_printer_raw` now applies the #1726 bitmask # cleanup to the cached state, so the test fixture # must declare both loaded slots — same shape the # real printer sends. "tray_exist_bits": "11", }, } } ).encode(), ) await asyncio.sleep(0.01) # 2. Tray-targeted incremental: unit 0 / tray 0 state changed. # Unit 1 is not in the update — must survive. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": {"ams": [{"id": "0", "tray": [{"id": "0", "state": "11"}]}]}, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() units = {u["id"]: u for u in cached["ams"]["ams"]} # Unit 0 keeps its tray_type from the pushall + picks up the new state. assert units["0"]["tray"][0]["tray_type"] == "PLA" assert units["0"]["tray"][0]["state"] == "11" # Unit 1 survives the incremental. assert "1" in units assert units["1"]["tray"][0]["tray_type"] == "PETG" await bridge.stop() @pytest.mark.asyncio async def test_tray_exist_bits_clears_empty_slots_in_slicer_cache(self): """#1726 (reported by @needo37): the bridge cache forwards the real printer's raw AMS payload to the slicer. Without the empty-slot cleanup that bambu_mqtt.py applies to Bambuddy's internal state, the cached units carried stale `tray_type` / `tray_color` / `tray_info_idx` for slots whose `tray_exist_bits` bit was 0 — and BambuStudio's Sync rendered those empty slots as phantom loaded filaments. After the fix the bridge runs the same shared ``apply_tray_exist_bits`` helper before storing the cache. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # Pushall: AMS 0 has slots 0/1/2/3; only slots 1, 2, 3 are loaded. # Slot 0 carries stale data (RFID/color/material from a previously # loaded spool). `tray_exist_bits` = 0xe = 0b1110 → bit 0 unset. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ { "id": "0", "tray": [ { "id": "0", "tray_type": "PLA", "tray_color": "FF0000FF", "tray_info_idx": "GFL00", "tag_uid": "1234567890abcdef", "tray_uuid": "abcdef1234567890abcdef1234567890", "remain": 75, "state": "11", }, {"id": "1", "tray_type": "PETG", "tray_color": "00FF00FF"}, {"id": "2", "tray_type": "ABS", "tray_color": "0000FFFF"}, {"id": "3", "tray_type": "TPU", "tray_color": "FFFF00FF"}, ], } ], "tray_exist_bits": "e", }, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() slot0 = cached["ams"]["ams"][0]["tray"][0] # Empty slot: stale per-tray fields wiped, state promoted to 9. assert slot0["state"] == 9, "empty slot must be promoted to state=9" assert slot0["tray_type"] == "" assert slot0["tray_color"] == "" assert slot0["tray_info_idx"] == "" assert slot0["tag_uid"] == "0000000000000000" assert slot0["tray_uuid"] == "00000000000000000000000000000000" assert slot0["remain"] == 0 # Loaded slots preserved. assert cached["ams"]["ams"][0]["tray"][1]["tray_type"] == "PETG" assert cached["ams"]["ams"][0]["tray"][2]["tray_type"] == "ABS" assert cached["ams"]["ams"][0]["tray"][3]["tray_type"] == "TPU" await bridge.stop() @pytest.mark.asyncio async def test_tray_exist_bits_shutdown_guard_preserves_cache(self): """#765 shutdown guard mirrored at the bridge: when the printer powers off it sends all-zero `tray_exist_bits` paired with `power_on_flag=False`. Wiping the cache on that pattern would propagate phantom empties to every slicer reconnect until the printer powers back on and pushes a real state. Skip cleanup on the shutdown-shaped payload.""" server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Normal pushall — all four slots loaded. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ { "id": "0", "tray": [ {"id": str(i), "tray_type": "PLA", "tray_color": f"{i:02x}{i:02x}{i:02x}FF"} for i in range(4) ], } ], "tray_exist_bits": "f", "power_on_flag": True, }, } } ).encode(), ) await asyncio.sleep(0.01) # 2. Shutdown-shaped push: tray_exist_bits=0 + power_on_flag=False. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "tray_exist_bits": "0", "power_on_flag": False, }, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() for i in range(4): assert cached["ams"]["ams"][0]["tray"][i]["tray_type"] == "PLA", f"slot {i} must survive the shutdown push" await bridge.stop() @pytest.mark.asyncio async def test_tray_exist_bits_skips_ams_ht_units(self): """AMS-HT units (id >= 128) use a separate addressing scheme and must not be touched by the bitmask cleanup — bit math at global_bit = ams_id * 4 + tray_id would overrun normal AMS bits. Pin the skip so future AMS-HT support doesn't accidentally wipe loaded HT slots. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ { "id": "128", "tray": [ {"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"}, ], } ], "tray_exist_bits": "0", "power_on_flag": True, }, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() ht_slot = cached["ams"]["ams"][0]["tray"][0] # tray_exist_bits="0" alone would normally wipe — but AMS-HT is # skipped, so the HT slot keeps its loaded data. assert ht_slot["tray_type"] == "PLA" await bridge.stop() @pytest.mark.asyncio async def test_partial_ams_tray_update_preserves_other_trays(self): """Same shape as the unit-level test but at the tray level. AMS unit 0 has four trays; the incremental only mentions tray 0. Trays 1-3 must survive intact.""" server = _make_server() bridge = _make_bridge(server) await bridge.start() bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": { "ams": [ { "id": "0", "tray": [ {"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"}, {"id": "1", "tray_type": "PETG", "tray_color": "00FF00FF"}, {"id": "2", "tray_type": "ABS", "tray_color": "0000FFFF"}, {"id": "3", "tray_type": "TPU", "tray_color": "FFFF00FF"}, ], } ], }, } } ).encode(), ) await asyncio.sleep(0.01) bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": {"ams": [{"id": "0", "tray": [{"id": "0", "state": "11"}]}]}, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() trays = {t["id"]: t for t in cached["ams"]["ams"][0]["tray"]} assert trays["0"]["tray_type"] == "PLA" assert trays["0"]["state"] == "11" # Trays not mentioned in the incremental survive intact. assert trays["1"]["tray_type"] == "PETG" assert trays["2"]["tray_type"] == "ABS" assert trays["3"]["tray_type"] == "TPU" await bridge.stop() @pytest.mark.asyncio async def test_incoming_ams_update_replaces_cached_ams(self): """Counterpart to the #1371 fix: preservation only kicks in when the incoming push OMITS a sticky key. When the printer DOES send a fresh `ams` value (e.g. on a pushall, or when AMS state genuinely changes), that value must take effect — the preservation must not shadow real updates. """ server = _make_server() bridge = _make_bridge(server) await bridge.start() # 1. Initial state: PLA in tray 0. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": {"ams": [{"id": "0", "tray": [{"id": "0", "tray_type": "PLA"}]}]}, } } ).encode(), ) await asyncio.sleep(0.01) # 2. Fresh push with PETG — must replace, not get shadowed by the old PLA. bridge._on_printer_raw( f"device/{H2D_SERIAL}/report", json.dumps( { "print": { "command": "push_status", "ams": {"ams": [{"id": "0", "tray": [{"id": "0", "tray_type": "PETG"}]}]}, } } ).encode(), ) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["ams"]["ams"][0]["tray"][0]["tray_type"] == "PETG" await bridge.stop() # --------------------------------------------------------------------------- # Caching: get_version response # --------------------------------------------------------------------------- class TestVersionCache: @pytest.mark.asyncio async def test_get_version_response_caches_modules(self): server = _make_server() bridge = _make_bridge(server) await bridge.start() payload = json.dumps( { "info": { "command": "get_version", "module": [ {"name": "ota", "sn": H2D_SERIAL, "sw_ver": "01.03.00.00"}, {"name": "n3f/0", "sn": "AMS_HW_1", "sw_ver": "04.00.21.87"}, ], } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) modules = bridge.get_latest_version_modules() assert modules is not None assert len(modules) == 2 # Device-level sn rewritten; AMS-hardware sn left alone. assert modules[0]["sn"] == VP_SERIAL assert modules[1]["sn"] == "AMS_HW_1" await bridge.stop() # --------------------------------------------------------------------------- # Selective fan-out (everything that's not push_status / get_version) # --------------------------------------------------------------------------- class TestCommandResponseFanout: @pytest.mark.asyncio async def test_extrusion_cali_get_response_is_fanned_out(self): """Slicer's extrusion_cali_get goes to the printer; the printer's response must reach the slicer or BambuStudio's pre-flight blocks Send.""" server = _make_server() server.push_raw_to_clients = AsyncMock() bridge = _make_bridge(server) await bridge.start() body = json.dumps({"print": {"command": "extrusion_cali_get", "filaments": []}}).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", body) await asyncio.sleep(0.01) server.push_raw_to_clients.assert_awaited_once() topic, _payload = server.push_raw_to_clients.await_args.args assert topic == f"device/{VP_SERIAL}/report" await bridge.stop() # --------------------------------------------------------------------------- # Forwarding: slicer → printer # --------------------------------------------------------------------------- class TestForwardToPrinter: @pytest.mark.asyncio async def test_forward_publishes_to_real_serial_request_topic(self): target = _make_paho_client() bridge = _make_bridge(_make_server(), target) await bridge.start() ok = bridge.forward_to_printer({"print": {"command": "stop"}}) assert ok is True target.publish_raw.assert_called_once() topic, payload = target.publish_raw.call_args.args assert topic == f"device/{H2D_SERIAL}/request" assert json.loads(payload) == {"print": {"command": "stop"}} await bridge.stop() @pytest.mark.asyncio async def test_forward_returns_false_when_not_bound(self): pm = MagicMock() pm.get_client = MagicMock(return_value=None) bridge = MQTTBridge( vp_id=1, vp_name="vp1", vp_serial=VP_SERIAL, target_printer_id=42, mqtt_server=_make_server(), printer_manager=pm, ) await bridge.start() assert bridge.forward_to_printer({"print": {"command": "stop"}}) is False await bridge.stop() # --------------------------------------------------------------------------- # SimpleMQTTServer status response: cached-as-base # --------------------------------------------------------------------------- def _capture_published(server: SimpleMQTTServer) -> list: """Wrap _publish_to_report to capture (topic, payload_dict).""" published: list = [] async def _capture(writer, payload, serial="", log_event=True): published.append((serial or server.serial, payload)) server._publish_to_report = _capture # type: ignore[assignment] return published class TestStatusReportCachedAsBase: """`_send_status_report` sends near-byte-identical real data when bridge cache exists.""" def _capture_published(self, server: SimpleMQTTServer): return _capture_published(server) @pytest.mark.asyncio async def test_uses_real_cache_when_bridge_active(self): server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = { "command": "push_status", "msg": 0, "ams": {"ams": [{"id": "0"}]}, "device": {"extruder": {"info": [{"id": 0}, {"id": 1}]}}, "nozzle_diameter": "0.4", "nozzle_type": "HH01", # real H2D value, not synthetic 'hardened_steel' } server.set_bridge(bridge) published = self._capture_published(server) await server._send_status_report(MagicMock()) assert len(published) == 1 _serial, payload = published[0] # AMS / device / nozzle_type all from cache assert payload["print"]["nozzle_type"] == "HH01" assert payload["print"]["device"]["extruder"]["info"][1]["id"] == 1 # Protocol fields under our control assert payload["print"]["command"] == "push_status" assert payload["print"]["gcode_state"] == "IDLE" @pytest.mark.asyncio async def test_falls_back_to_synthetic_when_no_cache(self): server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = None server.set_bridge(bridge) published = self._capture_published(server) await server._send_status_report(MagicMock()) assert len(published) == 1 _serial, payload = published[0] # Synthetic baseline has stub fields like nozzle_type='hardened_steel' # and a `storage` field that the real H2D doesn't push. assert payload["print"]["nozzle_type"] == "hardened_steel" assert "storage" in payload["print"] @pytest.mark.asyncio async def test_storage_indicators_overlaid_for_send_preflight(self): """#1228: P1S/A1-class firmware doesn't always include the SD/storage fields BambuStudio's "Send" pre-flight reads. Without these the slicer rejects with 'storage needs to be inserted' before even attempting FTP. The cached-as-base path now overlays them so the pre-flight passes regardless of what the real printer reports. """ server = _make_server() bridge = MagicMock() # Real P1S push without SD card inserted: home_flag has other bits set # but the SD bit (0x100) is clear; sdcard is False; no storage field. bridge.get_latest_print_state.return_value = { "command": "push_status", "msg": 0, "home_flag": 0x42, "sdcard": False, } server.set_bridge(bridge) published = self._capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] # SD bit ORed onto whatever was there — other bits preserved. assert payload["print"]["home_flag"] & 0x100 == 0x100 assert payload["print"]["home_flag"] & 0x42 == 0x42 # Force-set so a False from the printer doesn't trip the pre-flight. assert payload["print"]["sdcard"] is True # storage was missing — the overlay must inject a non-empty default. assert "storage" in payload["print"] assert payload["print"]["storage"]["free"] > 0 assert payload["print"]["storage"]["total"] > 0 @pytest.mark.asyncio async def test_storage_indicators_preserve_real_storage_when_present(self): """When the real printer DOES report a storage block, pass it through unchanged (the overlay only fills in the missing field, not overrides). """ server = _make_server() bridge = MagicMock() real_storage = {"free": 12345, "total": 67890} bridge.get_latest_print_state.return_value = { "command": "push_status", "msg": 0, "home_flag": 0x100, # SD bit already set on the real printer "sdcard": True, "storage": real_storage, } server.set_bridge(bridge) published = self._capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] # SD bit OR is idempotent — already-set bit stays set. assert payload["print"]["home_flag"] == 0x100 assert payload["print"]["sdcard"] is True # Real values pass through, NOT the synthetic defaults. assert payload["print"]["storage"] == real_storage @pytest.mark.asyncio async def test_overrides_protocol_fields_even_when_cache_present(self): """Cached value's gcode_state must NOT win over our local upload-state-machine value.""" server = _make_server() server._gcode_state = "PREPARE" server._current_file = "foo.3mf" bridge = MagicMock() bridge.get_latest_print_state.return_value = { "command": "push_status", "gcode_state": "IDLE", # printer is idle; we are mid-FTP-upload "gcode_file": "", "gcode_file_prepare_percent": "0", } server.set_bridge(bridge) published = self._capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["gcode_state"] == "PREPARE" assert payload["print"]["gcode_file"] == "foo.3mf" # --------------------------------------------------------------------------- # Live print progress (#1887 / #1558) # --------------------------------------------------------------------------- def _printing_cache(**overrides) -> dict: """Bridge cache for a target printer that is mid-print.""" cache = { "command": "push_status", "msg": 0, "gcode_state": "RUNNING", "gcode_file": "Metadata/plate_1.gcode", "subtask_name": "benchy", "mc_print_stage": "2", "mc_percent": 47, "mc_remaining_time": 3600, "stg": [1, 2, 3], "stg_cur": 14, "layer_num": 120, "total_layer_num": 250, "print_error": 0, } cache.update(overrides) return cache class TestLiveProgressMirror: """The VP mirrors the target printer's progress without ever looking busy. Both slicers gate the Device-tab progress panel and the Send button on the same predicate — `MachineObject::is_in_printing()`, i.e. gcode_state in {RUNNING, PAUSE, SLICING, PREPARE}. Reporting the printer's real state shows progress but blocks Send for as long as it prints (#1558); zeroing everything keeps Send alive but shows nothing (#1887). FINISH is the one state that does both: StatusPanel renders on `is_in_printing() || print_status == "FINISH"`, while SelectMachineDialog only blocks on `is_in_printing()`. """ @pytest.mark.asyncio async def test_progress_mirrored_while_target_prints(self): """#1887: the numbers the slicer needs come straight from the cache.""" server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache() server.set_bridge(bridge) published = _capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["mc_print_stage"] == "2" assert payload["print"]["mc_percent"] == 47 assert payload["print"]["mc_remaining_time"] == 3600 assert payload["print"]["stg"] == [1, 2, 3] assert payload["print"]["stg_cur"] == 14 assert payload["print"]["layer_num"] == 120 assert payload["print"]["total_layer_num"] == 250 # The job the printer is really running, not the VP's last upload. assert payload["print"]["subtask_name"] == "benchy" @pytest.mark.asyncio @pytest.mark.parametrize("target_state", ["RUNNING", "PAUSE"]) async def test_mirror_never_reports_a_printing_gcode_state(self, target_state): """#1558 guard: any state in `is_in_printing()` disables the Send button. This is the assertion that keeps the mirror honest — it may show the printer's numbers, but it must never claim the VP itself is printing. """ server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache(gcode_state=target_state) server.set_bridge(bridge) published = _capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["gcode_state"] == "FINISH" assert payload["print"]["gcode_state"] not in ("RUNNING", "PAUSE", "SLICING", "PREPARE") @pytest.mark.asyncio async def test_progress_zeroed_while_target_idle(self): """Nothing to mirror — the VP's own upload state owns the report.""" server = _make_server() server.set_gcode_state("FINISH", filename="foo.3mf", prepare_percent="100") server._state_changed_at = time.monotonic() - 60 # settled long ago bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache(gcode_state="IDLE", mc_percent=0, layer_num=0) server.set_bridge(bridge) published = _capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["gcode_state"] == "FINISH" # the VP's own, not mirrored assert payload["print"]["subtask_name"] == "foo" assert payload["print"]["mc_print_stage"] == "" assert payload["print"]["stg"] == [] assert payload["print"]["total_layer_num"] == 0 @pytest.mark.asyncio async def test_progress_zeroed_while_upload_in_flight(self): """A job being handed over outranks the mirror. The slicer is watching its own PREPARE → FINISH cycle here; feeding it the printer's progress mid-handshake would contradict the PREPARE it is waiting on. """ server = _make_server() server.set_gcode_state("PREPARE", filename="bar.3mf", prepare_percent="0") bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache() server.set_bridge(bridge) published = _capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["gcode_state"] == "PREPARE" assert payload["print"]["gcode_file"] == "bar.3mf" assert payload["print"]["subtask_name"] == "bar" assert payload["print"]["mc_percent"] == 0 assert payload["print"]["layer_num"] == 0 @pytest.mark.asyncio async def test_upload_settle_window_keeps_the_slicers_own_filename(self): """#1658: the send modal releases on FINISH carrying the name it uploaded. Swapping in the printer's filename while that handshake is still in flight wedges the slicer at "Downloading", so the mirror waits. """ server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache() server.set_bridge(bridge) published = _capture_published(server) server.set_gcode_state("FINISH", filename="bar.3mf", prepare_percent="100") await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["subtask_name"] == "bar" assert payload["print"]["mc_percent"] == 0 @pytest.mark.asyncio async def test_mirror_resumes_once_the_upload_has_settled(self): """Same VP as above, once the slicer has had its FINISH.""" server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache() server.set_bridge(bridge) published = _capture_published(server) server.set_gcode_state("FINISH", filename="bar.3mf", prepare_percent="100") server._state_changed_at = time.monotonic() - _UPLOAD_SETTLE_SECONDS - 1 await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["subtask_name"] == "benchy" assert payload["print"]["mc_percent"] == 47 @pytest.mark.asyncio async def test_print_error_never_mirrored(self): """A fault on the printer must not raise a modal error dialog in the slicer. The VP is not the machine that threw it — Bambuddy's own printer card reports the fault. """ server = _make_server() bridge = MagicMock() bridge.get_latest_print_state.return_value = _printing_cache(print_error=515) server.set_bridge(bridge) published = _capture_published(server) await server._send_status_report(MagicMock()) _serial, payload = published[0] assert payload["print"]["print_error"] == 0 assert payload["print"]["mc_percent"] == 47 # the rest still mirrors # --------------------------------------------------------------------------- # Wire format # --------------------------------------------------------------------------- class TestWireFormat: """BambuStudio's Send pre-flight rejects compact JSON — must match real printer's indented format (32K bytes for an idle H2D vs 14K compact).""" @pytest.mark.asyncio async def test_publish_uses_indent_4_json_format(self): server = _make_server() captured: list = [] async def _capture_drain(): pass writer = MagicMock() writer.write = lambda data: captured.append(data) writer.drain = AsyncMock() await server._publish_to_report(writer, {"print": {"command": "push_status", "ams": {}}}) body = b"".join(captured) assert b'\n "print"' in body, "publish_to_report must use indent=4 JSON" @pytest.mark.asyncio async def test_publish_records_bridge_to_slicer_event_by_default(self, monkeypatch): """#1622 round 3: every bridge-synthesised reply (info.get_version answer, project_file ack, on-demand pushall response) must show up in the cmd.jsonl trace under the ``bridge_to_slicer`` direction so a P1S↔H2D diff captures the fingerprint the slicer reads back from us.""" server = _make_server() writer = MagicMock() writer.write = lambda data: None writer.drain = AsyncMock() recorded: list = [] monkeypatch.setattr( "backend.app.services.virtual_printer.mqtt_server.append_event", lambda vp_name, direction, topic, payload: recorded.append((vp_name, direction, topic, payload)), ) payload = {"info": {"command": "get_version", "sequence_id": "0"}} await server._publish_to_report(writer, payload) assert len(recorded) == 1 vp_name, direction, topic, recorded_payload = recorded[0] assert direction == "bridge_to_slicer" assert topic.endswith("/report") assert recorded_payload == payload @pytest.mark.asyncio async def test_publish_skips_event_when_log_event_false(self, monkeypatch): """The 1Hz periodic-push path passes ``log_event=False`` so dump_wire's snapshot stays the canonical record of cache shape and the cmd.jsonl isn't flooded with ~60 lines/min per VP.""" server = _make_server() writer = MagicMock() writer.write = lambda data: None writer.drain = AsyncMock() recorded: list = [] monkeypatch.setattr( "backend.app.services.virtual_printer.mqtt_server.append_event", lambda *args, **kwargs: recorded.append(args), ) await server._publish_to_report(writer, {"print": {"command": "push_status"}}, log_event=False) assert recorded == [] # --------------------------------------------------------------------------- # Routing: _handle_publish # --------------------------------------------------------------------------- class TestPublishRouting: """Slicer-issued commands: project_file/gcode_file handled locally, everything else forwarded to the real printer.""" def _build_publish_payload(self, topic: str, body: bytes) -> bytes: topic_bytes = topic.encode("utf-8") return bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF]) + topic_bytes + body def _attach_active_bridge(self, server: SimpleMQTTServer) -> MagicMock: bridge = MagicMock() bridge.is_active = True bridge.forward_to_printer = MagicMock(return_value=True) server.set_bridge(bridge) return bridge @pytest.mark.asyncio async def test_project_file_handled_locally_not_forwarded(self): server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps({"print": {"command": "project_file", "subtask_name": "f", "sequence_id": "1"}}).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) with patch.object(server, "_send_print_response", new=AsyncMock()) as mock_resp: await server._handle_publish(0x30, payload, writer, "client1") bridge.forward_to_printer.assert_not_called() mock_resp.assert_awaited_once() @pytest.mark.asyncio async def test_gcode_file_handled_locally_not_forwarded(self): server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps({"print": {"command": "gcode_file", "subtask_name": "f.gcode", "sequence_id": "1"}}).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) with patch.object(server, "_send_print_response", new=AsyncMock()): await server._handle_publish(0x30, payload, writer, "client1") bridge.forward_to_printer.assert_not_called() @pytest.mark.asyncio async def test_pushall_handled_locally_not_forwarded(self): server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps({"pushing": {"command": "pushall", "sequence_id": "0"}}).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) with patch.object(server, "_send_status_report", new=AsyncMock()) as mock_status: await server._handle_publish(0x30, payload, writer, "client1") # Synthetic answer fires (fast, low latency); no forwarding (the # cache already mirrors what the printer would respond with). bridge.forward_to_printer.assert_not_called() mock_status.assert_awaited_once() @pytest.mark.asyncio async def test_get_version_handled_locally_not_forwarded(self): server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps({"info": {"command": "get_version", "sequence_id": "1"}}).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) with patch.object(server, "_send_version_response", new=AsyncMock()) as mock_ver: await server._handle_publish(0x30, payload, writer, "client1") bridge.forward_to_printer.assert_not_called() mock_ver.assert_awaited_once() @pytest.mark.asyncio async def test_extrusion_cali_get_is_forwarded(self): """extrusion_cali_get fetches per-filament k-profiles — must reach the printer.""" server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps( { "print": { "command": "extrusion_cali_get", "filament_id": "", "nozzle_diameter": "0.4", "sequence_id": "5", } } ).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) await server._handle_publish(0x30, payload, writer, "client1") bridge.forward_to_printer.assert_called_once() forwarded = bridge.forward_to_printer.call_args.args[0] assert forwarded["print"]["command"] == "extrusion_cali_get" @pytest.mark.asyncio async def test_print_stop_is_forwarded(self): server = _make_server() bridge = self._attach_active_bridge(server) writer = MagicMock() writer.write = MagicMock() writer.drain = AsyncMock() body = json.dumps({"print": {"command": "stop", "sequence_id": "5"}}).encode() payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body) await server._handle_publish(0x30, payload, writer, "client1") bridge.forward_to_printer.assert_called_once() # --------------------------------------------------------------------------- # IP encoding helper # --------------------------------------------------------------------------- class TestIpEncoding: def test_le_uint32_matches_real_h2d_capture(self): # 192.168.255.133 captured from real H2D's net.info[0].ip = 2248124608 assert _ip_to_uint32_le("192.168.255.133") == 2248124608 def test_vp_ip_round_trip(self): assert _ip_to_uint32_le("192.168.255.16") == 285190336 def test_invalid_ip_raises(self): with pytest.raises(ValueError): _ip_to_uint32_le("not.an.ip.actually") class TestHostnameResolution: """#1429 follow-up: users who configured the printer by FQDN (common on LANs with router-provided DNS like `p1s.fritz.box`) hit `invalid IPv4` on the encoder and the rewrite never armed — slicer kept FTPing direct to the real printer. The bridge now resolves hostname→IPv4 first.""" def test_pass_through_for_valid_ipv4(self): assert _resolve_target_to_ipv4("192.168.1.50") == "192.168.1.50" def test_empty_returns_none(self): assert _resolve_target_to_ipv4("") is None assert _resolve_target_to_ipv4(None) is None # type: ignore[arg-type] def test_hostname_resolves_via_getaddrinfo(self): with patch( "backend.app.services.virtual_printer.mqtt_bridge.socket.getaddrinfo", return_value=[(2, 1, 6, "", ("192.168.3.153", 0))], ) as mock_gai: assert _resolve_target_to_ipv4("p1s.fritz.box") == "192.168.3.153" # AF_INET filter prevents an IPv6-only result from being picked, # since net.info[*].ip is a uint32 LE that can't carry v6. assert mock_gai.call_args.kwargs.get("family") == socket.AF_INET def test_dns_failure_returns_none(self): with patch( "backend.app.services.virtual_printer.mqtt_bridge.socket.getaddrinfo", side_effect=OSError("Name or service not known"), ): assert _resolve_target_to_ipv4("nope.invalid") is None def test_fqdn_target_arms_encoding(self, caplog): """End-to-end: a client whose `ip_address` is an FQDN should arm the bridge once DNS resolves, and the cached rewrite uses the resolved IPv4 (not the hostname string) for the `net.info[].ip` encoding.""" server = _make_server(bind_address=VP_IP) bridge = _make_bridge(server) client = _make_paho_client(ip="p1s.fritz.box") bridge._target_client = client with ( patch( "backend.app.services.virtual_printer.mqtt_bridge.socket.getaddrinfo", return_value=[(2, 1, 6, "", (H2D_IP, 0))], ), caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"), ): bridge._refresh_ip_encoding() assert bridge._target_ip_uint32_le == _ip_to_uint32_le(H2D_IP) assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP) armed = [r for r in caplog.records if "MQTT bridge IP encoding armed" in r.getMessage()] assert len(armed) == 1 # Operator should see configured→resolved in the log line so a # bad-DNS regression is immediately legible. assert "p1s.fritz.box→192.168.255.133" in armed[0].getMessage() # --------------------------------------------------------------------------- # Auto-resolve fallback for default-config (bind_address = "0.0.0.0") # --------------------------------------------------------------------------- class TestBindAddressAutoResolve: """#1429 residual: VPs created without a dedicated bind IP run on `bind_address=0.0.0.0`. The original fix's `_refresh_ip_encoding` early-returned on 0.0.0.0, so the rewrite never armed and `net.info[].ip` kept leaking the real printer IP. Now the bridge auto-resolves a host interface in the printer's subnet and uses that as the VP IP.""" @pytest.mark.asyncio async def test_rewrite_arms_via_auto_resolved_host_ip(self): """When bind_address is 0.0.0.0, fall back to the host interface in the target printer's subnet and rewrite to that IP.""" server = _make_server(bind_address="0.0.0.0") # nosec B104 bridge = _make_bridge(server) with patch( "backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target", return_value=VP_IP, ): await bridge.start() h2d_le = _ip_to_uint32_le(H2D_IP) vp_le = _ip_to_uint32_le(VP_IP) payload = json.dumps( { "print": { "command": "push_status", "net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}]}, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) cached = bridge.get_latest_print_state() assert cached["net"]["info"][0]["ip"] == vp_le assert bridge._vp_ip_uint32_le == vp_le await bridge.stop() @pytest.mark.asyncio async def test_rewrite_disabled_when_no_matching_host_interface(self): """If no host interface shares a subnet with the printer, the bridge cannot pick a sensible VP IP — leave encoding unarmed and let the push through unrewritten (no crash, no wrong rewrite).""" server = _make_server(bind_address="") bridge = _make_bridge(server) with patch( "backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target", return_value=None, ): await bridge.start() h2d_le = _ip_to_uint32_le(H2D_IP) payload = json.dumps( { "print": { "command": "push_status", "net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}]}, } } ).encode() bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload) await asyncio.sleep(0.01) assert bridge._vp_ip_uint32_le is None assert bridge._target_ip_uint32_le is None await bridge.stop() @pytest.mark.asyncio async def test_explicit_bind_ip_takes_precedence_over_auto_resolve(self): """Auto-resolve only kicks in when bind_address is empty/0.0.0.0; an explicitly-set bind IP must be used verbatim even if there's also a same-subnet host interface.""" server = _make_server(bind_address=VP_IP) bridge = _make_bridge(server) # Auto-resolver would have returned a DIFFERENT IP — we must not use it. with patch( "backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target", return_value="10.99.99.99", ): await bridge.start() assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP) await bridge.stop() def test_resolve_helper_returns_none_for_unreachable_target(self): """The helper itself must be defensive — if `find_interface_for_ip` raises or returns None, we get None (no crash).""" with patch( "backend.app.services.network_utils.find_interface_for_ip", return_value=None, ): assert _resolve_host_interface_for_target("203.0.113.1") is None class TestNotArmedDiagnosticLogging: """#1429 follow-up: every silent early-return in `_refresh_ip_encoding` now emits one INFO line explaining WHY the rewrite couldn't arm. Throttled to one line per state change so an idle unarmed bridge doesn't spam the log every 30s tick. Cleared on arm so a future failure re-emits. """ def test_no_client_logs_once(self, caplog): bridge = _make_bridge(_make_server()) # Force the "no client" path: bridge starts with _target_client=None. assert bridge._target_client is None with caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"): bridge._refresh_ip_encoding() bridge._refresh_ip_encoding() # 2nd tick — same reason, must NOT re-log. bridge._refresh_ip_encoding() not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()] assert len(not_armed) == 1 assert "target_client is None" in not_armed[0].getMessage() def test_missing_target_ip_logs_specific_reason(self, caplog): bridge = _make_bridge(_make_server()) # Manually attach a client with no ip_address (simulates pre-DHCP). client = _make_paho_client() client.ip_address = "" bridge._target_client = client with caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"): bridge._refresh_ip_encoding() not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()] assert len(not_armed) == 1 assert "no ip_address" in not_armed[0].getMessage() def test_no_matching_host_interface_logs_specific_reason(self, caplog): server = _make_server(bind_address="0.0.0.0") # nosec B104 bridge = _make_bridge(server) with ( patch( "backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target", return_value=None, ), caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"), ): bridge._target_client = _make_paho_client() bridge._refresh_ip_encoding() not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()] assert len(not_armed) == 1 msg = not_armed[0].getMessage() assert H2D_IP in msg assert "no host interface" in msg def test_unresolvable_target_logs_reason(self, caplog): """When `ip_address` isn't a valid IPv4 *and* doesn't resolve via DNS, the bridge must report a single concrete not-armed reason naming the configured value — operator can then see exactly what input failed.""" server = _make_server(bind_address=VP_IP) bridge = _make_bridge(server) client = _make_paho_client() client.ip_address = "not.an.ip" bridge._target_client = client with ( patch( "backend.app.services.virtual_printer.mqtt_bridge.socket.getaddrinfo", side_effect=OSError("nodename nor servname provided"), ), caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"), ): bridge._refresh_ip_encoding() not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()] assert len(not_armed) == 1 assert "could not resolve printer host 'not.an.ip'" in not_armed[0].getMessage() def test_successful_arm_clears_dedup_so_future_failure_relogs(self, caplog): """After a successful arm, the dedup must reset so a subsequent regression (e.g. printer client unbinds) re-emits the diagnostic line instead of being silenced by the previous failure reason.""" bridge = _make_bridge(_make_server(bind_address=VP_IP)) bridge._target_client = _make_paho_client() with caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"): bridge._refresh_ip_encoding() # arms assert bridge._not_armed_reason is None # Simulate a regression — target_client drops away. bridge._target_client = None bridge._refresh_ip_encoding() bridge._refresh_ip_encoding() # 2nd same-reason tick must not re-log not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()] assert len(not_armed) == 1 # the post-arm failure armed = [r for r in caplog.records if "MQTT bridge IP encoding armed" in r.getMessage()] assert len(armed) == 1