mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-01 03:31:25 +02:00
maziggy
ff53e62ef8
Backend: - pytest configuration with async support and coverage - Unit tests for notification service (23 tests) - Unit tests for smart plug manager (12 tests) - Unit tests for archive service (16 tests) - Integration tests for API endpoints - Fix: notifications now send immediately (digest is summary only) Frontend: - Vitest configuration with jsdom and coverage - MSW for API mocking - Component tests for Toggle, Button, Card, ConfirmModal (77 tests) - Test utilities with custom render wrapper CI/CD: - GitHub Actions workflow for automated testing - Backend lint, unit tests, integration tests - Frontend lint, type-check, unit tests, build
779 lines
28 KiB
Python
779 lines
28 KiB
Python
"""Unit tests for PrinterManager service.
|
|
|
|
Tests printer connection management, status tracking, and print control.
|
|
"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock
|
|
from datetime import datetime
|
|
|
|
from backend.app.services.printer_manager import (
|
|
PrinterManager,
|
|
printer_state_to_dict,
|
|
init_printer_connections,
|
|
)
|
|
|
|
|
|
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")
|
|
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
|
|
|
|
# ========================================================================
|
|
# 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()
|
|
|
|
|
|
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 = {}
|
|
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."""
|
|
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 result["vt_tray"] is not None
|
|
assert result["vt_tray"]["id"] == 254
|
|
assert result["vt_tray"]["tray_color"] == "00FF00"
|
|
assert result["vt_tray"]["tray_type"] == "PETG"
|
|
|
|
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_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
|
|
|
|
|
|
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()
|