vinez/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-08-11 00:30:12 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions
bambuddy/backend/tests/unit/services/test_virtual_printer.py
maziggy 2e45893dd5 feat(print-options): add "Auto" state to bed levelling, flow & nozzle-offset calibration 
Bed levelling, flow calibration, and nozzle-offset calibration were on/off
only, so the sole way to run bed levelling was to force a full level before
every print. Bambu Studio has always offered a third "Auto" state that lets
the printer skip the calibration when it was done recently -- the state most
users actually want. Make these three options tri-state (off/on/auto),
defaulting to auto, and leave vibration/layer-inspect/timelapse as on/off
(Bambu Studio exposes no auto for those).

Wire encoding follows Bambu Studio's source exactly: each option sends a JSON
bool (true only for "on") plus a companion int -- off=0, on=1, auto=2. The
bool fields stay booleans (the #1478 H2S regression); only the companion int
widened from {0,1} to {0,1,2}. #1721's observation that stage 8/39 stays
queued when sending 2 is the auto contract (queued, skipped at runtime if
recent), not a broken "off".

- schemas: TriState = Literal[off/on/auto] with a BeforeValidator coercing
  legacy bool / 0-1 / true-false so old clients and un-migrated rows validate
- model + migration: boolean columns -> String; SQLite via column affinity +
  data backfill, PostgreSQL via ALTER COLUMN TYPE guarded on information_schema
  (verified on both dialects); settings rows normalised true/false -> on/off
- MQTT: start_print takes the tri-state strings and emits the paired bool+int
- Virtual Printer: reconstructs the slicer's auto/on/off from the int companion
  (auto_bed_leveling / extrude_cali_flag) in both capture paths
- frontend: CalibrationMode type; off/auto/on segmented controls in the print
  dialog, queue bulk-edit, and Settings -> Workflow; calibrationMode_* strings
  in all 11 locales
2026-07-20 17:55:56 +02:00

4070 lines
156 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Unit tests for Virtual Printer services.
Tests the virtual printer manager, FTP server, and SSDP server components.
"""
import asyncio
import json
import zipfile
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
def _write_3mf_with_filaments(file_path: Path, filaments: list[dict], plate_index: int = 1) -> None:
"""Build a minimal 3MF zip with `Metadata/slice_info.config` carrying the
given per-slot filament entries. Each `filaments` dict needs `id`, `type`,
`color`, `used_g`. Used by the #1188 VP queue-mode tests below."""
filament_xml = "".join(
f'<filament id="{f["id"]}" type="{f["type"]}" color="{f["color"]}" '
f'used_g="{f["used_g"]}" tray_info_idx="{f.get("tray_info_idx", "")}"/>'
for f in filaments
)
config = (
'<?xml version="1.0" encoding="utf-8"?>'
"<config>"
f'<plate><metadata key="index" value="{plate_index}"/>'
f"{filament_xml}"
"</plate>"
"</config>"
)
with zipfile.ZipFile(file_path, "w") as zf:
zf.writestr("Metadata/slice_info.config", config)
# Plate gcode is referenced for plate-id detection in the VP path —
# presence is enough; contents don't matter.
zf.writestr(f"Metadata/plate_{plate_index}.gcode", "; gcode\n")
class TestVirtualPrinterInstance:
"""Tests for VirtualPrinterInstance class."""
@pytest.fixture
def instance(self, tmp_path):
"""Create a VirtualPrinterInstance with test defaults."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
return VirtualPrinterInstance(
vp_id=1,
name="TestPrinter",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
# ========================================================================
# Tests for instance properties
# ========================================================================
def test_instance_stores_parameters(self, instance):
"""Verify constructor stores parameters correctly."""
assert instance.id == 1
assert instance.name == "TestPrinter"
assert instance.mode == "archive"
assert instance.model == "C11"
assert instance.access_code == "12345678"
assert instance.serial_suffix == "391800001"
def test_instance_serial_property(self, instance):
"""Verify serial is generated from model prefix + suffix."""
# C11 = P1P, prefix = 01S00A
assert instance.serial == "01S00A391800001"
def test_instance_serial_x1c(self, tmp_path):
"""Verify X1C serial uses correct prefix."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=2,
name="X1C",
mode="archive",
model="BL-P001",
access_code="12345678",
serial_suffix="391800002",
base_dir=tmp_path,
)
assert inst.serial == "00M00A391800002"
def test_instance_is_proxy_false(self, instance):
"""Verify is_proxy is False for non-proxy mode."""
assert instance.is_proxy is False
def test_instance_is_proxy_true(self, tmp_path):
"""Verify is_proxy is True for proxy mode."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=3,
name="Proxy",
mode="proxy",
model="C11",
access_code="",
serial_suffix="391800003",
target_printer_ip="192.168.1.100",
base_dir=tmp_path,
)
assert inst.is_proxy is True
def test_instance_is_running_with_active_tasks(self, instance):
"""Verify is_running is True when tasks are active."""
mock_task = MagicMock()
mock_task.done.return_value = False
instance._tasks = [mock_task]
assert instance.is_running is True
def test_instance_is_running_with_no_tasks(self, instance):
"""Verify is_running is False when no tasks."""
assert instance.is_running is False
def test_instance_creates_directories(self, instance, tmp_path):
"""Verify instance creates upload and cert directories."""
assert (tmp_path / "uploads" / "1").exists()
assert (tmp_path / "uploads" / "1" / "cache").exists()
assert (tmp_path / "certs" / "1").exists()
# ========================================================================
# Tests for status
# ========================================================================
def test_get_status_returns_correct_format(self, instance):
"""Verify get_status returns expected fields."""
instance._pending_files = {"file1.3mf": Path("/tmp/file1.3mf")} # nosec B108
mock_task = MagicMock(done=MagicMock(return_value=False))
instance._tasks = [mock_task]
status = instance.get_status()
assert status["running"] is True
assert status["pending_files"] == 1
def test_get_status_not_running(self, instance):
"""Verify get_status when no tasks."""
status = instance.get_status()
assert status["running"] is False
assert status["pending_files"] == 0
# ========================================================================
# Tests for file handling
# ========================================================================
@pytest.mark.asyncio
async def test_on_file_received_adds_to_pending(self, instance):
"""Verify received file is added to pending list in review mode."""
instance.mode = "review"
file_path = Path("/tmp/test.3mf") # nosec B108
with patch.object(instance, "_queue_file", new_callable=AsyncMock) as mock_queue:
await instance.on_file_received(file_path, "192.168.1.100")
assert "test.3mf" in instance._pending_files
mock_queue.assert_called_once()
@pytest.mark.asyncio
async def test_on_file_received_archives_immediately(self, instance):
"""Verify file is archived in immediate mode."""
file_path = Path("/tmp/test.3mf") # nosec B108
with patch.object(instance, "_archive_file", new_callable=AsyncMock) as mock_archive:
await instance.on_file_received(file_path, "192.168.1.100")
mock_archive.assert_called_once_with(file_path, "192.168.1.100")
@pytest.mark.asyncio
async def test_on_file_received_signals_FINISH_to_slicer(self, instance):
"""Regression #1280: when a slicer's Print flow uploads to a non-proxy VP,
the VP must transition gcode_state PREPARE → FINISH so the slicer's
in-flight-job lock releases. Going PREPARE → IDLE wedges Orca at
"Downloading...(0%)" and blocks the next dispatch with "busy with
another print job".
Send-flow slicers don't watch the post-upload state, so this is a
no-op behavior change for them.
"""
instance.mode = "archive"
instance._mqtt = MagicMock()
instance._mqtt.set_gcode_state = MagicMock()
file_path = Path("/tmp/test.3mf") # nosec B108
with patch.object(instance, "_archive_file", new_callable=AsyncMock):
await instance.on_file_received(file_path, "192.168.1.100")
instance._mqtt.set_gcode_state.assert_called_once_with("FINISH", filename="test.3mf", prepare_percent="100")
@pytest.mark.asyncio
async def test_on_file_received_non_3mf_does_not_touch_state(self, instance):
"""Non-3MF uploads (e.g., a job's auxiliary files) must not transition
the visible state — the slicer is only tracking the .3mf upload."""
instance.mode = "archive"
instance._mqtt = MagicMock()
instance._mqtt.set_gcode_state = MagicMock()
file_path = Path("/tmp/test.gcode") # nosec B108
with patch.object(instance, "_archive_file", new_callable=AsyncMock):
await instance.on_file_received(file_path, "192.168.1.100")
instance._mqtt.set_gcode_state.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_command_schedules_finish_release_non_proxy(self, instance):
"""#1658: Bambu Studio 2.7.x flipped the slicer's Send flow to
FTP → FTP → MQTT project_file. Under that order the synthetic
project_file ack overwrites the FINISH set by #1280 in
``on_file_received`` back to PREPARE, leaving the slicer's
"Downloading" modal stuck forever. Re-firing FINISH a moment after
the ack releases the modal.
"""
instance.mode = "archive"
instance._mqtt = MagicMock()
instance._mqtt.set_gcode_state = MagicMock()
with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
await instance.on_print_command("test.3mf", {"command": "project_file"})
mock_delayed.assert_called_once()
# First positional arg should be the filename; second is the delay seconds.
args = mock_delayed.call_args.args
assert args[0] == "test.3mf"
assert isinstance(args[1], int | float)
@pytest.mark.asyncio
async def test_on_print_command_proxy_mode_does_not_reschedule_finish(self, instance):
"""Proxy mode hands push_status straight from the real printer through
the bridge. Re-firing a synthetic FINISH would clobber a real
PREPARE / RUNNING transition coming back from the printer, so the
scheduler is exempt for proxy mode."""
instance.mode = "proxy"
instance._mqtt = MagicMock()
with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
await instance.on_print_command("test.3mf", {"command": "project_file"})
mock_delayed.assert_not_called()
@pytest.mark.asyncio
async def test_on_print_command_no_mqtt_does_not_schedule(self, instance):
"""If the MQTT server isn't running yet (transient race during boot),
the scheduler must skip silently — no AttributeError, no orphan task."""
instance.mode = "queue"
instance._mqtt = None
# Should not raise.
await instance.on_print_command("test.3mf", {"command": "project_file"})
assert instance._finish_release_task is None
@pytest.mark.asyncio
async def test_schedule_finish_release_cancels_previous_timer(self, instance):
"""A slicer that fires project_file twice (e.g. retry after a transient
FTP hiccup) must only result in one FINISH transition — the earlier
in-flight timer is cancelled when the next one is scheduled."""
instance.mode = "queue"
instance._mqtt = MagicMock()
instance._schedule_finish_release("first.3mf", delay=10.0)
first_task = instance._finish_release_task
assert first_task is not None
instance._schedule_finish_release("second.3mf", delay=10.0)
second_task = instance._finish_release_task
assert second_task is not first_task
# Give the loop one tick so the cancelled task settles.
await asyncio.sleep(0)
assert first_task.cancelled() or first_task.done()
# Clean up the still-pending second task so the test doesn't leak it.
second_task.cancel()
try:
await second_task
except asyncio.CancelledError:
pass
@pytest.mark.asyncio
async def test_delayed_finish_release_sets_finish_state(self, instance):
"""End-to-end: after the delay elapses, set_gcode_state is called with
FINISH and prepare_percent=100, matching the wire-format the slicer's
Print flow consumes to release "Downloading"."""
instance._mqtt = MagicMock()
instance._mqtt.set_gcode_state = MagicMock()
await instance._delayed_finish_release("queued.3mf", delay=0.0)
instance._mqtt.set_gcode_state.assert_called_once_with("FINISH", filename="queued.3mf", prepare_percent="100")
@pytest.mark.asyncio
async def test_on_print_command_no_filename_does_not_schedule(self, instance):
"""A project_file command without a subtask_name (defensive — real
slicers always send one) must not schedule a no-op FINISH that would
carry an empty filename on the next 1 Hz push."""
instance.mode = "queue"
instance._mqtt = MagicMock()
with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
await instance.on_print_command("", {"command": "project_file"})
mock_delayed.assert_not_called()
@pytest.mark.asyncio
async def test_archive_file_skips_non_3mf(self, instance):
"""Verify non-3MF files are skipped and cleaned up."""
instance._session_factory = MagicMock()
instance._pending_files["verify_job"] = Path("/tmp/verify_job") # nosec B108
with patch("pathlib.Path.unlink"):
await instance._archive_file(Path("/tmp/verify_job"), "192.168.1.100") # nosec B108
assert "verify_job" not in instance._pending_files
@pytest.mark.asyncio
async def test_archive_file_broadcasts_archive_created(self, tmp_path):
"""#1282: VP immediate-mode archives must broadcast archive_created so
the Archives page refreshes without a tab switch. Real-printer prints
get this via main.py's MQTT print_start handler; the VP path used to
skip the broadcast entirely."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=30,
name="ImmediateBroadcast",
mode="archive",
model="C12",
access_code="12345678",
serial_suffix="391800030",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 99
mock_archive.printer_id = None
mock_archive.filename = "test.3mf"
mock_archive.print_name = "test"
mock_archive.status = "archived"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
patch(
"backend.app.core.websocket.ws_manager.send_archive_created",
new_callable=AsyncMock,
) as mock_broadcast,
):
await inst._archive_file(file_path, "192.168.1.100")
mock_broadcast.assert_awaited_once()
payload = mock_broadcast.await_args.args[0]
assert payload["id"] == 99
assert payload["filename"] == "test.3mf"
assert payload["status"] == "archived"
# ========================================================================
# Tests for auto_dispatch
# ========================================================================
def test_auto_dispatch_defaults_to_true(self, tmp_path):
"""Verify auto_dispatch defaults to True when not specified."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=10,
name="DefaultDispatch",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800010",
base_dir=tmp_path,
)
assert inst.auto_dispatch is True
@pytest.mark.asyncio
async def test_add_to_print_queue_with_auto_dispatch_on(self, tmp_path):
"""Verify queue items have manual_start=False when auto_dispatch=True."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
added_items = []
def capture_add(item):
added_items.append(item)
mock_db.add = MagicMock(side_effect=capture_add)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=11,
name="AutoDispatchOn",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800011",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
# Create a temp 3mf file
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.manual_start is False
@pytest.mark.asyncio
async def test_add_to_print_queue_broadcasts_archive_created(self, tmp_path):
"""#1282: VP queue-mode uploads must broadcast archive_created so the
Archives page picks up the new entry live. Pre-fix the page only
refreshed when the user manually switched tabs."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
mock_db.add = MagicMock()
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=31,
name="QueueBroadcast",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800031",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 77
mock_archive.printer_id = None
mock_archive.filename = "test.3mf"
mock_archive.print_name = "test"
mock_archive.status = "archived"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
patch(
"backend.app.core.websocket.ws_manager.send_archive_created",
new_callable=AsyncMock,
) as mock_broadcast,
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
mock_broadcast.assert_awaited_once()
payload = mock_broadcast.await_args.args[0]
assert payload["id"] == 77
assert payload["print_name"] == "test"
assert payload["status"] == "archived"
@pytest.mark.asyncio
async def test_add_to_print_queue_with_auto_dispatch_off(self, tmp_path):
"""Verify queue items have manual_start=True when auto_dispatch=False."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
added_items = []
def capture_add(item):
added_items.append(item)
mock_db.add = MagicMock(side_effect=capture_add)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=12,
name="AutoDispatchOff",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800012",
auto_dispatch=False,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
# Create a temp 3mf file
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.manual_start is True
@pytest.mark.asyncio
async def test_add_to_print_queue_gcode_injection_on(self, tmp_path):
"""#1516: queue items opt into injection when the VP has gcode_injection=True."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
added_items = []
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=13,
name="InjectOn",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800013",
gcode_injection=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
assert added_items[0].gcode_injection is True
@pytest.mark.asyncio
async def test_add_to_print_queue_gcode_injection_off_by_default(self, tmp_path):
"""#1516: queue items do NOT inject when the VP leaves gcode_injection at its default."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
added_items = []
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=14,
name="InjectOff",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800014",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
assert added_items[0].gcode_injection is False
@pytest.mark.asyncio
async def test_add_to_print_queue_uses_workflow_defaults_from_settings(self, tmp_path):
"""#1235: VP queue-mode constructed PrintQueueItem without specifying
bed_levelling / flow_cali / vibration_cali / layer_inspect / timelapse,
so SQLAlchemy applied the column-level defaults and ignored the user's
workflow preferences entirely. Every print sent from the slicer to the
VP came through with the OPPOSITE of what the workflow page said,
forcing the user to edit each queue item by hand.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=22,
name="DefaultsTest",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800022",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# The reporter set every workflow default to the OPPOSITE of the model's
# column default. Pre-fix the column defaults won; with the fix the
# settings values must flow through to the queue item exactly as stored.
settings_map = {
"virtual_printer_archive_name_source": None,
# Legacy boolean-string rows still coerce (false->off, true->on).
"default_bed_levelling": "false", # tri-state default: auto
"default_flow_cali": "true", # tri-state default: auto
"default_vibration_cali": "false", # model default: True
"default_layer_inspect": "true", # model default: False
"default_timelapse": "true", # model default: False
}
async def fake_get_setting(_db, key):
return settings_map.get(key)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new=fake_get_setting,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.bed_levelling == "off", "default_bed_levelling=false must flow through"
assert queue_item.flow_cali == "on", "default_flow_cali=true must flow through"
assert queue_item.vibration_cali is False, "default_vibration_cali=false must flow through"
assert queue_item.layer_inspect is True, "default_layer_inspect=true must flow through"
assert queue_item.timelapse is True, "default_timelapse=true must flow through"
@pytest.mark.asyncio
async def test_add_to_print_queue_falls_back_to_schema_defaults_when_unset(self, tmp_path):
"""#1235 fallback: when no workflow setting is in the DB, the queue
item should use the AppSettings (Pydantic) defaults — same values
the user sees in the workflow page on a fresh install.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=23,
name="FreshInstallDefaults",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800023",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None, # No settings → fall back to schema defaults
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
# These must match the AppSettings (Pydantic) defaults in schemas/settings.py
assert queue_item.bed_levelling == "auto"
assert queue_item.flow_cali == "auto"
assert queue_item.vibration_cali is True
assert queue_item.layer_inspect is False
assert queue_item.timelapse is False
@pytest.mark.asyncio
async def test_add_to_print_queue_inherits_slicer_print_options(self, tmp_path):
"""#1403: VP-queue items used to fall back to `default_timelapse` even
though the slicer's MQTT `project_file` command carries the user's
actual choice. Capture-via-`on_print_command` flow lets the user's
slicer toggle reach the queue item.
Settings here have timelapse OFF; the slicer's MQTT capture has it ON.
After the fix the queue item must reflect the slicer's choice.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=24,
name="SlicerInherits",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800024",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# Pre-populate the capture as if MQTT `project_file` arrived already.
# Settings (below) deliberately have timelapse OFF — only the slicer
# capture should drive the resulting queue item.
await inst.on_print_command(
file_path.name,
{
"command": "project_file",
"timelapse": True,
"bed_leveling": False, # Note: MQTT field is single-L `bed_leveling`
"flow_cali": True,
"vibration_cali": False,
"layer_inspect": True,
},
)
settings_map = {
"virtual_printer_archive_name_source": None,
"default_bed_levelling": "true",
"default_flow_cali": "false",
"default_vibration_cali": "true",
"default_layer_inspect": "false",
"default_timelapse": "false",
}
async def fake_get_setting(_db, key):
return settings_map.get(key)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new=fake_get_setting,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.timelapse is True, "Slicer's timelapse=True must override settings.default_timelapse=False"
assert queue_item.bed_levelling == "off", "Slicer's bed_leveling=False must override default_bed_levelling"
assert queue_item.flow_cali == "on"
assert queue_item.vibration_cali is False
assert queue_item.layer_inspect is True
# Capture is consumed — no lingering state for the next print of the same name.
assert file_path.name not in inst._slicer_print_options
@pytest.mark.asyncio
async def test_add_to_print_queue_coerces_slicer_integer_zero_one(self, tmp_path):
"""#1403: H-family firmwares carry calibration flags as integers
(0/1) rather than booleans. The capture must coerce both shapes so
H-family-sliced jobs through the VP queue work the same as P1/X1.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=25,
name="SlicerIntegers",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800025",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
await inst.on_print_command(
file_path.name,
{"command": "project_file", "timelapse": 1, "bed_leveling": 0, "flow_cali": 1},
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.timelapse is True, "integer 1 must coerce to True"
assert queue_item.bed_levelling == "off", "integer 0 must coerce to off"
assert queue_item.flow_cali == "on"
@pytest.mark.asyncio
async def test_add_to_print_queue_captures_slicer_auto_from_int_companion(self, tmp_path):
"""The slicer's tri-state rides on the int companion (auto_bed_leveling /
extrude_cali_flag). When the slicer picks "Auto" it sends bed_leveling
false + auto_bed_leveling 2; the VP must record "auto", not "off".
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=26,
name="SlicerAuto",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800026",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
await inst.on_print_command(
file_path.name,
{
"command": "project_file",
"bed_leveling": False,
"auto_bed_leveling": 2,
"flow_cali": False,
"extrude_cali_flag": 2,
},
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.bed_levelling == "auto", "auto_bed_leveling=2 must record 'auto'"
assert queue_item.flow_cali == "auto", "extrude_cali_flag=2 must record 'auto'"
@pytest.mark.asyncio
async def test_add_to_print_queue_populates_required_filament_types(self, tmp_path):
"""#1188: VP queue-mode used to create PrintQueueItems with no
filament fields, so the scheduler fell through to model-only matching
and dispatched onto whatever printer was free regardless of loaded
colour. ``required_filament_types`` is populated unconditionally
(cheap, helps the scheduler validate type even without
``force_color_match``) — pin that contract here."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=21,
name="Reqs",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800021",
auto_dispatch=True,
queue_force_color_match=False, # off → only required_filament_types
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "multi.3mf"
_write_3mf_with_filaments(
file_path,
[
{"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "12.3"},
{"id": "2", "type": "PETG", "color": "#000000", "used_g": "4.5"},
# used_g=0 → not actually consumed by this plate, must be ignored
{"id": "3", "type": "ABS", "color": "#FF0000", "used_g": "0"},
],
plate_index=1,
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "multi"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
# Type-only fallback always populated. Sorted, deduped, no zero-use ABS.
assert queue_item.required_filament_types is not None
assert json.loads(queue_item.required_filament_types) == ["PETG", "PLA"]
# Setting off → no force_color_match overrides leaked.
assert queue_item.filament_overrides is None
@pytest.mark.asyncio
async def test_add_to_print_queue_force_color_match_writes_overrides(self, tmp_path):
"""#1188 core fix: when the per-VP ``queue_force_color_match`` toggle
is on, every consumed slot lands as a ``filament_overrides`` entry
with ``force_color_match: true``. This is the field the scheduler
keys on (``print_scheduler.py:512``) — without it, slot-by-slot
type+color matching never runs."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=22,
name="ForceColor",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800022",
auto_dispatch=True,
queue_force_color_match=True, # on
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "forced.3mf"
_write_3mf_with_filaments(
file_path,
[
{"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "10.0"},
{"id": "2", "type": "PLA", "color": "#FF00FF", "used_g": "5.0"},
],
plate_index=1,
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "forced"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
assert queue_item.filament_overrides is not None
overrides = json.loads(queue_item.filament_overrides)
assert overrides == [
{"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "force_color_match": True},
{"slot_id": 2, "type": "PLA", "color": "#FF00FF", "force_color_match": True},
]
# required_filament_types still populated alongside overrides.
assert json.loads(queue_item.required_filament_types) == ["PLA"]
@pytest.mark.asyncio
async def test_add_to_print_queue_force_color_match_skips_when_3mf_unparseable(self, tmp_path):
"""A malformed or fake-bytes 3MF must not crash the upload path —
we just write the queue item with no filament fields and let the
scheduler fall back to model-only matching (the pre-#1188 default).
Regression guard for the existing fake-bytes happy-path tests."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=23,
name="Unparseable",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800023",
auto_dispatch=True,
queue_force_color_match=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "bad.3mf"
file_path.write_bytes(b"not a real 3mf zip")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "bad"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
queue_item = added_items[0]
# No filament data extractable → both fields stay None (graceful
# fallback to model-only scheduling).
assert queue_item.required_filament_types is None
assert queue_item.filament_overrides is None
# ========================================================================
# Tests for archive_name_source setting (#1152)
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.parametrize(
("setting_value", "expected_prefer_filename"),
[
("filename", True),
("metadata", False),
(None, False), # Default when setting unset
("", False), # Defensive: empty string is not "filename"
],
)
async def test_archive_file_passes_prefer_filename_per_setting(
self, tmp_path, setting_value, expected_prefer_filename
):
"""_archive_file reads `virtual_printer_archive_name_source` and forwards
prefer_filename_for_name=True only when it equals 'filename' (#1152)."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=20,
name="NameSource",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800020",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "user-renamed-job.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "user-renamed-job"
archive_print_mock = AsyncMock(return_value=mock_archive)
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=setting_value,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
archive_print_mock,
),
):
await inst._archive_file(file_path, "192.168.1.100")
assert archive_print_mock.await_count == 1
kwargs = archive_print_mock.await_args.kwargs
assert kwargs.get("prefer_filename_for_name") is expected_prefer_filename
# ========================================================================
# Tests for failure-path cleanup (#audit-R2-1)
# ========================================================================
#
# All three file handlers (_archive_file, _queue_file, _add_to_print_queue)
# previously only popped _pending_files and unlinked the temp file on the
# success branch. Failure paths leaked the marker (blocking same-name
# retries via the FTP layer) and the temp file on disk. The cleanup must
# ALWAYS run, even when archival / queue insert raises.
@pytest.mark.asyncio
async def test_archive_file_failure_path_pops_pending_and_unlinks(self, tmp_path):
"""When the archive layer raises, `_pending_files[filename]` must still
be popped and the temp file must be unlinked. Otherwise the FTP layer's
same-name retry guard would silently reject the slicer's next attempt
and the upload_dir would accumulate ghost files."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=40,
name="ArchiveFailCleanup",
mode="archive",
model="C12",
access_code="12345678",
serial_suffix="391800040",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "cleanup-archive.3mf"
file_path.write_bytes(b"fake3mf")
inst._pending_files[file_path.name] = file_path
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
side_effect=RuntimeError("archive blew up"),
),
):
await inst._archive_file(file_path, "192.168.1.100")
assert file_path.name not in inst._pending_files
assert not file_path.exists()
@pytest.mark.asyncio
async def test_queue_file_failure_path_pops_pending_and_unlinks(self, tmp_path):
"""Same invariant for _queue_file: a DB error during PendingUpload
insert must not leak the in-flight marker or the temp file."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
# Commit raises — emulating a DB connectivity error.
mock_db.add = MagicMock()
mock_db.commit = AsyncMock(side_effect=RuntimeError("db unreachable"))
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=41,
name="QueueFailCleanup",
mode="review",
model="C12",
access_code="12345678",
serial_suffix="391800041",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "cleanup-queue.3mf"
file_path.write_bytes(b"fake3mf")
inst._pending_files[file_path.name] = file_path
await inst._queue_file(file_path, "192.168.1.100")
assert file_path.name not in inst._pending_files
assert not file_path.exists()
@pytest.mark.asyncio
async def test_add_to_print_queue_failure_path_pops_pending_and_unlinks(self, tmp_path):
"""Same invariant for _add_to_print_queue: a DB error or archive
failure must not leak the in-flight marker or the temp file."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
mock_db.add = MagicMock()
mock_db.commit = AsyncMock()
mock_db.execute = AsyncMock(side_effect=RuntimeError("queue insert blew up"))
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=42,
name="DispatchFailCleanup",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800042",
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "cleanup-dispatch.3mf"
file_path.write_bytes(b"fake3mf")
inst._pending_files[file_path.name] = file_path
with patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert file_path.name not in inst._pending_files
assert not file_path.exists()
# ========================================================================
# Test for position=MAX+1 (audit-R2)
# ========================================================================
@pytest.mark.asyncio
async def test_add_to_print_queue_position_picks_max_plus_one(self, tmp_path):
"""VP-queue items previously got hardcoded `position=1`, colliding
with existing items at position 1 and producing non-deterministic
execution order. Now the position is chosen by `MAX(position)+1`
against the target queue, matching the canonical `POST /print-queue/`
path."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
# Capture the inserted PrintQueueItem so we can assert on .position.
added_items: list = []
class _RecordingDb:
def __init__(self):
self.add = lambda item: added_items.append(item)
self.commit = AsyncMock()
async def execute(self, query): # noqa: ARG002
"""Return a stub result whose `.scalar()` reports the existing
MAX(position) for the target. Returning 7 means the new item
should land at 8."""
result = MagicMock()
result.scalar = MagicMock(return_value=7)
return result
mock_db = _RecordingDb()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=43,
name="PositionMaxPlusOne",
mode="queue",
model="C12",
access_code="12345678",
serial_suffix="391800043",
target_printer_id=99,
auto_dispatch=True,
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "next-position.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 555
mock_archive.printer_id = None
mock_archive.filename = "next-position.3mf"
mock_archive.print_name = "next-position"
mock_archive.status = "archived"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
patch(
"backend.app.core.websocket.ws_manager.send_archive_created",
new_callable=AsyncMock,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
# One queue item was added.
assert len(added_items) == 1
queue_item = added_items[0]
# Position = max(7) + 1 = 8 — NOT the legacy hardcoded 1.
assert queue_item.position == 8
@pytest.mark.asyncio
async def test_add_to_print_queue_multi_plate_send_all_enqueues_one_per_plate(self, tmp_path):
"""#1733: BambuStudio / OrcaSlicer "Send All" of a multi-plate project
uploads ONE 3MF containing every plate. Pre-fix only the first plate
index was extracted and one queue item was created; plates 2..N were
silently dropped. Post-fix every `<plate>` block in `slice_info.config`
produces its own PrintQueueItem with the correct ``plate_id``, sharing
the same backing archive, with consecutive positions for plate-order
execution.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items: list = []
class _RecordingDb:
def __init__(self):
# Capture inserted items as they're added; assign a fake .id
# on flush so the manager's logger doesn't see None.
self._next_id = 1000
def _add(item):
added_items.append(item)
self.add = _add
self.commit = AsyncMock()
async def execute(self, query): # noqa: ARG002
"""Return MAX(position) = 0 so plate items land at 1, 2, 3."""
result = MagicMock()
result.scalar = MagicMock(return_value=0)
return result
async def flush(self):
# Mimic the FK populate so queue_item.id is available after add().
for item in added_items:
if getattr(item, "id", None) is None:
item.id = self._next_id
self._next_id += 1
mock_db = _RecordingDb()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=44,
name="MultiPlateSendAll",
mode="queue",
model="O1D", # H2D — matches the live VP H2D-1 Proxy in #1733
access_code="12345678",
serial_suffix="391800044",
target_printer_id=1,
auto_dispatch=False, # manual_start, mirrors the live VP
base_dir=tmp_path,
session_factory=mock_session_factory,
)
# Build a 3MF with three plates baked into slice_info.config —
# mirrors what BambuStudio / OrcaSlicer's "Send All" puts on the wire.
file_path = tmp_path / "Cube.gcode.3mf"
_write_3mf_with_filaments(
file_path, [{"id": 1, "type": "PLA", "color": "#000000", "used_g": "15.61"}], plate_index=1
)
# Append plate 2 and 3 blocks to slice_info.config to mimic Send All.
with zipfile.ZipFile(file_path, "r") as zf:
existing = zf.read("Metadata/slice_info.config").decode()
# Inject two additional <plate> blocks (indices 2 and 3) inside <config>.
multi_plate_config = existing.replace(
"</config>",
(
'<plate><metadata key="index" value="2"/>'
'<filament id="2" type="PETG" color="#FB0207" used_g="14.45"/>'
"</plate>"
'<plate><metadata key="index" value="3"/>'
'<filament id="3" type="PLA" color="#FFFFFF" used_g="12.10"/>'
"</plate>"
"</config>"
),
)
# Repack the zip with the expanded slice_info.config.
import io as _io
buf = _io.BytesIO()
with zipfile.ZipFile(file_path, "r") as src, zipfile.ZipFile(buf, "w") as dst:
for name in src.namelist():
if name == "Metadata/slice_info.config":
dst.writestr(name, multi_plate_config)
else:
dst.writestr(name, src.read(name))
# Plate-2 and plate-3 gcode payloads so `extract_filament_requirements`
# has something to read for each — contents irrelevant, presence matters.
dst.writestr("Metadata/plate_2.gcode", "; plate 2 gcode\n")
dst.writestr("Metadata/plate_3.gcode", "; plate 3 gcode\n")
file_path.write_bytes(buf.getvalue())
mock_archive = MagicMock()
mock_archive.id = 999
mock_archive.printer_id = None
mock_archive.filename = "Cube.gcode.3mf"
mock_archive.print_name = "Cube"
mock_archive.status = "archived"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
patch(
"backend.app.core.websocket.ws_manager.send_archive_created",
new_callable=AsyncMock,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
# Three queue items, one per plate, with the correct plate_id and
# consecutive positions starting at MAX(position)+1 = 1.
assert len(added_items) == 3, f"Expected 3 queue items for 3-plate Send All, got {len(added_items)}"
plate_ids = [q.plate_id for q in added_items]
assert plate_ids == [1, 2, 3], f"plate_ids should preserve slice_info order, got {plate_ids}"
positions = [q.position for q in added_items]
assert positions == [1, 2, 3], f"positions should be consecutive, got {positions}"
archive_ids = {q.archive_id for q in added_items}
assert archive_ids == {999}, f"All queue items must share the single backing archive, got {archive_ids}"
# auto_dispatch=False on the VP → every item is manual_start.
assert all(q.manual_start for q in added_items)
@pytest.mark.asyncio
async def test_add_to_print_queue_captures_nozzle_mapping(self, tmp_path):
"""#1780: BambuStudio's project_file for H2C rack-swap (O1C2) sends
per-filament physical nozzle position IDs in `nozzle_mapping`. VP
intake must store it as a JSON string on the queue item so the
dispatcher can replay it. Without this the H2C firmware falls back
to "last matching nozzle" auto-pick and ignores the user's slicer
choice.
"""
import json as _json
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=42,
name="H2CRack",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="391800042",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# Pre-populate as if BS's project_file arrived. Wire shape matches
# BambuStudio's PrintJob params: nozzle_mapping = 32-entry array of
# per-filament physical nozzle position IDs (verified via H2C wire
# capture). The slicer-side `nozzles_info` field that the original
# #1780 attempt also looked for was never actually sent — it has
# been dropped from the capture path entirely.
await inst.on_print_command(
file_path.name,
{
"command": "project_file",
"nozzle_mapping": [16, -1, -1, 1, -1, -1, -1, -1],
},
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
item = added_items[0]
assert item.nozzle_mapping is not None
assert _json.loads(item.nozzle_mapping) == [16, -1, -1, 1, -1, -1, -1, -1]
@pytest.mark.asyncio
async def test_add_to_print_queue_no_nozzle_mapping_when_slicer_omits(self, tmp_path):
"""#1780: every model other than O1C2 sends no nozzle_mapping — the
queue item must carry NULL, not an empty list. NULL is what the
dispatch layer keys off of to skip the injection entirely on non-
rack-swap printers.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=43,
name="NotH2C",
mode="queue",
model="C11",
access_code="12345678",
serial_suffix="391800043",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# X1C-style slicer command — no nozzle fields.
await inst.on_print_command(
file_path.name,
{"command": "project_file", "timelapse": False, "bed_leveling": True},
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
item = added_items[0]
assert item.nozzle_mapping is None
@pytest.mark.asyncio
async def test_add_to_print_queue_nozzle_pick_replicated_across_plates(self, tmp_path, monkeypatch):
"""#1780 × #1697/#1188: a multi-plate Send All from BS must stamp the
same nozzle_mapping on every plate's queue item, not only the first.
Mirrors the per-plate stamping for gcode_injection,
filament_overrides, etc.
"""
import json as _json
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items = []
mock_db = AsyncMock()
mock_db.add = MagicMock(side_effect=added_items.append)
mock_db.flush = AsyncMock()
mock_db.commit = AsyncMock()
mock_db.execute = AsyncMock()
mock_db.execute.return_value.scalar.return_value = None
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=44,
name="H2CMultiPlate",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="391800044",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# Force 3 plates so the queue loop runs three times.
monkeypatch.setattr(inst, "_extract_plate_ids", lambda _p: [1, 2, 3])
await inst.on_print_command(
file_path.name,
{
"command": "project_file",
"nozzle_mapping": [16, 0],
},
)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 3
for item in added_items:
assert _json.loads(item.nozzle_mapping) == [16, 0]
@pytest.mark.asyncio
async def test_on_print_command_late_mqtt_retroactively_stamps_queue_item(self, tmp_path):
"""#1780 round 3: Bambu Studio's MQTT project_file can arrive AFTER
`_add_to_print_queue` already gave up waiting (observed at 2.085 s
on H2C wireless setups). The queue item was committed with settings
defaults; the slicer's nozzle_mapping + workflow flags must be
patched onto it when MQTT lands, otherwise the H2C firmware falls
back to auto-pick.
"""
import json as _json
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items: list = []
mock_db = AsyncMock()
mock_db.add = MagicMock(
side_effect=lambda item: (added_items.append(item), setattr(item, "id", 100 + len(added_items)))[0]
)
async def _flush():
# added_items[-1].id was set by `add`; nothing else to do.
return None
mock_db.flush = AsyncMock(side_effect=_flush)
mock_db.commit = AsyncMock()
# First execute() call (the position-max SELECT inside _add_to_print_queue)
# returns None; second (the eligible-pending SELECT in
# _restamp_recent_queue_item) returns the committed queue id; third
# (the UPDATE) is fire-and-forget.
position_max_result = MagicMock()
position_max_result.scalar = MagicMock(return_value=None)
select_pending_result = MagicMock()
select_pending_result.all = MagicMock(return_value=[(101,)])
update_result = MagicMock()
mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=99,
name="LateMQTT",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="391800099",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
# MQTT server presence enables the wait_for path; we don't actually
# use any methods on it.
inst._mqtt = MagicMock()
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
# 1. _add_to_print_queue runs WITHOUT a prior on_print_command —
# the wait_for times out (settings-default fallback) and the
# queue item is committed.
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
# Shorten the wait so the test isn't slow.
patch(
"backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
0.05,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
assert len(added_items) == 1
assert added_items[0].nozzle_mapping is None # MQTT was never received
assert file_path.name in inst._recent_queue_items
# 2. MQTT project_file arrives AFTER the wait expired — must
# retroactively patch the queue item.
await inst.on_print_command(
file_path.name,
{
"command": "project_file",
"file": file_path.name,
"nozzle_mapping": [16, -1, -1, 1],
"timelapse": True,
"bed_leveling": False,
},
)
# The UPDATE call is the third execute. Inspect its values.
update_call = mock_db.execute.await_args_list[2]
update_stmt = update_call.args[0]
compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
params = dict(compiled.params)
assert _json.loads(params["nozzle_mapping"]) == [16, -1, -1, 1]
assert params["timelapse"] is True
assert params["bed_levelling"] == "off" # MQTT bed_leveling → column bed_levelling (tri-state)
# Recent-queue tracking dict is cleared after the patch.
assert file_path.name not in inst._recent_queue_items
@pytest.mark.asyncio
async def test_add_to_print_queue_catches_mqtt_stashed_post_wait_timeout(self, tmp_path):
"""The actual race-window scenario: wait_for times out, then MQTT
arrives and stashes options AFTER the wait but BEFORE the
post-commit re-check. The post-commit pop must catch it.
"""
import json as _json
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
added_items: list = []
mock_db = AsyncMock()
mock_db.add = MagicMock(
side_effect=lambda item: (added_items.append(item), setattr(item, "id", 300 + len(added_items)))[0]
)
mock_db.flush = AsyncMock()
mock_db.commit = AsyncMock()
position_max_result = MagicMock()
position_max_result.scalar = MagicMock(return_value=None)
select_pending_result = MagicMock()
select_pending_result.all = MagicMock(return_value=[(301,)])
update_result = MagicMock()
mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=96,
name="RaceCommitYield",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="391800096",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
inst._mqtt = MagicMock()
file_path = tmp_path / "test.3mf"
file_path.write_bytes(b"fake3mf")
# Stash MQTT data on the FIRST commit (simulating MQTT arrival
# during _add_to_print_queue's commit yield); _restamp also calls
# db.commit later, so we one-shot the side effect.
commit_calls = {"n": 0}
async def _delayed_stash(*_args, **_kwargs):
commit_calls["n"] += 1
if commit_calls["n"] == 1:
inst._slicer_print_options[file_path.name] = {
"command": "project_file",
"file": file_path.name,
"nozzle_mapping": [0, 16, -1, -1],
"timelapse": False,
}
return None
mock_db.commit = AsyncMock(side_effect=_delayed_stash)
mock_archive = MagicMock()
mock_archive.id = 1
mock_archive.print_name = "test"
with (
patch(
"backend.app.api.routes.settings.get_setting",
new_callable=AsyncMock,
return_value=None,
),
patch(
"backend.app.services.archive.ArchiveService.archive_print",
new_callable=AsyncMock,
return_value=mock_archive,
),
patch(
"backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
0.05,
),
):
await inst._add_to_print_queue(file_path, "192.168.1.100")
# Queue item INSERTed with defaults (wait timed out, no slicer_opts).
assert len(added_items) == 1
# But the post-commit pop caught the late stash and applied the
# slicer nozzle_mapping via _restamp's UPDATE.
update_call = mock_db.execute.await_args_list[2]
update_stmt = update_call.args[0]
compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
params = dict(compiled.params)
assert _json.loads(params["nozzle_mapping"]) == [0, 16, -1, -1]
assert params["timelapse"] is False
# _recent_queue_items entry was consumed by the post-commit
# _restamp call.
assert file_path.name not in inst._recent_queue_items
# And the stash is empty.
assert file_path.name not in inst._slicer_print_options
@pytest.mark.asyncio
async def test_on_print_command_late_mqtt_skips_already_dispatched_item(self, tmp_path):
"""Once the scheduler has picked the queue item up (status != pending),
the retroactive patch is a no-op — racing the dispatcher would be
unsafe.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
mock_db = AsyncMock()
# The eligible-pending SELECT returns nothing — item is no longer pending.
empty_result = MagicMock()
empty_result.all = MagicMock(return_value=[])
mock_db.execute = AsyncMock(return_value=empty_result)
mock_db.commit = AsyncMock()
mock_session_factory = MagicMock()
mock_session_ctx = AsyncMock()
mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
mock_session_factory.return_value = mock_session_ctx
inst = VirtualPrinterInstance(
vp_id=98,
name="LateMQTTDispatched",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="391800098",
base_dir=tmp_path,
session_factory=mock_session_factory,
)
inst._mqtt = MagicMock()
# Pre-seed the recent-queue dict — pretend _add_to_print_queue just
# committed item id 42.
inst._recent_queue_items["test.3mf"] = ([42], 1_000_000.0)
# Drive _restamp via on_print_command on the late-MQTT path.
with patch("backend.app.services.virtual_printer.manager.time.monotonic", return_value=1_000_001.0):
await inst.on_print_command(
"test.3mf",
{
"command": "project_file",
"file": "test.3mf",
"nozzle_mapping": [16, -1],
},
)
# No UPDATE was issued — only the eligibility SELECT ran.
assert mock_db.execute.await_count == 1
mock_db.commit.assert_not_awaited()
assert "test.3mf" not in inst._recent_queue_items
class TestVirtualPrinterManager:
"""Tests for VirtualPrinterManager orchestrator."""
@pytest.fixture
def manager(self):
"""Create a VirtualPrinterManager instance."""
from backend.app.services.virtual_printer.manager import VirtualPrinterManager
return VirtualPrinterManager()
def test_manager_starts_empty(self, manager):
"""Verify manager starts with no instances."""
assert len(manager._instances) == 0
assert manager.is_enabled is False
def test_manager_get_status_empty(self, manager):
"""Verify get_status returns disabled state when no instances."""
status = manager.get_status()
assert status["enabled"] is False
assert status["running"] is False
assert status["mode"] == "archive"
def test_manager_is_enabled_with_instance(self, manager, tmp_path):
"""Verify is_enabled is True when instances exist."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="Test",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
manager._instances[1] = inst
assert manager.is_enabled is True
@pytest.mark.asyncio
async def test_manager_remove_instance_server(self, manager, tmp_path):
"""Verify remove_instance stops and removes a server-mode instance."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="Test",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
await manager.remove_instance(1)
assert 1 not in manager._instances
inst.stop_server.assert_called_once()
@pytest.mark.asyncio
async def test_manager_remove_instance_proxy(self, manager, tmp_path):
"""Verify remove_instance stops proxy-mode instance."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=2,
name="Proxy",
mode="proxy",
model="C11",
access_code="",
serial_suffix="391800002",
target_printer_ip="192.168.1.100",
base_dir=tmp_path,
)
inst.stop_proxy = AsyncMock()
manager._instances[2] = inst
await manager.remove_instance(2)
assert 2 not in manager._instances
inst.stop_proxy.assert_called_once()
def test_manager_get_status_with_instance(self, manager, tmp_path):
"""Verify legacy get_status returns first instance data."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="Bambuddy",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
mock_task = MagicMock(done=MagicMock(return_value=False))
inst._tasks = [mock_task]
inst._pending_files = {"file1.3mf": Path("/tmp/file1.3mf")} # nosec B108
manager._instances[1] = inst
status = manager.get_status()
assert status["enabled"] is True
assert status["running"] is True
assert status["mode"] == "archive"
assert status["name"] == "Bambuddy"
assert status["serial"] == "01S00A391800001"
assert status["model"] == "C11"
assert status["model_name"] == "P1P"
assert status["pending_files"] == 1
def test_manager_get_all_status(self, manager, tmp_path):
"""Verify get_all_status returns status for all instances."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
for i in range(1, 3):
inst = VirtualPrinterInstance(
vp_id=i,
name=f"VP{i}",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix=f"39180000{i}",
base_dir=tmp_path,
)
manager._instances[i] = inst
statuses = manager.get_all_status()
assert len(statuses) == 2
assert statuses[0]["name"] == "VP1"
assert statuses[1]["name"] == "VP2"
@pytest.mark.asyncio
async def test_manager_stop_all(self, manager, tmp_path):
"""Verify stop_all removes all instances."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
for i in range(1, 3):
inst = VirtualPrinterInstance(
vp_id=i,
name=f"VP{i}",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix=f"39180000{i}",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[i] = inst
await manager.stop_all()
assert len(manager._instances) == 0
# ========================================================================
# Tests for sync_from_db config change detection
# ========================================================================
def _make_db_vp(self, **overrides):
"""Create a mock VirtualPrinter DB object."""
defaults = {
"id": 1,
"name": "TestVP",
"enabled": True,
"mode": "archive",
"model": "C11",
"access_code": "12345678",
"serial_suffix": "391800001",
"bind_ip": "",
"remote_interface_ip": "",
"target_printer_id": None,
"auto_dispatch": True,
"tailscale_disabled": True, # Opt-in default (#1070 UX fix)
"queue_force_color_match": False, # default — must be explicit so MagicMock truthiness doesn't trip the change detector
"gcode_injection": False, # same reason as above
"position": 0,
}
defaults.update(overrides)
vp = MagicMock()
for k, v in defaults.items():
setattr(vp, k, v)
return vp
def _setup_sync_mocks(self, manager, enabled_vps, tmp_path):
"""Wire up session_factory mock for sync_from_db."""
mock_result = MagicMock()
mock_result.scalars.return_value.all.return_value = enabled_vps
mock_db = AsyncMock()
mock_db.execute = AsyncMock(return_value=mock_result)
mock_db.__aenter__ = AsyncMock(return_value=mock_db)
mock_db.__aexit__ = AsyncMock(return_value=False)
manager._session_factory = MagicMock(return_value=mock_db)
manager._base_dir = tmp_path
@pytest.mark.asyncio
async def test_sync_from_db_restarts_on_mode_change(self, manager, tmp_path):
"""Verify sync_from_db restarts VP when mode changes."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
# DB says mode changed to "review"
db_vp = self._make_db_vp(mode="review")
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
# Patch VirtualPrinterInstance to prevent actual start
with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
mock_new = MagicMock()
mock_new.start_server = AsyncMock()
MockInst.return_value = mock_new
await manager.sync_from_db()
mock_remove.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_sync_from_db_restarts_on_access_code_change(self, manager, tmp_path):
"""Verify sync_from_db restarts VP when access_code changes."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
db_vp = self._make_db_vp(access_code="newcode99")
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
mock_new = MagicMock()
mock_new.start_server = AsyncMock()
MockInst.return_value = mock_new
await manager.sync_from_db()
mock_remove.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_sync_from_db_skips_unchanged_instance(self, manager, tmp_path):
"""Verify sync_from_db does NOT restart when config is identical."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
manager._instances[1] = inst
# DB matches running config exactly
db_vp = self._make_db_vp()
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
await manager.sync_from_db()
mock_remove.assert_not_called()
@pytest.mark.asyncio
async def test_sync_from_db_restarts_on_bind_ip_change(self, manager, tmp_path):
"""Verify sync_from_db restarts VP when bind_ip changes."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
bind_ip="192.168.1.10",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
db_vp = self._make_db_vp(bind_ip="192.168.1.20")
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
mock_new = MagicMock()
mock_new.start_server = AsyncMock()
MockInst.return_value = mock_new
await manager.sync_from_db()
mock_remove.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_sync_from_db_restarts_on_model_change(self, manager, tmp_path):
"""Verify sync_from_db restarts VP when model changes."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
db_vp = self._make_db_vp(model="C12")
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
mock_new = MagicMock()
mock_new.start_server = AsyncMock()
MockInst.return_value = mock_new
await manager.sync_from_db()
mock_remove.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_sync_from_db_does_not_restart_on_tailscale_toggle(self, manager, tmp_path):
"""Flipping tailscale_disabled is purely informational — must NOT trigger a restart.
Cert provisioning was removed; the toggle only governs whether the VP card surfaces
the host's Tailscale IP/FQDN to the user. No service needs to reload, so changing
it through sync_from_db should leave any running instance untouched.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
tailscale_disabled=False,
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
db_vp = self._make_db_vp(tailscale_disabled=True)
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
await manager.sync_from_db()
mock_remove.assert_not_called()
@pytest.mark.asyncio
async def test_sync_from_db_restarts_on_gcode_injection_toggle(self, manager, tmp_path):
"""Toggling gcode_injection in the DB must restart the running instance.
Without this, the in-memory ``self.gcode_injection`` keeps its old value
and ``_add_to_print_queue`` stamps the stale flag on every new queue
item — so disabling injection in the UI silently has no effect until
the process restarts.
"""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=1,
name="TestVP",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800001",
gcode_injection=True,
base_dir=tmp_path,
)
inst.stop_server = AsyncMock()
manager._instances[1] = inst
db_vp = self._make_db_vp(gcode_injection=False)
self._setup_sync_mocks(manager, [db_vp], tmp_path)
with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
mock_new = MagicMock()
mock_new.start_server = AsyncMock()
MockInst.return_value = mock_new
await manager.sync_from_db()
mock_remove.assert_called_once_with(1)
class TestFTPSession:
"""Tests for FTP session handling."""
@pytest.fixture
def mock_reader(self):
"""Create a mock StreamReader."""
reader = AsyncMock()
return reader
@pytest.fixture
def mock_writer(self):
"""Create a mock StreamWriter."""
writer = MagicMock()
writer.get_extra_info = MagicMock(return_value=("192.168.1.100", 12345))
writer.write = MagicMock()
writer.drain = AsyncMock()
writer.close = MagicMock()
writer.wait_closed = AsyncMock()
writer.is_closing = MagicMock(return_value=False)
return writer
@pytest.fixture
def ssl_context(self):
"""Create a mock SSL context."""
return MagicMock()
@pytest.fixture
def session(self, mock_reader, mock_writer, ssl_context, tmp_path):
"""Create an FTPSession instance."""
from backend.app.services.virtual_printer.ftp_server import FTPSession
return FTPSession(
reader=mock_reader,
writer=mock_writer,
upload_dir=tmp_path,
access_code="12345678",
ssl_context=ssl_context,
on_file_received=None,
)
# ========================================================================
# Tests for authentication
# ========================================================================
@pytest.mark.asyncio
async def test_user_command_accepts_bblp(self, session):
"""Verify USER command accepts bblp user."""
await session.cmd_USER("bblp")
assert session.username == "bblp"
@pytest.mark.asyncio
async def test_pass_command_authenticates(self, session):
"""Verify PASS command authenticates with correct code."""
session.username = "bblp"
await session.cmd_PASS("12345678")
assert session.authenticated is True
@pytest.mark.asyncio
async def test_pass_command_rejects_wrong_code(self, session):
"""Verify PASS command rejects wrong access code."""
session.username = "bblp"
await session.cmd_PASS("wrongcode")
assert session.authenticated is False
# ========================================================================
# Tests for FTP commands
# ========================================================================
@pytest.mark.asyncio
async def test_syst_command(self, session):
"""Verify SYST returns UNIX type."""
await session.cmd_SYST("")
session.writer.write.assert_called()
call_args = session.writer.write.call_args[0][0].decode()
assert "215" in call_args
assert "UNIX" in call_args
@pytest.mark.asyncio
async def test_pwd_command_requires_auth(self, session):
"""Verify PWD requires authentication."""
session.authenticated = False
await session.cmd_PWD("")
call_args = session.writer.write.call_args[0][0].decode()
assert "530" in call_args
@pytest.mark.asyncio
async def test_pwd_command_when_authenticated(self, session):
"""Verify PWD returns root directory when authenticated."""
session.authenticated = True
await session.cmd_PWD("")
call_args = session.writer.write.call_args[0][0].decode()
assert "257" in call_args
@pytest.mark.asyncio
async def test_type_command_sets_binary(self, session):
"""Verify TYPE I sets binary mode."""
session.authenticated = True
await session.cmd_TYPE("I")
assert session.transfer_type == "I"
@pytest.mark.asyncio
async def test_pbsz_command(self, session):
"""Verify PBSZ returns success."""
await session.cmd_PBSZ("0")
call_args = session.writer.write.call_args[0][0].decode()
assert "200" in call_args
@pytest.mark.asyncio
async def test_prot_command_accepts_p(self, session):
"""Verify PROT P is accepted."""
await session.cmd_PROT("P")
call_args = session.writer.write.call_args[0][0].decode()
assert "200" in call_args
@pytest.mark.asyncio
async def test_quit_command(self, session):
"""Verify QUIT sends goodbye and raises CancelledError."""
with pytest.raises(asyncio.CancelledError):
await session.cmd_QUIT("")
class TestSSDPServer:
"""Tests for Virtual Printer SSDP server."""
@pytest.fixture
def ssdp_server(self):
"""Create a VirtualPrinterSSDPServer instance."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
return VirtualPrinterSSDPServer(
serial="TEST123",
name="TestPrinter",
model="BL-P001",
)
# ========================================================================
# Tests for SSDP response
# ========================================================================
def test_build_notify_message(self, ssdp_server):
"""Verify NOTIFY packet contains required headers."""
# Set a known IP for testing
ssdp_server._local_ip = "192.168.1.100"
message = ssdp_server._build_notify_message()
assert b"NOTIFY" in message
assert b"DevName.bambu.com: TestPrinter" in message
assert b"USN: TEST123" in message
def test_build_response_message(self, ssdp_server):
"""Verify response packet contains required headers."""
# Set a known IP for testing
ssdp_server._local_ip = "192.168.1.100"
message = ssdp_server._build_response_message()
assert b"HTTP/1.1 200 OK" in message
assert b"DevName.bambu.com: TestPrinter" in message
assert b"USN: TEST123" in message
def test_ssdp_server_uses_correct_model(self, ssdp_server):
"""Verify SSDP server uses the provided model."""
ssdp_server._local_ip = "192.168.1.100"
message = ssdp_server._build_notify_message()
assert b"DevModel.bambu.com: BL-P001" in message
# ========================================================================
# Tests for advertise_ip parameter
# ========================================================================
def test_advertise_ip_sets_local_ip(self):
"""Verify advertise_ip overrides auto-detection."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
server = VirtualPrinterSSDPServer(
serial="TEST123",
name="TestPrinter",
model="BL-P001",
advertise_ip="10.0.0.50",
)
assert server._local_ip == "10.0.0.50"
def test_advertise_ip_empty_string_uses_auto_detect(self):
"""Verify empty advertise_ip falls back to auto-detection."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
server = VirtualPrinterSSDPServer(
serial="TEST123",
name="TestPrinter",
model="BL-P001",
advertise_ip="",
)
assert server._local_ip is None
def test_advertise_ip_in_notify_message(self):
"""Verify NOTIFY message uses the advertise_ip."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
server = VirtualPrinterSSDPServer(
serial="TEST123",
name="TestPrinter",
model="BL-P001",
advertise_ip="10.0.0.50",
)
message = server._build_notify_message()
assert b"Location: 10.0.0.50" in message
def test_advertise_ip_in_response_message(self):
"""Verify M-SEARCH response uses the advertise_ip."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
server = VirtualPrinterSSDPServer(
serial="TEST123",
name="TestPrinter",
model="BL-P001",
advertise_ip="10.0.0.50",
)
message = server._build_response_message()
assert b"Location: 10.0.0.50" in message
def test_default_no_advertise_ip(self):
"""Verify default constructor has None local_ip (auto-detect)."""
from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer
server = VirtualPrinterSSDPServer()
assert server._local_ip is None
class TestCertificateService:
"""Tests for TLS certificate generation."""
@pytest.fixture
def cert_service(self, tmp_path):
"""Create a CertificateService instance."""
from backend.app.services.virtual_printer.certificate import CertificateService
return CertificateService(cert_dir=tmp_path, serial="TEST123")
def test_generate_certificates(self, cert_service, tmp_path):
"""Verify certificates are generated correctly."""
cert_path, key_path = cert_service.generate_certificates()
assert cert_path.exists()
assert key_path.exists()
# Verify certificate content
cert_content = cert_path.read_text()
assert "BEGIN CERTIFICATE" in cert_content
key_content = key_path.read_text()
assert "BEGIN" in key_content and "KEY" in key_content
def test_certificates_reused_if_exist(self, cert_service):
"""Verify existing certificates are reused."""
# First generation
cert_path1, key_path1 = cert_service.generate_certificates()
mtime1 = cert_path1.stat().st_mtime
# Second call should reuse (via ensure_certificates)
cert_path2, key_path2 = cert_service.ensure_certificates()
mtime2 = cert_path2.stat().st_mtime
assert mtime1 == mtime2 # File wasn't regenerated
def test_delete_certificates(self, cert_service):
"""Verify certificates can be deleted."""
cert_service.generate_certificates()
assert cert_service.cert_path.exists()
assert cert_service.key_path.exists()
cert_service.delete_certificates()
assert not cert_service.cert_path.exists()
assert not cert_service.key_path.exists()
def test_ensure_creates_if_not_exist(self, cert_service):
"""Verify ensure_certificates generates if not existing."""
assert not cert_service.cert_path.exists()
cert_path, key_path = cert_service.ensure_certificates()
assert cert_path.exists()
assert key_path.exists()
class TestBindServer:
"""Tests for BindServer (port 3002 bind/detect protocol)."""
@pytest.fixture
def bind_server(self):
"""Create a BindServer instance."""
from backend.app.services.virtual_printer.bind_server import BindServer
return BindServer(
serial="09400A391800001",
model="O1D",
name="Bambuddy",
)
def test_build_frame(self, bind_server):
"""Verify frame building produces correct format."""
payload = {"login": {"command": "detect"}}
frame = bind_server._build_frame(payload)
# Header: 0xA5A5
assert frame[:2] == b"\xa5\xa5"
# Trailer: 0xA7A7
assert frame[-2:] == b"\xa7\xa7"
# Length field is total message size (LE uint16)
import struct
total_len = struct.unpack_from("<H", frame, 2)[0]
assert total_len == len(frame)
# JSON payload is between header and trailer
import json
json_bytes = frame[4:-2]
parsed = json.loads(json_bytes)
assert parsed == payload
def test_parse_frame_valid(self, bind_server):
"""Verify valid frame parsing extracts JSON correctly."""
import json
import struct
payload = {"login": {"command": "detect", "sequence_id": "20000"}}
json_bytes = json.dumps(payload, separators=(",", ":")).encode()
total_len = 4 + len(json_bytes) + 2
frame = b"\xa5\xa5" + struct.pack("<H", total_len) + json_bytes + b"\xa7\xa7"
result = bind_server._parse_frame(frame)
assert result is not None
assert result["login"]["command"] == "detect"
assert result["login"]["sequence_id"] == "20000"
def test_parse_frame_invalid_header(self, bind_server):
"""Verify invalid header returns None."""
result = bind_server._parse_frame(b"\xbb\xbb\x06\x00{}\xa7\xa7")
assert result is None
def test_parse_frame_invalid_trailer(self, bind_server):
"""Verify invalid trailer returns None."""
result = bind_server._parse_frame(b"\xa5\xa5\x06\x00{}\xbb\xbb")
assert result is None
def test_parse_frame_too_short(self, bind_server):
"""Verify short data returns None."""
result = bind_server._parse_frame(b"\xa5\xa5\x00")
assert result is None
def test_parse_frame_invalid_json(self, bind_server):
"""Verify invalid JSON returns None."""
import struct
bad_json = b"not json"
total_len = 4 + len(bad_json) + 2
frame = b"\xa5\xa5" + struct.pack("<H", total_len) + bad_json + b"\xa7\xa7"
result = bind_server._parse_frame(frame)
assert result is None
def test_build_frame_roundtrip(self, bind_server):
"""Verify build_frame output can be parsed back."""
payload = {
"login": {
"bind": "free",
"command": "detect",
"connect": "lan",
"dev_cap": 1,
"id": "09400A391800001",
"model": "O1D",
"name": "Bambuddy",
"sequence_id": 3021,
"version": "01.00.00.00",
}
}
frame = bind_server._build_frame(payload)
parsed = bind_server._parse_frame(frame)
assert parsed is not None
assert parsed["login"]["id"] == "09400A391800001"
assert parsed["login"]["model"] == "O1D"
assert parsed["login"]["name"] == "Bambuddy"
assert parsed["login"]["bind"] == "free"
def test_bind_server_stores_config(self, bind_server):
"""Verify bind server stores serial, model, name."""
assert bind_server.serial == "09400A391800001"
assert bind_server.model == "O1D"
assert bind_server.name == "Bambuddy"
assert bind_server.version == "01.00.00.00"
def test_bind_server_custom_version(self):
"""Verify custom firmware version is stored."""
from backend.app.services.virtual_printer.bind_server import BindServer
server = BindServer(
serial="TEST123",
model="C13",
name="Test",
version="02.03.04.05",
)
assert server.version == "02.03.04.05"
def test_bind_ports_constant(self):
"""Verify BIND_PORTS includes both 3000 and 3002 for slicer compatibility."""
from backend.app.services.virtual_printer.bind_server import BIND_PORTS
assert 3000 in BIND_PORTS
assert 3002 in BIND_PORTS
def test_bind_server_initializes_empty_servers_list(self, bind_server):
"""Verify bind server starts with empty servers list."""
assert bind_server._servers == []
assert bind_server._running is False
class TestSlicerProxyManager:
"""Tests for SlicerProxyManager (proxy mode)."""
@pytest.fixture
def proxy_manager(self, tmp_path):
"""Create a SlicerProxyManager instance."""
from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager
# Create dummy cert files
cert_path = tmp_path / "cert.pem"
key_path = tmp_path / "key.pem"
cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
# Split string to avoid pre-commit hook false positive on test data
key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")
return SlicerProxyManager(
target_host="192.168.1.100",
cert_path=cert_path,
key_path=key_path,
)
def test_proxy_manager_initializes_ports(self, proxy_manager):
"""Verify proxy manager has correct port constants."""
# FTP proxy uses privileged port 990 to match what Bambu Studio expects
assert proxy_manager.LOCAL_FTP_PORT == 990
assert proxy_manager.LOCAL_MQTT_PORT == 8883
assert proxy_manager.PRINTER_FTP_PORT == 990
assert proxy_manager.PRINTER_MQTT_PORT == 8883
assert proxy_manager.PRINTER_FILE_TRANSFER_PORT == 6000
assert proxy_manager.PRINTER_RTSP_PORT == 322
# Auxiliary ports: undocumented proprietary ports for A1/P1S etc.
assert proxy_manager.PRINTER_AUX_PORTS == [2024, 2025, 2026]
# Bind ports: both 3000 and 3002 for slicer compatibility
assert proxy_manager.PRINTER_BIND_PORTS == [3000, 3002]
# FTP data port range for transparent EPSV proxying
assert proxy_manager.FTP_DATA_PORT_MIN == 50000
assert proxy_manager.FTP_DATA_PORT_MAX == 50100
def test_proxy_manager_stores_target_host(self, proxy_manager):
"""Verify proxy manager stores target host."""
assert proxy_manager.target_host == "192.168.1.100"
def test_get_status_before_start(self, proxy_manager):
"""Verify get_status returns zeros before start."""
status = proxy_manager.get_status()
assert status["running"] is False
assert status["ftp_connections"] == 0
assert status["mqtt_connections"] == 0
@pytest.mark.asyncio
async def test_proxy_start_creates_transparent_proxies(self, tmp_path):
"""Verify start() uses TCPProxy for FTP/FileTransfer/RTSP and TLSProxy only for MQTT.
The transparent proxy architecture preserves end-to-end TLS between
slicer and printer for all protocols except MQTT, which must be
TLS-terminated to rewrite the printer's IP in MQTT payloads.
"""
from unittest.mock import patch
from backend.app.services.virtual_printer.tcp_proxy import (
SlicerProxyManager,
TCPProxy,
TLSProxy,
)
cert_path = tmp_path / "cert.pem"
key_path = tmp_path / "key.pem"
cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")
mgr = SlicerProxyManager(
target_host="192.168.1.100",
cert_path=cert_path,
key_path=key_path,
bind_address="10.0.0.1",
)
# Mock asyncio.create_task and asyncio.gather to prevent actual
# server start. Close every coroutine handed to gather — otherwise
# the ~110 run_with_logging() coros built inside start() are
# garbage-collected unfinalized and surface later as
# PytestUnraisableExceptionWarning at random in other tests.
async def _close_pending(*coros, **_):
for c in coros:
if asyncio.iscoroutine(c):
c.close()
with (
patch("asyncio.create_task") as mock_create_task,
patch("asyncio.gather", side_effect=_close_pending),
patch.object(SlicerProxyManager, "_log_activity"),
):
mock_create_task.return_value = MagicMock()
await mgr.start()
# FTP, FileTransfer, RTSP should be TCPProxy (transparent)
assert isinstance(mgr._ftp_proxy, TCPProxy), "FTP should be TCPProxy (transparent)"
assert isinstance(mgr._file_transfer_proxy, TCPProxy), "FileTransfer should be TCPProxy"
assert isinstance(mgr._rtsp_proxy, TCPProxy), "RTSP should be TCPProxy"
# MQTT should be TLSProxy (TLS-terminated for IP rewriting)
assert isinstance(mgr._mqtt_proxy, TLSProxy), "MQTT should be TLSProxy (TLS-terminated)"
# Auxiliary ports (2024-2026) should be TCPProxy (transparent)
assert len(mgr._aux_proxies) == 3, "Should have 3 aux port proxies"
for ap in mgr._aux_proxies:
assert isinstance(ap, TCPProxy), "Aux proxies should be TCPProxy"
assert mgr._aux_proxies[0].listen_port == 2024
assert mgr._aux_proxies[0].target_port == 2024
assert mgr._aux_proxies[2].listen_port == 2026
# FTP data ports should be pre-created as TCPProxy instances
assert len(mgr._ftp_data_proxies) == 101 # 50000-50100 inclusive
for dp in mgr._ftp_data_proxies:
assert isinstance(dp, TCPProxy), "FTP data proxies should be TCPProxy"
# Verify FTP data proxies target the same port on the printer
first_dp = mgr._ftp_data_proxies[0]
assert first_dp.listen_port == 50000
assert first_dp.target_port == 50000
assert first_dp.target_host == "192.168.1.100"
last_dp = mgr._ftp_data_proxies[-1]
assert last_dp.listen_port == 50100
assert last_dp.target_port == 50100
def test_proxy_manager_mqtt_has_ip_rewriting(self, tmp_path):
"""Verify MQTT proxy is configured with IP rewriting when bind_address is set."""
from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager
cert_path = tmp_path / "cert.pem"
key_path = tmp_path / "key.pem"
cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")
mgr = SlicerProxyManager(
target_host="192.168.1.100",
cert_path=cert_path,
key_path=key_path,
bind_address="10.0.0.1",
)
# Before start, proxies are None — verify constructor stores rewrite config
assert mgr.bind_address == "10.0.0.1"
assert mgr.target_host == "192.168.1.100"
class TestSSDPProxy:
"""Tests for SSDPProxy (cross-network SSDP relay)."""
@pytest.fixture
def ssdp_proxy(self):
"""Create an SSDPProxy instance."""
from backend.app.services.virtual_printer.ssdp_server import SSDPProxy
return SSDPProxy(
local_interface_ip="192.168.1.100",
remote_interface_ip="10.0.0.100",
target_printer_ip="192.168.1.50",
)
def test_ssdp_proxy_stores_interface_ips(self, ssdp_proxy):
"""Verify SSDPProxy stores interface IPs correctly."""
assert ssdp_proxy.local_interface_ip == "192.168.1.100"
assert ssdp_proxy.remote_interface_ip == "10.0.0.100"
assert ssdp_proxy.target_printer_ip == "192.168.1.50"
def test_rewrite_ssdp_location(self, ssdp_proxy):
"""Verify SSDP Location header is rewritten to remote interface IP."""
original_packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: TestPrinter\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
# Location should be changed to remote interface IP
assert b"Location: 10.0.0.100" in rewritten
assert b"Location: 192.168.1.50" not in rewritten
# Other headers should be preserved
assert b"DevName.bambu.com: TestPrinter" in rewritten
def test_rewrite_ssdp_location_case_insensitive(self, ssdp_proxy):
"""Verify SSDP Location rewrite is case insensitive."""
original_packet = b"NOTIFY * HTTP/1.1\r\nlocation: 192.168.1.50\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
assert b"10.0.0.100" in rewritten
def test_rewrite_ssdp_location_no_match(self, ssdp_proxy):
"""Verify packet without Location header is returned unchanged."""
original_packet = b"NOTIFY * HTTP/1.1\r\nDevName.bambu.com: Test\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
# No Location header, but _rewrite_ssdp logs a warning and returns as-is
assert b"DevName.bambu.com: Test" in rewritten
def test_parse_ssdp_message(self, ssdp_proxy):
"""Verify SSDP message parsing extracts headers."""
packet = (
b"NOTIFY * HTTP/1.1\r\n"
b"Location: 192.168.1.50\r\n"
b"DevName.bambu.com: TestPrinter\r\n"
b"DevModel.bambu.com: BL-P001\r\n"
b"\r\n"
)
headers = ssdp_proxy._parse_ssdp_message(packet)
assert headers["location"] == "192.168.1.50"
assert headers["devname.bambu.com"] == "TestPrinter"
assert headers["devmodel.bambu.com"] == "BL-P001"
class TestVirtualPrinterManagerDirectories:
"""Tests for VirtualPrinterManager directory management."""
def test_ensure_base_directories_creates_subdirs(self, tmp_path):
"""Verify _ensure_base_directories creates required base directories."""
from backend.app.services.virtual_printer.manager import VirtualPrinterManager
manager = VirtualPrinterManager()
manager._base_dir = tmp_path / "virtual_printer"
manager._ensure_base_directories()
assert (tmp_path / "virtual_printer").exists()
assert (tmp_path / "virtual_printer" / "uploads").exists()
assert (tmp_path / "virtual_printer" / "certs").exists()
def test_ensure_base_directories_handles_permission_error(self, tmp_path, caplog):
"""Verify _ensure_base_directories logs error on permission failure."""
import logging
from backend.app.services.virtual_printer.manager import VirtualPrinterManager
manager = VirtualPrinterManager()
vp_dir = tmp_path / "virtual_printer"
manager._base_dir = vp_dir
original_mkdir = type(vp_dir).mkdir
def mock_mkdir(self, *args, **kwargs):
if "virtual_printer" in str(self):
raise PermissionError("Permission denied")
return original_mkdir(self, *args, **kwargs)
with caplog.at_level(logging.ERROR), patch.object(type(vp_dir), "mkdir", mock_mkdir):
manager._ensure_base_directories()
assert "Permission denied" in caplog.text
def test_instance_creates_per_vp_directories(self, tmp_path):
"""Verify VirtualPrinterInstance creates per-VP upload and cert dirs."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
VirtualPrinterInstance(
vp_id=42,
name="Test",
mode="archive",
model="C11",
access_code="12345678",
serial_suffix="391800042",
base_dir=tmp_path,
)
assert (tmp_path / "uploads" / "42").exists()
assert (tmp_path / "uploads" / "42" / "cache").exists()
assert (tmp_path / "certs" / "42").exists()
class TestVirtualPrinterCameraPassthrough:
"""Tests for the non-proxy VP camera pass-through port selection (#1868).
The camera port must follow the TARGET printer's model, not the VP's
spoofed model: RTSP models (X1/X2/H2/P2S) use 322, chamber-image models
(A1/P1) use 6000. Before the fix, the manager hardcoded 322, so P1S /
A1 targets got a 322 listener with no upstream and BambuStudio /
OrcaSlicer Liveview failed with error `[2:-10061]` (the reporter's
symptom in #1868).
"""
@staticmethod
def _patch_start_server(monkeypatch):
"""Patch every non-camera constructor start_server touches so the
test only exercises the camera-port branch. Returns the list that
``TCPProxy`` calls get captured into."""
from backend.app.services.virtual_printer import manager as vp_manager
tcp_calls: list[dict] = []
class FakeTCPProxy:
def __init__(self, **kwargs):
tcp_calls.append(kwargs)
self.kwargs = kwargs
async def start(self):
return None
async def stop(self):
return None
# Camera pass-through is the only TCPProxy start_server constructs.
monkeypatch.setattr(vp_manager, "TCPProxy", FakeTCPProxy)
# No-op every other service constructor + resolve_cert_and_advertise.
# start_server() awaits `.ready.wait()` on FTP/MQTT/Bind/SSDP after
# spawning them, so each fake instance must carry an already-set
# asyncio.Event as `.ready` — a plain MagicMock returns another
# MagicMock for `.wait()`, which `asyncio.gather` then rejects with
# `TypeError: An asyncio.Future, a coroutine or an awaitable is
# required`.
def _service_factory():
def make(*args, **kwargs):
inst = MagicMock()
inst.start = AsyncMock(return_value=None)
inst.stop = AsyncMock(return_value=None)
ready = asyncio.Event()
ready.set()
inst.ready = ready
return inst
return make
for name in (
"VirtualPrinterFTPServer",
"SimpleMQTTServer",
"MQTTBridge",
"BindServer",
"VirtualPrinterSSDPServer",
"SSDPProxy",
):
monkeypatch.setattr(vp_manager, name, _service_factory())
return tcp_calls
async def _run_start_server(
self,
tmp_path,
monkeypatch,
*,
target_model: str,
target_ip: str = "192.168.1.100",
) -> list[dict]:
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
tcp_calls = self._patch_start_server(monkeypatch)
instance = VirtualPrinterInstance(
vp_id=99,
name="CamTest",
mode="archive",
model="BL-P001", # VP's spoofed identity — irrelevant here
access_code="12345678",
serial_suffix="391800099",
target_printer_id=7,
base_dir=tmp_path,
)
# printer_manager stub returns a client with the model + ip we want.
client = MagicMock()
client.ip_address = target_ip
client.model = target_model
printer_manager = MagicMock()
printer_manager.get_client.return_value = client
instance._printer_manager = printer_manager
# Cert / advertise resolution — start_server calls this early. Patch
# to a fixed tuple so no filesystem I/O is required.
monkeypatch.setattr(
instance,
"_resolve_cert_and_advertise",
lambda: (Path("/tmp/cert.pem"), Path("/tmp/key.pem"), "192.168.1.1"), # nosec B108
)
try:
await instance.start_server()
finally:
for task in instance._tasks:
task.cancel()
await asyncio.gather(*instance._tasks, return_exceptions=True)
return tcp_calls
@pytest.mark.asyncio
async def test_rtsp_model_p2s_opens_port_322(self, tmp_path, monkeypatch):
calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P2S")
assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls), (
f"Expected 322 pass-through for RTSP model P2S, got {calls}"
)
@pytest.mark.asyncio
async def test_chamber_image_model_p1s_opens_port_6000(self, tmp_path, monkeypatch):
"""#1868 regression guard: P1S target must expose 6000, not 322."""
calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P1S")
assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls), (
f"Expected 6000 pass-through for chamber-image model P1S (#1868), got {calls}"
)
assert not any(c["listen_port"] == 322 for c in calls), f"P1S should NOT get a 322 listener, got {calls}"
@pytest.mark.asyncio
async def test_chamber_image_model_a1_opens_port_6000(self, tmp_path, monkeypatch):
calls = await self._run_start_server(tmp_path, monkeypatch, target_model="A1")
assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls)
@pytest.mark.asyncio
async def test_rtsp_model_x1c_opens_port_322(self, tmp_path, monkeypatch):
calls = await self._run_start_server(tmp_path, monkeypatch, target_model="X1C")
assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls)
class TestVirtualPrinterInstanceProxyMode:
"""Tests for VirtualPrinterInstance proxy mode."""
@pytest.fixture
def proxy_instance(self, tmp_path):
"""Create a proxy-mode VirtualPrinterInstance."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
return VirtualPrinterInstance(
vp_id=10,
name="ProxyTest",
mode="proxy",
model="C11",
access_code="",
serial_suffix="391800010",
target_printer_ip="192.168.1.100",
target_printer_serial="01P00A000000001",
base_dir=tmp_path,
)
def test_proxy_instance_properties(self, proxy_instance):
"""Verify proxy instance stores config correctly."""
assert proxy_instance.is_proxy is True
assert proxy_instance.mode == "proxy"
assert proxy_instance.target_printer_ip == "192.168.1.100"
assert proxy_instance.target_printer_serial == "01P00A000000001"
def test_proxy_instance_does_not_require_access_code(self, proxy_instance):
"""Verify proxy mode can have empty access code."""
assert proxy_instance.access_code == ""
def test_get_status_proxy_includes_proxy_fields(self, proxy_instance):
"""Verify get_status includes proxy fields when proxy is active."""
mock_proxy = MagicMock()
mock_proxy.get_status.return_value = {
"running": True,
"ftp_port": 990,
"mqtt_port": 8883,
"ftp_connections": 1,
"mqtt_connections": 2,
"target_host": "192.168.1.100",
}
proxy_instance._proxy = mock_proxy
status = proxy_instance.get_status()
assert "proxy" in status
assert status["proxy"]["ftp_port"] == 990
assert status["proxy"]["mqtt_connections"] == 2
def test_proxy_instance_stores_remote_interface(self, tmp_path):
"""Verify proxy instance stores remote_interface_ip."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=11,
name="Proxy2",
mode="proxy",
model="C11",
access_code="",
serial_suffix="391800011",
target_printer_ip="192.168.1.100",
remote_interface_ip="10.0.0.50",
base_dir=tmp_path,
)
assert inst.remote_interface_ip == "10.0.0.50"
class TestVirtualPrinterInstanceIPOverride:
"""Tests for remote_interface_ip and bind_ip on VirtualPrinterInstance."""
@pytest.fixture
def instance_with_remote_ip(self, tmp_path):
"""Create an instance with remote_interface_ip set."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
return VirtualPrinterInstance(
vp_id=20,
name="IPTest",
mode="archive",
model="BL-P001",
access_code="12345678",
serial_suffix="391800020",
bind_ip="192.168.1.50",
remote_interface_ip="10.0.0.50",
base_dir=tmp_path,
)
def test_instance_stores_bind_ip(self, instance_with_remote_ip):
"""Verify bind_ip is stored."""
assert instance_with_remote_ip.bind_ip == "192.168.1.50"
def test_instance_stores_remote_interface_ip(self, instance_with_remote_ip):
"""Verify remote_interface_ip is stored."""
assert instance_with_remote_ip.remote_interface_ip == "10.0.0.50"
def test_generate_certificates_includes_remote_and_bind_ip(self, instance_with_remote_ip):
"""Verify generate_certificates passes remote_interface_ip and bind_ip as SANs."""
with (
patch.object(instance_with_remote_ip._cert_service, "delete_printer_certificate"),
patch.object(
instance_with_remote_ip._cert_service,
"generate_certificates",
return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")), # nosec B108
) as mock_gen,
):
instance_with_remote_ip.generate_certificates()
mock_gen.assert_called_once_with(additional_ips=["10.0.0.50", "192.168.1.50"])
def test_generate_certificates_no_remote_ip(self, tmp_path):
"""Verify generate_certificates passes only bind_ip when no remote_interface_ip."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=21,
name="NoRemote",
mode="archive",
model="BL-P001",
access_code="12345678",
serial_suffix="391800021",
bind_ip="192.168.1.50",
base_dir=tmp_path,
)
with (
patch.object(inst._cert_service, "delete_printer_certificate"),
patch.object(
inst._cert_service,
"generate_certificates",
return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")), # nosec B108
) as mock_gen,
):
inst.generate_certificates()
mock_gen.assert_called_once_with(additional_ips=["192.168.1.50"])
def test_generate_certificates_no_ips(self, tmp_path):
"""Verify generate_certificates passes None when no IPs configured."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=22,
name="NoIPs",
mode="archive",
model="BL-P001",
access_code="12345678",
serial_suffix="391800022",
base_dir=tmp_path,
)
with (
patch.object(inst._cert_service, "delete_printer_certificate"),
patch.object(
inst._cert_service,
"generate_certificates",
return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")), # nosec B108
) as mock_gen,
):
inst.generate_certificates()
mock_gen.assert_called_once_with(additional_ips=None)
class TestBindServer:
"""Tests for the BindServer (port 3002 bind/detect protocol)."""
@pytest.fixture
def bind_server(self):
"""Create a BindServer instance."""
from backend.app.services.virtual_printer.bind_server import BindServer
return BindServer(
serial="01S00C000000001",
model="BL-P001",
name="Bambuddy",
)
def test_build_frame(self, bind_server):
"""Verify frame format: 0xA5A5 + len(u16le) + JSON + 0xA7A7."""
payload = {"login": {"command": "detect"}}
frame = bind_server._build_frame(payload)
assert frame[:2] == b"\xa5\xa5"
assert frame[-2:] == b"\xa7\xa7"
# Length field is total message size
import struct
total_len = struct.unpack_from("<H", frame, 2)[0]
assert total_len == len(frame)
# JSON payload is between header and trailer
import json
json_bytes = frame[4:-2]
parsed = json.loads(json_bytes)
assert parsed == payload
def test_parse_frame_valid(self, bind_server):
"""Verify valid frame parsing."""
frame = bind_server._build_frame({"login": {"command": "detect", "sequence_id": "20000"}})
result = bind_server._parse_frame(frame)
assert result is not None
assert result["login"]["command"] == "detect"
assert result["login"]["sequence_id"] == "20000"
def test_parse_frame_invalid_header(self, bind_server):
"""Verify invalid header returns None."""
frame = b"\xb5\xb5\x10\x00" + b'{"login":{}}' + b"\xa7\xa7"
assert bind_server._parse_frame(frame) is None
def test_parse_frame_invalid_trailer(self, bind_server):
"""Verify invalid trailer returns None."""
frame = b"\xa5\xa5\x10\x00" + b'{"login":{}}' + b"\xb7\xb7"
assert bind_server._parse_frame(frame) is None
def test_parse_frame_too_short(self, bind_server):
"""Verify short data returns None."""
assert bind_server._parse_frame(b"\xa5\xa5\x00") is None
assert bind_server._parse_frame(b"") is None
def test_parse_frame_invalid_json(self, bind_server):
"""Verify invalid JSON returns None."""
import struct
bad_json = b"not json"
total_len = 4 + len(bad_json) + 2
frame = b"\xa5\xa5" + struct.pack("<H", total_len) + bad_json + b"\xa7\xa7"
assert bind_server._parse_frame(frame) is None
def test_build_frame_roundtrip(self, bind_server):
"""Verify build then parse roundtrip."""
original = {"login": {"bind": "free", "command": "detect", "id": "01S00C000000001"}}
frame = bind_server._build_frame(original)
parsed = bind_server._parse_frame(frame)
assert parsed == original
def test_bind_server_stores_config(self, bind_server):
"""Verify config is stored correctly."""
assert bind_server.serial == "01S00C000000001"
assert bind_server.model == "BL-P001"
assert bind_server.name == "Bambuddy"
assert bind_server.version == "01.00.00.00"
def test_bind_server_custom_version(self):
"""Verify custom firmware version is stored."""
from backend.app.services.virtual_printer.bind_server import BindServer
server = BindServer(
serial="01S00C000000001",
model="BL-P001",
name="Bambuddy",
version="01.09.00.10",
)
assert server.version == "01.09.00.10"
def test_bind_ports_includes_both(self):
"""Verify BIND_PORTS includes both 3000 and 3002 for slicer compatibility."""
from backend.app.services.virtual_printer.bind_server import BIND_PORTS
assert 3000 in BIND_PORTS
assert 3002 in BIND_PORTS
def test_bind_server_initializes_empty_servers_list(self, bind_server):
"""Verify bind server starts with empty servers list."""
assert bind_server._servers == []
assert bind_server._running is False
@pytest.mark.asyncio
async def test_start_server_creates_bind_server(self, tmp_path):
"""Verify start_server creates BindServer with correct params."""
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
inst = VirtualPrinterInstance(
vp_id=99,
name="Bambuddy",
mode="archive",
model="BL-P001",
access_code="12345678",
serial_suffix="391800099",
bind_ip="192.168.1.50",
base_dir=tmp_path,
)
# Each mocked child service exposes a real asyncio.Event for the
# readiness barrier added in start_server (set on instantiation so
# the barrier returns immediately in tests).
ready_event = asyncio.Event()
ready_event.set()
def with_ready(*_args, **_kwargs):
child = MagicMock()
child.ready = ready_event
return child
with (
patch(
"backend.app.services.virtual_printer.manager.VirtualPrinterSSDPServer",
side_effect=with_ready,
),
patch(
"backend.app.services.virtual_printer.manager.VirtualPrinterFTPServer",
side_effect=with_ready,
),
patch(
"backend.app.services.virtual_printer.manager.SimpleMQTTServer",
side_effect=with_ready,
),
patch(
"backend.app.services.virtual_printer.manager.BindServer",
side_effect=with_ready,
) as mock_bind_cls,
patch.object(inst._cert_service, "delete_printer_certificate"),
patch.object(
inst._cert_service,
"generate_certificates",
return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")), # nosec B108
),
):
await inst.start_server()
mock_bind_cls.assert_called_once_with(
serial=inst.serial,
model="BL-P001",
name="Bambuddy",
bind_address="192.168.1.50",
cert_path=Path("/tmp/cert.pem"), # nosec B108
key_path=Path("/tmp/key.pem"), # nosec B108
)
class TestResolveModelCodes:
"""Tests for model code resolution (display name → SSDP code)."""
def test_display_name_to_model_code_maps_all_models(self):
"""Verify reverse mapping covers all VIRTUAL_PRINTER_MODELS entries."""
from backend.app.services.virtual_printer.manager import DISPLAY_NAME_TO_MODEL_CODE, VIRTUAL_PRINTER_MODELS
for _code, display_name in VIRTUAL_PRINTER_MODELS.items():
assert display_name in DISPLAY_NAME_TO_MODEL_CODE
# For non-duplicate display names, should map back to a valid code
assert DISPLAY_NAME_TO_MODEL_CODE[display_name] in VIRTUAL_PRINTER_MODELS
def test_resolve_printer_model_with_ssdp_code(self):
"""SSDP codes pass through unchanged."""
from backend.app.api.routes.virtual_printers import _resolve_printer_model
assert _resolve_printer_model("BL-P001") == "BL-P001"
assert _resolve_printer_model("O1D") == "O1D"
assert _resolve_printer_model("N2S") == "N2S"
def test_resolve_printer_model_with_display_name(self):
"""Display names resolve to SSDP codes."""
from backend.app.api.routes.virtual_printers import _resolve_printer_model
assert _resolve_printer_model("X1C") == "BL-P001"
assert _resolve_printer_model("H2D") == "O1D"
assert _resolve_printer_model("A1") == "N2S"
assert _resolve_printer_model("P1S") == "C12"
def test_resolve_printer_model_with_none_or_unknown(self):
"""None and unknown values return None."""
from backend.app.api.routes.virtual_printers import _resolve_printer_model
assert _resolve_printer_model(None) is None
assert _resolve_printer_model("UnknownModel") is None
class TestMqttIpRewrite:
"""Tests for TLSProxy._rewrite_mqtt_ip() MQTT packet IP rewriting."""
@staticmethod
def _build_mqtt_publish(topic: str, payload: bytes) -> bytes:
"""Build a minimal MQTT PUBLISH packet."""
# PUBLISH fixed header: type 3, no flags
topic_bytes = topic.encode("utf-8")
# Variable header: topic length (2 bytes) + topic
var_header = len(topic_bytes).to_bytes(2, "big") + topic_bytes
body = var_header + payload
# Encode remaining length
remaining = len(body)
header = bytearray([0x30]) # PUBLISH, QoS 0
while True:
encoded_byte = remaining % 128
remaining //= 128
if remaining > 0:
encoded_byte |= 0x80
header.append(encoded_byte)
if remaining == 0:
break
return bytes(header) + body
@staticmethod
def _build_mqtt_pingreq() -> bytes:
"""Build an MQTT PINGREQ packet (2 bytes, no payload)."""
return b"\xc0\x00"
def test_rewrite_ip_in_publish(self):
"""IP string in PUBLISH payload is rewritten."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
payload = b'{"rtsp_url":"rtsps://192.168.1.100:322/live"}'
packet = self._build_mqtt_publish("device/status", payload)
result, buf = TLSProxy._rewrite_mqtt_ip(packet, b"192.168.1.100", b"10.0.0.1", bytearray())
assert b"10.0.0.1" in result
assert b"192.168.1.100" not in result
def test_no_rewrite_when_ip_absent(self):
"""Packets without the target IP are passed through unchanged."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
payload = b'{"status":"idle"}'
packet = self._build_mqtt_publish("device/status", payload)
result, buf = TLSProxy._rewrite_mqtt_ip(packet, b"192.168.1.100", b"10.0.0.1", bytearray())
assert result == packet
def test_non_publish_packets_unchanged(self):
"""Non-PUBLISH packets (e.g. PINGREQ) are never rewritten."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
pingreq = self._build_mqtt_pingreq()
result, buf = TLSProxy._rewrite_mqtt_ip(pingreq, b"192.168.1.100", b"10.0.0.1", bytearray())
assert result == pingreq
def test_rewrite_preserves_packet_framing(self):
"""Rewritten packet has valid MQTT remaining length."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
# Use IPs of different lengths to test length re-encoding
old_ip = b"192.168.255.133" # 15 bytes
new_ip = b"10.0.0.1" # 8 bytes
payload = b'{"ip":"192.168.255.133"}'
packet = self._build_mqtt_publish("device/status", payload)
result, buf = TLSProxy._rewrite_mqtt_ip(packet, old_ip, new_ip, bytearray())
# Parse the result to verify framing
assert result[0] == 0x30 # PUBLISH header byte
# Decode remaining length
pos = 1
remaining = 0
multiplier = 1
while True:
b = result[pos]
pos += 1
remaining += (b & 0x7F) * multiplier
multiplier *= 128
if (b & 0x80) == 0:
break
# Remaining length should match actual data
assert pos + remaining == len(result)
assert new_ip in result
def test_incomplete_packet_buffered(self):
"""Incomplete packet at end of chunk is buffered for next call."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
payload = b'{"ip":"192.168.1.100"}'
packet = self._build_mqtt_publish("device/status", payload)
# Split packet in the middle
half = len(packet) // 2
chunk1 = packet[:half]
chunk2 = packet[half:]
result1, buf = TLSProxy._rewrite_mqtt_ip(chunk1, b"192.168.1.100", b"10.0.0.1", bytearray())
# First chunk should be buffered (incomplete packet)
assert len(buf) > 0
result2, buf = TLSProxy._rewrite_mqtt_ip(chunk2, b"192.168.1.100", b"10.0.0.1", buf)
# Second chunk completes the packet, IP should be rewritten
combined = result1 + result2
assert b"10.0.0.1" in combined
assert b"192.168.1.100" not in combined
def test_multiple_packets_in_one_chunk(self):
"""Multiple MQTT packets in a single chunk are all processed."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
payload1 = b'{"ip":"192.168.1.100"}'
payload2 = b'{"other":"data"}'
packet1 = self._build_mqtt_publish("topic1", payload1)
packet2 = self._build_mqtt_publish("topic2", payload2)
combined = packet1 + packet2
result, buf = TLSProxy._rewrite_mqtt_ip(combined, b"192.168.1.100", b"10.0.0.1", bytearray())
assert b"10.0.0.1" in result
assert b"192.168.1.100" not in result
# Second packet should still be present
assert b"other" in result
def test_extra_replacements(self):
"""Extra replacement pairs (e.g. integer IP) are also applied."""
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
payload = b'{"net":{"info":[{"ip":2248124608}]}}'
packet = self._build_mqtt_publish("device/status", payload)
result, buf = TLSProxy._rewrite_mqtt_ip(
packet,
b"NOMATCH",
b"NOREPLACE",
bytearray(),
extra_replacements=[(b"2248124608", b"285190336")],
)
assert b"285190336" in result
assert b"2248124608" not in result
class TestIpToLeIntBytes:
"""Tests for TLSProxy._ip_to_le_int_bytes() integer IP conversion."""
def test_converts_ip_to_le_int(self):
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
assert TLSProxy._ip_to_le_int_bytes("192.168.255.133") == b"2248124608"
assert TLSProxy._ip_to_le_int_bytes("192.168.255.16") == b"285190336"
assert TLSProxy._ip_to_le_int_bytes("10.0.0.1") == b"16777226"
def test_roundtrip(self):
"""Verify the integer converts back to the correct IP."""
import struct
from backend.app.services.virtual_printer.tcp_proxy import TLSProxy
for ip in ["192.168.1.1", "10.0.0.1", "172.16.0.100", "192.168.255.133"]:
le_int = int(TLSProxy._ip_to_le_int_bytes(ip))
parts = ip.split(".")
expected = struct.unpack("<I", bytes(int(p) for p in parts))[0]
assert le_int == expected
class TestSSDPProxyName:
"""Tests for SSDPProxy VP name rewriting."""
@pytest.fixture
def ssdp_proxy_with_name(self):
from backend.app.services.virtual_printer.ssdp_server import SSDPProxy
return SSDPProxy(
local_interface_ip="192.168.1.100",
remote_interface_ip="10.0.0.100",
target_printer_ip="192.168.1.50",
name="H2D-1 Proxy",
)
@pytest.fixture
def ssdp_proxy_without_name(self):
from backend.app.services.virtual_printer.ssdp_server import SSDPProxy
return SSDPProxy(
local_interface_ip="192.168.1.100",
remote_interface_ip="10.0.0.100",
target_printer_ip="192.168.1.50",
)
def test_rewrite_uses_configured_name(self, ssdp_proxy_with_name):
"""When name is set, DevName is replaced entirely."""
packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: RealPrinter\r\nDevBind.bambu.com: cloud\r\n\r\n"
rewritten = ssdp_proxy_with_name._rewrite_ssdp(packet)
assert b"DevName.bambu.com: H2D-1 Proxy" in rewritten
assert b"RealPrinter" not in rewritten
def test_rewrite_appends_proxy_without_name(self, ssdp_proxy_without_name):
"""When no name is set, ' - Proxy' is appended to the real name."""
packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: RealPrinter\r\nDevBind.bambu.com: cloud\r\n\r\n"
rewritten = ssdp_proxy_without_name._rewrite_ssdp(packet)
assert b"DevName.bambu.com: RealPrinter - Proxy" in rewritten
class TestVPProjectFileStashKey:
"""Regression: `on_print_command` MUST stash slicer options under the
FTP filename (`data["file"]`, with extension), NOT under `filename`
(the slicer's `subtask_name`, bare).
#1780 root cause (real bundle, 2026-06-21): BambuStudio sends
`subtask_name = "Model_Name"` (bare) and `file = "Model_Name.gcode.3mf"`
(with extension). `_add_to_print_queue` looks up the stash under
`file_path.name` from the FTP receive side, which always has the
extension. If the stash uses `subtask_name`, lookup misses → every
captured slicer field (bed_leveling, flow_cali, vibration_cali,
layer_inspect, timelapse, nozzle_mapping) silently falls back to
settings defaults on every Bambu Studio "Send" upload.
`filename` (subtask_name) must still flow to `_schedule_finish_release`
untouched — push_status echoes it back as gcode_file / subtask_name and
the slicer matches against its own local subtask_name there. So
`on_print_command` keeps `filename` for state-feedback but derives the
stash key from `data["file"]`.
"""
@pytest.fixture
def instance(self, tmp_path):
from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
return VirtualPrinterInstance(
vp_id=99,
name="StashKeyTest",
mode="queue",
model="O1C2",
access_code="12345678",
serial_suffix="999999999",
base_dir=tmp_path,
)
@pytest.mark.asyncio
async def test_stash_key_uses_file_field_not_subtask_name(self, instance):
"""BambuStudio's real wire shape: `subtask_name` ≠ `file`.
on_print_command must stash under `data["file"]` so the FTP-side
`_add_to_print_queue` lookup matches.
"""
# mqtt_server.py:_handle_publish hands the bare subtask_name as
# `filename` and the full print_data body as `data`. The FTP filename
# lives in `data["file"]`.
await instance.on_print_command(
"Filament_Track_Switch_Holder", # subtask_name (bare)
{
"command": "project_file",
"subtask_name": "Filament_Track_Switch_Holder",
"file": "Filament_Track_Switch_Holder.gcode.3mf",
"nozzle_mapping": [16, -1, -1, 1],
},
)
# Stash MUST be under the FTP filename, not the bare subtask_name.
# `_add_to_print_queue` does `_slicer_print_options.pop(file_path.name, None)`
# where file_path.name == "Filament_Track_Switch_Holder.gcode.3mf".
assert "Filament_Track_Switch_Holder.gcode.3mf" in instance._slicer_print_options
assert "Filament_Track_Switch_Holder" not in instance._slicer_print_options
# Body must carry nozzle_mapping verbatim.
stashed = instance._slicer_print_options["Filament_Track_Switch_Holder.gcode.3mf"]
assert stashed["nozzle_mapping"] == [16, -1, -1, 1]
@pytest.mark.asyncio
async def test_stash_key_falls_back_to_filename_when_file_absent(self, instance):
"""Defensive fallback: a slicer that omits the `file` field entirely
(legacy / non-3MF) must fall back to `filename` (subtask_name), not
leave the stash unkeyed."""
await instance.on_print_command(
"BareName",
{
"command": "project_file",
"subtask_name": "BareName",
# no "file" field
},
)
assert "BareName" in instance._slicer_print_options
@pytest.mark.asyncio
async def test_stash_key_signals_event_under_file_key(self, instance):
"""`_add_to_print_queue` registers a wait-event under `file_path.name`
when the slicer's project_file arrives late. on_print_command must
signal THAT event (keyed by the FTP filename), not one keyed by
subtask_name — else the waiter times out even though the stash is
present and addressable."""
import asyncio
ftp_filename = "Filament_Track_Switch_Holder.gcode.3mf"
event = asyncio.Event()
instance._slicer_print_options_events[ftp_filename] = event
await instance.on_print_command(
"Filament_Track_Switch_Holder", # bare subtask_name
{
"command": "project_file",
"subtask_name": "Filament_Track_Switch_Holder",
"file": ftp_filename,
},
)
# Event keyed by FTP filename must fire even though on_print_command
# was called with the bare subtask_name.
assert event.is_set()
Reference in a new issue View git blame Copy permalink