"""Unit tests for PrinterManager service. Tests printer connection management, status tracking, and print control. """ import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest from backend.app.services.printer_manager import ( PrinterManager, get_derived_status_name, has_stg_cur_idle_bug, init_printer_connections, parse_plate_id, printer_state_to_dict, supports_chamber_temp, supports_drying, ) class TestPrinterManager: """Tests for PrinterManager class.""" @pytest.fixture def manager(self): """Create a fresh PrinterManager instance.""" return PrinterManager() @pytest.fixture def mock_printer(self): """Create a mock Printer object.""" printer = MagicMock() printer.id = 1 printer.ip_address = "192.168.1.100" printer.serial_number = "00M09A123456789" printer.access_code = "12345678" printer.is_active = True return printer @pytest.fixture def mock_client(self): """Create a mock BambuMQTTClient.""" client = MagicMock() client.state = MagicMock() client.state.connected = True client.state.state = "IDLE" client.state.progress = 0 client.state.temperatures = {"nozzle": 25, "bed": 25} client.state.raw_data = {} client.logging_enabled = False return client # ======================================================================== # Tests for initialization # ======================================================================== def test_init_creates_empty_clients_dict(self, manager): """Verify manager initializes with empty clients dict.""" assert manager._clients == {} def test_init_callbacks_are_none(self, manager): """Verify all callbacks are initially None.""" assert manager._on_print_start is None assert manager._on_print_complete is None assert manager._on_status_change is None assert manager._on_ams_change is None def test_init_loop_is_none(self, manager): """Verify event loop is initially None.""" assert manager._loop is None # ======================================================================== # Tests for callback setters # ======================================================================== def test_set_event_loop(self, manager): """Verify event loop can be set.""" mock_loop = MagicMock() manager.set_event_loop(mock_loop) assert manager._loop == mock_loop def test_set_print_start_callback(self, manager): """Verify print start callback can be set.""" callback = MagicMock() manager.set_print_start_callback(callback) assert manager._on_print_start == callback def test_set_print_complete_callback(self, manager): """Verify print complete callback can be set.""" callback = MagicMock() manager.set_print_complete_callback(callback) assert manager._on_print_complete == callback def test_set_status_change_callback(self, manager): """Verify status change callback can be set.""" callback = MagicMock() manager.set_status_change_callback(callback) assert manager._on_status_change == callback def test_set_ams_change_callback(self, manager): """Verify AMS change callback can be set.""" callback = MagicMock() manager.set_ams_change_callback(callback) assert manager._on_ams_change == callback # ======================================================================== # Tests for _schedule_async # ======================================================================== def test_schedule_async_with_running_loop(self, manager): """Verify async coroutine is scheduled when loop is running.""" mock_loop = MagicMock() mock_loop.is_running.return_value = True manager._loop = mock_loop async def dummy_coro(): pass coro = dummy_coro() manager._schedule_async(coro) mock_loop.is_running.assert_called_once() # Clean up the coroutine coro.close() def test_schedule_async_without_loop(self, manager): """Verify nothing happens when no loop is set.""" async def dummy_coro(): pass coro = dummy_coro() # Should not raise manager._schedule_async(coro) coro.close() def test_schedule_async_with_stopped_loop(self, manager): """Verify nothing happens when loop is not running.""" mock_loop = MagicMock() mock_loop.is_running.return_value = False manager._loop = mock_loop async def dummy_coro(): pass coro = dummy_coro() manager._schedule_async(coro) coro.close() # ======================================================================== # Tests for connect_printer # ======================================================================== @pytest.mark.asyncio async def test_connect_printer_creates_client(self, manager, mock_printer): """Verify connecting creates an MQTT client.""" with patch("backend.app.services.printer_manager.BambuMQTTClient") as MockClient: mock_instance = MagicMock() mock_instance.state = MagicMock() mock_instance.state.connected = True MockClient.return_value = mock_instance result = await manager.connect_printer(mock_printer) MockClient.assert_called_once() mock_instance.connect.assert_called_once() assert mock_printer.id in manager._clients assert result is True @pytest.mark.asyncio async def test_connect_printer_disconnects_existing(self, manager, mock_printer, mock_client): """Verify connecting disconnects existing client first.""" manager._clients[mock_printer.id] = mock_client with patch("backend.app.services.printer_manager.BambuMQTTClient") as MockClient: new_client = MagicMock() new_client.state = MagicMock() new_client.state.connected = True MockClient.return_value = new_client await manager.connect_printer(mock_printer) mock_client.disconnect.assert_called_once() @pytest.mark.asyncio async def test_connect_printer_returns_false_on_failure(self, manager, mock_printer): """Verify returns False when connection fails.""" with patch("backend.app.services.printer_manager.BambuMQTTClient") as MockClient: mock_instance = MagicMock() mock_instance.state = MagicMock() mock_instance.state.connected = False MockClient.return_value = mock_instance result = await manager.connect_printer(mock_printer) assert result is False # ======================================================================== # Tests for disconnect_printer # ======================================================================== def test_disconnect_printer_removes_client(self, manager, mock_client): """Verify disconnecting removes and disconnects client.""" manager._clients[1] = mock_client manager.disconnect_printer(1) mock_client.disconnect.assert_called_once() assert 1 not in manager._clients def test_disconnect_printer_handles_missing(self, manager): """Verify disconnecting non-existent printer doesn't raise.""" manager.disconnect_printer(999) # Should not raise # ======================================================================== # Tests for disconnect_all # ======================================================================== def test_disconnect_all_disconnects_all_clients(self, manager): """Verify all clients are disconnected.""" client1 = MagicMock() client2 = MagicMock() manager._clients[1] = client1 manager._clients[2] = client2 manager.disconnect_all() client1.disconnect.assert_called_once() client2.disconnect.assert_called_once() assert len(manager._clients) == 0 # ======================================================================== # Tests for get_status # ======================================================================== def test_get_status_returns_state(self, manager, mock_client): """Verify get_status returns client state.""" manager._clients[1] = mock_client result = manager.get_status(1) mock_client.check_staleness.assert_called_once() assert result == mock_client.state def test_get_status_returns_none_for_unknown(self, manager): """Verify get_status returns None for unknown printer.""" result = manager.get_status(999) assert result is None # ======================================================================== # Tests for get_all_statuses # ======================================================================== def test_get_all_statuses_returns_all(self, manager): """Verify all statuses are returned.""" client1 = MagicMock() client1.state = MagicMock(connected=True) client2 = MagicMock() client2.state = MagicMock(connected=False) manager._clients[1] = client1 manager._clients[2] = client2 result = manager.get_all_statuses() assert len(result) == 2 assert 1 in result assert 2 in result client1.check_staleness.assert_called_once() client2.check_staleness.assert_called_once() # ======================================================================== # Tests for is_connected # ======================================================================== def test_is_connected_returns_true(self, manager, mock_client): """Verify is_connected returns True for connected printer.""" mock_client.check_staleness.return_value = True manager._clients[1] = mock_client result = manager.is_connected(1) assert result is True def test_is_connected_returns_false_for_unknown(self, manager): """Verify is_connected returns False for unknown printer.""" result = manager.is_connected(999) assert result is False # ======================================================================== # Tests for get_client # ======================================================================== def test_get_client_returns_client(self, manager, mock_client): """Verify get_client returns the client.""" manager._clients[1] = mock_client result = manager.get_client(1) assert result == mock_client def test_get_client_returns_none_for_unknown(self, manager): """Verify get_client returns None for unknown printer.""" result = manager.get_client(999) assert result is None # ======================================================================== # Tests for mark_printer_offline # ======================================================================== def test_mark_printer_offline_updates_state(self, manager, mock_client): """Verify mark_printer_offline updates client state.""" mock_client.state.connected = True manager._clients[1] = mock_client manager.mark_printer_offline(1) assert mock_client.state.connected is False assert mock_client.state.state == "unknown" def test_mark_printer_offline_triggers_callback(self, manager, mock_client): """Verify mark_printer_offline triggers status callback.""" mock_client.state.connected = True manager._clients[1] = mock_client # Callback must return a coroutine async def async_callback(printer_id, state): pass manager._on_status_change = async_callback # Need a running loop for callback mock_loop = MagicMock() mock_loop.is_running.return_value = True manager._loop = mock_loop manager.mark_printer_offline(1) # Callback should be scheduled via run_coroutine_threadsafe mock_loop.is_running.assert_called() # State should be updated assert mock_client.state.connected is False def test_mark_printer_offline_handles_unknown(self, manager): """Verify mark_printer_offline handles unknown printer.""" manager.mark_printer_offline(999) # Should not raise def test_mark_printer_offline_skips_already_offline(self, manager, mock_client): """Verify mark_printer_offline skips already offline printer.""" mock_client.state.connected = False manager._clients[1] = mock_client manager.mark_printer_offline(1) # State should remain unchanged assert mock_client.state.connected is False # ======================================================================== # Tests for start_print # ======================================================================== def test_start_print_calls_client(self, manager, mock_client): """Verify start_print calls client method.""" mock_client.start_print.return_value = True manager._clients[1] = mock_client result = manager.start_print(1, "test.gcode") mock_client.start_print.assert_called_once_with( "test.gcode", 1, ams_mapping=None, timelapse=False, bed_levelling=True, flow_cali=False, vibration_cali=True, layer_inspect=False, use_ams=True, ) assert result is True def test_start_print_returns_false_for_unknown(self, manager): """Verify start_print returns False for unknown printer.""" result = manager.start_print(999, "test.gcode") assert result is False def test_start_print_logs_print_command_with_caller(self, manager, mock_client, caplog): """Verify start_print logs PRINT COMMAND with caller info (#374).""" mock_client.start_print.return_value = True manager._clients[1] = mock_client with caplog.at_level(logging.INFO, logger="backend.app.services.printer_manager"): manager.start_print(1, "benchy.3mf") print_cmd_logs = [r for r in caplog.records if "PRINT COMMAND" in r.message] assert len(print_cmd_logs) == 1 log_msg = print_cmd_logs[0].message assert "printer=1" in log_msg assert "file=benchy.3mf" in log_msg assert "caller=" in log_msg def test_start_print_logs_even_when_printer_unknown(self, manager, caplog): """Verify PRINT COMMAND is logged even for unknown printers (#374).""" with caplog.at_level(logging.INFO, logger="backend.app.services.printer_manager"): result = manager.start_print(999, "ghost.3mf") assert result is False print_cmd_logs = [r for r in caplog.records if "PRINT COMMAND" in r.message] assert len(print_cmd_logs) == 1 # ======================================================================== # Tests for stop_print # ======================================================================== def test_stop_print_calls_client(self, manager, mock_client): """Verify stop_print calls client method.""" mock_client.stop_print.return_value = True manager._clients[1] = mock_client result = manager.stop_print(1) mock_client.stop_print.assert_called_once() assert result is True def test_stop_print_returns_false_for_unknown(self, manager): """Verify stop_print returns False for unknown printer.""" result = manager.stop_print(999) assert result is False # ======================================================================== # Tests for wait_for_cooldown # ======================================================================== @pytest.mark.asyncio async def test_wait_for_cooldown_returns_true_when_cool(self, manager, mock_client): """Verify wait_for_cooldown returns True when printer is cool.""" mock_client.state.connected = True mock_client.state.temperatures = {"nozzle": 40, "bed": 30} mock_client.check_staleness.return_value = True manager._clients[1] = mock_client result = await manager.wait_for_cooldown(1, target_temp=50) assert result is True @pytest.mark.asyncio async def test_wait_for_cooldown_returns_false_on_disconnect(self, manager, mock_client): """Verify wait_for_cooldown returns False when printer disconnects.""" mock_client.state.connected = False mock_client.check_staleness.return_value = False manager._clients[1] = mock_client result = await manager.wait_for_cooldown(1, target_temp=50, timeout=1) assert result is False @pytest.mark.asyncio async def test_wait_for_cooldown_returns_false_for_unknown(self, manager): """Verify wait_for_cooldown returns False for unknown printer.""" result = await manager.wait_for_cooldown(999, target_temp=50, timeout=1) assert result is False @pytest.mark.asyncio async def test_wait_for_cooldown_checks_both_nozzles(self, manager, mock_client): """Verify wait_for_cooldown checks both nozzles for dual extruders.""" mock_client.state.connected = True mock_client.state.temperatures = {"nozzle": 40, "nozzle_2": 45, "bed": 30} mock_client.check_staleness.return_value = True manager._clients[1] = mock_client result = await manager.wait_for_cooldown(1, target_temp=50) assert result is True # ======================================================================== # Tests for logging methods # ======================================================================== def test_enable_logging_calls_client(self, manager, mock_client): """Verify enable_logging calls client method.""" manager._clients[1] = mock_client result = manager.enable_logging(1, True) mock_client.enable_logging.assert_called_once_with(True) assert result is True def test_enable_logging_returns_false_for_unknown(self, manager): """Verify enable_logging returns False for unknown printer.""" result = manager.enable_logging(999, True) assert result is False def test_get_logs_returns_logs(self, manager, mock_client): """Verify get_logs returns client logs.""" mock_logs = [MagicMock(), MagicMock()] mock_client.get_logs.return_value = mock_logs manager._clients[1] = mock_client result = manager.get_logs(1) assert result == mock_logs def test_get_logs_returns_empty_for_unknown(self, manager): """Verify get_logs returns empty list for unknown printer.""" result = manager.get_logs(999) assert result == [] def test_clear_logs_calls_client(self, manager, mock_client): """Verify clear_logs calls client method.""" manager._clients[1] = mock_client result = manager.clear_logs(1) mock_client.clear_logs.assert_called_once() assert result is True def test_clear_logs_returns_false_for_unknown(self, manager): """Verify clear_logs returns False for unknown printer.""" result = manager.clear_logs(999) assert result is False def test_is_logging_enabled_returns_status(self, manager, mock_client): """Verify is_logging_enabled returns client status.""" mock_client.logging_enabled = True manager._clients[1] = mock_client result = manager.is_logging_enabled(1) assert result is True def test_is_logging_enabled_returns_false_for_unknown(self, manager): """Verify is_logging_enabled returns False for unknown printer.""" result = manager.is_logging_enabled(999) assert result is False # ======================================================================== # Tests for request_status_update # ======================================================================== def test_request_status_update_calls_client(self, manager, mock_client): """Verify request_status_update calls client method.""" mock_client.request_status_update.return_value = True manager._clients[1] = mock_client result = manager.request_status_update(1) mock_client.request_status_update.assert_called_once() assert result is True def test_request_status_update_returns_false_for_unknown(self, manager): """Verify request_status_update returns False for unknown printer.""" result = manager.request_status_update(999) assert result is False # ======================================================================== # Tests for test_connection # ======================================================================== @pytest.mark.asyncio async def test_test_connection_success(self, manager): """Verify test_connection returns success on connection.""" with patch("backend.app.services.printer_manager.BambuMQTTClient") as MockClient: mock_instance = MagicMock() mock_instance.state = MagicMock() mock_instance.state.connected = True mock_instance.state.state = "IDLE" mock_instance.state.raw_data = {"device_model": "X1C"} MockClient.return_value = mock_instance result = await manager.test_connection("192.168.1.100", "00M09A123456789", "12345678") assert result["success"] is True assert result["state"] == "IDLE" assert result["model"] == "X1C" mock_instance.disconnect.assert_called_once() @pytest.mark.asyncio async def test_test_connection_failure(self, manager): """Verify test_connection returns failure on connection error.""" with patch("backend.app.services.printer_manager.BambuMQTTClient") as MockClient: mock_instance = MagicMock() mock_instance.state = MagicMock() mock_instance.state.connected = False MockClient.return_value = mock_instance result = await manager.test_connection("192.168.1.100", "00M09A123456789", "12345678") assert result["success"] is False assert result["state"] is None mock_instance.disconnect.assert_called_once() # ======================================================================== # Tests for current print user tracking (Issue #206) # ======================================================================== def test_set_current_print_user(self, manager): """Verify current print user can be set.""" manager.set_current_print_user(1, 42, "testuser") assert 1 in manager._current_print_user assert manager._current_print_user[1]["user_id"] == 42 assert manager._current_print_user[1]["username"] == "testuser" def test_get_current_print_user_returns_user(self, manager): """Verify get_current_print_user returns the stored user.""" manager.set_current_print_user(1, 42, "testuser") result = manager.get_current_print_user(1) assert result is not None assert result["user_id"] == 42 assert result["username"] == "testuser" def test_get_current_print_user_returns_none_for_unknown(self, manager): """Verify get_current_print_user returns None for unknown printer.""" result = manager.get_current_print_user(999) assert result is None def test_clear_current_print_user(self, manager): """Verify current print user can be cleared.""" manager.set_current_print_user(1, 42, "testuser") manager.clear_current_print_user(1) result = manager.get_current_print_user(1) assert result is None def test_clear_current_print_user_no_error_for_unknown(self, manager): """Verify clearing unknown printer doesn't raise error.""" # Should not raise manager.clear_current_print_user(999) def test_set_current_print_user_overwrites_existing(self, manager): """Verify setting user overwrites existing value.""" manager.set_current_print_user(1, 42, "user1") manager.set_current_print_user(1, 99, "user2") result = manager.get_current_print_user(1) assert result["user_id"] == 99 assert result["username"] == "user2" def test_multiple_printers_have_separate_users(self, manager): """Verify each printer tracks its own user separately.""" manager.set_current_print_user(1, 42, "user1") manager.set_current_print_user(2, 99, "user2") result1 = manager.get_current_print_user(1) result2 = manager.get_current_print_user(2) assert result1["username"] == "user1" assert result2["username"] == "user2" class TestPrinterStateToDict: """Tests for printer_state_to_dict helper function.""" @pytest.fixture def mock_state(self): """Create a mock PrinterState.""" state = MagicMock() state.connected = True state.state = "RUNNING" state.current_print = "test.3mf" state.subtask_name = "Test Print" state.gcode_file = "/sdcard/test.gcode" state.progress = 50 state.remaining_time = 3600 state.layer_num = 10 state.total_layers = 20 state.temperatures = {"nozzle": 200, "bed": 60} state.hms_errors = [] state.ams_status_main = 0 state.ams_status_sub = 0 state.tray_now = "1" state.wifi_signal = -50 state.raw_data = {} state.stg_cur = -1 # No calibration stage active state.firmware_version = None return state def test_basic_conversion(self, mock_state): """Verify basic state fields are converted.""" result = printer_state_to_dict(mock_state) assert result["connected"] is True assert result["state"] == "RUNNING" assert result["progress"] == 50 assert result["temperatures"] == {"nozzle": 200, "bed": 60} def test_ams_data_parsing(self, mock_state): """Verify AMS data is parsed correctly.""" mock_state.raw_data = { "ams": [ { "id": 0, "humidity_raw": 45, "temp": 25, "tray": [ { "id": 0, "tray_color": "FF0000", "tray_type": "PLA", "tray_sub_brands": "Generic", "remain": 80, "k": 0.5, "tag_uid": "ABC123", "tray_uuid": "uuid-123", } ], } ] } result = printer_state_to_dict(mock_state) assert result["ams"] is not None assert len(result["ams"]) == 1 assert result["ams"][0]["humidity"] == 45 assert len(result["ams"][0]["tray"]) == 1 assert result["ams"][0]["tray"][0]["tray_color"] == "FF0000" def test_empty_tag_uid_becomes_none(self, mock_state): """Verify empty tag_uid is converted to None.""" mock_state.raw_data = { "ams": [ { "id": 0, "tray": [ { "id": 0, "tag_uid": "", "tray_uuid": "00000000000000000000000000000000", } ], } ] } result = printer_state_to_dict(mock_state) assert result["ams"][0]["tray"][0]["tag_uid"] is None assert result["ams"][0]["tray"][0]["tray_uuid"] is None def test_zero_tag_uid_becomes_none(self, mock_state): """Verify zero tag_uid is converted to None.""" mock_state.raw_data = { "ams": [ { "id": 0, "tray": [ { "id": 0, "tag_uid": "0000000000000000", } ], } ] } result = printer_state_to_dict(mock_state) assert result["ams"][0]["tray"][0]["tag_uid"] is None def test_vt_tray_parsing(self, mock_state): """Verify virtual tray is parsed correctly as a list.""" mock_state.raw_data = { "vt_tray": [ { "tray_color": "00FF00", "tray_type": "PETG", "tray_sub_brands": "Generic", "remain": 60, "tag_uid": "VT123", } ] } result = printer_state_to_dict(mock_state) assert isinstance(result["vt_tray"], list) assert len(result["vt_tray"]) == 1 assert result["vt_tray"][0]["id"] == 254 assert result["vt_tray"][0]["tray_color"] == "00FF00" assert result["vt_tray"][0]["tray_type"] == "PETG" def test_vt_tray_dict_normalized_to_list(self, mock_state): """Verify vt_tray as a raw dict (from MQTT) is normalized to a list.""" mock_state.raw_data = { "vt_tray": { "id": "254", "tray_color": "FF0000", "tray_type": "PLA", "tray_sub_brands": "Generic", "tag_uid": "0000000000000000", "tray_uuid": "00000000000000000000000000000000", "remain": 0, } } result = printer_state_to_dict(mock_state) assert isinstance(result["vt_tray"], list) assert len(result["vt_tray"]) == 1 assert result["vt_tray"][0]["tray_color"] == "FF0000" assert result["vt_tray"][0]["tray_type"] == "PLA" assert result["vt_tray"][0]["tag_uid"] is None assert result["vt_tray"][0]["tray_uuid"] is None def test_vt_tray_non_list_non_dict_ignored(self, mock_state): """Verify unexpected vt_tray types (e.g. string) produce empty list.""" mock_state.raw_data = {"vt_tray": "unexpected_string"} result = printer_state_to_dict(mock_state) assert result["vt_tray"] == [] def test_hms_errors_conversion(self, mock_state): """Verify HMS errors are converted correctly.""" error = MagicMock() error.code = "0700_0100" error.attr = 1 error.module = "AMS" error.severity = 2 mock_state.hms_errors = [error] result = printer_state_to_dict(mock_state) assert len(result["hms_errors"]) == 1 assert result["hms_errors"][0]["code"] == "0700_0100" assert result["hms_errors"][0]["module"] == "AMS" def test_cover_url_added_for_running_print(self, mock_state): """Verify cover_url is added for running prints.""" result = printer_state_to_dict(mock_state, printer_id=1) assert result["cover_url"] == "/api/v1/printers/1/cover" def test_current_plate_id_extracted_from_gcode_file(self, mock_state): """Verify current_plate_id is parsed from a Bambu plate path (#881).""" mock_state.gcode_file = "/Metadata/plate_3.gcode" result = printer_state_to_dict(mock_state) assert result["current_plate_id"] == 3 def test_current_plate_id_none_when_no_plate_segment(self, mock_state): """Verify current_plate_id stays None when gcode_file has no plate marker.""" mock_state.gcode_file = "/sdcard/test.gcode" result = printer_state_to_dict(mock_state) assert result["current_plate_id"] is None def test_cover_url_none_when_not_running(self, mock_state): """Verify cover_url is None when not printing.""" mock_state.state = "IDLE" result = printer_state_to_dict(mock_state, printer_id=1) assert result["cover_url"] is None def test_ams_ht_detection(self, mock_state): """Verify AMS-HT is detected (1 tray vs 4).""" mock_state.raw_data = { "ams": [ { "id": 0, "tray": [{"id": 0}], # Only 1 tray = AMS-HT } ] } result = printer_state_to_dict(mock_state) assert result["ams"][0]["is_ams_ht"] is True def test_regular_ams_detection(self, mock_state): """Verify regular AMS is detected (4 trays).""" mock_state.raw_data = {"ams": [{"id": 0, "tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]}]} result = printer_state_to_dict(mock_state) assert result["ams"][0]["is_ams_ht"] is False def test_chamber_temp_filtered_for_p1s(self, mock_state): """Verify chamber temperature is filtered out for P1S (no chamber sensor).""" mock_state.temperatures = { "nozzle": 200, "bed": 60, "chamber": 5, "chamber_target": 0, "chamber_heating": False, } result = printer_state_to_dict(mock_state, model="P1S") assert "chamber" not in result["temperatures"] assert "chamber_target" not in result["temperatures"] assert "chamber_heating" not in result["temperatures"] assert result["temperatures"]["nozzle"] == 200 assert result["temperatures"]["bed"] == 60 def test_chamber_temp_kept_for_x1c(self, mock_state): """Verify chamber temperature is kept for X1C (has chamber sensor).""" mock_state.temperatures = { "nozzle": 200, "bed": 60, "chamber": 25, "chamber_target": 45, "chamber_heating": True, } result = printer_state_to_dict(mock_state, model="X1C") assert result["temperatures"]["chamber"] == 25 assert result["temperatures"]["chamber_target"] == 45 assert result["temperatures"]["chamber_heating"] is True def test_chamber_temp_filtered_for_a1(self, mock_state): """Verify chamber temperature is filtered out for A1 (no chamber sensor).""" mock_state.temperatures = {"nozzle": 200, "bed": 60, "chamber": 5} result = printer_state_to_dict(mock_state, model="A1") assert "chamber" not in result["temperatures"] def test_chamber_temp_kept_when_no_model(self, mock_state): """Verify chamber temperature is kept when model is not specified (conservative approach).""" mock_state.temperatures = {"nozzle": 200, "bed": 60, "chamber": 25} result = printer_state_to_dict(mock_state) # No model specified # When model is unknown, we can't filter - leave as is # Actually supports_chamber_temp returns False for None, so it will filter # Let's check the actual behavior assert "chamber" not in result["temperatures"] def test_ams_drying_fields_included(self, mock_state): """Verify AMS drying fields (dry_time, module_type) are included in output.""" mock_state.raw_data = { "ams": [ { "id": 0, "dry_time": 42, "module_type": "n3f", "tray": [ { "id": 0, "tray_color": "FF0000", "tray_type": "PLA", "drying_temp": 55, "drying_time": 240, } ], } ] } result = printer_state_to_dict(mock_state) ams_unit = result["ams"][0] assert ams_unit["dry_time"] == 42 assert ams_unit["module_type"] == "n3f" # Tray-level drying fields tray = ams_unit["tray"][0] assert tray["drying_temp"] == 55 assert tray["drying_time"] == 240 def test_awaiting_plate_clear_defaults_false(self, mock_state): """Without a printer_id, awaiting_plate_clear is False (no lookup possible).""" result = printer_state_to_dict(mock_state) assert result["awaiting_plate_clear"] is False def test_awaiting_plate_clear_surfaced_when_set(self, mock_state): """With printer_id, awaiting_plate_clear reflects PrinterManager state. Regression: PR #939 left this flag off the WebSocket payload, so the "Clear Plate" button only appeared after the 30 s REST fallback poll. """ from backend.app.services.printer_manager import printer_manager printer_manager.set_awaiting_plate_clear(12345, True) try: result = printer_state_to_dict(mock_state, printer_id=12345) assert result["awaiting_plate_clear"] is True finally: printer_manager.set_awaiting_plate_clear(12345, False) def test_name_and_model_surfaced_when_registered(self, mock_state): """Registered PrinterInfo name + model arg should land in the WS payload. Regression for #963 follow-up: without this, the gcode viewer's printer selector had to wait on a /printers fetch before it could render real names, and the initial WS snapshot showed "Printer 1" fallbacks. """ from backend.app.services.printer_manager import PrinterInfo, printer_manager # Register a stub PrinterInfo; the real manager writes this on connect. printer_manager._printer_info[98765] = PrinterInfo(name="My X1C", serial_number="01S00-0") try: result = printer_state_to_dict(mock_state, printer_id=98765, model="X1C") assert result["name"] == "My X1C" assert result["model"] == "X1C" finally: printer_manager._printer_info.pop(98765, None) def test_name_and_model_absent_when_no_printer_id(self, mock_state): """Without a printer_id (unusual callsites), name/model keys stay absent. The consumers (gcode viewer, frontend card) tolerate missing keys; what they can't tolerate is an unrelated printer's name accidentally leaking into a status meant for a different one. """ result = printer_state_to_dict(mock_state) assert "name" not in result assert "model" not in result def test_model_absent_when_arg_is_none(self, mock_state): """`model` arg=None must not plant a `model` key at all. If the arg is None, callers didn't know the model yet; emitting a `model: null` field would overwrite a good value cached client-side. """ from backend.app.services.printer_manager import PrinterInfo, printer_manager printer_manager._printer_info[55555] = PrinterInfo(name="N", serial_number="S") try: result = printer_state_to_dict(mock_state, printer_id=55555, model=None) assert "model" not in result assert result["name"] == "N" finally: printer_manager._printer_info.pop(55555, None) class TestStatusKeyDryingDedup: """Regression tests for WebSocket dedup including drying fields. The WebSocket broadcast deduplication uses printer_state_to_dict output to detect changes. If drying fields (like dry_time) are missing from the dict, changes to those fields won't trigger broadcasts. """ def test_dry_time_change_changes_status_key(self): """Verify dry_time is present in AMS unit data so dedup can detect changes.""" state = MagicMock() state.connected = True state.state = "IDLE" state.current_print = None state.subtask_name = None state.gcode_file = None state.progress = 0 state.remaining_time = 0 state.layer_num = 0 state.total_layers = 0 state.temperatures = {"nozzle": 25, "bed": 25} state.hms_errors = [] state.ams_status_main = 0 state.ams_status_sub = 0 state.tray_now = None state.wifi_signal = -50 state.stg_cur = -1 # First state: drying active with 30 minutes remaining state.raw_data = {"ams": [{"id": 0, "dry_time": 30, "module_type": "n3f", "tray": [{"id": 0}]}]} result1 = printer_state_to_dict(state) # Second state: drying time decreased state.raw_data = {"ams": [{"id": 0, "dry_time": 29, "module_type": "n3f", "tray": [{"id": 0}]}]} result2 = printer_state_to_dict(state) # The dicts should differ — dry_time changed assert result1["ams"][0]["dry_time"] == 30 assert result2["ams"][0]["dry_time"] == 29 assert result1["ams"] != result2["ams"] class TestSupportsChamberTemp: """Tests for supports_chamber_temp helper function.""" def test_x1_series_supported(self): """Verify X1 series printers support chamber temp.""" assert supports_chamber_temp("X1") is True assert supports_chamber_temp("X1C") is True assert supports_chamber_temp("X1E") is True def test_p2_series_supported(self): """Verify P2 series printers support chamber temp.""" assert supports_chamber_temp("P2S") is True def test_h2_series_supported(self): """Verify H2 series printers support chamber temp.""" assert supports_chamber_temp("H2C") is True assert supports_chamber_temp("H2D") is True assert supports_chamber_temp("H2DPRO") is True assert supports_chamber_temp("H2S") is True def test_p1_series_not_supported(self): """Verify P1 series printers do NOT support chamber temp.""" assert supports_chamber_temp("P1P") is False assert supports_chamber_temp("P1S") is False def test_a1_series_not_supported(self): """Verify A1 series printers do NOT support chamber temp.""" assert supports_chamber_temp("A1") is False assert supports_chamber_temp("A1MINI") is False def test_none_model_not_supported(self): """Verify None model returns False.""" assert supports_chamber_temp(None) is False def test_case_insensitive(self): """Verify model matching is case-insensitive.""" assert supports_chamber_temp("x1c") is True assert supports_chamber_temp("X1c") is True assert supports_chamber_temp("p1s") is False def test_internal_model_codes_supported(self): """Verify internal model codes from MQTT/SSDP are recognized.""" # X1/X1C assert supports_chamber_temp("BL-P001") is True # X1E assert supports_chamber_temp("C13") is True # H2D assert supports_chamber_temp("O1D") is True # H2C assert supports_chamber_temp("O1C") is True # H2S assert supports_chamber_temp("O1S") is True # H2D Pro assert supports_chamber_temp("O1E") is True # P2S assert supports_chamber_temp("N7") is True def test_internal_model_codes_not_supported(self): """Verify A1/P1 internal codes are NOT supported.""" # P1P assert supports_chamber_temp("C11") is False # P1S assert supports_chamber_temp("C12") is False # A1 assert supports_chamber_temp("N2S") is False # A1 Mini assert supports_chamber_temp("N1") is False class TestSupportsDrying: """Tests for supports_drying helper function.""" def test_known_supported_with_firmware(self): """Verify known models with sufficient firmware return True.""" assert supports_drying("X1C", "01.09.00.00") is True assert supports_drying("P1S", "01.08.00.00") is True assert supports_drying("H2D", "01.02.30.00") is True assert supports_drying("H2S", "01.02.00.00") is True assert supports_drying("P2S", "01.02.00.00") is True assert supports_drying("N7", "01.02.00.00") is True def test_known_supported_old_firmware(self): """Verify known models with old firmware return False.""" assert supports_drying("X1C", "01.08.00.00") is False assert supports_drying("P1S", "01.07.00.00") is False assert supports_drying("H2S", "01.01.00.00") is False assert supports_drying("P2S", "01.01.99.99") is False assert supports_drying("N7", "01.01.99.99") is False def test_known_supported_no_firmware(self): """Verify known models with no firmware return False.""" assert supports_drying("X1C", None) is False assert supports_drying("P2S", None) is False def test_unsupported_models(self): """Verify models without AMS drying support return False regardless of firmware.""" for model in ["A1", "A1MINI", "A1-MINI", "H2C", "N1", "N2S"]: assert supports_drying(model, "99.99.99.99") is False, f"Expected False for {model}" def test_unknown_models_allowed(self): """Verify unknown models are allowed (graceful fallback). Models not in the unsupported set AND not matching any known firmware-gated model substring get the benefit of the doubt and return True. "H2D Pro" contains "H2D" so it IS firmware-gated (needs firmware). """ # Truly unknown models: no substring match in _DRYING_MIN_FIRMWARE assert supports_drying("FUTURE_MODEL", None) is True # X1E contains "X1" substring, so it IS firmware-gated assert supports_drying("X1E", "01.09.00.00") is True # H2D Pro contains "H2D" substring, so it IS firmware-gated assert supports_drying("H2D Pro", "01.02.30.00") is True def test_none_model(self): """Verify None model returns False.""" assert supports_drying(None, "01.09.00.00") is False def test_case_insensitive(self): """Verify model matching is case-insensitive.""" assert supports_drying("x1c", "01.09.00.00") is True assert supports_drying("p2s", "01.02.00.00") is True assert supports_drying("a1", "99.99.99.99") is False class TestGetDerivedStatusName: """Tests for get_derived_status_name function.""" def test_stg_cur_255_returns_none(self): """Verify stg_cur=255 (A1/P1 idle) returns None, not 'Unknown stage (255)'.""" state = MagicMock() state.stg_cur = 255 state.state = "IDLE" result = get_derived_status_name(state) assert result is None def test_stg_cur_negative_one_returns_none_when_idle(self): """Verify stg_cur=-1 (X1 idle) returns None.""" state = MagicMock() state.stg_cur = -1 state.state = "IDLE" result = get_derived_status_name(state) assert result is None def test_valid_stage_returns_name(self): """Verify valid stg_cur values return stage name.""" state = MagicMock() state.stg_cur = 1 # Auto bed leveling result = get_derived_status_name(state) assert result == "Auto bed leveling" def test_stg_cur_zero_returns_printing(self): """Verify stg_cur=0 returns 'Printing' when no model specified.""" state = MagicMock() state.stg_cur = 0 result = get_derived_status_name(state) assert result == "Printing" def test_a1_idle_with_stg_cur_zero_returns_none(self): """Verify A1 with IDLE state and stg_cur=0 returns None (bug workaround).""" state = MagicMock() state.stg_cur = 0 state.state = "IDLE" # Test various A1 model names for model in ["A1", "A1 Mini", "A1-Mini", "A1MINI", "N1", "N2S"]: result = get_derived_status_name(state, model) assert result is None, f"Expected None for model {model}" def test_a1_running_with_stg_cur_zero_returns_printing(self): """Verify A1 with RUNNING state and stg_cur=0 still returns 'Printing'.""" state = MagicMock() state.stg_cur = 0 state.state = "RUNNING" result = get_derived_status_name(state, "A1") assert result == "Printing" def test_non_a1_idle_with_stg_cur_zero_returns_printing(self): """Verify non-A1 models with IDLE and stg_cur=0 still return 'Printing'.""" state = MagicMock() state.stg_cur = 0 state.state = "IDLE" # X1C should not get the workaround result = get_derived_status_name(state, "X1C") assert result == "Printing" class TestHasStgCurIdleBug: """Tests for has_stg_cur_idle_bug function.""" def test_a1_models_return_true(self): """Verify A1 model variants return True.""" assert has_stg_cur_idle_bug("A1") is True assert has_stg_cur_idle_bug("A1 Mini") is True assert has_stg_cur_idle_bug("A1-Mini") is True assert has_stg_cur_idle_bug("A1MINI") is True assert has_stg_cur_idle_bug("a1") is True # case insensitive assert has_stg_cur_idle_bug("a1 mini") is True def test_p1_models_return_true(self): """Verify P1P/P1S model variants return True.""" assert has_stg_cur_idle_bug("P1P") is True assert has_stg_cur_idle_bug("P1S") is True assert has_stg_cur_idle_bug("p1p") is True # case insensitive def test_internal_codes_return_true(self): """Verify internal model codes return True.""" assert has_stg_cur_idle_bug("N1") is True # A1 Mini assert has_stg_cur_idle_bug("N2S") is True # A1 assert has_stg_cur_idle_bug("C11") is True # P1P assert has_stg_cur_idle_bug("C12") is True # P1S def test_non_affected_models_return_false(self): """Verify non-affected models return False.""" assert has_stg_cur_idle_bug("X1C") is False assert has_stg_cur_idle_bug("X1") is False assert has_stg_cur_idle_bug("H2D") is False def test_none_model_returns_false(self): """Verify None model returns False.""" assert has_stg_cur_idle_bug(None) is False def test_empty_model_returns_false(self): """Verify empty model returns False.""" assert has_stg_cur_idle_bug("") is False class TestInitPrinterConnections: """Tests for init_printer_connections function.""" @pytest.mark.asyncio async def test_connects_all_active_printers(self): """Verify all active printers are connected.""" mock_db = AsyncMock() mock_printer1 = MagicMock(id=1, is_active=True) mock_printer2 = MagicMock(id=2, is_active=True) mock_result = MagicMock() mock_result.scalars.return_value.all.return_value = [mock_printer1, mock_printer2] mock_db.execute.return_value = mock_result with patch("backend.app.services.printer_manager.printer_manager") as mock_manager: mock_manager.connect_printer = AsyncMock() await init_printer_connections(mock_db) assert mock_manager.connect_printer.call_count == 2 @pytest.mark.asyncio async def test_handles_empty_printer_list(self): """Verify empty printer list is handled.""" mock_db = AsyncMock() mock_result = MagicMock() mock_result.scalars.return_value.all.return_value = [] mock_db.execute.return_value = mock_result with patch("backend.app.services.printer_manager.printer_manager") as mock_manager: mock_manager.connect_printer = AsyncMock() await init_printer_connections(mock_db) mock_manager.connect_printer.assert_not_called() class TestAmsChangeCallback: """Tests for AMS change callback functionality.""" @pytest.fixture def manager(self): """Create a fresh PrinterManager instance.""" return PrinterManager() def test_ams_change_callback_is_triggered(self, manager): """Verify AMS change callback is called when AMS data changes.""" callback = MagicMock() manager.set_ams_change_callback(callback) # Verify callback was set assert manager._on_ams_change == callback def test_ams_change_callback_receives_correct_data(self, manager): """Verify AMS change callback receives the correct AMS data format.""" received_data = [] def capture_callback(printer_id, ams_data): received_data.append((printer_id, ams_data)) manager.set_ams_change_callback(capture_callback) # The callback should accept printer_id and ams_data # This tests the callback signature assert manager._on_ams_change is not None assert callable(manager._on_ams_change) class TestParsePlateId: """Tests for parse_plate_id() — active-print plate extraction from gcode paths. Regression coverage for #881 follow-up: the REST /status endpoint and the WebSocket push path both use this helper, so they must agree on the plate number the frontend sees. """ def test_bambu_metadata_path(self): # Canonical path that Bambu Studio / OrcaSlicer stamp into the 3MF. assert parse_plate_id("/Metadata/plate_2.gcode") == 2 def test_plate_one(self): assert parse_plate_id("/Metadata/plate_1.gcode") == 1 def test_double_digit_plate(self): assert parse_plate_id("/Metadata/plate_12.gcode") == 12 def test_none_input(self): assert parse_plate_id(None) is None def test_empty_string(self): assert parse_plate_id("") is None def test_path_without_plate_segment(self): # Some firmware / slicers report a bare filename without the plate marker. assert parse_plate_id("/upload/my-model.gcode") is None def test_similar_but_non_matching_names(self): # "plate.gcode" (no number) and "nameplate_2.gcode" (substring) must not # be mistaken for real plate markers. The regex anchors on `plate_`. assert parse_plate_id("/Metadata/plate.gcode") is None assert parse_plate_id("/plates/3.gcode") is None def test_substring_match_still_extracts(self): # The regex isn't anchored to the start of a segment — any occurrence # wins. This matches real Bambu paths where the segment is preceded by # arbitrary directory noise, and matches the equivalent frontend regex. assert parse_plate_id("/uploads/project/plate_5.gcode.md5") == 5 class TestResolvePlateId: """Tests for resolve_plate_id() — plate resolution with dispatch precedence. Regression coverage for #1166: P1S firmware 01.10.00.00 only puts the .3mf filename in print.gcode_file, so parse_plate_id() returns None and the printer card falls back to plate 1. When Bambuddy dispatches the print itself we know the right plate; resolve_plate_id() prefers that record over the gcode_file regex when subtask_name matches. """ def _make_state(self, **kwargs): from backend.app.services.bambu_mqtt import PrinterState state = PrinterState() for k, v in kwargs.items(): setattr(state, k, v) return state def test_dispatched_plate_wins_when_subtask_matches(self): # User dispatches plate 4 via Bambuddy. Printer reflects subtask_name # but firmware drops the plate path from gcode_file. Without the dispatch # record we'd default to plate 1. from backend.app.services.printer_manager import resolve_plate_id state = self._make_state( gcode_file="MyModel.3mf", # No plate path — firmware bug subtask_name="MyModel", dispatched_plate_id=4, dispatched_subtask="MyModel", ) assert resolve_plate_id(state) == 4 def test_dispatched_ignored_when_subtask_differs(self): # Bambuddy's dispatch record is for a previous print; the printer is # now running a different subtask (Studio-direct dispatch). The stale # record must not be used — fall back to gcode_file regex. from backend.app.services.printer_manager import resolve_plate_id state = self._make_state( gcode_file="/Metadata/plate_2.gcode", subtask_name="DifferentPrint", dispatched_plate_id=4, dispatched_subtask="MyModel", ) assert resolve_plate_id(state) == 2 def test_falls_back_to_gcode_regex_without_dispatch(self): # Studio-direct dispatch — no Bambuddy dispatch record. Existing logic # (parse_plate_id on gcode_file) must still work. from backend.app.services.printer_manager import resolve_plate_id state = self._make_state( gcode_file="/Metadata/plate_3.gcode", subtask_name="MyModel", ) assert resolve_plate_id(state) == 3 def test_returns_none_when_nothing_resolvable(self): # No dispatch record AND firmware swallowed the plate path. The route # uses this signal to invoke the 3MF-scan fallback. from backend.app.services.printer_manager import resolve_plate_id state = self._make_state( gcode_file="MyModel.3mf", subtask_name="MyModel", ) assert resolve_plate_id(state) is None def test_dispatched_subtask_required_to_avoid_false_match(self): # dispatched_plate_id without dispatched_subtask is incomplete — we # can't validate it points at the current print, so we ignore it. from backend.app.services.printer_manager import resolve_plate_id state = self._make_state( gcode_file="MyModel.3mf", subtask_name="MyModel", dispatched_plate_id=4, dispatched_subtask=None, ) assert resolve_plate_id(state) is None