mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
889c8bd87f
P1S 01.10.00.00 (and similar firmware revisions) only echo the .3mf
filename in print.gcode_file, dropping the Metadata/plate_N.gcode path.
The /cover route's regex falls back to plate 1 — and the printer card
shows the wrong plate's thumbnail on multi-plate prints.
Resolution order in the new resolve_plate_id() helper (used by both
the status route's current_plate_id and /cover):
1. The plate Bambuddy dispatched. start_print() now records
(dispatched_plate_id, dispatched_subtask) on PrinterState; the
subtask check rejects stale records from a previous Bambuddy
dispatch bleeding into a Studio-direct print on the same project.
2. plate_(\d+)\.gcode regex on state.gcode_file (existing behaviour
for firmware that does include the path).
3. After download, scan the 3MF for a unique Metadata/plate_*.gcode —
covers per-plate archives sliced separately in Studio without a
Bambuddy dispatch record.
4. Default to plate 1.
Cover-byte cache key simplified to (subtask_name, view_key) now that
plate resolution is late-bound. clear_cover_cache() already fires on
every print start, so re-dispatches with a different plate always
fetch a fresh thumbnail.
Bambuddy-dispatched prints additionally register the local archive
3MF in the cover cache at dispatch time, so /cover reads straight
from the archive directory and doesn't refetch the file over FTP
from a printer whose FTP server is busy serving the active print.
Coverage: 5 unit tests for resolve_plate_id, 4 unit tests for the
dispatch record on start_print, 2 integration tests for the cover
route (dispatch wins over plate-1 default; 3MF-scan fallback for
per-plate archive without dispatch record).
1503 lines
57 KiB
Python
1503 lines
57 KiB
Python
"""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_<num>`.
|
|
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
|