public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
bambuddy/backend/tests/unit/test_vp_mqtt_bridge.py
T
maziggy 9d35a4e8c6 Mark four test-side bandit false positives 
The PR gate reported four new alerts, all in test code. A fixture's
/tmp/x.3mf is a column value the migration under test UPDATEs, not a
path anything opens. Two f-string statements interpolate column names
from a literal list declared two lines above, with the id bound - a
column name cannot be a bind parameter, which is why it is written into
the string at all. The two joins move onto their own lines because a
trailing marker would have taken line 64 past the 120-character limit.

The fourth is a near miss rather than a finding: the line below it
already carries the marker, as do four other wildcard sites in the same
file. The wildcard is what that test exists to assert about.
2026-08-29 15:25:26 +02:00

2113 lines
87 KiB
Python
Raw Permalink Blame History

"""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_a2l_ams_lite_slots_survive_in_slicer_cache(self):
"""#2697 (reported by @qoatzelcoat): every A2L slot rendered as "?" in
BambuStudio through the VP, while Bambuddy's own AMS card was correct.
The A2L reports its AMS Lite as physical unit id 16 but packs the
presence bits at base 24. Bambuddy's internal path normalises 16 -> 6
before the cleanup runs, so it read the right bits; the bridge parses
the raw printer payload itself and still held 16, so the cleanup read
bits 64-67 — never set — and wiped all four slots in the cache the
slicer reads. A slicer-side filament pick reverted on the next 1 Hz
push for the same reason.
The cached units must keep the physical id 16: BambuStudio addresses
the Lite as 16 (it sends `ams_get_rfid {ams_id: 16}` through the VP).
"""
server = _make_server()
bridge = _make_bridge(server)
await bridge.start()
# Reporter's capture: tray_exist_bits 0x7000000 = bits 24/25/26 →
# slots 0, 1, 2 loaded, slot 3 empty.
bridge._on_printer_raw(
f"device/{H2D_SERIAL}/report",
json.dumps(
{
"print": {
"command": "push_status",
"ams": {
"ams": [
{
"id": "16",
"tray": [
{
"id": "0",
"state": 3,
"tray_type": "PLA",
"tray_sub_brands": "PLA Basic",
"tray_color": "C12E1FFF",
"tray_info_idx": "GFA00",
"remain": 100,
},
{"id": "1", "state": 3, "tray_type": "PETG", "tray_info_idx": "GFG00"},
{"id": "2", "state": 3, "tray_type": "ABS", "tray_info_idx": "GFB00"},
{"id": "3", "state": 3, "tray_type": "TPU", "tray_info_idx": "GFU00"},
],
}
],
"tray_exist_bits": "7000000",
},
}
}
).encode(),
)
await asyncio.sleep(0.01)
cached = bridge.get_latest_print_state()
unit = cached["ams"]["ams"][0]
# The slicer-facing cache keeps the PHYSICAL id — BambuStudio speaks 16.
assert unit["id"] == "16"
trays = unit["tray"]
assert trays[0]["tray_type"] == "PLA", "loaded slot wrongly cleared (bit base 64 regression)"
assert trays[1]["tray_type"] == "PETG"
assert trays[2]["tray_type"] == "ABS"
assert trays[0]["tray_info_idx"] == "GFA00"
# Slot 3 is genuinely empty and still gets the normal cleanup.
assert trays[3]["state"] == 9
assert trays[3]["tray_type"] == ""
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_clears_empty_ams_ht_unit(self):
"""AMS-HT (id 128-135) presence rides bit 16+(ams_id-128), so the bridge
cache clears an empty HT slot just like the internal AMS card — keeping
the slicer-facing view in sync (#1726, #2670). Loaded (bit 16 set) is
preserved; empty (bit 16 clear) is wiped.
"""
server = _make_server()
bridge = _make_bridge(server)
await bridge.start()
def _push(tray_exist_bits: str) -> dict:
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": tray_exist_bits,
"power_on_flag": True,
},
}
}
).encode(),
)
# Loaded: bit 16 set → HT slot preserved.
_push("10000")
await asyncio.sleep(0.01)
assert bridge.get_latest_print_state()["ams"]["ams"][0]["tray"][0]["tray_type"] == "PLA"
# Empty: bit 16 clear → HT slot wiped so the slicer sees no phantom spool.
_push("0")
await asyncio.sleep(0.01)
assert bridge.get_latest_print_state()["ams"]["ams"][0]["tray"][0]["tray_type"] == ""
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
# ---------------------------------------------------------------------------
# VIRTUAL_PRINTER_ADVERTISE_ADDRESS override (#2930)
# ---------------------------------------------------------------------------
class TestAdvertiseAddressOverride:
"""#2930: behind NAT — Docker bridge networking being the case that
prompted it — no local interface carries the address a slicer uses to
reach Bambuddy, so both the bind address and the auto-resolved host
interface put a container-private IP into `net.info[].ip` and the
slicer's FTP upload goes nowhere. The env override supplies that address
directly. It is opt-in precisely so that every install which does not
set it keeps the behaviour it has today.
"""
ENV = "VIRTUAL_PRINTER_ADVERTISE_ADDRESS"
HOST_IP = "192.168.1.50"
CONTAINER_IP = "172.24.0.2"
@staticmethod
def _bound_bridge(bind_address: str) -> MQTTBridge:
"""A bridge with its target client already attached, so
`_refresh_ip_encoding` reaches IP resolution instead of returning
early on an unbound client."""
bridge = _make_bridge(_make_server(bind_address=bind_address))
bridge._target_client = _make_paho_client()
return bridge
@pytest.mark.asyncio
async def test_override_wins_over_an_explicit_bind_address(self, monkeypatch):
"""The bind address is the container IP; the slicer has to be told the
host IP or its FTP connection has nowhere to land."""
monkeypatch.setenv(self.ENV, self.HOST_IP)
bridge = _make_bridge(_make_server(bind_address=self.CONTAINER_IP))
await bridge.start()
try:
payload = json.dumps(
{
"print": {
"command": "push_status",
"net": {"info": [{"ip": _ip_to_uint32_le(H2D_IP), "mask": 0xFFFFFF}]},
}
}
).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
await asyncio.sleep(0.01)
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(self.HOST_IP)
encoded = bridge.get_latest_print_state()["net"]["info"][0]["ip"]
assert encoded == _ip_to_uint32_le(self.HOST_IP)
# Decoded independently of the helper that produced it: asserting
# both sides with `_ip_to_uint32_le` would agree even if the byte
# order were wrong, and the slicer reads this field as LE.
assert socket.inet_ntoa(encoded.to_bytes(4, "little")) == self.HOST_IP
finally:
await bridge.stop()
@pytest.mark.asyncio
async def test_override_wins_over_auto_resolve(self, monkeypatch):
"""A VP left on the default 0.0.0.0 bind would otherwise auto-resolve a
host interface — inside a bridge-network container that resolves to the
container's own address, or to nothing at all."""
monkeypatch.setenv(self.ENV, self.HOST_IP)
bridge = self._bound_bridge("0.0.0.0") # nosec B104
with patch(
"backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target",
return_value=self.CONTAINER_IP,
):
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(self.HOST_IP)
def test_unset_leaves_the_bind_address_untouched(self, monkeypatch):
"""The whole point of making this opt-in: with nothing set, resolution
is byte-for-byte what it was before the override existed."""
monkeypatch.delenv(self.ENV, raising=False)
bridge = self._bound_bridge(VP_IP)
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP)
def test_invalid_override_falls_back_instead_of_disarming(self, caplog, monkeypatch):
"""A typo must not un-arm the rewrite. Unarmed means the *real printer
IP* reaches the slicer (#1429) — strictly worse than the wrong VP IP,
so an unusable override degrades to the bind address."""
monkeypatch.setenv(self.ENV, "192.168.1")
with caplog.at_level(logging.WARNING, logger="backend.app.services.virtual_printer.mqtt_bridge"):
bridge = self._bound_bridge(VP_IP)
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP)
warnings = [r for r in caplog.records if self.ENV in r.getMessage()]
assert len(warnings) == 1, "operator gets exactly one line naming the bad value"
assert "192.168.1" in warnings[0].getMessage()
def test_hostname_override_falls_back(self, monkeypatch):
"""`net.info[].ip` is a uint32 — a hostname cannot round-trip through
it, and resolving one here would pick an address the operator did not
choose. Fall back rather than guess."""
monkeypatch.setenv(self.ENV, "bambuddy.local")
bridge = self._bound_bridge(VP_IP)
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP)
def test_whitespace_only_override_is_ignored(self, monkeypatch):
"""`VIRTUAL_PRINTER_ADVERTISE_ADDRESS=` in a compose file is how people
leave a variable listed but unused — not a configuration error."""
monkeypatch.setenv(self.ENV, " ")
bridge = self._bound_bridge(VP_IP)
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP)
def test_surrounding_whitespace_is_tolerated(self, monkeypatch):
"""Copy-pasted compose values carry stray spaces; that should not cost
the user a warning and a silent fall back to the wrong address."""
monkeypatch.setenv(self.ENV, f" {self.HOST_IP} ")
bridge = self._bound_bridge(self.CONTAINER_IP)
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(self.HOST_IP)
def test_wildcard_override_falls_through_to_auto_resolve(self, monkeypatch):
"""0.0.0.0 is a bind address, never a destination — treat it as unset
rather than encoding it and sending the slicer to 0.0.0.0."""
monkeypatch.setenv(self.ENV, "0.0.0.0") # nosec B104
bridge = self._bound_bridge("0.0.0.0") # nosec B104
with patch(
"backend.app.services.virtual_printer.mqtt_bridge._resolve_host_interface_for_target",
return_value=VP_IP,
):
bridge._refresh_ip_encoding()
assert bridge._vp_ip_uint32_le == _ip_to_uint32_le(VP_IP)
def test_armed_log_names_the_override_as_the_source(self, caplog, monkeypatch):
"""An operator reading the log has to be able to tell where the
advertised IP came from, otherwise a stale variable is invisible."""
monkeypatch.setenv(self.ENV, self.HOST_IP)
bridge = self._bound_bridge(self.CONTAINER_IP)
with caplog.at_level(logging.INFO, logger="backend.app.services.virtual_printer.mqtt_bridge"):
bridge._refresh_ip_encoding()
armed = [r for r in caplog.records if "IP encoding armed" in r.getMessage()]
assert len(armed) == 1
assert self.ENV in armed[0].getMessage()
assert self.HOST_IP in armed[0].getMessage()
def test_not_armed_message_points_at_the_override(self, caplog, monkeypatch):
"""The bridge-network install that has NOT set the variable is exactly
the one that cannot auto-resolve — the diagnostic has to name the
remedy, or the operator sees only that nothing works."""
monkeypatch.delenv(self.ENV, raising=False)
bridge = self._bound_bridge("0.0.0.0") # nosec B104
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._refresh_ip_encoding()
not_armed = [r for r in caplog.records if "NOT armed" in r.getMessage()]
assert len(not_armed) == 1
assert self.ENV in not_armed[0].getMessage()
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
Reference in New Issue View Git Blame Copy Permalink