Files
bambuddy/backend/tests/unit/services/test_bambu_mqtt.py
T
maziggy 899c2c6480 revert(printers): remove SD card badge entirely
Four attempts at making the printer-card SD badge stable on H2D all failed:
  the final straw was powering on an A1 causing every connected H2D to flip to
  red simultaneously. Bambu firmware SD signaling is not reliably derivable
  from MQTT — the legacy `sdcard` field is sporadic and inconsistently typed,
  and home_flag bits 8-9 are cleared on heartbeat pushes regardless of card
  state with no clean way to distinguish heartbeats from full status reports.

  Remove the badge from the Printers page card and the Printer Info modal,
  drop `sdcard` from the frontend PrinterStatus type, and strip all home_flag
  derivation and heartbeat-handling code from the MQTT parser.

  `state.sdcard` is retained on the backend and populated only from a plain
  truthy read of the `sdcard` field, because firmware_update.py uses it as a
  precondition before starting firmware installs.
2026-04-14 18:41:46 +02:00

3619 lines
138 KiB
Python

"""
Tests for the BambuMQTTClient service.
These tests focus on timelapse tracking during prints.
"""
import json
import pytest
class TestTimelapseTracking:
"""Tests for timelapse state tracking during prints."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_timelapse_flag_initializes_to_false(self, mqtt_client):
"""Verify _timelapse_during_print starts as False."""
assert mqtt_client._timelapse_during_print is False
def test_timelapse_flag_set_when_timelapse_active_during_running(self, mqtt_client):
"""Verify timelapse flag is set when timelapse is active while printing."""
# Simulate print running
mqtt_client._was_running = True
mqtt_client.state.timelapse = False
# Simulate xcam data showing timelapse is enabled
xcam_data = {"timelapse": "enable"}
mqtt_client._parse_xcam_data(xcam_data)
assert mqtt_client.state.timelapse is True
assert mqtt_client._timelapse_during_print is True
def test_timelapse_flag_not_set_when_not_running(self, mqtt_client):
"""Verify timelapse flag is NOT set when printer not running."""
# Printer is idle (not running)
mqtt_client._was_running = False
mqtt_client.state.timelapse = False
# Timelapse is enabled but we're not printing
xcam_data = {"timelapse": "enable"}
mqtt_client._parse_xcam_data(xcam_data)
assert mqtt_client.state.timelapse is True
# Flag should NOT be set since we're not printing
assert mqtt_client._timelapse_during_print is False
def test_timelapse_flag_persists_after_timelapse_stops(self, mqtt_client):
"""Verify timelapse flag stays True even after recording stops."""
# Simulate print running with timelapse
mqtt_client._was_running = True
# Enable timelapse during print
xcam_data = {"timelapse": "enable"}
mqtt_client._parse_xcam_data(xcam_data)
assert mqtt_client._timelapse_during_print is True
# Disable timelapse (recording stops at end of print)
xcam_data = {"timelapse": "disable"}
mqtt_client._parse_xcam_data(xcam_data)
# Flag should still be True (persists until reset)
assert mqtt_client.state.timelapse is False
assert mqtt_client._timelapse_during_print is True
def test_timelapse_flag_from_print_data(self, mqtt_client):
"""Verify timelapse flag is set from print data (not just xcam)."""
# Simulate print running
mqtt_client._was_running = True
mqtt_client.state.timelapse = False
mqtt_client._timelapse_during_print = False
# Manually test the timelapse parsing logic from _parse_print_data
# This tests the "timelapse" field in the main print data
data = {"timelapse": True}
mqtt_client.state.timelapse = data["timelapse"] is True
if mqtt_client.state.timelapse and mqtt_client._was_running:
mqtt_client._timelapse_during_print = True
assert mqtt_client._timelapse_during_print is True
class TestPrintCompletionWithTimelapse:
"""Tests for print completion including timelapse flag."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_print_complete_includes_timelapse_flag(self, mqtt_client):
"""Verify print complete callback includes timelapse_was_active."""
# Set up completion callback
callback_data = {}
def on_complete(data):
callback_data.update(data)
mqtt_client.on_print_complete = on_complete
# Simulate a print that had timelapse active
mqtt_client._was_running = True
mqtt_client._completion_triggered = False
mqtt_client._timelapse_during_print = True
mqtt_client._previous_gcode_state = "RUNNING"
mqtt_client._previous_gcode_file = "test.gcode"
mqtt_client.state.subtask_name = "Test Print"
# Simulate print finish
mqtt_client.state.state = "FINISH"
# Manually trigger the completion logic (simplified)
# In real code this happens in _parse_print_data
should_trigger = (
mqtt_client.state.state in ("FINISH", "FAILED")
and not mqtt_client._completion_triggered
and mqtt_client.on_print_complete
and mqtt_client._previous_gcode_state == "RUNNING"
)
if should_trigger:
status = "completed" if mqtt_client.state.state == "FINISH" else "failed"
timelapse_was_active = mqtt_client._timelapse_during_print
mqtt_client._completion_triggered = True
mqtt_client._was_running = False
mqtt_client._timelapse_during_print = False
mqtt_client.on_print_complete(
{
"status": status,
"filename": mqtt_client._previous_gcode_file,
"subtask_name": mqtt_client.state.subtask_name,
"timelapse_was_active": timelapse_was_active,
}
)
assert "timelapse_was_active" in callback_data
assert callback_data["timelapse_was_active"] is True
def test_print_complete_timelapse_flag_false_when_no_timelapse(self, mqtt_client):
"""Verify timelapse_was_active is False when no timelapse during print."""
callback_data = {}
def on_complete(data):
callback_data.update(data)
mqtt_client.on_print_complete = on_complete
# Print without timelapse
mqtt_client._was_running = True
mqtt_client._completion_triggered = False
mqtt_client._timelapse_during_print = False # No timelapse
mqtt_client._previous_gcode_state = "RUNNING"
mqtt_client._previous_gcode_file = "test.gcode"
mqtt_client.state.subtask_name = "Test Print"
mqtt_client.state.state = "FINISH"
# Trigger completion
timelapse_was_active = mqtt_client._timelapse_during_print
mqtt_client.on_print_complete(
{
"status": "completed",
"filename": mqtt_client._previous_gcode_file,
"subtask_name": mqtt_client.state.subtask_name,
"timelapse_was_active": timelapse_was_active,
}
)
assert callback_data["timelapse_was_active"] is False
def test_timelapse_flag_reset_after_completion(self, mqtt_client):
"""Verify _timelapse_during_print is reset after print completion."""
mqtt_client._timelapse_during_print = True
mqtt_client._was_running = True
mqtt_client._completion_triggered = False
# Simulate completion reset
mqtt_client._completion_triggered = True
mqtt_client._was_running = False
mqtt_client._timelapse_during_print = False
assert mqtt_client._timelapse_during_print is False
class TestRealisticMessageFlow:
"""Tests that simulate realistic MQTT message sequences.
These tests process messages through _process_message to test the full flow,
including the order of xcam parsing vs state detection.
"""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_timelapse_detected_at_print_start_in_same_message(self, mqtt_client):
"""Test that timelapse is detected when xcam and state come in same message.
This is the critical race condition test - xcam data is parsed BEFORE
state detection, so the timelapse flag must be set AFTER _was_running is True.
"""
# Callbacks to track events
start_callback_data = {}
def on_start(data):
start_callback_data.update(data)
mqtt_client.on_print_start = on_start
# Initial state - idle
mqtt_client._was_running = False
mqtt_client._timelapse_during_print = False
mqtt_client._previous_gcode_state = None
# Simulate first message when print starts - contains both xcam and gcode_state
# This is the realistic scenario from the printer
# NOTE: Real MQTT messages wrap print data inside a "print" key
payload = {
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test_print.gcode",
"subtask_name": "Test_Print",
"xcam": {
"timelapse": "enable", # Timelapse is enabled in this print
"printing_monitor": True,
},
"mc_percent": 0,
"mc_remaining_time": 3600,
}
}
# Process the message (this is what happens in real MQTT flow)
mqtt_client._process_message(payload)
# Verify timelapse was detected even though xcam is parsed before state
assert mqtt_client._was_running is True, "_was_running should be True after RUNNING state"
assert mqtt_client.state.timelapse is True, "state.timelapse should be True"
assert mqtt_client._timelapse_during_print is True, (
"timelapse_during_print should be True when timelapse is in the same message as RUNNING state"
)
def test_timelapse_not_detected_when_disabled(self, mqtt_client):
"""Test that timelapse is NOT detected when disabled in xcam data."""
mqtt_client.on_print_start = lambda data: None
# Initial state - idle
mqtt_client._was_running = False
mqtt_client._timelapse_during_print = False
mqtt_client._previous_gcode_state = None
# Print starts without timelapse
payload = {
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test_print.gcode",
"subtask_name": "Test_Print",
"xcam": {
"timelapse": "disable", # Timelapse is disabled
"printing_monitor": True,
},
}
}
mqtt_client._process_message(payload)
assert mqtt_client._was_running is True
assert mqtt_client.state.timelapse is False
assert mqtt_client._timelapse_during_print is False
def test_timelapse_detected_when_enabled_after_print_start(self, mqtt_client):
"""Test timelapse detected when enabled in a message after print starts."""
mqtt_client.on_print_start = lambda data: None
# First message - print starts without timelapse info
payload_start = {
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test_print.gcode",
"subtask_name": "Test_Print",
}
}
mqtt_client._process_message(payload_start)
assert mqtt_client._was_running is True
assert mqtt_client._timelapse_during_print is False # Not detected yet
# Second message - xcam data arrives with timelapse enabled
payload_xcam = {
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test_print.gcode",
"subtask_name": "Test_Print",
"xcam": {
"timelapse": "enable",
},
}
}
mqtt_client._process_message(payload_xcam)
# Now timelapse should be detected because _was_running is already True
assert mqtt_client._timelapse_during_print is True
def test_print_complete_includes_timelapse_flag_full_flow(self, mqtt_client):
"""Test full print lifecycle with timelapse - from start to completion."""
start_data = {}
complete_data = {}
def on_start(data):
start_data.update(data)
def on_complete(data):
complete_data.update(data)
mqtt_client.on_print_start = on_start
mqtt_client.on_print_complete = on_complete
# 1. Print starts with timelapse
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
"xcam": {"timelapse": "enable"},
}
}
)
assert mqtt_client._timelapse_during_print is True
assert "subtask_name" in start_data
# 2. Print continues (multiple messages)
for _ in range(3):
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
"mc_percent": 50,
}
}
)
# Timelapse flag should still be True
assert mqtt_client._timelapse_during_print is True
# 3. Print completes
mqtt_client._process_message(
{
"print": {
"gcode_state": "FINISH",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
# Verify completion callback received timelapse flag
assert "timelapse_was_active" in complete_data
assert complete_data["timelapse_was_active"] is True
assert complete_data["status"] == "completed"
# Flags should be reset after completion
assert mqtt_client._timelapse_during_print is False
assert mqtt_client._was_running is False
def test_print_failed_includes_timelapse_flag(self, mqtt_client):
"""Test that failed print also includes timelapse flag."""
complete_data = {}
def on_complete(data):
complete_data.update(data)
mqtt_client.on_print_start = lambda data: None
mqtt_client.on_print_complete = on_complete
# Start with timelapse
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
"xcam": {"timelapse": "enable"},
}
}
)
# Print fails
mqtt_client._process_message(
{
"print": {
"gcode_state": "FAILED",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
assert complete_data["timelapse_was_active"] is True
assert complete_data["status"] == "failed"
class TestAMSDataMerging:
"""Tests for AMS data merging, particularly handling empty slots."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_empty_slot_clears_tray_type(self, mqtt_client):
"""Test that empty slot update clears tray_type (Issue #147).
When a spool is removed from an old AMS, the printer sends empty values.
These must overwrite the previous values to show the slot as empty.
"""
# Initial state: AMS unit with a loaded spool
initial_ams = {
"ams": [
{
"id": 0,
"tray": [
{
"id": 0,
"tray_type": "PLA",
"tray_sub_brands": "Bambu PLA Basic",
"tray_color": "FF0000",
"tag_uid": "1234567890ABCDEF",
"remain": 80,
}
],
}
]
}
mqtt_client._handle_ams_data(initial_ams)
# Verify initial state
ams_data = mqtt_client.state.raw_data.get("ams", [])
assert len(ams_data) == 1
tray = ams_data[0]["tray"][0]
assert tray["tray_type"] == "PLA"
assert tray["tray_color"] == "FF0000"
# Now simulate spool removal - printer sends empty values
empty_update = {
"ams": [
{
"id": 0,
"tray": [
{
"id": 0,
"tray_type": "", # Empty = slot is empty
"tray_sub_brands": "",
"tray_color": "",
"tag_uid": "0000000000000000", # Zero UID
"remain": 0,
}
],
}
]
}
mqtt_client._handle_ams_data(empty_update)
# Verify empty values were applied (not ignored by merge logic)
ams_data = mqtt_client.state.raw_data.get("ams", [])
tray = ams_data[0]["tray"][0]
assert tray["tray_type"] == "", "tray_type should be cleared when slot is empty"
assert tray["tray_color"] == "", "tray_color should be cleared when slot is empty"
assert tray["tray_sub_brands"] == "", "tray_sub_brands should be cleared"
assert tray["tag_uid"] == "0000000000000000", "tag_uid should be cleared"
def test_partial_update_preserves_other_fields(self, mqtt_client):
"""Test that partial updates still preserve non-slot-status fields."""
# Initial state with full data
initial_ams = {
"ams": [
{
"id": 0,
"humidity": "3",
"temp": "25.5",
"tray": [
{
"id": 0,
"tray_type": "PLA",
"tray_color": "00FF00",
"remain": 90,
"k": 0.02,
}
],
}
]
}
mqtt_client._handle_ams_data(initial_ams)
# Partial update - only remain changes
partial_update = {
"ams": [
{
"id": 0,
"tray": [
{
"id": 0,
"remain": 85, # Only this changed
}
],
}
]
}
mqtt_client._handle_ams_data(partial_update)
# Verify remain was updated but other fields preserved
ams_data = mqtt_client.state.raw_data.get("ams", [])
tray = ams_data[0]["tray"][0]
assert tray["remain"] == 85, "remain should be updated"
assert tray["tray_type"] == "PLA", "tray_type should be preserved"
assert tray["tray_color"] == "00FF00", "tray_color should be preserved"
assert tray["k"] == 0.02, "k should be preserved"
def test_tray_exist_bits_clears_empty_slots(self, mqtt_client):
"""Test that tray_exist_bits clears slots marked as empty (Issue #147).
New AMS models (AMS 2 Pro) don't send empty tray data when a spool is removed.
Instead, they update tray_exist_bits to indicate which slots have spools.
"""
# Initial state: AMS 0 and AMS 1 with loaded spools
initial_ams = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_type": "PLA", "tray_color": "FF0000", "remain": 80},
{"id": 1, "tray_type": "PETG", "tray_color": "00FF00", "remain": 60},
{"id": 2, "tray_type": "ABS", "tray_color": "0000FF", "remain": 40},
{"id": 3, "tray_type": "TPU", "tray_color": "FFFF00", "remain": 20},
],
},
{
"id": 1,
"tray": [
{"id": 0, "tray_type": "PLA", "tray_color": "FFFFFF", "remain": 90},
{"id": 1, "tray_type": "PLA", "tray_color": "000000", "remain": 70},
{"id": 2, "tray_type": "PLA", "tray_color": "FF00FF", "remain": 50},
{"id": 3, "tray_type": "PLA", "tray_color": "00FFFF", "remain": 30},
],
},
],
"tray_exist_bits": "ff", # All 8 slots have spools (0xFF = 11111111)
}
mqtt_client._handle_ams_data(initial_ams)
# Verify initial state
ams_data = mqtt_client.state.raw_data.get("ams", [])
assert ams_data[1]["tray"][3]["tray_type"] == "PLA" # AMS 1 slot 3 (B4) has spool
# Now simulate spool removal from AMS 1 slot 3 (B4)
# tray_exist_bits: 0x7f = 01111111 (bit 7 = 0 means AMS 1 slot 3 is empty)
update_ams = {
"ams": [
{"id": 0, "tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]},
{"id": 1, "tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]},
],
"tray_exist_bits": "7f", # Bit 7 = 0 -> AMS 1 slot 3 is empty
}
mqtt_client._handle_ams_data(update_ams)
# Verify AMS 1 slot 3 was cleared
ams_data = mqtt_client.state.raw_data.get("ams", [])
b4_tray = ams_data[1]["tray"][3]
assert b4_tray["tray_type"] == "", "tray_type should be cleared for empty slot"
assert b4_tray["remain"] == 0, "remain should be 0 for empty slot"
# Verify other slots are preserved
assert ams_data[0]["tray"][0]["tray_type"] == "PLA", "A1 should still have PLA"
assert ams_data[1]["tray"][0]["tray_type"] == "PLA", "B1 should still have PLA"
def test_shutdown_message_preserves_ams_data(self, mqtt_client):
"""Printer shutdown (power_on_flag=False) must not wipe AMS slot data (#765).
When a printer shuts down it sends a final MQTT message with
tray_exist_bits='0' and power_on_flag=False. This all-zero value
previously caused every slot to be cleared, which then triggered
auto-unlink of all spool assignments on reconnect.
"""
# Initial state: two AMS units with loaded spools
initial_ams = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF", "remain": 80},
{"id": 1, "tray_type": "PETG", "tray_color": "00FF00FF", "remain": 60},
],
},
{
"id": 1,
"tray": [
{"id": 0, "tray_type": "PETG", "tray_color": "DBDDD9FF", "remain": 90},
{"id": 1, "tray_type": "PETG", "tray_color": "67DB25FF", "remain": 70},
],
},
],
"tray_exist_bits": "33", # Slots 0,1 of each AMS (0b00110011)
"power_on_flag": True,
}
mqtt_client._handle_ams_data(initial_ams)
# Verify initial state
ams_data = mqtt_client.state.raw_data["ams"]
assert ams_data[0]["tray"][0]["tray_type"] == "PLA"
assert ams_data[1]["tray"][0]["tray_type"] == "PETG"
# Simulate printer shutdown — all-zero bits with power_on_flag=False
shutdown_ams = {
"ams_exist_bits": "0",
"tray_exist_bits": "0",
"power_on_flag": False,
"insert_flag": False,
"tray_now": "0",
"version": 0,
}
mqtt_client._handle_ams_data(shutdown_ams)
# AMS slot data MUST be preserved — shutdown should not clear it
ams_data = mqtt_client.state.raw_data["ams"]
assert ams_data[0]["tray"][0]["tray_type"] == "PLA", "Shutdown must not clear AMS 0 slot 0"
assert ams_data[0]["tray"][0]["tray_color"] == "FF0000FF", "Shutdown must not clear AMS 0 slot 0 color"
assert ams_data[0]["tray"][1]["tray_type"] == "PETG", "Shutdown must not clear AMS 0 slot 1"
assert ams_data[1]["tray"][0]["tray_type"] == "PETG", "Shutdown must not clear AMS 1 slot 0"
assert ams_data[1]["tray"][1]["tray_type"] == "PETG", "Shutdown must not clear AMS 1 slot 1"
def test_genuine_removal_still_clears_with_power_on(self, mqtt_client):
"""Genuine spool removal (power_on_flag=True) must still clear slot data.
Ensures the #765 fix doesn't break normal spool removal detection.
"""
# Initial state: AMS with loaded spool
initial_ams = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_type": "PLA", "tray_color": "FF0000", "remain": 80},
{"id": 1, "tray_type": "PETG", "tray_color": "00FF00", "remain": 60},
],
},
],
"tray_exist_bits": "3", # Both slots occupied (0b11)
"power_on_flag": True,
}
mqtt_client._handle_ams_data(initial_ams)
# Spool removed from slot 1 while printer is running
removal_ams = {
"ams": [
{
"id": 0,
"tray": [{"id": 0}, {"id": 1}],
},
],
"tray_exist_bits": "1", # Only slot 0 occupied (0b01)
"power_on_flag": True,
}
mqtt_client._handle_ams_data(removal_ams)
# Slot 0 preserved, slot 1 cleared
ams_data = mqtt_client.state.raw_data["ams"]
assert ams_data[0]["tray"][0]["tray_type"] == "PLA", "Slot 0 should be preserved"
assert ams_data[0]["tray"][1]["tray_type"] == "", "Slot 1 should be cleared on removal"
assert ams_data[0]["tray"][1]["tray_color"] == "", "Slot 1 color should be cleared"
def test_power_on_flag_defaults_true_when_absent(self, mqtt_client):
"""When power_on_flag is not in the MQTT data, clearing must proceed normally.
Ensures backwards compatibility with firmware that doesn't send power_on_flag.
"""
# Initial state
initial_ams = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_type": "PLA", "tray_color": "FF0000", "remain": 80},
],
},
],
"tray_exist_bits": "1",
}
mqtt_client._handle_ams_data(initial_ams)
# Update WITHOUT power_on_flag — should still clear when bit=0
update_ams = {
"ams": [{"id": 0, "tray": [{"id": 0}]}],
"tray_exist_bits": "0",
# No power_on_flag key at all
}
mqtt_client._handle_ams_data(update_ams)
ams_data = mqtt_client.state.raw_data["ams"]
assert ams_data[0]["tray"][0]["tray_type"] == "", (
"Without power_on_flag, clearing should proceed (defaults to True)"
)
class TestAMSTrayStateClearning:
"""Tests for AMS tray state-based clearing (#784).
Some printers (e.g. H2D) only send {id, state} in incremental MQTT
updates when a tray is not fully loaded. state=11 means loaded;
other values (9=empty, 10=spool present but filament not in feeder)
should clear stale tray data that was set from an earlier pushall.
"""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_H2D",
access_code="12345678",
)
return client
def _seed_loaded_tray(self, mqtt_client):
"""Seed AMS 0 with a fully loaded tray (state=11) and an empty slot."""
initial = {
"ams": [
{
"id": 0,
"tray": [
{
"id": 0,
"tray_type": "PETG",
"tray_sub_brands": "PETG HF",
"tray_color": "00FF00FF",
"tray_id_name": "A00-G1",
"tray_info_idx": "GFG99",
"tag_uid": "AABBCCDD11223344",
"tray_uuid": "AABBCCDD11223344AABBCCDD11223344",
"remain": 75,
"k": 0.02,
"cali_idx": 5,
"state": 11,
},
{
"id": 1,
"tray_type": "PLA",
"tray_color": "FF0000FF",
"remain": 50,
"state": 11,
},
],
}
],
"power_on_flag": False, # H2D always sends False
}
mqtt_client._handle_ams_data(initial)
ams = mqtt_client.state.raw_data["ams"]
assert ams[0]["tray"][0]["tray_type"] == "PETG"
assert ams[0]["tray"][1]["tray_type"] == "PLA"
def test_state_10_clears_stale_tray_data(self, mqtt_client):
"""Incremental update with state=10 (spool present, not loaded) clears tray."""
self._seed_loaded_tray(mqtt_client)
# H2D sends only {id, state} when filament is retracted
update = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "state": 10},
{"id": 1, "state": 11}, # slot 1 still loaded
],
}
],
"power_on_flag": False,
}
mqtt_client._handle_ams_data(update)
ams = mqtt_client.state.raw_data["ams"]
tray0 = ams[0]["tray"][0]
tray1 = ams[0]["tray"][1]
# Tray 0 should be cleared
assert tray0["tray_type"] == "", "tray_type must be cleared on state=10"
assert tray0["tray_color"] == "", "tray_color must be cleared"
assert tray0["tray_sub_brands"] == "", "tray_sub_brands must be cleared"
assert tray0["tray_id_name"] == "", "tray_id_name must be cleared"
assert tray0["tray_info_idx"] == "", "tray_info_idx must be cleared"
assert tray0["tag_uid"] == "0000000000000000", "tag_uid must be cleared"
assert tray0["tray_uuid"] == "00000000000000000000000000000000", "tray_uuid must be cleared"
assert tray0["remain"] == 0, "remain must be 0"
assert tray0["k"] is None, "k must be cleared"
assert tray0["cali_idx"] is None, "cali_idx must be cleared"
assert tray0["state"] == 10, "state should be preserved"
# Tray 1 should be untouched
assert tray1["tray_type"] == "PLA", "Loaded slot must be preserved"
assert tray1["remain"] == 50
def test_state_9_clears_stale_tray_data(self, mqtt_client):
"""Incremental update with state=9 (empty, no spool) clears tray."""
self._seed_loaded_tray(mqtt_client)
update = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "state": 9},
{"id": 1, "state": 11},
],
}
],
"power_on_flag": False,
}
mqtt_client._handle_ams_data(update)
tray0 = mqtt_client.state.raw_data["ams"][0]["tray"][0]
assert tray0["tray_type"] == "", "state=9 must clear tray_type"
assert tray0["remain"] == 0
def test_state_11_preserves_tray_data(self, mqtt_client):
"""Incremental update with state=11 (loaded) must NOT clear tray."""
self._seed_loaded_tray(mqtt_client)
update = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "state": 11},
{"id": 1, "state": 11},
],
}
],
"power_on_flag": False,
}
mqtt_client._handle_ams_data(update)
tray0 = mqtt_client.state.raw_data["ams"][0]["tray"][0]
assert tray0["tray_type"] == "PETG", "state=11 must preserve tray data"
assert tray0["tray_color"] == "00FF00FF"
assert tray0["remain"] == 75
def test_no_clearing_when_tray_type_already_empty(self, mqtt_client):
"""Don't re-clear a tray that's already empty (avoids log spam)."""
self._seed_loaded_tray(mqtt_client)
# First unload clears
update = {
"ams": [{"id": 0, "tray": [{"id": 0, "state": 10}, {"id": 1, "state": 11}]}],
"power_on_flag": False,
}
mqtt_client._handle_ams_data(update)
assert mqtt_client.state.raw_data["ams"][0]["tray"][0]["tray_type"] == ""
# Second identical update should not trigger clearing again
# (merged_tray.get("tray_type") is already empty/falsy)
mqtt_client._handle_ams_data(update)
assert mqtt_client.state.raw_data["ams"][0]["tray"][0]["tray_type"] == ""
def test_reload_after_unload_restores_data(self, mqtt_client):
"""After clearing via state=10, a full update with state=11 restores data."""
self._seed_loaded_tray(mqtt_client)
# Unload
mqtt_client._handle_ams_data(
{
"ams": [{"id": 0, "tray": [{"id": 0, "state": 10}, {"id": 1, "state": 11}]}],
"power_on_flag": False,
}
)
assert mqtt_client.state.raw_data["ams"][0]["tray"][0]["tray_type"] == ""
# Reload — full tray data arrives again
mqtt_client._handle_ams_data(
{
"ams": [
{
"id": 0,
"tray": [
{
"id": 0,
"tray_type": "PETG",
"tray_sub_brands": "PETG HF",
"tray_color": "00FF00FF",
"remain": 75,
"state": 11,
},
{"id": 1, "state": 11},
],
}
],
"power_on_flag": False,
}
)
tray0 = mqtt_client.state.raw_data["ams"][0]["tray"][0]
assert tray0["tray_type"] == "PETG", "Reload must restore tray data"
assert tray0["tray_color"] == "00FF00FF"
assert tray0["remain"] == 75
class TestNozzleRackData:
"""Tests for nozzle rack data parsing from H2 series device.nozzle.info."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_h2c_nozzle_rack_populated_with_8_entries(self, mqtt_client):
"""H2C provides 8 nozzle entries: IDs 0,1 (L/R hotend) + 16-21 (rack)."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{
"id": 0,
"type": "HS",
"diameter": "0.4",
"wear": 5,
"stat": 1,
"max_temp": 300,
"serial_number": "SN-L",
},
{
"id": 1,
"type": "HS",
"diameter": "0.4",
"wear": 3,
"stat": 0,
"max_temp": 300,
"serial_number": "SN-R",
},
{
"id": 16,
"type": "HS",
"diameter": "0.4",
"wear": 10,
"stat": 0,
"max_temp": 300,
"serial_number": "SN-16",
},
{
"id": 17,
"type": "HH01",
"diameter": "0.6",
"wear": 0,
"stat": 0,
"max_temp": 300,
"serial_number": "SN-17",
},
{
"id": 18,
"type": "HS",
"diameter": "0.4",
"wear": 2,
"stat": 0,
"max_temp": 300,
"serial_number": "SN-18",
},
{
"id": 19,
"type": "",
"diameter": "",
"wear": None,
"stat": None,
"max_temp": 0,
"serial_number": "",
},
{
"id": 20,
"type": "",
"diameter": "",
"wear": None,
"stat": None,
"max_temp": 0,
"serial_number": "",
},
{
"id": 21,
"type": "",
"diameter": "",
"wear": None,
"stat": None,
"max_temp": 0,
"serial_number": "",
},
]
}
}
}
}
mqtt_client._process_message(payload)
assert len(mqtt_client.state.nozzle_rack) == 8
ids = [n["id"] for n in mqtt_client.state.nozzle_rack]
assert ids == [0, 1, 16, 17, 18, 19, 20, 21]
def test_h2d_nozzle_rack_populated_with_2_entries(self, mqtt_client):
"""H2D provides 2 nozzle entries: IDs 0,1 (L/R hotend) — no rack slots."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{
"id": 0,
"type": "HS",
"diameter": "0.4",
"wear": 5,
"stat": 1,
"max_temp": 300,
"serial_number": "SN-L",
},
{
"id": 1,
"type": "HS",
"diameter": "0.4",
"wear": 3,
"stat": 1,
"max_temp": 300,
"serial_number": "SN-R",
},
]
}
}
}
}
mqtt_client._process_message(payload)
assert len(mqtt_client.state.nozzle_rack) == 2
ids = [n["id"] for n in mqtt_client.state.nozzle_rack]
assert ids == [0, 1]
def test_single_nozzle_h2s_populated(self, mqtt_client):
"""H2S provides 1 nozzle entry: ID 0 only — single nozzle printer."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{
"id": 0,
"type": "HS",
"diameter": "0.4",
"wear": 2,
"stat": 1,
"max_temp": 300,
"serial_number": "SN-0",
},
]
}
}
}
}
mqtt_client._process_message(payload)
assert len(mqtt_client.state.nozzle_rack) == 1
assert mqtt_client.state.nozzle_rack[0]["id"] == 0
def test_empty_nozzle_info_does_not_populate_rack(self, mqtt_client):
"""Empty nozzle info list should not populate nozzle_rack."""
payload = {"print": {"device": {"nozzle": {"info": []}}}}
mqtt_client._process_message(payload)
assert mqtt_client.state.nozzle_rack == []
def test_nozzle_rack_sorted_by_id(self, mqtt_client):
"""Nozzle rack entries should be sorted by ID regardless of input order."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{"id": 17, "type": "HS", "diameter": "0.6"},
{"id": 0, "type": "HS", "diameter": "0.4"},
{"id": 16, "type": "HS", "diameter": "0.4"},
{"id": 1, "type": "HS", "diameter": "0.4"},
]
}
}
}
}
mqtt_client._process_message(payload)
ids = [n["id"] for n in mqtt_client.state.nozzle_rack]
assert ids == [0, 1, 16, 17]
def test_nozzle_rack_field_mapping(self, mqtt_client):
"""Verify field mapping from MQTT nozzle_info to nozzle_rack dict keys."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{
"id": 16,
"type": "HH01",
"diameter": "0.6",
"wear": 15,
"stat": 0,
"max_temp": 320,
"serial_number": "SN-ABC123",
"filament_colour": "FF8800",
"filament_id": "F42",
"tray_type": "ABS",
}
]
}
}
}
}
mqtt_client._process_message(payload)
slot = mqtt_client.state.nozzle_rack[0]
assert slot["id"] == 16
assert slot["type"] == "HH01"
assert slot["diameter"] == "0.6"
assert slot["wear"] == 15
assert slot["stat"] == 0
assert slot["max_temp"] == 320
assert slot["serial_number"] == "SN-ABC123"
assert slot["filament_color"] == "FF8800"
assert slot["filament_id"] == "F42"
assert slot["filament_type"] == "ABS"
def test_nozzle_info_updates_nozzle_state(self, mqtt_client):
"""Nozzle info for IDs 0,1 should also update nozzle state (type/diameter)."""
payload = {
"print": {
"device": {
"nozzle": {
"info": [
{"id": 0, "type": "HS", "diameter": "0.4"},
{"id": 1, "type": "HH01", "diameter": "0.6"},
]
}
}
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.nozzles[0].nozzle_type == "HS"
assert mqtt_client.state.nozzles[0].nozzle_diameter == "0.4"
assert mqtt_client.state.nozzles[1].nozzle_type == "HH01"
assert mqtt_client.state.nozzles[1].nozzle_diameter == "0.6"
class TestRequestTopicFailSafe:
"""Tests for graceful degradation when broker rejects request topic subscription."""
@pytest.fixture(autouse=True)
def clear_request_topic_cache(self):
"""Clear class-level cache before each test to avoid cross-test pollution."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
BambuMQTTClient._request_topic_cache.clear()
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_request_topic_supported_by_default(self, mqtt_client):
"""Request topic subscription is attempted by default."""
assert mqtt_client._request_topic_supported is True
assert mqtt_client._request_topic_confirmed is False
def test_on_subscribe_confirms_success(self, mqtt_client):
"""Successful SUBACK marks request topic as confirmed."""
from paho.mqtt.reasoncodes import ReasonCode
mqtt_client._request_topic_sub_mid = 42
rc = ReasonCode(9, identifier=0) # SUBACK packetType=9, QoS 0 = success
mqtt_client._on_subscribe(None, None, 42, [rc], None)
assert mqtt_client._request_topic_confirmed is True
assert mqtt_client._request_topic_supported is True
assert mqtt_client._request_topic_sub_mid is None
assert mqtt_client._request_topic_sub_time == 0.0
def test_on_subscribe_detects_rejection(self, mqtt_client):
"""SUBACK with failure code disables request topic."""
from paho.mqtt.reasoncodes import ReasonCode
mqtt_client._request_topic_sub_mid = 42
rc = ReasonCode(9, identifier=0x80) # SUBACK packetType=9, 0x80 = failure
mqtt_client._on_subscribe(None, None, 42, [rc], None)
assert mqtt_client._request_topic_supported is False
assert mqtt_client._request_topic_confirmed is False
def test_on_subscribe_ignores_other_mids(self, mqtt_client):
"""SUBACK for other subscriptions (e.g. report topic) is ignored."""
from paho.mqtt.reasoncodes import ReasonCode
mqtt_client._request_topic_sub_mid = 42
rc = ReasonCode(9, identifier=0x80)
mqtt_client._on_subscribe(None, None, 99, [rc], None)
# Not affected — mid doesn't match
assert mqtt_client._request_topic_supported is True
def test_disconnect_after_subscription_disables_topic(self, mqtt_client):
"""Disconnect within 10s of subscription attempt disables request topic."""
import time
mqtt_client._request_topic_sub_time = time.time()
mqtt_client._request_topic_confirmed = False
mqtt_client._last_message_time = 0.0
mqtt_client._on_disconnect(None, None)
assert mqtt_client._request_topic_supported is False
assert mqtt_client._request_topic_sub_time == 0.0
def test_disconnect_after_confirmation_does_not_disable(self, mqtt_client):
"""Disconnect after SUBACK confirmation keeps request topic enabled."""
import time
mqtt_client._request_topic_sub_time = time.time()
mqtt_client._request_topic_confirmed = True
mqtt_client._last_message_time = 0.0
mqtt_client._on_disconnect(None, None)
assert mqtt_client._request_topic_supported is True
def test_late_disconnect_does_not_disable(self, mqtt_client):
"""Disconnect long after subscription (>10s) doesn't blame request topic."""
import time
mqtt_client._request_topic_sub_time = time.time() - 30.0
mqtt_client._request_topic_confirmed = False
mqtt_client._last_message_time = 0.0
mqtt_client._on_disconnect(None, None)
assert mqtt_client._request_topic_supported is True
def test_on_connect_skips_request_topic_when_unsupported(self, mqtt_client):
"""After marking unsupported, reconnect skips request topic subscription."""
mqtt_client._request_topic_supported = False
subscribe_calls = []
mock_client = type(
"MockClient",
(),
{
"subscribe": lambda self, topic: subscribe_calls.append(topic) or (0, 1),
},
)()
mqtt_client._on_connect(mock_client, None, None, 0)
# Only report topic subscribed, not request topic
assert len(subscribe_calls) == 1
assert subscribe_calls[0] == mqtt_client.topic_subscribe
def test_cache_persists_across_instances(self):
"""New client instance inherits request topic unsupported state from cache."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client1 = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_CACHE",
access_code="12345678",
)
assert client1._request_topic_supported is True
# Simulate disconnect-after-subscribe disabling the topic
client1._request_topic_sub_time = __import__("time").time()
client1._request_topic_confirmed = False
client1._last_message_time = 0.0
client1._on_disconnect(None, None)
assert client1._request_topic_supported is False
# New instance for same serial should inherit the cached state
client2 = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_CACHE",
access_code="12345678",
)
assert client2._request_topic_supported is False
def test_cache_does_not_affect_different_serial(self):
"""Cache is per-serial — different printer is unaffected."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
BambuMQTTClient._request_topic_cache["SERIAL_A"] = False
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="SERIAL_B",
access_code="12345678",
)
assert client._request_topic_supported is True
def test_cache_updated_on_suback_success(self):
"""Successful SUBACK caches positive confirmation."""
from paho.mqtt.reasoncodes import ReasonCode
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_SUBACK",
access_code="12345678",
)
client._request_topic_sub_mid = 42
rc = ReasonCode(9, identifier=0) # Success
client._on_subscribe(None, None, 42, [rc], None)
assert BambuMQTTClient._request_topic_cache["TEST_SUBACK"] is True
def test_cache_updated_on_suback_rejection(self):
"""SUBACK rejection caches negative state."""
from paho.mqtt.reasoncodes import ReasonCode
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_REJECT",
access_code="12345678",
)
client._request_topic_sub_mid = 42
rc = ReasonCode(9, identifier=0x80) # Failure
client._on_subscribe(None, None, 42, [rc], None)
assert BambuMQTTClient._request_topic_cache["TEST_REJECT"] is False
class TestRequestTopicAmsMapping:
"""Tests for capturing ams_mapping from the MQTT request topic."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_captured_ams_mapping_initializes_to_none(self, mqtt_client):
"""Verify _captured_ams_mapping starts as None."""
assert mqtt_client._captured_ams_mapping is None
def test_handle_request_message_captures_ams_mapping(self, mqtt_client):
"""project_file command with ams_mapping stores the mapping."""
data = {
"print": {
"command": "project_file",
"ams_mapping": [0, 4, -1, -1],
"url": "ftp://192.168.1.100/test.3mf",
}
}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping == [0, 4, -1, -1]
def test_handle_request_message_ignores_non_print_commands(self, mqtt_client):
"""Non-project_file commands don't store ams_mapping."""
data = {
"print": {
"command": "pause",
}
}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping is None
def test_handle_request_message_ignores_missing_ams_mapping(self, mqtt_client):
"""project_file command without ams_mapping doesn't store anything."""
data = {
"print": {
"command": "project_file",
"url": "ftp://192.168.1.100/test.3mf",
}
}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping is None
def test_handle_request_message_ignores_non_dict_print(self, mqtt_client):
"""Non-dict print value is safely ignored."""
data = {"print": "not_a_dict"}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping is None
def test_handle_request_message_ignores_missing_print(self, mqtt_client):
"""Message without print key is safely ignored."""
data = {"pushing": {"command": "pushall"}}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping is None
def test_captured_mapping_overwrites_previous(self, mqtt_client):
"""A new print command overwrites a previously captured mapping."""
mqtt_client._captured_ams_mapping = [0, -1, -1, -1]
data = {
"print": {
"command": "project_file",
"ams_mapping": [4, 8, -1, -1],
}
}
mqtt_client._handle_request_message(data)
assert mqtt_client._captured_ams_mapping == [4, 8, -1, -1]
def test_print_start_callback_includes_ams_mapping(self, mqtt_client):
"""on_print_start callback data includes captured ams_mapping."""
start_data = {}
def on_start(data):
start_data.update(data)
mqtt_client.on_print_start = on_start
mqtt_client._captured_ams_mapping = [0, 4, -1, -1]
# Trigger print start
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
assert start_data.get("ams_mapping") == [0, 4, -1, -1]
def test_print_start_callback_ams_mapping_none_when_not_captured(self, mqtt_client):
"""on_print_start callback has ams_mapping=None when no mapping captured."""
start_data = {}
def on_start(data):
start_data.update(data)
mqtt_client.on_print_start = on_start
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
assert "ams_mapping" in start_data
assert start_data["ams_mapping"] is None
def test_print_complete_callback_includes_ams_mapping(self, mqtt_client):
"""on_print_complete callback data includes captured ams_mapping."""
complete_data = {}
def on_complete(data):
complete_data.update(data)
mqtt_client.on_print_start = lambda d: None
mqtt_client.on_print_complete = on_complete
mqtt_client._captured_ams_mapping = [0, 9, -1, -1]
# Start print
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
# Complete print
mqtt_client._process_message(
{
"print": {
"gcode_state": "FINISH",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
assert complete_data.get("ams_mapping") == [0, 9, -1, -1]
def test_captured_mapping_cleared_after_print_complete(self, mqtt_client):
"""_captured_ams_mapping is reset to None after print completion."""
mqtt_client.on_print_start = lambda d: None
mqtt_client.on_print_complete = lambda d: None
mqtt_client._captured_ams_mapping = [0, 4, -1, -1]
# Start print
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
# Complete print
mqtt_client._process_message(
{
"print": {
"gcode_state": "FINISH",
"gcode_file": "/data/Metadata/test.gcode",
"subtask_name": "Test",
}
}
)
assert mqtt_client._captured_ams_mapping is None
def test_full_flow_capture_and_deliver(self, mqtt_client):
"""Full flow: slicer sends print command → MQTT captures mapping → completion delivers it."""
complete_data = {}
def on_complete(data):
complete_data.update(data)
mqtt_client.on_print_start = lambda d: None
mqtt_client.on_print_complete = on_complete
# 1. Slicer sends print command (captured from request topic)
mqtt_client._handle_request_message(
{
"print": {
"command": "project_file",
"ams_mapping": [4, 9, -1, -1],
"url": "ftp://192.168.1.100/model.3mf",
}
}
)
assert mqtt_client._captured_ams_mapping == [4, 9, -1, -1]
# 2. Printer reports RUNNING
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "/data/Metadata/model.gcode",
"subtask_name": "Model",
}
}
)
# 3. Printer reports FINISH
mqtt_client._process_message(
{
"print": {
"gcode_state": "FINISH",
"gcode_file": "/data/Metadata/model.gcode",
"subtask_name": "Model",
}
}
)
assert complete_data["ams_mapping"] == [4, 9, -1, -1]
assert complete_data["status"] == "completed"
# Mapping cleared after completion
assert mqtt_client._captured_ams_mapping is None
# ---------------------------------------------------------------------------
# tray_now disambiguation helpers
# ---------------------------------------------------------------------------
def _ams_payload(tray_now, ams_units=None, tray_exist_bits=None, ams_exist_bits=None):
"""Build minimal print.ams payload for tray_now disambiguation tests."""
ams = {"tray_now": str(tray_now)}
if ams_units is not None:
ams["ams"] = ams_units
if tray_exist_bits is not None:
ams["tray_exist_bits"] = tray_exist_bits
if ams_exist_bits is not None:
ams["ams_exist_bits"] = ams_exist_bits
return {"print": {"ams": ams}}
def _extruder_info_payload(extruders):
"""Build device.extruder.info payload (dual-nozzle detection + snow).
Each entry in *extruders* is a dict with at least ``id`` and ``snow``.
"""
return {
"print": {
"device": {
"extruder": {
"info": extruders,
}
}
}
}
def _extruder_state_payload(state_val):
"""Build device.extruder.state payload (active extruder via bit 8)."""
return {
"print": {
"device": {
"extruder": {
"state": state_val,
}
}
}
}
# ---------------------------------------------------------------------------
# 1. Single-nozzle X1E — direct passthrough
# ---------------------------------------------------------------------------
class TestTrayNowSingleNozzleX1E:
"""Single-nozzle, 1 AMS — tray_now is a direct passthrough."""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_X1E",
access_code="12345678",
)
def test_tray_now_direct_passthrough_slot_0_to_3(self, mqtt_client):
"""Each tray_now 0-3 maps 1:1 on single-nozzle printers."""
for slot in range(4):
mqtt_client._process_message(_ams_payload(slot))
assert mqtt_client.state.tray_now == slot
def test_tray_now_255_means_unloaded(self, mqtt_client):
"""tray_now=255 means no filament loaded."""
mqtt_client._process_message(_ams_payload(255))
assert mqtt_client.state.tray_now == 255
def test_single_extruder_does_not_trigger_dual_nozzle(self, mqtt_client):
"""device.extruder.info with 1 entry must NOT set _is_dual_nozzle."""
mqtt_client._process_message(_extruder_info_payload([{"id": 0, "snow": 0xFF00FF}]))
assert mqtt_client._is_dual_nozzle is False
def test_last_loaded_tray_survives_unload(self, mqtt_client):
"""Load tray 2, unload → last_loaded_tray stays 2."""
mqtt_client._process_message(_ams_payload(2))
assert mqtt_client.state.last_loaded_tray == 2
mqtt_client._process_message(_ams_payload(255))
assert mqtt_client.state.tray_now == 255
assert mqtt_client.state.last_loaded_tray == 2
# ---------------------------------------------------------------------------
# 2. Single-nozzle P2S — multiple AMS, global IDs pass through
# ---------------------------------------------------------------------------
class TestTrayNowSingleNozzleP2S:
"""Single-nozzle, 2 AMS — tray_now > 3 passes through as global ID."""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_P2S",
access_code="12345678",
)
def test_tray_now_ams1_global_ids_4_to_7(self, mqtt_client):
"""tray_now 4-7 are global IDs for AMS 1 on single-nozzle printers."""
for global_id in range(4, 8):
mqtt_client._process_message(_ams_payload(global_id))
assert mqtt_client.state.tray_now == global_id
def test_tray_change_across_ams_units(self, mqtt_client):
"""Switch from AMS 0 slot 1 → AMS 1 slot 2 (global 6)."""
mqtt_client._process_message(_ams_payload(1))
assert mqtt_client.state.tray_now == 1
mqtt_client._process_message(_ams_payload(6))
assert mqtt_client.state.tray_now == 6
# ---------------------------------------------------------------------------
# 2b. Single-nozzle P2S — multi-AMS local slot disambiguation (#420)
# ---------------------------------------------------------------------------
class TestTrayNowP2SMultiAmsDisambiguation:
"""P2S firmware sends local slot IDs (0-3) in tray_now even with dual AMS.
When ams_exist_bits indicates >1 AMS unit and tray_now is 0-3, the backend
should use the MQTT mapping field (snow-encoded) to resolve the correct
global tray ID.
"""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_P2S_DUAL",
access_code="12345678",
)
return client
def test_resolves_ams1_slot1_from_mapping(self, mqtt_client):
"""tray_now=1 with mapping=[257] → global ID 5 (AMS1-T1).
257 snow-decoded: ams_hw_id=1, slot=1 → global 1*4+1=5.
"""
# Set mapping field in raw_data (as the MQTT handler would)
mqtt_client.state.raw_data["mapping"] = [257]
mqtt_client._process_message(
_ams_payload(1, ams_exist_bits="3") # '3' = 0b11 → AMS 0 and 1
)
assert mqtt_client.state.tray_now == 5
def test_resolves_ams1_slot0_from_mapping(self, mqtt_client):
"""tray_now=0 with mapping=[256] → global ID 4 (AMS1-T0).
256 snow-decoded: ams_hw_id=1, slot=0 → global 1*4+0=4.
"""
mqtt_client.state.raw_data["mapping"] = [256]
mqtt_client._process_message(_ams_payload(0, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 4
def test_resolves_ams1_slot3_from_mapping(self, mqtt_client):
"""tray_now=3 with mapping=[259] → global ID 7 (AMS1-T3).
259 snow-decoded: ams_hw_id=1, slot=3 → global 1*4+3=7.
"""
mqtt_client.state.raw_data["mapping"] = [259]
mqtt_client._process_message(_ams_payload(3, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 7
def test_ams0_slot_unchanged_when_mapping_confirms_ams0(self, mqtt_client):
"""tray_now=1 with mapping=[1] → stays 1 (AMS0-T1).
1 snow-decoded: ams_hw_id=0, slot=1 → global 0*4+1=1.
"""
mqtt_client.state.raw_data["mapping"] = [1]
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 1
def test_multicolor_resolves_ams1_from_multi_entry_mapping(self, mqtt_client):
"""Multi-color print: mapping=[0, 257] → tray_now=1 resolves to AMS1-T1 (5).
Entry 0: ams_hw_id=0, slot=0 (local 0) — doesn't match tray_now=1.
Entry 257: ams_hw_id=1, slot=1 (local 1) — matches tray_now=1 → global 5.
"""
mqtt_client.state.raw_data["mapping"] = [0, 257]
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 5
def test_multicolor_four_slot_mapping(self, mqtt_client):
"""mapping=[65535, 65535, 65535, 257] → tray_now=1 resolves to global 5.
Only entry 257 has local slot=1, other entries are unmapped (65535).
Reproduces exact data from issue #420 support package.
"""
mqtt_client.state.raw_data["mapping"] = [65535, 65535, 65535, 257]
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 5
def test_ambiguous_mapping_falls_back_to_local_slot(self, mqtt_client):
"""Two AMS units with same local slot in mapping → ambiguous, keep local slot.
mapping=[1, 257]: both have local slot 1 (AMS0-T1 and AMS1-T1).
Cannot disambiguate → fall back to tray_now=1.
"""
mqtt_client.state.raw_data["mapping"] = [1, 257]
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 1
def test_no_mapping_falls_back_to_local_slot(self, mqtt_client):
"""No mapping field available → fall back to raw tray_now."""
# No mapping in raw_data (e.g. manual filament load, not during print)
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 1
def test_empty_mapping_falls_back_to_local_slot(self, mqtt_client):
"""Empty mapping list → fall back to raw tray_now."""
mqtt_client.state.raw_data["mapping"] = []
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 1
def test_single_ams_passthrough(self, mqtt_client):
"""Single AMS (ams_exist_bits='1') → tray_now 0-3 is direct global ID."""
mqtt_client._process_message(_ams_payload(2, ams_exist_bits="1"))
assert mqtt_client.state.tray_now == 2
def test_no_ams_exist_bits_passthrough(self, mqtt_client):
"""No ams_exist_bits in payload → fall back to raw tray_now."""
mqtt_client._process_message(_ams_payload(1))
assert mqtt_client.state.tray_now == 1
def test_tray_now_255_unaffected_by_multi_ams(self, mqtt_client):
"""tray_now=255 (unloaded) passes through regardless of AMS count."""
mqtt_client.state.raw_data["mapping"] = [257]
mqtt_client._process_message(_ams_payload(255, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 255
def test_tray_now_above_3_unaffected(self, mqtt_client):
"""tray_now > 3 is already a global ID and passes through directly."""
mqtt_client._process_message(_ams_payload(6, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 6
def test_last_loaded_tray_uses_resolved_global_id(self, mqtt_client):
"""last_loaded_tray should reflect the resolved global ID, not local slot."""
mqtt_client.state.raw_data["mapping"] = [257]
mqtt_client.state.state = "RUNNING"
mqtt_client._process_message(_ams_payload(1, ams_exist_bits="3"))
assert mqtt_client.state.tray_now == 5
assert mqtt_client.state.last_loaded_tray == 5
class TestResolveLocalSlotFromMapping:
"""Unit tests for _resolve_local_slot_from_mapping static method."""
def test_single_match_ams0(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, [1]) == 1
def test_single_match_ams1(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
# 257 = 1*256 + 1 → AMS1 slot1 → global 5
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, [257]) == 5
def test_single_match_ams2(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
# 514 = 2*256 + 2 → AMS2 slot2 → global 10
assert BambuMQTTClient._resolve_local_slot_from_mapping(2, [514]) == 10
def test_unmapped_entries_skipped(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, [65535, 65535, 65535, 257]) == 5
def test_no_match_returns_none(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
# mapping has slot 0 only, looking for slot 2
assert BambuMQTTClient._resolve_local_slot_from_mapping(2, [0]) is None
def test_ambiguous_returns_none(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
# Both AMS0 slot1 (1) and AMS1 slot1 (257) → ambiguous
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, [1, 257]) is None
def test_none_mapping_returns_none(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, None) is None
def test_empty_mapping_returns_none(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
assert BambuMQTTClient._resolve_local_slot_from_mapping(1, []) is None
def test_ams_ht_slot0_match(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
# AMS-HT id=128: snow = 128*256 + 0 = 32768
assert BambuMQTTClient._resolve_local_slot_from_mapping(0, [32768]) == 128
# ---------------------------------------------------------------------------
# 3. H2D Pro — initial state detection
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DSetup:
"""H2D Pro initial state detection."""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_H2D",
access_code="12345678",
)
def test_dual_nozzle_detected_from_extruder_info(self, mqtt_client):
"""2 entries in device.extruder.info → _is_dual_nozzle=True."""
mqtt_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 0xFF00FF},
]
)
)
assert mqtt_client._is_dual_nozzle is True
def test_ams_extruder_map_parsed_from_info_field(self, mqtt_client):
"""AMS info field is hex: 0x2003 → ext 0 (right), 0x2104 → ext 1 (left)."""
# MQTT sends info as string; BambuStudio parses as hex via stoull(str, 16)
ams_units = [
{"id": 0, "info": "2003", "tray": [{"id": i} for i in range(4)]},
{"id": 128, "info": "2104", "tray": [{"id": 0}]},
]
payload = {
"print": {
"ams": {
"ams": ams_units,
"tray_now": "255",
"tray_exist_bits": "1000f",
},
}
}
mqtt_client._process_message(payload)
# 0x2003: bits 8-11 = (0x2003 >> 8) & 0xF = 0x20 & 0xF = 0 → extruder 0 (right)
# 0x2104: bits 8-11 = (0x2104 >> 8) & 0xF = 0x21 & 0xF = 1 → extruder 1 (left)
assert mqtt_client.state.ams_extruder_map == {"0": 0, "128": 1}
def test_ams_extruder_map_real_h2d_values(self, mqtt_client):
"""Real H2D MQTT values: AMS2 Pro on right, AMS-HT on left."""
ams_units = [
{"id": 0, "info": "10001003", "tray": [{"id": i} for i in range(4)]},
{"id": 128, "info": "10002104", "tray": [{"id": 0}]},
]
payload = {
"print": {
"ams": {
"ams": ams_units,
"tray_now": "255",
"tray_exist_bits": "1000a",
},
}
}
mqtt_client._process_message(payload)
# 0x10001003: bits 8-11 = (0x10001003 >> 8) & 0xF = 0x10 & 0xF = 0 → right
# 0x10002104: bits 8-11 = (0x10002104 >> 8) & 0xF = 0x21 & 0xF = 1 → left
assert mqtt_client.state.ams_extruder_map == {"0": 0, "128": 1}
def test_ams_extruder_map_skips_uninitialized(self, mqtt_client):
"""extruder_id 0xE means uninitialized AMS — should be skipped."""
ams_units = [
{"id": 0, "info": "e03", "tray": [{"id": i} for i in range(4)]},
]
payload = {
"print": {
"ams": {
"ams": ams_units,
"tray_now": "255",
"tray_exist_bits": "f",
},
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.ams_extruder_map == {}
def test_ams_extruder_map_partial_update_preserves_entries(self, mqtt_client):
"""Partial MQTT update with one AMS should not overwrite other entries."""
# First: full update with both AMS units
full_payload = {
"print": {
"ams": {
"ams": [
{"id": 0, "info": "2003", "tray": [{"id": i} for i in range(4)]},
{"id": 128, "info": "2104", "tray": [{"id": 0}]},
],
"tray_now": "255",
"tray_exist_bits": "1000f",
},
}
}
mqtt_client._process_message(full_payload)
assert mqtt_client.state.ams_extruder_map == {"0": 0, "128": 1}
# Then: partial update with only AMS 0 (no info field this time)
partial_payload = {
"print": {
"ams": {
"ams": [
{"id": 0, "tray": [{"id": 0, "remain": 50}]},
],
"tray_now": "0",
"tray_exist_bits": "1000f",
},
}
}
mqtt_client._process_message(partial_payload)
# Both entries should still be present
assert mqtt_client.state.ams_extruder_map == {"0": 0, "128": 1}
def test_dual_nozzle_detection_before_ams_in_same_message(self, mqtt_client):
"""Dual-nozzle detection at line 538 happens before _handle_ams_data() at line 549.
If both arrive in the same message, tray_now disambiguation already uses dual-nozzle logic.
"""
payload = {
"print": {
"device": {
"extruder": {
"info": [
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 0xFF00FF},
],
"state": 0x0001,
}
},
"ams": {
"ams": [
{"id": 0, "info": "2003", "tray": [{"id": i} for i in range(4)]},
],
"tray_now": "2",
"tray_exist_bits": "f",
},
}
}
mqtt_client._process_message(payload)
# Dual-nozzle was detected; AMS 0 on right extruder (active by default);
# snow is 0xFF00FF (unloaded), so falls through to ams_extruder_map fallback.
# Single AMS on extruder 0 → global_id = 0*4+2 = 2
assert mqtt_client._is_dual_nozzle is True
assert mqtt_client.state.tray_now == 2
# ---------------------------------------------------------------------------
# Shared H2D fixture for classes 4-8
# ---------------------------------------------------------------------------
class _H2DFixtureMixin:
"""Mixin providing a pre-configured H2D Pro client."""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_H2D",
access_code="12345678",
)
@pytest.fixture
def h2d_client(self, mqtt_client):
"""Pre-configure as H2D Pro: dual-nozzle + ams_extruder_map."""
mqtt_client._process_message(
{
"print": {
"device": {
"extruder": {
"info": [
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 0xFF00FF},
],
"state": 0x0001, # right extruder active
}
},
"ams": {
"ams": [
{"id": 0, "info": "2003", "tray": [{"id": i} for i in range(4)]},
{"id": 128, "info": "2104", "tray": [{"id": 0}]},
],
"tray_now": "255",
"tray_exist_bits": "1000f",
},
}
}
)
assert mqtt_client._is_dual_nozzle is True
assert mqtt_client.state.ams_extruder_map == {"0": 0, "128": 1}
return mqtt_client
# ---------------------------------------------------------------------------
# 4. H2D Snow field disambiguation
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DSnow(_H2DFixtureMixin):
"""Snow field disambiguation (primary path)."""
def test_snow_disambiguates_ams0_slot(self, h2d_client):
"""snow ext[0]=AMS 0 slot 2, tray_now='2' → global 2."""
# Send snow update FIRST (snow is parsed AFTER tray_now in the same message,
# so we need it in a prior message).
snow_val = 0 << 8 | 2 # AMS 0 slot 2 = raw 2
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": snow_val},
{"id": 1, "snow": 0xFF00FF},
]
)
)
assert h2d_client.state.h2d_extruder_snow.get(0) == 2
# Now send tray_now=2
h2d_client._process_message(_ams_payload(2))
assert h2d_client.state.tray_now == 2
def test_snow_disambiguates_ams_ht_to_128(self, h2d_client):
"""snow ext[1]=AMS HT (128), left active, tray_now='0' → global 128."""
# Snow: extruder 1 → AMS 128 slot 0
snow_val = 128 << 8 | 0 # = 32768
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": snow_val},
]
)
)
assert h2d_client.state.h2d_extruder_snow.get(1) == 128
# Switch to left extruder
h2d_client._process_message(_extruder_state_payload(0x0100))
assert h2d_client.state.active_extruder == 1
# tray_now="0" with left extruder active, snow says AMS HT (128)
# AMS HT snow_slot = 0 (single slot), parsed_tray_now = 0 → match
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
def test_snow_updates_h2d_extruder_snow_state(self, h2d_client):
"""Verify state.h2d_extruder_snow dict is populated correctly."""
snow_ext0 = 1 << 8 | 3 # AMS 1 slot 3 → global 7
snow_ext1 = 0 << 8 | 0 # AMS 0 slot 0 → global 0
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": snow_ext0},
{"id": 1, "snow": snow_ext1},
]
)
)
assert h2d_client.state.h2d_extruder_snow[0] == 7
assert h2d_client.state.h2d_extruder_snow[1] == 0
def test_snow_unloaded_value(self, h2d_client):
"""snow=0xFFFF (ams_id=255, slot=255) → 255 (unloaded)."""
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFFFF},
{"id": 1, "snow": 0xFFFF},
]
)
)
assert h2d_client.state.h2d_extruder_snow[0] == 255
assert h2d_client.state.h2d_extruder_snow[1] == 255
def test_snow_initial_sentinel_not_stored(self, h2d_client):
"""snow=0xFF00FF (firmware initial sentinel) is not parsed into h2d_extruder_snow."""
# 0xFF00FF has ams_id=0xFF00=65280 which doesn't match any branch
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 0xFF00FF},
]
)
)
# Snow dict should remain empty (no matching branch)
assert h2d_client.state.h2d_extruder_snow == {}
# ---------------------------------------------------------------------------
# 5. H2D Pending target disambiguation
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DPendingTarget(_H2DFixtureMixin):
"""Pending target disambiguation (when Bambuddy initiates load)."""
def test_pending_target_matches_slot(self, h2d_client):
"""pending=5, tray_now='1' (5%4=1 matches) → tray_now=5."""
h2d_client.state.pending_tray_target = 5
h2d_client._process_message(_ams_payload(1))
assert h2d_client.state.tray_now == 5
assert h2d_client.state.pending_tray_target is None # cleared
def test_pending_target_slot_mismatch(self, h2d_client):
"""pending=5, tray_now='2' → uses raw slot, clears pending."""
h2d_client.state.pending_tray_target = 5
h2d_client._process_message(_ams_payload(2))
# Slot 2 != 5%4=1 → mismatch, uses raw slot 2
assert h2d_client.state.tray_now == 2
assert h2d_client.state.pending_tray_target is None
def test_pending_target_takes_priority_over_snow(self, h2d_client):
"""When both pending and snow are set, pending wins."""
# Set up snow for extruder 0 → AMS 0 slot 1 → global 1
snow_val = 0 << 8 | 1
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": snow_val},
{"id": 1, "snow": 0xFF00FF},
]
)
)
assert h2d_client.state.h2d_extruder_snow.get(0) == 1
# Set pending target to AMS 1 slot 1 (global 5)
h2d_client.state.pending_tray_target = 5
# tray_now="1" — matches pending (5%4=1), pending should win over snow
h2d_client._process_message(_ams_payload(1))
assert h2d_client.state.tray_now == 5
# ---------------------------------------------------------------------------
# 6. H2D ams_extruder_map fallback
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DFallback(_H2DFixtureMixin):
"""ams_extruder_map fallback (no pending, no snow)."""
def test_single_ams_on_extruder_computes_global_id(self, h2d_client):
"""AMS 0 on right extruder, tray_now='2' → 0*4+2=2."""
# h2d_client has snow=0xFF00FF (unloaded) by default, so snow path skips
h2d_client._process_message(_ams_payload(2))
# AMS 0 is the only AMS on extruder 0 (right, active by default)
# Fallback: single AMS → global = 0*4+2 = 2
assert h2d_client.state.tray_now == 2
def test_multiple_ams_keeps_current_if_valid(self, h2d_client):
"""Current tray matches slot → keeps it (multi-AMS on same extruder)."""
# Set up: two AMS units on the same extruder (right, ext 0)
h2d_client.state.ams_extruder_map = {"0": 0, "1": 0}
# Pre-set tray_now=5 (AMS 1 slot 1) — current_ams=1 which is in ams_on_extruder
h2d_client.state.tray_now = 5
# tray_now="1" → 5%4=1 matches → keep current=5
h2d_client._process_message(_ams_payload(1))
assert h2d_client.state.tray_now == 5
def test_no_ams_on_extruder_uses_raw_slot(self, h2d_client):
"""No AMS mapped to the active extruder → raw slot as global ID."""
# All AMS on left extruder, but right is active
h2d_client.state.ams_extruder_map = {"0": 1, "128": 1}
h2d_client._process_message(_ams_payload(2))
assert h2d_client.state.tray_now == 2
def test_single_ams_ht_on_extruder_returns_unit_id(self, h2d_client):
"""AMS-HT 128 alone on left extruder, slot 0 → global ID 128 (not 512)."""
# Switch to left extruder (where AMS-HT 128 is mapped)
h2d_client._process_message(_extruder_state_payload(0x0100))
# Only AMS-HT 128 on left extruder; no snow available
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
def test_single_ams_ht_ignores_nonzero_slot(self, h2d_client):
"""AMS-HT has single slot; even if printer reports slot 1, global ID = unit ID."""
h2d_client.state.ams_extruder_map = {"129": 0}
h2d_client._process_message(_ams_payload(1))
# AMS-HT 129: global ID = 129, not 129*4+1=517
assert h2d_client.state.tray_now == 129
def test_multiple_ams_keeps_current_ams_ht(self, h2d_client):
"""Current tray is AMS-HT 128, slot 0 reported → keeps 128."""
h2d_client.state.ams_extruder_map = {"0": 0, "128": 0}
h2d_client.state.tray_now = 128
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
def test_multiple_ams_slot_nonzero_excludes_ams_ht(self, h2d_client):
"""Slot > 0 eliminates AMS-HT candidates; single regular AMS left → resolves."""
# AMS 0 + AMS-HT 128 both on right extruder
h2d_client.state.ams_extruder_map = {"0": 0, "128": 0}
h2d_client.state.tray_now = 255 # no current match
# Slot 2 → can't be AMS-HT → only AMS 0 → global = 0*4+2 = 2
h2d_client._process_message(_ams_payload(2))
assert h2d_client.state.tray_now == 2
def test_multiple_ams_slot_nonzero_narrows_to_single_ht_excluded(self, h2d_client):
"""Two regular AMS + one AMS-HT, slot > 0 → AMS-HT excluded but still ambiguous."""
h2d_client.state.ams_extruder_map = {"0": 0, "1": 0, "128": 0}
h2d_client.state.tray_now = 255
# Slot 3 → excludes AMS-HT, but AMS 0 and AMS 1 both remain → ambiguous
h2d_client._process_message(_ams_payload(3))
assert h2d_client.state.tray_now == 3 # raw slot fallback
# ---------------------------------------------------------------------------
# 6b. H2D last_loaded_tray validation
# ---------------------------------------------------------------------------
class TestLastLoadedTrayValidation(_H2DFixtureMixin):
"""last_loaded_tray only stores physically valid tray IDs."""
def test_regular_ams_tray_stored(self, h2d_client):
"""Valid regular AMS tray (0-15) → stored in last_loaded_tray."""
h2d_client.state.tray_now = 7
# Trigger tray_now processing via AMS message
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 1 << 8 | 3}, # AMS 1 slot 3 → global 7
{"id": 1, "snow": 0xFF00FF},
]
)
)
h2d_client._process_message(_ams_payload(3))
assert h2d_client.state.tray_now == 7
assert h2d_client.state.last_loaded_tray == 7
def test_ams_ht_tray_stored(self, h2d_client):
"""Valid AMS-HT tray (128-135) → stored in last_loaded_tray."""
h2d_client._process_message(_extruder_state_payload(0x0100))
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 128 << 8 | 0},
]
)
)
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
assert h2d_client.state.last_loaded_tray == 128
def test_unloaded_not_stored(self, h2d_client):
"""tray_now=255 (unloaded) → last_loaded_tray unchanged."""
h2d_client.state.last_loaded_tray = 5
h2d_client._process_message(_ams_payload(255))
assert h2d_client.state.tray_now == 255
assert h2d_client.state.last_loaded_tray == 5
# ---------------------------------------------------------------------------
# 7. H2D Active extruder switching
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DActiveExtruder(_H2DFixtureMixin):
"""Active extruder switching via device.extruder.state bit 8."""
def test_active_extruder_right_by_default(self, h2d_client):
"""Initial state.active_extruder == 0 (right)."""
assert h2d_client.state.active_extruder == 0
def test_extruder_state_bit8_switches_to_left(self, h2d_client):
"""state=0x100 → active_extruder=1 (left)."""
h2d_client._process_message(_extruder_state_payload(0x0100))
assert h2d_client.state.active_extruder == 1
def test_extruder_state_bit8_switches_back_to_right(self, h2d_client):
"""Cycle 0 → 1 → 0."""
h2d_client._process_message(_extruder_state_payload(0x0100))
assert h2d_client.state.active_extruder == 1
h2d_client._process_message(_extruder_state_payload(0x0001))
assert h2d_client.state.active_extruder == 0
def test_extruder_switch_changes_tray_disambiguation(self, h2d_client):
"""Snow on both extruders; switching active changes which snow is used."""
# Snow: ext 0 → AMS 0 slot 1 (global 1), ext 1 → AMS 128 slot 0 (global 128)
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0 << 8 | 1}, # AMS 0 slot 1 → global 1
{"id": 1, "snow": 128 << 8 | 0}, # AMS HT → global 128
]
)
)
# Right active (default) — tray_now="1" → snow ext[0] says global 1
h2d_client._process_message(_ams_payload(1))
assert h2d_client.state.tray_now == 1
# Switch to left
h2d_client._process_message(_extruder_state_payload(0x0100))
# Left active — tray_now="0" → snow ext[1] says AMS HT (128), slot 0 matches
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
# ---------------------------------------------------------------------------
# 8. H2D Full multi-message sequences
# ---------------------------------------------------------------------------
class TestTrayNowDualNozzleH2DFullSequence(_H2DFixtureMixin):
"""Multi-message sequences simulating real H2D Pro prints."""
def test_h2d_right_nozzle_ams0_lifecycle(self, h2d_client):
"""Setup → load AMS 0 slot 1 → verify tray_now=1."""
# Snow update: extruder 0 loading AMS 0 slot 1
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0 << 8 | 1},
{"id": 1, "snow": 0xFF00FF},
]
)
)
# Printer reports tray_now="1"
h2d_client._process_message(_ams_payload(1))
assert h2d_client.state.tray_now == 1
assert h2d_client.state.last_loaded_tray == 1
def test_h2d_left_nozzle_ams_ht_lifecycle(self, h2d_client):
"""Setup → switch left → load AMS HT → verify tray_now=128."""
# Switch to left extruder
h2d_client._process_message(_extruder_state_payload(0x0100))
# Snow: ext 1 → AMS HT slot 0
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0xFF00FF},
{"id": 1, "snow": 128 << 8 | 0},
]
)
)
# Printer reports tray_now="0" (AMS HT single slot)
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
assert h2d_client.state.last_loaded_tray == 128
def test_h2d_multi_color_alternating_nozzles(self, h2d_client):
"""Multi-color print alternating between right and left nozzles.
Sequence:
1. Right loads AMS 0 slot 0 (tray=0)
2. Switch left, load AMS HT (tray=128)
3. Switch right, snow updates, load AMS 0 slot 2 (tray=2)
4. Unload (255)
"""
# Step 1: Right extruder loads AMS 0 slot 0
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0 << 8 | 0},
{"id": 1, "snow": 0xFF00FF},
]
)
)
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 0
# Step 2: Switch to left, load AMS HT
h2d_client._process_message(_extruder_state_payload(0x0100))
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0 << 8 | 0},
{"id": 1, "snow": 128 << 8 | 0},
]
)
)
h2d_client._process_message(_ams_payload(0))
assert h2d_client.state.tray_now == 128
# Step 3: Switch back to right, load AMS 0 slot 2
h2d_client._process_message(_extruder_state_payload(0x0001))
h2d_client._process_message(
_extruder_info_payload(
[
{"id": 0, "snow": 0 << 8 | 2},
{"id": 1, "snow": 128 << 8 | 0},
]
)
)
h2d_client._process_message(_ams_payload(2))
assert h2d_client.state.tray_now == 2
# Step 4: Unload
h2d_client._process_message(_ams_payload(255))
assert h2d_client.state.tray_now == 255
assert h2d_client.state.last_loaded_tray == 2
class TestTrayChangeLog:
"""Tests for tray_change_log tracking during prints (mid-print tray switch)."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TRAYLOG1",
access_code="12345678",
)
return client
def test_tray_change_log_defaults_empty(self, mqtt_client):
"""tray_change_log starts as an empty list."""
assert mqtt_client.state.tray_change_log == []
def test_tray_change_log_seeded_on_print_start(self, mqtt_client):
"""Print start clears log and seeds with initial tray at layer 0."""
mqtt_client.state.tray_now = 2
mqtt_client.state.last_loaded_tray = 2
mqtt_client._previous_gcode_state = "IDLE"
# Transition to RUNNING via _process_message
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "test.3mf",
}
}
)
assert mqtt_client.state.tray_change_log == [(2, 0)]
def test_tray_change_log_cleared_on_new_print(self, mqtt_client):
"""Old log entries are cleared when a new print starts."""
mqtt_client.state.tray_change_log = [(5, 0), (3, 100)]
mqtt_client.state.tray_now = 1
mqtt_client.state.last_loaded_tray = 1
mqtt_client._previous_gcode_state = "IDLE"
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"gcode_file": "new.3mf",
}
}
)
assert mqtt_client.state.tray_change_log == [(1, 0)]
def test_tray_change_recorded_during_running(self, mqtt_client):
"""Tray change while RUNNING is appended to the log."""
mqtt_client.state.state = "RUNNING"
mqtt_client.state.layer_num = 50
mqtt_client.state.last_loaded_tray = 0
mqtt_client.state.tray_change_log = [(0, 0)]
# Simulate tray_now update via AMS data
mqtt_client.state.tray_now = 1
# Trigger the tracking code path
tn = mqtt_client.state.tray_now
if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
mqtt_client.state.last_loaded_tray = tn
assert mqtt_client.state.tray_change_log == [(0, 0), (1, 50)]
def test_tray_change_not_recorded_when_idle(self, mqtt_client):
"""Tray changes while IDLE are NOT logged."""
mqtt_client.state.state = "IDLE"
mqtt_client.state.layer_num = 0
mqtt_client.state.last_loaded_tray = 0
mqtt_client.state.tray_change_log = []
mqtt_client.state.tray_now = 3
tn = mqtt_client.state.tray_now
if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
mqtt_client.state.last_loaded_tray = tn
assert mqtt_client.state.tray_change_log == []
def test_tray_change_recorded_during_pause(self, mqtt_client):
"""Tray change while PAUSE is also logged (AMS can swap during pause)."""
mqtt_client.state.state = "PAUSE"
mqtt_client.state.layer_num = 75
mqtt_client.state.last_loaded_tray = 2
mqtt_client.state.tray_change_log = [(2, 0)]
mqtt_client.state.tray_now = 5
tn = mqtt_client.state.tray_now
if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
mqtt_client.state.last_loaded_tray = tn
assert mqtt_client.state.tray_change_log == [(2, 0), (5, 75)]
def test_same_tray_not_logged_twice(self, mqtt_client):
"""Same tray value doesn't create duplicate log entries."""
mqtt_client.state.state = "RUNNING"
mqtt_client.state.layer_num = 30
mqtt_client.state.last_loaded_tray = 2
mqtt_client.state.tray_change_log = [(2, 0)]
# Same tray again
mqtt_client.state.tray_now = 2
tn = mqtt_client.state.tray_now
if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
mqtt_client.state.last_loaded_tray = tn
assert mqtt_client.state.tray_change_log == [(2, 0)]
def test_multiple_tray_changes(self, mqtt_client):
"""Multiple tray changes create a full history."""
mqtt_client.state.state = "RUNNING"
mqtt_client.state.last_loaded_tray = 0
mqtt_client.state.tray_change_log = [(0, 0)]
changes = [(1, 50), (3, 120), (0, 200)]
for tray, layer in changes:
mqtt_client.state.tray_now = tray
mqtt_client.state.layer_num = layer
tn = mqtt_client.state.tray_now
if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
mqtt_client.state.last_loaded_tray = tn
assert mqtt_client.state.tray_change_log == [(0, 0), (1, 50), (3, 120), (0, 200)]
class TestDeveloperModeDetection:
"""Tests for developer LAN mode detection from MQTT 'fun' field."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient instance for testing."""
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_developer_mode_initially_none(self, mqtt_client):
"""Verify developer_mode starts as None (unknown)."""
assert mqtt_client.state.developer_mode is None
def test_developer_mode_on_when_bit_clear(self, mqtt_client):
"""Verify developer_mode is True when bit 0x20000000 is clear."""
# Bit 29 clear in lower 32 bits = developer mode ON
payload = {
"print": {
"gcode_state": "IDLE",
"fun": "1C8187FF9CFF",
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.developer_mode is True
def test_developer_mode_off_when_bit_set(self, mqtt_client):
"""Verify developer_mode is False when bit 0x20000000 is set."""
# Bit 29 set in lower 32 bits = developer mode OFF (encryption required)
payload = {
"print": {
"gcode_state": "IDLE",
"fun": "1C81A7FF9CFF",
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.developer_mode is False
def test_developer_mode_exact_bit_check(self, mqtt_client):
"""Verify only bit 0x20000000 matters, not other bits."""
# 0x20000000 in hex = bit 29. Set ONLY that bit.
payload = {
"print": {
"gcode_state": "IDLE",
"fun": "000020000000",
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.developer_mode is False
# All zeros = all bits clear = developer mode ON
payload["print"]["fun"] = "000000000000"
mqtt_client._process_message(payload)
assert mqtt_client.state.developer_mode is True
def test_developer_mode_invalid_fun_ignored(self, mqtt_client):
"""Verify invalid fun values don't crash or change state."""
mqtt_client.state.developer_mode = True
payload = {
"print": {
"gcode_state": "IDLE",
"fun": "not_a_hex_value",
}
}
mqtt_client._process_message(payload)
# Should remain unchanged
assert mqtt_client.state.developer_mode is True
def test_developer_mode_missing_fun_preserves_state(self, mqtt_client):
"""Verify messages without fun field don't reset developer_mode."""
mqtt_client.state.developer_mode = False
payload = {
"print": {
"gcode_state": "RUNNING",
"mc_percent": 50,
}
}
mqtt_client._process_message(payload)
assert mqtt_client.state.developer_mode is False
def test_developer_mode_persists_across_messages(self, mqtt_client):
"""Verify developer_mode set by fun persists across messages without fun."""
# First message sets developer_mode
mqtt_client._process_message(
{
"print": {
"gcode_state": "IDLE",
"fun": "3EC1AFFF9CFF",
}
}
)
assert mqtt_client.state.developer_mode is False
# Subsequent messages without fun don't change it
for _ in range(3):
mqtt_client._process_message(
{
"print": {
"gcode_state": "RUNNING",
"mc_percent": 50,
}
}
)
assert mqtt_client.state.developer_mode is False
class TestDeveloperModeProbeTimeout:
"""Tests for developer mode probe timeout, retry, and forced reconnect (#887).
When a printer's MQTT session is half-broken (sends status but ignores
commands), the developer mode probe gets no response. The timeout logic
retries once, then force-closes the socket on the second failure.
"""
@pytest.fixture
def mqtt_client(self):
import time
from unittest.mock import MagicMock
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
# Simulate connected state with a mock MQTT client
client.state.connected = True
mock_paho = MagicMock()
mock_paho.socket.return_value = MagicMock()
client._client = mock_paho
# Set connect time in the past so the 5s probe delay is satisfied
client._connect_time = time.monotonic() - 10.0
return client
def _make_pushall_data(self):
"""Create a print data dict with >30 keys (triggers probe) and no 'fun' field."""
return {f"key_{i}": i for i in range(35)}
def test_first_timeout_allows_retry(self, mqtt_client):
"""After first probe timeout, _dev_mode_probed resets to allow retry."""
import time
data = self._make_pushall_data()
# First pushall triggers the probe
mqtt_client._update_state(data)
assert mqtt_client._dev_mode_probed is True
assert mqtt_client._dev_mode_probe_seq is not None
assert mqtt_client.state.developer_mode is None
# Simulate 11 seconds passing
mqtt_client._dev_mode_probe_time = time.monotonic() - 11.0
# Next status message detects the timeout
mqtt_client._update_state(data)
assert mqtt_client._dev_mode_probe_failures == 1
assert mqtt_client._dev_mode_probe_seq is None
# Should allow retry on next full message
assert mqtt_client._dev_mode_probed is False
# Connection should NOT be force-closed after 1 failure
assert mqtt_client.state.connected is True
def test_second_timeout_forces_reconnect(self, mqtt_client):
"""After two consecutive probe timeouts, force-close the socket."""
import time
data = self._make_pushall_data()
state_change_called = []
mqtt_client.on_state_change = lambda s: state_change_called.append(True)
# First probe + timeout
mqtt_client._update_state(data)
mqtt_client._dev_mode_probe_time = time.monotonic() - 11.0
mqtt_client._update_state(data)
assert mqtt_client._dev_mode_probe_failures == 1
# Second probe (retry) + timeout
mqtt_client._update_state(data) # triggers new probe
assert mqtt_client._dev_mode_probed is True
mqtt_client._dev_mode_probe_time = time.monotonic() - 11.0
mqtt_client._update_state(data) # detects second timeout
assert mqtt_client._dev_mode_probe_failures == 2
assert mqtt_client.state.connected is False
assert mqtt_client._stale_reconnecting is True
# Socket should have been closed
mqtt_client._client.socket().close.assert_called()
# on_state_change should have been called
assert len(state_change_called) > 0
def test_successful_probe_resets_failure_counter(self, mqtt_client):
"""A probe response after a previous failure resets the counter."""
import time
data = self._make_pushall_data()
# First probe + timeout → failure=1
mqtt_client._update_state(data)
seq = mqtt_client._dev_mode_probe_seq
mqtt_client._dev_mode_probe_time = time.monotonic() - 11.0
mqtt_client._update_state(data)
assert mqtt_client._dev_mode_probe_failures == 1
# Retry probe
mqtt_client._update_state(data)
new_seq = mqtt_client._dev_mode_probe_seq
assert new_seq is not None
assert new_seq != seq
# Simulate successful response
mqtt_client._handle_dev_mode_probe_response(
{
"command": "ams_filament_setting",
"sequence_id": new_seq,
"result": "success",
}
)
assert mqtt_client._dev_mode_probe_failures == 0
assert mqtt_client.state.developer_mode is True
assert mqtt_client._dev_mode_probe_seq is None
def test_no_timeout_when_probe_not_sent(self, mqtt_client):
"""The timeout branch is only entered when a probe is pending."""
# No probe sent — _dev_mode_probed is False, _dev_mode_probe_seq is None
data = {"gcode_state": "IDLE", "mc_percent": 0} # < 30 keys
mqtt_client._update_state(data)
assert mqtt_client._dev_mode_probe_failures == 0
def test_on_connect_resets_probe_state_but_preserves_developer_mode(self, mqtt_client):
"""_on_connect resets probe tracking but preserves cached developer_mode."""
import time
mqtt_client._dev_mode_probed = True
mqtt_client._dev_mode_probe_seq = "42"
mqtt_client._dev_mode_probe_time = time.monotonic()
mqtt_client._dev_mode_probe_failures = 2
mqtt_client.state.developer_mode = True
# subscribe() must return (result, mid) tuple
mqtt_client._client.subscribe.return_value = (0, 1)
mqtt_client._on_connect(mqtt_client._client, None, None, 0)
# developer_mode is preserved across reconnects (#887)
assert mqtt_client.state.developer_mode is True
assert mqtt_client._dev_mode_probed is False
assert mqtt_client._dev_mode_probe_seq is None
assert mqtt_client._dev_mode_probe_time == 0.0
assert mqtt_client._dev_mode_probe_failures == 0
assert mqtt_client._connect_time > 0
def test_probe_deferred_when_connect_too_recent(self, mqtt_client):
"""Probe is deferred if less than 5s have passed since _on_connect."""
import time
data = self._make_pushall_data()
# Set connect time to 1 second ago — too recent for probe
mqtt_client._connect_time = time.monotonic() - 1.0
mqtt_client._update_state(data)
# Pushall seen, so needs_probe is set, but probe NOT fired yet
assert mqtt_client._dev_mode_needs_probe is True
assert mqtt_client._dev_mode_probed is False
assert mqtt_client._dev_mode_probe_seq is None
def test_probe_fires_after_delay(self, mqtt_client):
"""Probe fires once 5s have passed since _on_connect."""
import time
data = self._make_pushall_data()
# Set connect time to 6 seconds ago — delay satisfied
mqtt_client._connect_time = time.monotonic() - 6.0
mqtt_client._update_state(data)
# Probe should have fired
assert mqtt_client._dev_mode_needs_probe is True
assert mqtt_client._dev_mode_probed is True
assert mqtt_client._dev_mode_probe_seq is not None
def test_probe_fires_on_incremental_after_delay(self, mqtt_client):
"""After seeing a pushall within 5s, probe fires on later incremental message."""
import time
pushall_data = self._make_pushall_data()
incremental_data = {"gcode_state": "IDLE", "mc_percent": 0} # < 30 keys
# Pushall arrives 1s after connect — too early for probe
mqtt_client._connect_time = time.monotonic() - 1.0
mqtt_client._update_state(pushall_data)
assert mqtt_client._dev_mode_needs_probe is True
assert mqtt_client._dev_mode_probed is False
# 5s later, an incremental update arrives — probe fires now
mqtt_client._connect_time = time.monotonic() - 6.0
mqtt_client._update_state(incremental_data)
assert mqtt_client._dev_mode_probed is True
assert mqtt_client._dev_mode_probe_seq is not None
def test_no_reprobe_when_developer_mode_cached(self, mqtt_client):
"""Auto-reconnect preserves developer_mode, skipping reprobe."""
import time
data = self._make_pushall_data()
# Simulate known developer_mode from previous connection
mqtt_client.state.developer_mode = True
mqtt_client._connect_time = time.monotonic() - 10.0
mqtt_client._update_state(data)
# Should NOT probe — developer_mode is already known
assert mqtt_client._dev_mode_needs_probe is False
assert mqtt_client._dev_mode_probed is False
assert mqtt_client._dev_mode_probe_seq is None
assert mqtt_client.state.developer_mode is True
def test_on_connect_resets_needs_probe(self, mqtt_client):
"""_on_connect resets _dev_mode_needs_probe for a clean start."""
mqtt_client._dev_mode_needs_probe = True
mqtt_client._client.subscribe.return_value = (0, 1)
mqtt_client._on_connect(mqtt_client._client, None, None, 0)
assert mqtt_client._dev_mode_needs_probe is False
class TestVtTrayNormalization:
"""Tests for vt_tray dict→list normalization in _update_state.
MQTT sends vt_tray as a dict for single-slot printers, but all consumers
expect a list. _update_state must normalize it before any callback can
read raw_data, because the dev-mode probe may release the GIL and let
the event loop read the partially-updated state.
"""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
return client
def test_vt_tray_dict_normalized_in_update_state(self, mqtt_client):
"""Verify _update_state wraps a raw vt_tray dict into a list."""
vt_dict = {
"id": "254",
"tray_color": "FF0000",
"tray_type": "PLA",
"tag_uid": "0000000000000000",
"tray_uuid": "00000000000000000000000000000000",
}
data = {"gcode_state": "IDLE", "vt_tray": vt_dict}
mqtt_client._update_state(data)
stored = mqtt_client.state.raw_data.get("vt_tray")
assert isinstance(stored, list)
assert len(stored) == 1
assert stored[0]["tray_color"] == "FF0000"
def test_vt_tray_list_unchanged_in_update_state(self, mqtt_client):
"""Verify _update_state keeps an already-list vt_tray unchanged."""
vt_list = [
{"id": "254", "tray_type": "PLA"},
{"id": "255", "tray_type": "PETG"},
]
data = {"gcode_state": "IDLE", "vt_tray": vt_list}
mqtt_client._update_state(data)
stored = mqtt_client.state.raw_data.get("vt_tray")
assert isinstance(stored, list)
assert len(stored) == 2
def test_preserved_vt_tray_restored_before_probe(self, mqtt_client):
"""Verify preserved vt_tray is restored before dev-mode probe runs.
On the first message, the incremental handler wraps vt_tray into a list
and stores it. _update_state then replaces raw_data with the full data
dict, but must restore preserved fields BEFORE the probe publishes
(which can release the GIL).
"""
# Simulate: incremental handler already stored a wrapped list
mqtt_client.state.raw_data = {
"vt_tray": [{"id": "254", "tray_type": "PLA", "tray_color": "00FF00"}],
}
# Now _update_state runs with new data that has vt_tray as dict
new_data = {
"gcode_state": "IDLE",
"vt_tray": {"id": "254", "tray_type": "PETG", "tray_color": "FF0000"},
}
mqtt_client._update_state(new_data)
# The preserved list (PLA/green) should take priority over new data
stored = mqtt_client.state.raw_data["vt_tray"]
assert isinstance(stored, list)
assert stored[0]["tray_type"] == "PLA"
assert stored[0]["tray_color"] == "00FF00"
def test_first_message_vt_tray_dict_becomes_list(self, mqtt_client):
"""Verify on the very first message, vt_tray dict is still a list.
When there's no previously preserved data, the normalized dict should
remain as a list in raw_data.
"""
# raw_data starts empty — no preserved vt_tray
mqtt_client.state.raw_data = {}
data = {
"gcode_state": "IDLE",
"vt_tray": {"id": "254", "tray_type": "ABS"},
}
mqtt_client._update_state(data)
stored = mqtt_client.state.raw_data["vt_tray"]
assert isinstance(stored, list)
assert stored[0]["tray_type"] == "ABS"
class TestSendDryingCommand:
"""Tests for send_drying_command MQTT payload construction."""
@pytest.fixture
def mqtt_client(self):
"""Create a BambuMQTTClient with a mock MQTT client."""
from unittest.mock import MagicMock
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
client._client = MagicMock()
return client
def test_rotate_tray_false_by_default(self, mqtt_client):
"""Verify rotate_tray defaults to False in the MQTT payload."""
mqtt_client.send_drying_command(ams_id=0, temp=55, duration=4, mode=1, filament="PLA")
call_args = mqtt_client._client.publish.call_args
payload = json.loads(call_args[0][1])
assert payload["print"]["rotate_tray"] is False
def test_rotate_tray_true_when_enabled(self, mqtt_client):
"""Verify rotate_tray is True when explicitly enabled."""
mqtt_client.send_drying_command(ams_id=0, temp=55, duration=4, mode=1, filament="PLA", rotate_tray=True)
call_args = mqtt_client._client.publish.call_args
payload = json.loads(call_args[0][1])
assert payload["print"]["rotate_tray"] is True
def test_rotate_tray_false_on_stop(self, mqtt_client):
"""Verify rotate_tray is False when stopping drying (mode=0)."""
mqtt_client.send_drying_command(ams_id=0, temp=0, duration=0, mode=0)
call_args = mqtt_client._client.publish.call_args
payload = json.loads(call_args[0][1])
assert payload["print"]["rotate_tray"] is False
def test_all_required_fields_present(self, mqtt_client):
"""Verify all required MQTT fields are present in the drying command."""
mqtt_client.send_drying_command(ams_id=128, temp=75, duration=8, mode=1, filament="ABS", rotate_tray=True)
call_args = mqtt_client._client.publish.call_args
payload = json.loads(call_args[0][1])
cmd = payload["print"]
assert cmd["command"] == "ams_filament_drying"
assert cmd["ams_id"] == 128
assert cmd["temp"] == 75
assert cmd["duration"] == 8
assert cmd["mode"] == 1
assert cmd["rotate_tray"] is True
assert cmd["filament"] == "ABS"
assert cmd["cooling_temp"] == 20
assert cmd["humidity"] == 0
assert cmd["close_power_conflict"] is False
assert "sequence_id" in cmd
def test_publishes_with_qos_1(self, mqtt_client):
"""Verify drying commands are published with QoS 1."""
mqtt_client.send_drying_command(ams_id=0, temp=55, duration=4)
call_args = mqtt_client._client.publish.call_args
# qos may be positional arg [2] or keyword
qos = call_args.kwargs.get("qos", call_args[0][2] if len(call_args[0]) > 2 else None)
assert qos == 1
class TestStartPrintAmsMapping:
"""Tests for ams_mapping/ams_mapping2 construction in start_print().
BambuStudio converts virtual tray IDs (254/255) to -1 in the flat
ams_mapping and puts the real external spool info only in ams_mapping2.
Passing raw 254/255 in the flat array causes H2D firmware to fail
with 0700_8012 "Failed to get AMS mapping table".
"""
@pytest.fixture
def mqtt_client(self):
from unittest.mock import MagicMock
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST123",
access_code="12345678",
)
client._client = MagicMock()
client.state.connected = True
return client
def _get_published_command(self, mqtt_client):
"""Extract the parsed print command from the last publish call."""
call_args = mqtt_client._client.publish.call_args
return json.loads(call_args[0][1])["print"]
def test_regular_ams_trays_preserved_in_flat_mapping(self, mqtt_client):
"""Regular AMS tray IDs pass through unchanged in flat ams_mapping."""
mqtt_client.start_print("test.3mf", ams_mapping=[0, 5, 11])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [0, 5, 11]
assert cmd["ams_mapping2"] == [
{"ams_id": 0, "slot_id": 0},
{"ams_id": 1, "slot_id": 1},
{"ams_id": 2, "slot_id": 3},
]
def test_unmapped_slots(self, mqtt_client):
"""Unmapped slots (-1) produce -1 in flat and 0xFF/0xFF in mapping2."""
mqtt_client.start_print("test.3mf", ams_mapping=[-1, -1])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1, -1]
assert cmd["ams_mapping2"] == [
{"ams_id": 255, "slot_id": 255},
{"ams_id": 255, "slot_id": 255},
]
def test_external_main_nozzle_becomes_minus_one_in_flat(self, mqtt_client):
"""Virtual tray 255 (main nozzle) must be -1 in flat mapping."""
mqtt_client.start_print("test.3mf", ams_mapping=[255])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1]
assert cmd["ams_mapping2"] == [{"ams_id": 255, "slot_id": 0}]
def test_single_nozzle_external_spool_uses_main_id(self, mqtt_client):
"""Single-nozzle external spool (254) maps to ams_id=255 (VIRTUAL_TRAY_MAIN_ID).
Firmware reports tray_now=254 for external spool, but the print command
must use ams_id=255 in ams_mapping2. Sending 254 causes the firmware to
target AMS tray 0 instead of external spool (07FF_8012 error).
"""
mqtt_client.start_print("test.3mf", ams_mapping=[254])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1]
assert cmd["ams_mapping2"] == [{"ams_id": 255, "slot_id": 0}]
def test_h2d_external_spool_mixed_with_ams(self, mqtt_client):
"""H2D scenario: AMS trays + unmapped + external deputy nozzle."""
# Reproduces the exact scenario from issue #797:
# 5-slot 3MF, only slot 5 assigned to external deputy nozzle (254)
mqtt_client.start_print("test.3mf", ams_mapping=[-1, -1, -1, -1, 255])
cmd = self._get_published_command(mqtt_client)
# Flat mapping: all -1 (external converted, unmapped stay -1)
assert cmd["ams_mapping"] == [-1, -1, -1, -1, -1]
# Detailed mapping: unmapped slots use 0xFF, external uses real ams_id
assert cmd["ams_mapping2"] == [
{"ams_id": 255, "slot_id": 255},
{"ams_id": 255, "slot_id": 255},
{"ams_id": 255, "slot_id": 255},
{"ams_id": 255, "slot_id": 255},
{"ams_id": 255, "slot_id": 0},
]
def test_ams_ht_trays_preserved_in_flat_mapping(self, mqtt_client):
"""AMS-HT tray IDs (>=128) pass through in flat mapping."""
mqtt_client.start_print("test.3mf", ams_mapping=[128, 131])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [128, 131]
assert cmd["ams_mapping2"] == [
{"ams_id": 128, "slot_id": 0},
{"ams_id": 131, "slot_id": 0},
]
def test_non_h2d_both_external_maps_to_main_id(self, mqtt_client):
"""Non-H2D: both 254 and 255 map to ams_id=255 (single nozzle)."""
mqtt_client.start_print("test.3mf", ams_mapping=[254, 255])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1, -1]
assert cmd["ams_mapping2"] == [
{"ams_id": 255, "slot_id": 0},
{"ams_id": 255, "slot_id": 0},
]
def test_h2d_external_preserves_deputy_id(self, mqtt_client):
"""H2D dual-nozzle: 254 (deputy) stays 254, 255 (main) stays 255."""
mqtt_client.model = "H2D"
mqtt_client.start_print("test.3mf", ams_mapping=[254, 255])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1, -1]
assert cmd["ams_mapping2"] == [
{"ams_id": 254, "slot_id": 0},
{"ams_id": 255, "slot_id": 0},
]
def test_h2d_single_external_deputy(self, mqtt_client):
"""H2D: single external spool on deputy nozzle (254) keeps ams_id=254."""
mqtt_client.model = "H2D Pro"
mqtt_client.start_print("test.3mf", ams_mapping=[254])
cmd = self._get_published_command(mqtt_client)
assert cmd["ams_mapping"] == [-1]
assert cmd["ams_mapping2"] == [{"ams_id": 254, "slot_id": 0}]
def test_external_spool_only_sets_use_ams_false(self, mqtt_client):
"""Single external spool on non-H2D printer sets use_ams=False."""
mqtt_client.start_print("test.3mf", ams_mapping=[254], use_ams=True)
cmd = self._get_published_command(mqtt_client)
assert cmd["use_ams"] is False
def test_all_unmapped_sets_use_ams_false(self, mqtt_client):
"""All unmapped slots on non-H2D printer sets use_ams=False."""
mqtt_client.start_print("test.3mf", ams_mapping=[-1, -1], use_ams=True)
cmd = self._get_published_command(mqtt_client)
assert cmd["use_ams"] is False
def test_mixed_ams_and_external_keeps_use_ams_true(self, mqtt_client):
"""AMS tray + external spool keeps use_ams=True."""
mqtt_client.start_print("test.3mf", ams_mapping=[0, 254], use_ams=True)
cmd = self._get_published_command(mqtt_client)
assert cmd["use_ams"] is True
def test_h2d_both_external_keeps_use_ams_true(self, mqtt_client):
"""H2D with both external spools keeps use_ams=True (nozzle routing)."""
mqtt_client.model = "H2D"
mqtt_client.start_print("test.3mf", ams_mapping=[254, 255], use_ams=True)
cmd = self._get_published_command(mqtt_client)
assert cmd["use_ams"] is True
def test_empty_ams_mapping_keeps_use_ams_true(self, mqtt_client):
"""Empty ams_mapping list does not override use_ams."""
mqtt_client.start_print("test.3mf", ams_mapping=[], use_ams=True)
cmd = self._get_published_command(mqtt_client)
assert cmd["use_ams"] is True
def test_no_ams_mapping_omits_fields(self, mqtt_client):
"""When ams_mapping is None, neither field is in the command."""
mqtt_client.start_print("test.3mf", ams_mapping=None)
cmd = self._get_published_command(mqtt_client)
assert "ams_mapping" not in cmd
assert "ams_mapping2" not in cmd
class TestStaleReconnect:
"""Tests for stale connection detection and reconnect without UI bouncing."""
@pytest.fixture
def mqtt_client(self):
from backend.app.services.bambu_mqtt import BambuMQTTClient
client = BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST_STALE",
access_code="12345678",
)
return client
def test_check_staleness_sets_flag_and_broadcasts_once(self, mqtt_client):
"""check_staleness() should set connected=False, broadcast, and set _stale_reconnecting."""
import time
state_changes = []
mqtt_client.on_state_change = lambda s: state_changes.append(s.connected)
mqtt_client.state.connected = True
mqtt_client._last_message_time = time.time() - 120 # well past 60s threshold
result = mqtt_client.check_staleness()
assert result is False
assert mqtt_client.state.connected is False
assert mqtt_client._stale_reconnecting is True
assert state_changes == [False] # Exactly one broadcast
def test_check_staleness_noop_when_not_connected(self, mqtt_client):
"""check_staleness() should not set flag when already disconnected."""
import time
mqtt_client.state.connected = False
mqtt_client._last_message_time = time.time() - 120
mqtt_client.check_staleness()
assert mqtt_client._stale_reconnecting is False
def test_check_staleness_noop_when_not_stale(self, mqtt_client):
"""check_staleness() should not set flag when messages are recent."""
import time
mqtt_client.state.connected = True
mqtt_client._last_message_time = time.time() - 5 # 5s ago, well within 60s
result = mqtt_client.check_staleness()
assert result is True
assert mqtt_client.state.connected is True
assert mqtt_client._stale_reconnecting is False
def test_on_disconnect_skipped_during_stale_reconnect(self, mqtt_client):
"""_on_disconnect should not broadcast state when _stale_reconnecting is set."""
state_changes = []
mqtt_client.on_state_change = lambda s: state_changes.append(s.connected)
mqtt_client._stale_reconnecting = True
mqtt_client.state.connected = False
mqtt_client._on_disconnect(None, None)
# No state change broadcast — check_staleness() already did it
assert state_changes == []
assert mqtt_client.state.connected is False
def test_on_disconnect_fires_event_during_stale_reconnect(self, mqtt_client):
"""_on_disconnect must still fire _disconnection_event even during stale reconnect.
If disconnect() is called while _stale_reconnecting is True (e.g. user removes
the printer before paho reconnects), the event must fire so disconnect() doesn't hang.
"""
import threading
mqtt_client._stale_reconnecting = True
mqtt_client._disconnection_event = threading.Event()
mqtt_client._on_disconnect(None, None)
assert mqtt_client._disconnection_event.is_set()
def test_on_connect_clears_stale_reconnecting_flag(self, mqtt_client):
"""_on_connect should clear _stale_reconnecting and restore connected=True."""
mqtt_client._stale_reconnecting = True
mqtt_client.state.connected = False
subscribe_calls = []
mock_client = type(
"MockClient",
(),
{
"subscribe": lambda self, topic: subscribe_calls.append(topic) or (0, 1),
},
)()
mqtt_client._on_connect(mock_client, None, None, 0)
assert mqtt_client._stale_reconnecting is False
assert mqtt_client.state.connected is True
def test_full_stale_reconnect_cycle_no_bounce(self, mqtt_client):
"""Full cycle: stale → disconnect callback → reconnect. UI should see exactly one disconnect."""
import time
state_changes = []
mqtt_client.on_state_change = lambda s: state_changes.append(s.connected)
mqtt_client.state.connected = True
mqtt_client._last_message_time = time.time() - 120
# Step 1: Stale detection triggers
mqtt_client.check_staleness()
assert state_changes == [False]
# Step 2: Paho fires disconnect callback (from socket close)
mqtt_client._on_disconnect(None, None)
# Should NOT add another state change
assert state_changes == [False]
# Step 3: Paho reconnects
subscribe_calls = []
mock_client = type(
"MockClient",
(),
{
"subscribe": lambda self, topic: subscribe_calls.append(topic) or (0, 1),
},
)()
mqtt_client._on_connect(mock_client, None, None, 0)
assert state_changes == [False, True] # Now connected again
assert mqtt_client._stale_reconnecting is False
def test_spurious_disconnect_suppressed_when_recent_messages(self, mqtt_client):
"""Non-error disconnect with recent messages should be suppressed."""
import time
state_changes = []
mqtt_client.on_state_change = lambda s: state_changes.append(s.connected)
mqtt_client.state.connected = True
mqtt_client._last_message_time = time.time() - 3 # 3s ago
# Non-error disconnect (rc=None)
mqtt_client._on_disconnect(None, None)
assert state_changes == []
assert mqtt_client.state.connected is True
def test_error_disconnect_not_suppressed_despite_recent_messages(self, mqtt_client):
"""Error disconnect should always be processed, even with recent messages."""
import time
import paho.mqtt.client as mqtt
from paho.mqtt.reasoncodes import ReasonCode
state_changes = []
mqtt_client.on_state_change = lambda s: state_changes.append(s.connected)
mqtt_client.state.connected = True
mqtt_client._last_message_time = time.time() - 3 # 3s ago
# Error disconnect (rc.is_failure = True)
rc = ReasonCode(mqtt.CONNACK >> 4, identifier=0x80) # Failure code
mqtt_client._on_disconnect(None, None, rc=rc)
assert state_changes == [False]
assert mqtt_client.state.connected is False
class TestDoorOpenParsing:
"""Tests for enclosure door state parsing (X1 home_flag bit 23 vs others stat bit 23)."""
def _make_client(self, model: str):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST",
access_code="12345678",
model=model,
)
def test_x1c_door_open_from_home_flag(self):
client = self._make_client("X1C")
# bit 23 set
client._update_state({"home_flag": 0xC0E5CD98})
assert client.state.door_open is True
def test_x1c_door_closed_from_home_flag(self):
client = self._make_client("X1C")
client.state.door_open = True # start "open"
client._update_state({"home_flag": 0xC065CD98})
assert client.state.door_open is False
def test_x1c_ignores_stat_field(self):
# X1C must NOT use stat (bit 23 in stat is unrelated for X1)
client = self._make_client("X1C")
client._update_state({"home_flag": 0xC065CD98, "stat": "47A58000"})
assert client.state.door_open is False # home_flag wins
def test_h2d_door_open_from_stat(self):
client = self._make_client("H2D")
client._update_state({"stat": "640A58000"}) # bit 23 set
assert client.state.door_open is True
def test_h2d_door_closed_from_stat(self):
client = self._make_client("H2D")
client.state.door_open = True
client._update_state({"stat": "640258000"}) # bit 23 cleared
assert client.state.door_open is False
def test_h2d_ignores_home_flag(self):
# Non-X1 must NOT consume home_flag for door state
client = self._make_client("H2D")
client._update_state({"home_flag": 0xC0E5CD98, "stat": "640258000"})
assert client.state.door_open is False # stat wins
def test_invalid_stat_does_not_raise(self):
client = self._make_client("H2D")
client._update_state({"stat": "not-hex"})
assert client.state.door_open is False
class TestSdCardParsing:
"""SD-card state is only set from the top-level `sdcard` field (bool/int/
string variants). home_flag is NOT consulted — heartbeat pushes clear those
bits even when a card is inserted, and the prior badge feature was removed
entirely because no reliable heartbeat-vs-full-push heuristic existed."""
def _make_client(self, model: str = "H2D"):
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TEST",
access_code="12345678",
model=model,
)
def test_home_flag_alone_does_not_touch_sdcard(self):
client = self._make_client()
client.state.sdcard = True
for home_flag in (0x00000000, 0x00000100, 0x00000200):
client._update_state({"home_flag": home_flag})
assert client.state.sdcard is True
def test_sdcard_string_fallback_when_no_home_flag(self):
client = self._make_client()
client._update_state({"sdcard": "HAS_SDCARD_NORMAL"})
assert client.state.sdcard is True
def test_sdcard_int_fallback_when_no_home_flag(self):
# `1 is True` is False — the old strict check flapped here.
client = self._make_client()
client._update_state({"sdcard": 1})
assert client.state.sdcard is True
def test_sdcard_bool_fallback_when_no_home_flag(self):
client = self._make_client()
client._update_state({"sdcard": True})
assert client.state.sdcard is True
client._update_state({"sdcard": False})
assert client.state.sdcard is False