"""Integration tests for Printers API endpoints. Tests the full request/response cycle for /api/v1/printers/ endpoints. """ from unittest.mock import AsyncMock, MagicMock, patch from urllib.parse import unquote import pytest from httpx import AsyncClient from sqlalchemy import select from backend.app.core.config import settings from backend.app.services.printer_media import PrinterFilesJobStatus, PrinterFilesZipResult @pytest.fixture(autouse=True) def _mock_printer_test_connection(): """Default mock: connection test returns success. POST /printers/ now refuses to persist a printer when the MQTT connection probe fails (would otherwise leave an empty card in the dashboard for a mistyped access code). Existing tests assume the save succeeds, so we mock the probe green by default; the failure branch is exercised by a dedicated test below. """ with patch( "backend.app.services.printer_manager.printer_manager.test_connection", new=AsyncMock(return_value={"success": True, "state": "IDLE", "model": "X1C"}), ) as m: yield m class TestPrintersAPI: """Integration tests for /api/v1/printers/ endpoints.""" # ======================================================================== # List endpoints # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_list_printers_empty(self, async_client: AsyncClient): """Verify empty list is returned when no printers exist.""" response = await async_client.get("/api/v1/printers/") assert response.status_code == 200 assert response.json() == [] @pytest.mark.asyncio @pytest.mark.integration async def test_list_printers_with_data(self, async_client: AsyncClient, printer_factory, db_session): """Verify list returns existing printers.""" await printer_factory(name="Test Printer") response = await async_client.get("/api/v1/printers/") assert response.status_code == 200 data = response.json() assert len(data) >= 1 assert any(p["name"] == "Test Printer" for p in data) # ======================================================================== # Create endpoints # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer(self, async_client: AsyncClient): """Verify printer can be created.""" data = { "name": "New Printer", "serial_number": "00M09A111111111", "ip_address": "192.168.1.100", "access_code": "12345678", "is_active": True, "model": "X1C", } response = await async_client.post("/api/v1/printers/", json=data) assert response.status_code == 200 result = response.json() assert result["name"] == "New Printer" assert result["serial_number"] == "00M09A111111111" assert result["model"] == "X1C" @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer_with_hostname(self, async_client: AsyncClient): """Verify printer can be created with a hostname instead of IP address.""" data = { "name": "DNS Printer", "serial_number": "00M09A555555555", "ip_address": "printer.local", "access_code": "12345678", "model": "P1S", } response = await async_client.post("/api/v1/printers/", json=data) assert response.status_code == 200 result = response.json() assert result["name"] == "DNS Printer" assert result["ip_address"] == "printer.local" @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer_with_fqdn(self, async_client: AsyncClient): """Verify printer can be created with a fully qualified domain name.""" data = { "name": "FQDN Printer", "serial_number": "00M09A666666666", "ip_address": "my-printer.home.lan", "access_code": "12345678", "model": "X1C", } response = await async_client.post("/api/v1/printers/", json=data) assert response.status_code == 200 result = response.json() assert result["ip_address"] == "my-printer.home.lan" @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer_invalid_hostname(self, async_client: AsyncClient): """Verify invalid hostnames are rejected.""" data = { "name": "Bad Printer", "serial_number": "00M09A777777777", "ip_address": "-invalid", "access_code": "12345678", } response = await async_client.post("/api/v1/printers/", json=data) assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer_duplicate_serial(self, async_client: AsyncClient, printer_factory, db_session): """Verify duplicate serial number is rejected.""" await printer_factory(serial_number="00M09A222222222") data = { "name": "Duplicate Printer", "serial_number": "00M09A222222222", "ip_address": "192.168.1.101", "access_code": "12345678", } response = await async_client.post("/api/v1/printers/", json=data) # Should fail due to duplicate serial assert response.status_code in [400, 409, 422, 500] @pytest.mark.asyncio @pytest.mark.integration async def test_create_printer_rejects_when_mqtt_probe_fails(self, async_client: AsyncClient, db_session): """Wrong access code / unreachable IP must NOT persist the printer. Regression: users were reporting empty / never-connecting printer cards that traced back to a mistyped access code. The create route now runs an MQTT probe up front and returns 400 if it fails — the row is never written. """ data = { "name": "Bad Code Printer", "serial_number": "00M09A999999999", "ip_address": "192.168.1.250", "access_code": "WRONG-CODE", "is_active": True, "model": "X1C", } with patch( "backend.app.services.printer_manager.printer_manager.test_connection", new=AsyncMock(return_value={"success": False, "state": None, "model": None}), ): response = await async_client.post("/api/v1/printers/", json=data) assert response.status_code == 400 detail = response.json()["detail"] # Backend returns a stable code for the frontend i18n layer to map; # the message field is an English fallback for non-UI clients. assert detail["code"] == "printer_connection_failed" assert "connect" in detail["message"].lower() # And critically: the printer row was never persisted. from backend.app.models.printer import Printer result = await db_session.execute(select(Printer).where(Printer.serial_number == "00M09A999999999")) assert result.scalar_one_or_none() is None, "Failed-probe printer must not be persisted" # ======================================================================== # Get single endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer(self, async_client: AsyncClient, printer_factory, db_session): """Verify single printer can be retrieved.""" printer = await printer_factory(name="Get Test Printer") response = await async_client.get(f"/api/v1/printers/{printer.id}") assert response.status_code == 200 result = response.json() assert result["id"] == printer.id assert result["name"] == "Get Test Printer" @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.get("/api/v1/printers/9999") assert response.status_code == 404 # ======================================================================== # Update endpoints # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_update_printer_name(self, async_client: AsyncClient, printer_factory, db_session): """Verify printer name can be updated.""" printer = await printer_factory(name="Original Name") response = await async_client.patch(f"/api/v1/printers/{printer.id}", json={"name": "Updated Name"}) assert response.status_code == 200 assert response.json()["name"] == "Updated Name" @pytest.mark.asyncio @pytest.mark.integration async def test_update_printer_active_status(self, async_client: AsyncClient, printer_factory, db_session): """Verify printer active status can be updated.""" printer = await printer_factory(is_active=True) response = await async_client.patch(f"/api/v1/printers/{printer.id}", json={"is_active": False}) assert response.status_code == 200 assert response.json()["is_active"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_update_printer_auto_archive(self, async_client: AsyncClient, printer_factory, db_session): """Verify auto_archive setting can be updated.""" printer = await printer_factory(auto_archive=True) response = await async_client.patch(f"/api/v1/printers/{printer.id}", json={"auto_archive": False}) assert response.status_code == 200 assert response.json()["auto_archive"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_update_nonexistent_printer(self, async_client: AsyncClient): """Verify updating non-existent printer returns 404.""" response = await async_client.patch("/api/v1/printers/9999", json={"name": "New Name"}) assert response.status_code == 404 # ======================================================================== # Delete endpoints # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_delete_printer(self, async_client: AsyncClient, printer_factory, db_session): """Verify printer can be deleted.""" printer = await printer_factory() printer_id = printer.id response = await async_client.delete(f"/api/v1/printers/{printer_id}") assert response.status_code == 200 # Verify deleted response = await async_client.get(f"/api/v1/printers/{printer_id}") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_delete_printer_removes_scheduled_dryings( self, async_client: AsyncClient, printer_factory, db_session ): """Deleting a printer must also delete its scheduled_dryings rows, since SQLite doesn't enforce FK cascades (#2638). """ from backend.app.models.scheduled_drying import ScheduledDrying printer = await printer_factory() printer_id = printer.id row = ScheduledDrying(printer_id=printer_id, ams_id=0, temp=55, duration_hours=8) db_session.add(row) await db_session.commit() row_id = row.id response = await async_client.delete(f"/api/v1/printers/{printer_id}") assert response.status_code == 200 result = await db_session.execute(select(ScheduledDrying).where(ScheduledDrying.id == row_id)) assert result.scalar_one_or_none() is None @pytest.mark.asyncio @pytest.mark.integration async def test_delete_nonexistent_printer(self, async_client: AsyncClient): """Verify deleting non-existent printer returns 404.""" response = await async_client.delete("/api/v1/printers/9999") assert response.status_code == 404 # ======================================================================== # File download endpoint — non-ASCII filename regression (#1245) # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize( ("filename", "ascii_fallback"), [ ("龙泡泡石墩子_p2s_ok.gcode.3mf", "p2s_ok.gcode.3mf"), ("こんにちは.gcode.3mf", "gcode.3mf"), ("résumé.gcode.3mf", "rsum.gcode.3mf"), ("مرحبا.gcode.3mf", "gcode.3mf"), ("文件.3mf", "3mf"), ("hello.3mf", "hello.3mf"), ], ) async def test_download_printer_file_non_ascii_filename( self, async_client: AsyncClient, printer_factory, filename: str, ascii_fallback: str, db_session, tmp_path, ): """Non-ASCII filenames must not crash header encoding (issue #1245).""" printer = await printer_factory() file_bytes = b"fake 3mf content" staged = tmp_path / "single" / "printer-file" staged.parent.mkdir() staged.write_bytes(file_bytes) with patch( "backend.app.api.routes.printers.build_printer_file", new=AsyncMock( return_value=PrinterFilesZipResult( path=staged, requested=1, successful=1, failed_paths=(), total_bytes=len(file_bytes), ) ), ): response = await async_client.get( f"/api/v1/printers/{printer.id}/files/download", params={"path": f"/cache/{filename}"}, ) assert response.status_code == 200 assert response.content == file_bytes content_disposition = response.headers["content-disposition"] assert f'filename="{ascii_fallback}"' in content_disposition assert "filename*=UTF-8''" in content_disposition encoded_name = content_disposition.split("filename*=UTF-8''", 1)[1] assert unquote(encoded_name) == filename @pytest.mark.asyncio @pytest.mark.integration async def test_download_job_returns_immediately_and_prepared_token_streams_file_response( self, async_client: AsyncClient, printer_factory, tmp_path, monkeypatch, ): """The browser polls a short job request, then uses a native GET.""" from backend.app.core.auth import create_slicer_download_token printer = await printer_factory() monkeypatch.setattr(settings, "archive_dir", tmp_path / "archive") token = await create_slicer_download_token("printer-files", printer.id) bundle_dir = settings.archive_dir / "temp" / "printer-file-downloads" / f"{printer.id}-{token}" bundle_dir.mkdir(parents=True) zip_path = bundle_dir / "printer-files.zip" zip_path.write_bytes(b"disk-backed zip") with patch( "backend.app.api.routes.printers.start_printer_files_job", new=AsyncMock( return_value=PrinterFilesJobStatus( job_id="job-id", printer_id=printer.id, state="queued", requested=2, filename="Test Printer videos.zip", ) ), ): job_response = await async_client.post( f"/api/v1/printers/{printer.id}/files/download-job", json={ "paths": ["/ipcam/one.mp4", "/ipcam/two.mp4"], "sizes": {"/ipcam/one.mp4": 7, "/ipcam/two.mp4": 8}, "filename": "Test Printer videos.zip", "as_zip": True, }, ) assert job_response.status_code == 200 assert job_response.json() == { "job_id": "job-id", "printer_id": printer.id, "state": "queued", "requested": 2, "successful": 0, "failed": 0, "token": None, "filename": "Test Printer videos.zip", "message": None, } response = await async_client.get( f"/api/v1/printers/{printer.id}/files/dl/{token}/Test%20Printer%20videos.zip", ) assert response.status_code == 200 assert response.content == b"disk-backed zip" assert 'filename="Test Printer videos.zip"' in response.headers["content-disposition"] # Tokens are single-use, including after a successful large download. replay = await async_client.get( f"/api/v1/printers/{printer.id}/files/dl/{token}/Test%20Printer%20videos.zip", ) assert replay.status_code == 403 # The browser reaches this URL through an click and saves # whatever comes back under the name it was going to use, so a refusal # has to arrive as a legible file rather than as a JSON body landing on # the user's disk named .zip. assert replay.headers["content-type"].startswith("text/plain") assert 'filename="download-failed.txt"' in replay.headers["content-disposition"] assert b"expired" in replay.content @pytest.mark.asyncio @pytest.mark.integration async def test_bulk_download_token_rejects_another_printer_without_consuming_it( self, async_client: AsyncClient, printer_factory, tmp_path, monkeypatch, ): """A printer-A token cannot download through printer B and survives that attempt.""" printer_a = await printer_factory(name="Printer A", serial_number="TOKENBINDINGA01") printer_b = await printer_factory(name="Printer B", serial_number="TOKENBINDINGB01") monkeypatch.setattr(settings, "archive_dir", tmp_path / "archive") from backend.app.core.auth import create_slicer_download_token token = await create_slicer_download_token("printer-files", printer_a.id) bundle_dir = settings.archive_dir / "temp" / "printer-file-downloads" / f"{printer_a.id}-{token}" bundle_dir.mkdir(parents=True) zip_path = bundle_dir / "printer-files.zip" zip_path.write_bytes(b"bound zip") wrong_printer = await async_client.get( f"/api/v1/printers/{printer_b.id}/files/dl/{token}/printer-files.zip", ) assert wrong_printer.status_code == 403 correct_printer = await async_client.get( f"/api/v1/printers/{printer_a.id}/files/dl/{token}/printer-files.zip", ) assert correct_printer.status_code == 200 assert correct_printer.content == b"bound zip" @pytest.mark.asyncio @pytest.mark.integration async def test_legacy_bulk_download_accepts_paths_without_sizes( self, async_client: AsyncClient, printer_factory, tmp_path, ): """Existing API clients may omit the newer preflight size hints.""" printer = await printer_factory() bundle_dir = tmp_path / "legacy-bundle" bundle_dir.mkdir() zip_path = bundle_dir / "printer-files.zip" zip_path.write_bytes(b"legacy zip") build_zip = AsyncMock( return_value=PrinterFilesZipResult( path=zip_path, requested=2, successful=2, failed_paths=(), total_bytes=10, ) ) with patch("backend.app.api.routes.printers.build_printer_files_zip", new=build_zip): response = await async_client.post( f"/api/v1/printers/{printer.id}/files/download-zip", json={"paths": ["model.gcode", "model.gcode"]}, ) assert response.status_code == 200 assert response.content == b"legacy zip" assert build_zip.await_args.args[1:] == (["/model.gcode", "/model.gcode"], {}) assert build_zip.await_args.kwargs == {"preserve_paths": False, "allow_empty": True} @pytest.mark.asyncio @pytest.mark.integration async def test_legacy_bulk_download_empty_selection_keeps_400_contract( self, async_client: AsyncClient, printer_factory, ): printer = await printer_factory() response = await async_client.post( f"/api/v1/printers/{printer.id}/files/download-zip", json={"paths": []}, ) assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("transport", ["header", "bearer"]) async def test_printer_file_routes_enforce_api_key_printer_scope( self, async_client: AsyncClient, printer_factory, db_session, transport: str, ): """A key restricted to printer A must not list printer B's storage.""" from backend.app.core.auth import generate_api_key from backend.app.models.api_key import APIKey from backend.app.services.bambu_ftp import FileListResult printer_a = await printer_factory(name="Scoped A", serial_number="SCOPEA000000001") printer_b = await printer_factory(name="Scoped B", serial_number="SCOPEB000000001") setup = await async_client.post( "/api/v1/auth/setup", json={ "auth_enabled": True, "admin_username": "scopeadmin", "admin_password": "AdminPass1!", }, ) assert setup.status_code == 200, setup.text full_key, key_hash, key_prefix = generate_api_key() db_session.add( APIKey( name=f"printer-scope-{transport}", key_hash=key_hash, key_prefix=key_prefix, can_control_printer=True, printer_ids=[printer_a.id], enabled=True, ) ) await db_session.commit() headers = {"X-API-Key": full_key} if transport == "header" else {"Authorization": f"Bearer {full_key}"} listing = AsyncMock(return_value=FileListResult(files=[], available=True)) with patch("backend.app.api.routes.printers.list_files_result_async", new=listing): denied = await async_client.get( f"/api/v1/printers/{printer_b.id}/files", headers=headers, ) allowed = await async_client.get( f"/api/v1/printers/{printer_a.id}/files", headers=headers, ) assert denied.status_code == 403 assert allowed.status_code == 200 listing.assert_awaited_once() # ======================================================================== # Status endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer_status( self, async_client: AsyncClient, printer_factory, mock_printer_manager, db_session ): """Verify printer status can be retrieved.""" printer = await printer_factory() response = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert response.status_code == 200 result = response.json() assert "connected" in result assert "state" in result @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer_status_not_found(self, async_client: AsyncClient): """Verify 404 for status of non-existent printer.""" response = await async_client.get("/api/v1/printers/9999/status") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer_status_includes_fila_switch_when_installed( self, async_client: AsyncClient, printer_factory, db_session ): """When the FTS accessory is installed, the status response must include the fila_switch object with the routing arrays. See #1162. The accessory is detected from print.device.fila_switch in MQTT; we feed a PrinterState with FilaSwitchState(installed=True, ...) and confirm it survives the schema serialization round-trip. """ from unittest.mock import MagicMock, patch from backend.app.services.bambu_mqtt import FilaSwitchState, PrinterState printer = await printer_factory() state = PrinterState() state.connected = True state.state = "IDLE" state.fila_switch = FilaSwitchState( installed=True, in_slots=[-1, 2], out_extruders=[0, 1], stat=0, info=2, ) with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) response = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert response.status_code == 200 result = response.json() assert result["fila_switch"] is not None assert result["fila_switch"]["installed"] is True assert result["fila_switch"]["in_slots"] == [-1, 2] assert result["fila_switch"]["out_extruders"] == [0, 1] assert result["fila_switch"]["stat"] == 0 assert result["fila_switch"]["info"] == 2 @pytest.mark.asyncio @pytest.mark.integration async def test_status_reports_switch_readiness_on_the_first_load(self, async_client: AsyncClient, printer_factory): """``ready`` has to be computed by the REST route, not just the WebSocket. This response is what the page has before any push arrives. Leaving the field at its default would tell a correctly set-up machine that its switch is not set up, and the AMS menu refuses Load on that. """ from unittest.mock import MagicMock, patch from backend.app.services.bambu_mqtt import FilaSwitchState, PrinterState printer = await printer_factory() state = PrinterState() state.connected = True state.state = "IDLE" state.fila_switch = FilaSwitchState(installed=True) state.raw_data = {"ams": [{"id": "0", "tray": []}, {"id": "1", "tray": []}]} state.ams_switch_inlet = {"0": "A"} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) unbound = await async_client.get(f"/api/v1/printers/{printer.id}/status") state.ams_switch_inlet = {"0": "A", "1": "B"} bound = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert unbound.json()["fila_switch"]["ready"] is False assert bound.json()["fila_switch"]["ready"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_status_reports_which_hotend_holds_which_slot(self, async_client: AsyncClient, printer_factory): """Also needed on the first load: it decides which hotend Load may offer.""" from unittest.mock import MagicMock, patch from backend.app.services.bambu_mqtt import ExtruderSlot, PrinterState printer = await printer_factory() state = PrinterState() state.connected = True state.state = "IDLE" state.extruder_slots = { 0: ExtruderSlot(ams_id=0, slot_id=2, has_filament=True), 1: ExtruderSlot(), } with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) response = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert response.json()["extruder_slots"] == { "0": {"ams_id": 0, "slot_id": 2, "has_filament": True}, "1": {"ams_id": None, "slot_id": None, "has_filament": False}, } @pytest.mark.asyncio @pytest.mark.integration async def test_cover_uses_dispatched_plate_when_gcode_file_lacks_path( self, async_client: AsyncClient, printer_factory, db_session, tmp_path ): """When firmware drops the plate path from gcode_file (e.g. P1S 01.10.00.00, #1166), the dispatched-plate record must take precedence and serve plate 4's thumbnail instead of falling back to plate_1.png.""" import io import zipfile from unittest.mock import MagicMock, patch from backend.app.services.bambu_ftp import cache_3mf_download from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() # Build a 3MF that mimics a "true" multi-plate archive: thumbnails # for plates 1..4 are all present, gcode files for plates 1..4 are # all present. Without the dispatch record we'd default to plate_1.png. threemf_path = tmp_path / "MyModel.3mf" with zipfile.ZipFile(threemf_path, "w") as zf: for plate in range(1, 5): zf.writestr(f"Metadata/plate_{plate}.png", f"PLATE_{plate}_PNG".encode()) zf.writestr(f"Metadata/plate_{plate}.gcode", f"; plate {plate} gcode\n") cache_3mf_download(printer.id, "MyModel.3mf", threemf_path) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "MyModel" state.gcode_file = "MyModel.3mf" # firmware drops plate path state.dispatched_plate_id = 4 state.dispatched_subtask = "MyModel" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert response.status_code == 200 assert response.content == b"PLATE_4_PNG" @pytest.mark.asyncio @pytest.mark.integration async def test_cover_pick_view_serves_active_plate_object_mask( self, async_client: AsyncClient, printer_factory, db_session, tmp_path ): """The skip-items UI needs the slicer's exact object-ID mask, not an inferred bounding box, so a plate click resolves to the firmware ID.""" import zipfile from unittest.mock import MagicMock, patch from backend.app.services.bambu_ftp import cache_3mf_download from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() threemf_path = tmp_path / "PickMask.3mf" with zipfile.ZipFile(threemf_path, "w") as zf: zf.writestr("Metadata/pick_1.png", b"PICK_ONE") zf.writestr("Metadata/pick_3.png", b"PICK_THREE") zf.writestr("Metadata/plate_3.gcode", "; active plate\n") cache_3mf_download(printer.id, "PickMask.3mf", threemf_path) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "PickMask" state.gcode_file = "PickMask.3mf" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) response = await async_client.get(f"/api/v1/printers/{printer.id}/cover?view=pick") assert response.status_code == 200 assert response.content == b"PICK_THREE" @pytest.mark.asyncio @pytest.mark.integration async def test_cover_pick_view_404s_rather_than_serving_a_render( self, async_client: AsyncClient, printer_factory, db_session, tmp_path ): """A mask is coordinates, not decoration. Archives without pick_N.png (Handy jobs, older slicers) must 404 so the UI drops to the checklist — every other view's fallback to a rendered thumbnail would be decoded as object IDs here, and a click would skip an arbitrary object.""" import zipfile from unittest.mock import MagicMock, patch from backend.app.services.bambu_ftp import cache_3mf_download from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() threemf_path = tmp_path / "NoMask.3mf" with zipfile.ZipFile(threemf_path, "w") as zf: zf.writestr("Metadata/top_1.png", b"TOP_RENDER") zf.writestr("Metadata/plate_1.png", b"PLATE_RENDER") zf.writestr("Metadata/plate_1.gcode", "; active plate\n") cache_3mf_download(printer.id, "NoMask.3mf", threemf_path) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "NoMask" state.gcode_file = "NoMask.3mf" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) response = await async_client.get(f"/api/v1/printers/{printer.id}/cover?view=pick") assert response.status_code == 404 assert b"RENDER" not in response.content @pytest.mark.asyncio @pytest.mark.integration async def test_cover_pick_view_does_not_borrow_another_plates_mask( self, async_client: AsyncClient, printer_factory, db_session, tmp_path ): """Plate 1's mask over plate 3's layout resolves clicks to whatever occupied that pixel on a different plate, so the active plate's mask is the only acceptable answer.""" import zipfile from unittest.mock import MagicMock, patch from backend.app.services.bambu_ftp import cache_3mf_download from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() threemf_path = tmp_path / "OtherPlate.3mf" with zipfile.ZipFile(threemf_path, "w") as zf: zf.writestr("Metadata/pick_1.png", b"PICK_ONE") zf.writestr("Metadata/top_3.png", b"TOP_THREE") zf.writestr("Metadata/plate_3.gcode", "; active plate\n") cache_3mf_download(printer.id, "OtherPlate.3mf", threemf_path) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "OtherPlate" state.gcode_file = "OtherPlate.3mf" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) response = await async_client.get(f"/api/v1/printers/{printer.id}/cover?view=pick") assert response.status_code == 404 assert response.content != b"PICK_ONE" @pytest.mark.asyncio @pytest.mark.integration async def test_cover_3mf_scan_fallback_for_per_plate_archive( self, async_client: AsyncClient, printer_factory, db_session, tmp_path ): """Per-plate archives sliced separately in Bambu Studio contain a single Metadata/plate_N.gcode (the active plate) but bundle thumbnails for every plate. With no dispatch record (e.g. dispatched via Studio directly) and no plate path in gcode_file, the route must scan the 3MF and pick plate N's thumbnail. See #1166 option 4.""" import zipfile from unittest.mock import MagicMock, patch from backend.app.services.bambu_ftp import cache_3mf_download from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() # Per-plate archive: thumbnails for all plates, gcode for plate 3 only. threemf_path = tmp_path / "PerPlate.3mf" with zipfile.ZipFile(threemf_path, "w") as zf: for plate in range(1, 5): zf.writestr(f"Metadata/plate_{plate}.png", f"PLATE_{plate}_PNG".encode()) zf.writestr("Metadata/plate_3.gcode", "; only plate 3 has gcode\n") cache_3mf_download(printer.id, "PerPlate.3mf", threemf_path) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "PerPlate" state.gcode_file = "PerPlate.3mf" # No dispatch record (Studio-direct dispatch). with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert response.status_code == 200 assert response.content == b"PLATE_3_PNG" @pytest.mark.asyncio @pytest.mark.integration async def test_cover_negative_cache_skips_repeat_ftp_fanout( self, async_client: AsyncClient, printer_factory, db_session ): """#1420: when every FTP path returns 550 for the current subtask, the next request for the same subtask must short-circuit to 404 instead of replaying the 8-path FTP fan-out (which starves the printer's single FTP socket and flooded the user's logs).""" from unittest.mock import AsyncMock, MagicMock, patch from backend.app.api.routes.printers import _cover_404_cache, _cover_cache from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() _cover_cache.pop(printer.id, None) _cover_404_cache.pop(printer.id, None) state = PrinterState() state.connected = True state.state = "RUNNING" state.subtask_name = "OrphanPrint" state.gcode_file = "OrphanPrint.3mf" ftp_mock = AsyncMock(return_value=False) with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch("backend.app.api.routes.printers.download_file_try_paths_async", ftp_mock), ): mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) r1 = await async_client.get(f"/api/v1/printers/{printer.id}/cover") r2 = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert r1.status_code == 404 assert r2.status_code == 404 # First call retries internally; second call must short-circuit before FTP. first_call_count = ftp_mock.await_count assert first_call_count >= 1 # Second request didn't add to the count: the negative cache held. assert ftp_mock.await_count == first_call_count @pytest.mark.asyncio @pytest.mark.integration async def test_get_printer_status_omits_fila_switch_when_not_installed( self, async_client: AsyncClient, printer_factory, db_session ): """Without the FTS accessory, fila_switch must be null so the frontend keeps applying the per-extruder filter on regular dual-nozzle printers.""" from unittest.mock import MagicMock, patch from backend.app.services.bambu_mqtt import PrinterState printer = await printer_factory() state = PrinterState() state.connected = True state.state = "IDLE" # default fila_switch — installed = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_status = MagicMock(return_value=state) mock_pm.is_awaiting_plate_clear = MagicMock(return_value=False) response = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert response.status_code == 200 result = response.json() assert result["fila_switch"] is None # ======================================================================== # Test connection endpoint # ======================================================================== class TestCoverPoolHygiene: """Regression guard for the /cover DB-connection leak (issue #2572).""" def test_cover_does_not_hold_a_get_db_session(self): """The cover endpoint must NOT take a ``Depends(get_db)`` session. ``get_db`` is a ``yield`` dependency, so its session stays open for the whole request — including the 3MF cover download (up to 8 remote paths × retries with backoff, minutes under FTP contention), pinning one pooled DB connection ``idle in transaction`` the entire time. The endpoint fetches the printer in a short-lived ``async with async_session()`` and releases the connection before the FTP work. If someone re-adds a ``Depends(get_db)`` param, this fails. """ import inspect from backend.app.api.routes.printers import get_db, get_printer_cover for name, param in inspect.signature(get_printer_cover).parameters.items(): dependency = getattr(param.default, "dependency", None) assert dependency is not get_db, ( f"get_printer_cover re-introduced a get_db-held session via parameter {name!r} — " "it would stay open for the entire FTP cover download (issue #2572)" ) class TestPrinterDataIntegrity: """Tests for printer data integrity.""" @pytest.mark.asyncio @pytest.mark.integration async def test_printer_stores_all_fields(self, async_client: AsyncClient, printer_factory, db_session): """Verify printer stores all fields correctly.""" printer = await printer_factory( name="Full Test Printer", serial_number="00M09A444444444", ip_address="192.168.1.150", model="P1S", is_active=True, auto_archive=False, ) response = await async_client.get(f"/api/v1/printers/{printer.id}") assert response.status_code == 200 result = response.json() assert result["name"] == "Full Test Printer" assert result["serial_number"] == "00M09A444444444" assert result["ip_address"] == "192.168.1.150" assert result["model"] == "P1S" assert result["is_active"] is True assert result["auto_archive"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_printer_update_persists(self, async_client: AsyncClient, printer_factory, db_session): """CRITICAL: Verify printer updates persist.""" printer = await printer_factory(name="Original", is_active=True) # Update await async_client.patch(f"/api/v1/printers/{printer.id}", json={"name": "Updated", "is_active": False}) # Verify persistence response = await async_client.get(f"/api/v1/printers/{printer.id}") result = response.json() assert result["name"] == "Updated" assert result["is_active"] is False # ======================================================================== # Refresh status endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_refresh_status_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/refresh-status") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_refresh_status_not_connected(self, async_client: AsyncClient, printer_factory): """Verify 400 when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.request_status_update.return_value = False response = await async_client.post(f"/api/v1/printers/{printer.id}/refresh-status") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_refresh_status_success(self, async_client: AsyncClient, printer_factory): """Verify successful refresh request.""" printer = await printer_factory(name="Connected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.request_status_update.return_value = True response = await async_client.post(f"/api/v1/printers/{printer.id}/refresh-status") assert response.status_code == 200 assert response.json()["status"] == "refresh_requested" mock_pm.request_status_update.assert_called_once_with(printer.id) # ======================================================================== # Current print user endpoint (Issue #206) # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_get_current_print_user_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.get("/api/v1/printers/99999/current-print-user") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_get_current_print_user_returns_empty_when_no_user(self, async_client: AsyncClient, printer_factory): """Verify empty object returned when no user is tracked.""" printer = await printer_factory(name="Test Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_current_print_user.return_value = None response = await async_client.get(f"/api/v1/printers/{printer.id}/current-print-user") assert response.status_code == 200 assert response.json() == {} @pytest.mark.asyncio @pytest.mark.integration async def test_get_current_print_user_returns_user_info(self, async_client: AsyncClient, printer_factory): """Verify user info is returned when tracked.""" printer = await printer_factory(name="Test Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_current_print_user.return_value = {"user_id": 42, "username": "testuser"} response = await async_client.get(f"/api/v1/printers/{printer.id}/current-print-user") assert response.status_code == 200 result = response.json() assert result["user_id"] == 42 assert result["username"] == "testuser" class TestPrintControlAPI: """Integration tests for print control endpoints (stop, pause, resume).""" # ======================================================================== # Stop print endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_stop_print_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/print/stop") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_stop_print_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/print/stop") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_stop_print_success(self, async_client: AsyncClient, printer_factory): """Verify successful stop print request.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.stop_print.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/stop") assert response.status_code == 200 assert response.json()["success"] is True mock_client.stop_print.assert_called_once() # ======================================================================== # Pause print endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_pause_print_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/print/pause") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_pause_print_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/print/pause") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_pause_print_success(self, async_client: AsyncClient, printer_factory): """Verify successful pause print request.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.pause_print.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/pause") assert response.status_code == 200 assert response.json()["success"] is True mock_client.pause_print.assert_called_once() # ======================================================================== # Resume print endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_resume_print_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/print/resume") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_resume_print_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/print/resume") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_resume_print_success(self, async_client: AsyncClient, printer_factory): """Verify successful resume print request.""" printer = await printer_factory(name="Paused Printer") mock_client = MagicMock() mock_client.resume_print.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/resume") assert response.status_code == 200 assert response.json()["success"] is True mock_client.resume_print.assert_called_once() class TestAMSRefreshAPI: """Integration tests for AMS slot refresh endpoint.""" @pytest.mark.asyncio @pytest.mark.integration async def test_ams_refresh_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/ams/0/slot/0/refresh") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_ams_refresh_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/0/slot/0/refresh") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_ams_refresh_success(self, async_client: AsyncClient, printer_factory): """Verify successful AMS refresh request.""" printer = await printer_factory(name="Printer with AMS") mock_client = MagicMock() mock_client.ams_refresh_tray.return_value = (True, "Refreshing AMS 0 tray 1") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/0/slot/1/refresh") assert response.status_code == 200 result = response.json() assert result["success"] is True mock_client.ams_refresh_tray.assert_called_once_with(0, 1) @pytest.mark.asyncio @pytest.mark.integration async def test_ams_refresh_filament_loaded(self, async_client: AsyncClient, printer_factory): """Verify error when filament is loaded (can't refresh while loaded).""" printer = await printer_factory(name="Printer with AMS") mock_client = MagicMock() mock_client.ams_refresh_tray.return_value = (False, "Please unload filament first") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/0/slot/0/refresh") assert response.status_code == 400 assert "unload" in response.json()["detail"].lower() class TestAMSLoadUnloadAPI: """Integration tests for AMS load / unload endpoints (#891).""" # ── load ───────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.integration async def test_load_invalid_tray_id(self, async_client: AsyncClient, printer_factory): """tray_id outside {0..15, 254, 255} is rejected.""" printer = await printer_factory(name="P") response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=99") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_load_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/ams/load?tray_id=0") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_load_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="Disconnected") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=0") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_load_ams_slot_success(self, async_client: AsyncClient, printer_factory): """tray_id=5 → AMS 1 slot 2 (1-indexed in the message).""" printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_load_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=5") assert response.status_code == 200 mock_client.ams_load_filament.assert_called_once_with(5, extruder_id=None) assert "AMS 1" in response.json()["message"] @pytest.mark.asyncio @pytest.mark.integration async def test_load_external_left_success(self, async_client: AsyncClient, printer_factory): """tray_id=254 → external spool / Ext-L.""" printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_load_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=254") assert response.status_code == 200 mock_client.ams_load_filament.assert_called_once_with(254, extruder_id=None) assert "external" in response.json()["message"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_load_external_right_success(self, async_client: AsyncClient, printer_factory): """tray_id=255 → Ext-R on dual-nozzle H2D.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_load_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=255") assert response.status_code == 200 mock_client.ams_load_filament.assert_called_once_with(255, extruder_id=None) assert "Ext-R" in response.json()["message"] @pytest.mark.asyncio @pytest.mark.integration async def test_load_mqtt_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_load_filament.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=0") assert response.status_code == 500 assert "failed" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_load_forwards_the_chosen_hotend(self, async_client: AsyncClient, printer_factory): """A printer with a Filament Track Switch has to name the hotend to feed.""" printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_load_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=5&extruder_id=1") assert response.status_code == 200 mock_client.ams_load_filament.assert_called_once_with(5, extruder_id=1) @pytest.mark.asyncio @pytest.mark.integration async def test_load_rejects_a_hotend_that_does_not_exist(self, async_client: AsyncClient, printer_factory): """Only 0 and 1 are real hotends; anything else is a client bug.""" printer = await printer_factory(name="P") response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/load?tray_id=5&extruder_id=2") assert response.status_code == 422 # ── unload ─────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.integration async def test_unload_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/ams/unload") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_unload_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="Disconnected") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_unload_success(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_unload_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload") assert response.status_code == 200 mock_client.ams_unload_filament.assert_called_once_with(None) assert response.json()["success"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_unload_forwards_the_slot(self, async_client: AsyncClient, printer_factory): """The slot is what tells a dual-nozzle printer which hotend to unload.""" printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_unload_filament.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload?tray_id=2") assert response.status_code == 200 mock_client.ams_unload_filament.assert_called_once_with(2) @pytest.mark.asyncio @pytest.mark.integration async def test_unload_of_an_unloaded_slot_is_a_conflict_not_a_fault( self, async_client: AsyncClient, printer_factory ): """Clicking Unload on an idle slot is a no-op the operator can understand. A 500 would read as a broken printer; the menu is per-slot and picking a slot no hotend is fed from is an ordinary mistake. """ printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_unload_filament.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload?tray_id=2") assert response.status_code == 409 @pytest.mark.asyncio @pytest.mark.integration async def test_unload_rejects_an_invalid_slot(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P") response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload?tray_id=99") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_unload_mqtt_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P") mock_client = MagicMock() mock_client.ams_unload_filament.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/ams/unload") assert response.status_code == 500 assert "failed" in response.json()["detail"].lower() class TestConfigureAMSSlotAPI: """Integration tests for AMS slot configure endpoint — tray_info_idx resolution.""" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/0/configure", params={ "tray_info_idx": "GFL99", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FF0000FF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, }, ) assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_configure_with_gf_id_keeps_it(self, async_client: AsyncClient, printer_factory): """Standard Bambu GF* filament IDs are sent as-is.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = None # No existing state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFL05" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_builtin_empty_setting_id_is_derived(self, async_client: AsyncClient, printer_factory): """A built-in preset (GF* tray_info_idx, empty setting_id) gets a derived GFS* setting_id (#2604). The Configure AMS Slot modal sends built-in / local / Orca-generic presets with an empty setting_id. Publishing a filament-id-without-setting-id slot makes the printer treat it as half configured and revert to its previous profile, so the route must back-fill setting_id from the resolved tray_info_idx. """ printer = await printer_factory(name="X1C") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = None response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/1/configure", params={ "tray_info_idx": "GFB99", # Generic ABS, built-in "tray_type": "ABS", "tray_sub_brands": "Generic ABS", "tray_color": "000000FF", "nozzle_temp_min": 240, "nozzle_temp_max": 280, # setting_id intentionally omitted (empty) — as the modal sends it }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" assert call_kwargs.kwargs["setting_id"] == "GFSB99" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_generic_material_fallback_derives_setting_id( self, async_client: AsyncClient, printer_factory ): """When only a material is given, the generic tray_info_idx fallback still yields a setting_id (#2604).""" printer = await printer_factory(name="X1C") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_status = MagicMock() mock_status.raw_data = {"ams": {"ams": []}} # No existing tray to reuse with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/1/configure", params={ "tray_info_idx": "", # No preset id — route derives generic from material "tray_type": "ABS", "tray_sub_brands": "Generic ABS", "tray_color": "000000FF", "nozzle_temp_min": 240, "nozzle_temp_max": 280, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" assert call_kwargs.kwargs["setting_id"] == "GFSB99" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_pfus_sent_directly(self, async_client: AsyncClient, printer_factory): """PFUS* cloud-synced custom preset IDs are sent to the printer.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_status = MagicMock() mock_status.raw_data = {"ams": {"ams": []}} # No existing tray data with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "PFUS9ac902733670a9", "tray_type": "PLA", "tray_sub_brands": "Devil Design PLA", "tray_color": "FF0000FF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "PFUS9ac902733670a9" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_pfus_takes_priority_over_slot(self, async_client: AsyncClient, printer_factory): """Provided PFUS* preset takes priority over slot's existing preset.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True # Simulate slot already configured by slicer with cloud-synced preset mock_status = MagicMock() mock_status.raw_data = { "ams": { "ams": [ { "id": 2, "tray": [ { "id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA", "tray_color": "FF0000FF", } ], } ] } } with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "PFUS9ac902733670a9", "tray_type": "PLA", "tray_sub_brands": "Devil Design PLA", "tray_color": "FF0000FF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Provided preset wins over slot's existing one assert call_kwargs.kwargs["tray_info_idx"] == "PFUS9ac902733670a9" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_pfus_used_regardless_of_slot_material(self, async_client: AsyncClient, printer_factory): """Provided PFUS* preset is used even when slot has a different material.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True # Slot currently has PETG but user is configuring PLA mock_status = MagicMock() mock_status.raw_data = { "ams": { "ams": [ { "id": 2, "tray": [{"id": 3, "tray_info_idx": "GFG99", "tray_type": "PETG", "tray_color": "FFFFFFFF"}], } ] } } with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "PFUS9ac902733670a9", "tray_type": "PLA", "tray_sub_brands": "Devil Design PLA", "tray_color": "FF0000FF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Provided preset wins — slot's material is irrelevant assert call_kwargs.kwargs["tray_info_idx"] == "PFUS9ac902733670a9" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_empty_id_uses_generic(self, async_client: AsyncClient, printer_factory): """Empty tray_info_idx (local preset) is replaced with generic.""" printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_status = MagicMock() mock_status.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "", "tray_type": "PETG", "tray_sub_brands": "PETG Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 220, "nozzle_temp_max": 260, }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFG99" @pytest.mark.asyncio @pytest.mark.integration async def test_configure_pfus_preserves_setting_id_pair(self, async_client: AsyncClient, printer_factory): """Both tray_info_idx=PFUS* and setting_id=PFUS* are forwarded untouched. Pins the end-to-end contract the frontend #1053 fix relies on: when the user configures a slot with a custom cloud preset whose cloud detail has filament_id=null, the frontend sends the setting_id in BOTH fields and the backend must not collapse either to a generic GF* ID. """ printer = await printer_factory(name="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_status = MagicMock() mock_status.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_status response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/128/0/configure", params={ "tray_info_idx": "PFUSa8fb76f9733e3c", "tray_type": "ABS", "tray_sub_brands": "Sting3D ABS", "tray_color": "000000FF", "nozzle_temp_min": 240, "nozzle_temp_max": 280, "setting_id": "PFUSa8fb76f9733e3c", }, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "PFUSa8fb76f9733e3c" assert call_kwargs.kwargs["setting_id"] == "PFUSa8fb76f9733e3c" # Explicitly assert no generic-collapse happened for this HT slot. assert call_kwargs.kwargs["tray_info_idx"] != "GFB99" class TestSkipObjectsAPI: """Integration tests for skip objects endpoints.""" # ======================================================================== # Get printable objects endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_get_objects_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.get("/api/v1/printers/99999/print/objects") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_get_objects_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_get_objects_empty(self, async_client: AsyncClient, printer_factory): """Verify empty objects list when no print is active.""" printer = await printer_factory(name="Idle Printer") mock_client = MagicMock() mock_client.state.printable_objects = {} mock_client.state.skipped_objects = [] mock_client.state.state = "IDLE" mock_client.state.subtask_name = None # Prevent FTP download attempt with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 200 result = response.json() assert result["objects"] == [] assert result["total"] == 0 assert result["skipped_count"] == 0 assert result["is_printing"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_get_objects_with_data(self, async_client: AsyncClient, printer_factory): """Verify objects list when print is active.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.state.printable_objects = {100: "Part A", 200: "Part B", 300: "Part C"} mock_client.state.skipped_objects = [200] mock_client.state.state = "RUNNING" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 200 result = response.json() assert result["total"] == 3 assert result["skipped_count"] == 1 assert result["is_printing"] is True # Check objects have correct structure objects_by_id = {obj["id"]: obj for obj in result["objects"]} assert objects_by_id[100]["name"] == "Part A" assert objects_by_id[100]["skipped"] is False assert objects_by_id[200]["name"] == "Part B" assert objects_by_id[200]["skipped"] is True assert objects_by_id[300]["name"] == "Part C" assert objects_by_id[300]["skipped"] is False # ======================================================================== # Skip objects endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_get_objects_with_positions(self, async_client: AsyncClient, printer_factory): """Verify objects list includes position data when available.""" printer = await printer_factory(name="Printing Printer") # New format with position data mock_client = MagicMock() mock_client.state.printable_objects = { 100: {"name": "Part A", "x": 50.0, "y": 100.0}, 200: {"name": "Part B", "x": 150.0, "y": 100.0}, } mock_client.state.skipped_objects = [] mock_client.state.state = "RUNNING" with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 200 result = response.json() assert result["total"] == 2 # Check objects have position data objects_by_id = {obj["id"]: obj for obj in result["objects"]} assert objects_by_id[100]["name"] == "Part A" assert objects_by_id[100]["x"] == 50.0 assert objects_by_id[100]["y"] == 100.0 assert objects_by_id[200]["name"] == "Part B" assert objects_by_id[200]["x"] == 150.0 assert objects_by_id[200]["y"] == 100.0 # ======================================================================== # Skip objects endpoint # ======================================================================== @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/print/skip-objects", json=[100]) assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/print/skip-objects", json=[100]) assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_empty_list(self, async_client: AsyncClient, printer_factory): """Verify error when no object IDs provided.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.state.printable_objects = {100: "Part A"} mock_client.state.skipped_objects = [] with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/skip-objects", json=[]) assert response.status_code == 400 assert "no object" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_invalid_id(self, async_client: AsyncClient, printer_factory): """Verify error when object ID doesn't exist.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.state.printable_objects = {100: "Part A"} mock_client.state.skipped_objects = [] with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/skip-objects", json=[999]) assert response.status_code == 400 assert "invalid" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_success(self, async_client: AsyncClient, printer_factory): """Verify successful skip objects request.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.state.printable_objects = {100: "Part A", 200: "Part B"} mock_client.state.skipped_objects = [] mock_client.skip_objects.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/skip-objects", json=[100]) assert response.status_code == 200 result = response.json() assert result["success"] is True assert 100 in result["skipped_objects"] mock_client.skip_objects.assert_called_once_with([100]) @pytest.mark.asyncio @pytest.mark.integration async def test_skip_objects_multiple(self, async_client: AsyncClient, printer_factory): """Verify skipping multiple objects at once.""" printer = await printer_factory(name="Printing Printer") mock_client = MagicMock() mock_client.state.printable_objects = {100: "Part A", 200: "Part B", 300: "Part C"} mock_client.state.skipped_objects = [] mock_client.skip_objects.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/print/skip-objects", json=[100, 200]) assert response.status_code == 200 result = response.json() assert result["success"] is True assert 100 in result["skipped_objects"] assert 200 in result["skipped_objects"] mock_client.skip_objects.assert_called_once_with([100, 200]) class TestChamberLightAPI: """Integration tests for chamber light control endpoint.""" @pytest.mark.asyncio @pytest.mark.integration async def test_chamber_light_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/chamber-light?on=true") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_chamber_light_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/chamber-light?on=true") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_chamber_light_on_success(self, async_client: AsyncClient, printer_factory): """Verify successful chamber light on request.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.set_chamber_light.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/chamber-light?on=true") assert response.status_code == 200 result = response.json() assert result["success"] is True assert "on" in result["message"].lower() mock_client.set_chamber_light.assert_called_once_with(True) @pytest.mark.asyncio @pytest.mark.integration async def test_chamber_light_off_success(self, async_client: AsyncClient, printer_factory): """Verify successful chamber light off request.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.set_chamber_light.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/chamber-light?on=false") assert response.status_code == 200 result = response.json() assert result["success"] is True assert "off" in result["message"].lower() mock_client.set_chamber_light.assert_called_once_with(False) @pytest.mark.asyncio @pytest.mark.integration async def test_chamber_light_failure(self, async_client: AsyncClient, printer_factory): """Verify error handling when chamber light control fails.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.set_chamber_light.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/chamber-light?on=true") assert response.status_code == 500 assert "failed" in response.json()["detail"].lower() class TestAirductModeAPI: """Integration tests for the airduct mode endpoint (P2S/H2*).""" @pytest.mark.asyncio @pytest.mark.integration async def test_invalid_mode_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="P2S") response = await async_client.post(f"/api/v1/printers/{printer.id}/airduct-mode?mode=foo") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="P2S") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/airduct-mode?mode=cooling") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_cooling_success(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="P2S") mock_client = MagicMock() mock_client.set_airduct_mode.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/airduct-mode?mode=cooling") assert response.status_code == 200 assert response.json()["success"] is True mock_client.set_airduct_mode.assert_called_once_with("cooling") @pytest.mark.asyncio @pytest.mark.integration async def test_heating_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="P2S") mock_client = MagicMock() mock_client.set_airduct_mode.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/airduct-mode?mode=heating") assert response.status_code == 500 class TestClearHMSErrorsAPI: """Integration tests for clear HMS errors endpoint.""" @pytest.mark.asyncio @pytest.mark.integration async def test_clear_hms_errors_not_found(self, async_client: AsyncClient): """Verify 404 for non-existent printer.""" response = await async_client.post("/api/v1/printers/99999/hms/clear") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_clear_hms_errors_not_connected(self, async_client: AsyncClient, printer_factory): """Verify error when printer is not connected.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/clear") assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_clear_hms_errors_success(self, async_client: AsyncClient, printer_factory): """Verify successful clear HMS errors request.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.clear_hms_errors.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/clear") assert response.status_code == 200 result = response.json() assert result["success"] is True assert "cleared" in result["message"].lower() mock_client.clear_hms_errors.assert_called_once() @pytest.mark.asyncio @pytest.mark.integration async def test_clear_hms_errors_failure(self, async_client: AsyncClient, printer_factory): """Verify error handling when clear HMS errors fails.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.clear_hms_errors.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/clear") assert response.status_code == 500 assert "failed" in response.json()["detail"].lower() class TestExecuteHMSActionAPI: """Integration tests for the /hms/execute-action endpoint (#1743). Mirrors TestClearHMSErrorsAPI's shape — the two routes share the same permission gate, the same DB-lookup + client-existence flow, and the same dispatch-then-return-success pattern. The body-validation cases add coverage that the bare clear endpoint doesn't need. """ _VALID_BODY = { "print_error": "03008070", "action": "OK_BUTTON", "job_id": None, } @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_not_found(self, async_client: AsyncClient): """404 for a printer id that doesn't exist.""" response = await async_client.post("/api/v1/printers/99999/hms/execute-action", json=self._VALID_BODY) assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_not_connected(self, async_client: AsyncClient, printer_factory): """400 when the printer record exists but the MQTT client is offline.""" printer = await printer_factory(name="Disconnected Printer") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post( f"/api/v1/printers/{printer.id}/hms/execute-action", json=self._VALID_BODY ) assert response.status_code == 400 assert "not connected" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_success(self, async_client: AsyncClient, printer_factory): """200 happy path — dispatcher returns True AND the printer pushes at least one MQTT message into the ack-wait window. A fresh inbound message is the firmware's proof that the command landed (publish success is necessary but not sufficient; see #1830 §(3)).""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() # Pre-action state — paused with a fault, last message arrived at t=0. mock_client.state.state = "PAUSE" mock_client.state.print_error = 0x05008051 mock_client.state.hms_errors = [object()] mock_client._last_message_time = 100.0 def _act(*_a, **_kw): # Simulate the printer pushing a status update within the ack-wait # window. The pushall that follows every command is what produces # this — the actual state fields don't have to move (#1869: a # wrong-plate IGNORE_RESUME re-pauses with the same fault but the # printer DID push back). mock_client._last_message_time = 100.5 return True mock_client.execute_hms_action.side_effect = _act with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch("backend.app.api.routes.printers.HMS_ACTION_ACK_WAIT_SECONDS", 0.01), ): mock_pm.get_client.return_value = mock_client body = {"print_error": "07008029", "action": "FILAMENT_EXTRUDED", "job_id": "task-7"} response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/execute-action", json=body) assert response.status_code == 200 result = response.json() assert result["success"] is True assert "executed" in result["message"].lower() # Body args reach the client method in (print_error, action, job_id) order. mock_client.execute_hms_action.assert_called_once_with("07008029", "FILAMENT_EXTRUDED", "task-7") @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_dispatcher_failure(self, async_client: AsyncClient, printer_factory): """400 when the dispatcher returns False (unknown action, mid-flight disconnect).""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.execute_hms_action.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post( f"/api/v1/printers/{printer.id}/hms/execute-action", json=self._VALID_BODY ) assert response.status_code == 400 assert "failed" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_no_printer_ack_returns_502(self, async_client: AsyncClient, printer_factory): """502 when publish succeeded but no MQTT message arrives back within the ack-wait window. This is the silent-rejection failure mode #1830 identifies: the broker ACKs the publish at QoS 1 but the firmware drops the command (err mismatch, wrong shape, state mismatch). Surfacing this as 502 instead of 200 stops the UI from claiming success while the modal sticks.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.state.state = "PAUSE" mock_client.state.print_error = 0x05008051 mock_client.state.hms_errors = [object()] mock_client._last_message_time = 100.0 mock_client.execute_hms_action.return_value = True # publish "succeeded" # Crucially: _last_message_time does NOT advance → no inbound push. with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch("backend.app.api.routes.printers.HMS_ACTION_ACK_WAIT_SECONDS", 0.01), ): mock_pm.get_client.return_value = mock_client response = await async_client.post( f"/api/v1/printers/{printer.id}/hms/execute-action", json=self._VALID_BODY ) assert response.status_code == 502 assert "acknowledge" in response.json()["detail"].lower() @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_ignore_resume_repauses_within_window_still_acks( self, async_client: AsyncClient, printer_factory ): """200 when the printer ack'd the command but immediately re-paused with the same fault — e.g. wrong-plate IGNORE_RESUME (#1869). The previous (gcode_state, hms_errors-len) diff produced a false 502 because both fields round-tripped to their pre-publish values inside the ack window. Probing `_last_message_time` survives the round-trip because the printer's status push lands regardless of the eventual state.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.state.state = "PAUSE" mock_client.state.print_error = 0x05008051 mock_client.state.hms_errors = [object()] mock_client._last_message_time = 100.0 def _act(*_a, **_kw): # Printer ack'd, briefly resumed, re-detected the wrong plate, and # re-paused with the same fault. Net diff on state fields is zero, # but a fresh status push DID arrive. mock_client._last_message_time = 100.4 mock_client.state.state = "PAUSE" # round-tripped mock_client.state.hms_errors = [object()] # same length return True mock_client.execute_hms_action.side_effect = _act with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch("backend.app.api.routes.printers.HMS_ACTION_ACK_WAIT_SECONDS", 0.01), ): mock_pm.get_client.return_value = mock_client body = {"print_error": "05008051", "action": "IGNORE_RESUME", "job_id": None} response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/execute-action", json=body) assert response.status_code == 200 assert response.json()["success"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_accepts_16_char_full_code(self, async_client: AsyncClient, printer_factory): """200 for a 16-char full_code (hms[]-array-sourced fault). The schema's relaxed pattern allows both 8-char (print_error) and 16-char (hms[]) shapes.""" printer = await printer_factory(name="Test Printer") mock_client = MagicMock() mock_client.state.state = "RUNNING" mock_client.state.print_error = 0 mock_client.state.hms_errors = [object()] mock_client._last_message_time = 100.0 def _act(*_a, **_kw): mock_client._last_message_time = 100.4 return True mock_client.execute_hms_action.side_effect = _act with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch("backend.app.api.routes.printers.HMS_ACTION_ACK_WAIT_SECONDS", 0.01), ): mock_pm.get_client.return_value = mock_client body = {"print_error": "0C00030000020010", "action": "IGNORE_RESUME"} response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/execute-action", json=body) assert response.status_code == 200 mock_client.execute_hms_action.assert_called_once_with("0C00030000020010", "IGNORE_RESUME", None) @pytest.mark.asyncio @pytest.mark.integration async def test_execute_hms_action_rejects_malformed_print_error(self, async_client: AsyncClient, printer_factory): """422 when print_error fails the relaxed pattern (8 OR 16 hex chars). Lengths in between (9-15) and outside (7, 17+) are invalid; stray input can't reach the dispatcher's match statement.""" printer = await printer_factory(name="Test Printer") bad_bodies = [ {"print_error": "0300_8070", "action": "OK_BUTTON"}, # underscore {"print_error": "0300807", "action": "OK_BUTTON"}, # 7 chars (too short) {"print_error": "030080700", "action": "OK_BUTTON"}, # 9 chars (between) {"print_error": "030080700300807", "action": "OK_BUTTON"}, # 15 chars (between) {"print_error": "0300807003008070A", "action": "OK_BUTTON"}, # 17 chars (too long) {"print_error": "0300GGGG", "action": "OK_BUTTON"}, # non-hex ] for body in bad_bodies: response = await async_client.post(f"/api/v1/printers/{printer.id}/hms/execute-action", json=body) assert response.status_code == 422, body def _build_h2d_state(*, ams_id: int = 0, tray_id: int = 2, cali_idx: int = 5): """Build a MagicMock PrinterState for an H2D printer with a single BL spool tray. Used by both TestApplyPaAfterRefresh (Phase 13 P13-T-BE-1) and the K-profile persistence tests below. The tray data passes is_bambu_tag (32-char non-zero tray_uuid + non-empty tray_info_idx). """ nozzle = MagicMock(nozzle_diameter="0.4") state = MagicMock() state.nozzles = [nozzle] state.ams_extruder_map = {"0": 0} state.raw_data = { "ams": [ { "id": ams_id, "tray": [ { "id": tray_id, "tray_type": "PLA", "tag_uid": "AABBCC1122334400", "tray_uuid": "11223344556677880011223344556677", "tray_info_idx": "GFL05", "cali_idx": cali_idx, } ], } ] } return state def _patch_async_session_to(db_session): """Patch backend.app.core.database.async_session so calls inside the function under test reuse the test fixture's db_session. `_apply_pa_after_refresh` lazy-imports `from backend.app.core.database import async_session` at runtime (line 2849). When we patch the source module before the call, the lazy import picks up the patched object. Returns the patch context manager; use as `with _patch_async_session_to(db_session):`. Pattern verified against test_print_lifecycle.py:38-42. """ cm = AsyncMock() cm.__aenter__ = AsyncMock(return_value=db_session) cm.__aexit__ = AsyncMock(return_value=None) return patch("backend.app.core.database.async_session", return_value=cm) class TestApplyPaAfterRefresh: """Phase 13 P13-T-BE-1: _apply_pa_after_refresh K-profile cascade. Verifies the 3-stage cascade (local SpoolKProfile → Spoolman SpoolmanKProfile → live tray.cali_idx fallback) and the Bug A regression (kp.extruder, not kp.extruder_id, after the Phase 13 P13-2a fix). `_apply_pa_after_refresh` is a free function spawned via asyncio.create_task from the /ams-refresh endpoint. Tests call it directly because awaiting the spawned task in an HTTP test would require sleeping past the 5-second guard that delays MQTT until RFID re-read finishes. """ @pytest.mark.asyncio @pytest.mark.integration async def test_local_kp_match_sends_stored_cali_idx(self, db_session, printer_factory): """Local SpoolAssignment + matching SpoolKProfile → stored cali_idx wins.""" from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() spool = Spool(material="PLA", color_name="Red", rgba="FF0000FF") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, ) ) db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() kwargs = mock_client.extrusion_cali_sel.call_args.kwargs assert kwargs["cali_idx"] == 42 # stored profile, not 5 (live) @pytest.mark.asyncio @pytest.mark.integration async def test_local_no_kp_uses_live_cali_idx(self, db_session, printer_factory): """Local SpoolAssignment but no matching SpoolKProfile → live cali_idx (Stage 3).""" from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory() spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() kwargs = mock_client.extrusion_cali_sel.call_args.kwargs assert kwargs["cali_idx"] == 5 # live tray.cali_idx fallback @pytest.mark.asyncio @pytest.mark.integration async def test_spoolman_kp_when_no_local(self, db_session, printer_factory): """No local assignment + Spoolman SlotAssignment + SpoolmanKProfile → Spoolman cali_idx.""" from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.settings import Settings from backend.app.models.spoolman_k_profile import SpoolmanKProfile from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory() db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=99, ) ) db_session.add( SpoolmanKProfile( spoolman_spool_id=99, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.030, cali_idx=77, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() kwargs = mock_client.extrusion_cali_sel.call_args.kwargs assert kwargs["cali_idx"] == 77 # Spoolman stored profile @pytest.mark.asyncio @pytest.mark.integration async def test_spoolman_no_kp_uses_live(self, db_session, printer_factory): """Spoolman SlotAssignment but no SpoolmanKProfile → live cali_idx (Stage 3).""" from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.settings import Settings from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory() db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=99, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 5 @pytest.mark.asyncio @pytest.mark.integration async def test_no_assignment_uses_live(self, db_session, printer_factory): """No assignment of any kind + live cali_idx >= 0 → live fallback.""" from backend.app.api.routes.printers import _apply_pa_after_refresh printer = await printer_factory() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 5 @pytest.mark.asyncio @pytest.mark.integration async def test_negative_live_cali_idx_skipped(self, db_session, printer_factory): """No assignment + live cali_idx=-1 → no MQTT call (invalid live value).""" from backend.app.api.routes.printers import _apply_pa_after_refresh printer = await printer_factory() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=-1) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_no_assignment_no_live_cali_idx_no_call(self, db_session, printer_factory): """No assignment of any kind AND no live cali_idx in tray → no MQTT call. Distinct from test_negative_live_cali_idx_skipped: that test has cali_idx=-1 in raw_data; this one omits the field entirely (returns None from .get("cali_idx")). Both must result in no MQTT call. """ from backend.app.api.routes.printers import _apply_pa_after_refresh printer = await printer_factory() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) # State with NO cali_idx field on tray at all nozzle = MagicMock(nozzle_diameter="0.4") state = MagicMock() state.nozzles = [nozzle] state.ams_extruder_map = {"0": 0} state.raw_data = { "ams": [ { "id": 0, "tray": [ { "id": 2, "tray_type": "PLA", "tag_uid": "AABBCC1122334400", "tray_uuid": "11223344556677880011223344556677", "tray_info_idx": "GFL05", # cali_idx field intentionally omitted } ], } ] } with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_extruder_mismatch_uses_kp_as_fallback(self, db_session, printer_factory): """K-profile for extruder=1 but slot is extruder=0 → no exact match, but the kp is used as extruder-agnostic fallback rather than dropped. Hard-skipping on extruder mismatch was the previous behavior; in practice it caused stored K-profiles to be silently ignored whenever the AMS-extruder mapping had shifted (or when only one of the two extruders was ever calibrated for a given spool). The cascade now prefers an exact extruder match but falls back to any matching kp for the same printer + nozzle. """ from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, ) ) # K-profile is for extruder=1, but slot's ams_extruder_map["0"]=0 db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=1, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) # No exact extruder match, but the stored kp wins as the # extruder-agnostic fallback over live cali_idx=5. mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 42 @pytest.mark.asyncio @pytest.mark.integration async def test_extruder_exact_match_preferred_over_fallback( self, db_session, printer_factory, ): """When two kp rows exist, one with matching extruder and one without, the exact-extruder kp wins (extruder-agnostic fallback only fires when no exact match exists). """ from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, ) ) # Two kp rows: extruder=1 (mismatch w/ slot extruder=0) and extruder=0 (exact) db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=1, nozzle_diameter="0.4", k_value=0.030, cali_idx=99, ) ) db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) # Exact-extruder=0 kp wins (cali_idx=42), not the extruder=1 fallback (99) mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 42 @pytest.mark.asyncio @pytest.mark.integration async def test_regression_bug_a_kp_extruder_attr(self, db_session, printer_factory): """Regression test for Phase 13 P13-2a Bug A. Pre-fix Z.2910 used `kp.extruder_id` (AttributeError on SpoolKProfile, silently swallowed by outer try/except). On dual-nozzle printers with slot_extruder != None this caused the K-profile match loop to crash. After P13-2a the field name is correct: `kp.extruder`. """ from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, ) ) # extruder=0 matches slot_extruder=0 (from ams_extruder_map={"0":0}) db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) # If Bug A regressed (kp.extruder_id), the loop would AttributeError → silent fail # → no extrusion_cali_sel call. Post-fix the loop matches and sends cali_idx=42. mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 42 @pytest.mark.asyncio @pytest.mark.integration async def test_tag_fallback_finds_spool_when_assignment_missing( self, db_session, printer_factory, ): """Stage 1b regression for the maintainer's #2 reproducer on H2D: reset slot, trigger re-read → slot ends up on the default K-profile instead of the spool's stored profile. Setup mirrors the bug: - Spool has tray_uuid set (the RFID tag was registered earlier). - SpoolKProfile exists for that spool with cali_idx=42. - NO SpoolAssignment row — the reset deleted it before the re-read triggered _apply_pa_after_refresh, and tag-auto-detect has not re-created it yet within the 5 s sleep window. - Live tray.cali_idx=5 (firmware-default after the RFID re-read). Without Stage 1b the cascade falls through to Stage 3 and re-asserts the firmware-default cali_idx=5. With Stage 1b it locates the spool by the live tray's tray_uuid and applies the stored cali_idx=42. """ from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() # Spool with tray_uuid matching the one _build_h2d_state puts on the tray spool = Spool( material="PLA", color_name="Red", rgba="FF0000FF", tray_uuid="11223344556677880011223344556677", tag_uid="AABBCC1122334400", ) db_session.add(spool) await db_session.flush() # K-profile is bound to the spool, not to a slot db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) # NOTE: deliberately no SpoolAssignment — that's the bug condition. await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=5) with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) # Stage 1b should match the spool by tray_uuid → stored cali_idx=42 wins # over live cali_idx=5. Pre-fix this would have been 5 (firmware default). mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 42 @pytest.mark.asyncio @pytest.mark.integration async def test_tag_fallback_matches_by_tag_uid_when_uuid_zero( self, db_session, printer_factory, ): """Stage 1b: when tray_uuid is the zero sentinel but tag_uid is real, match by tag_uid. Older firmwares occasionally report a zero tray_uuid right after RFID re-read while the tag_uid is already populated.""" from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() # Spool indexed by tag_uid, not tray_uuid spool = Spool(material="PLA", tag_uid="AABBCC1122334400") db_session.add(spool) await db_session.flush() db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=99, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) # Build a state where the tray reports a real tag_uid but a zero tray_uuid # while still passing is_bambu_tag (tag_uid + tray_info_idx is sufficient). state = _build_h2d_state(cali_idx=5) state.raw_data["ams"][0]["tray"][0]["tray_uuid"] = "00000000000000000000000000000000" with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 99 @pytest.mark.asyncio @pytest.mark.integration async def test_tag_fallback_skipped_when_zero_sentinels( self, db_session, printer_factory, ): """Stage 1b: when both tray_uuid and tag_uid are zero sentinels, the fallback must not match any spool (would otherwise pick up an unrelated spool created with empty/zero tag fields). Falls through to Stage 3 live cali_idx as before. """ from backend.app.api.routes.printers import _apply_pa_after_refresh from backend.app.models.spool import Spool from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory() # Decoy spool with no tag info — must NOT match spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolKProfile( spool_id=spool.id, printer_id=printer.id, extruder=0, nozzle_diameter="0.4", k_value=0.025, cali_idx=42, ) ) await db_session.commit() mock_client = MagicMock() mock_client.extrusion_cali_sel = MagicMock(return_value=True) state = _build_h2d_state(cali_idx=7) # Force both tag fields to the zero sentinels but keep tray_info_idx # so is_bambu_tag still passes (preset present) state.raw_data["ams"][0]["tray"][0]["tag_uid"] = "0000000000000000" state.raw_data["ams"][0]["tray"][0]["tray_uuid"] = "00000000000000000000000000000000" with ( patch("backend.app.api.routes.printers.asyncio.sleep", AsyncMock()), patch("backend.app.api.routes.printers.printer_manager") as mock_pm, _patch_async_session_to(db_session), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = state # is_bambu_tag actually rejects both-zero + only-preset, so the # function returns early. We just want to confirm we didn't blow # up scanning for a tag-fallback spool. await _apply_pa_after_refresh(printer.id, ams_id=0, slot_id=2) # is_bambu_tag short-circuits early when both UID and UUID are zero, # so no MQTT call should fire and the decoy spool's cali_idx=42 must # NOT leak through. if mock_client.extrusion_cali_sel.called: assert mock_client.extrusion_cali_sel.call_args.kwargs["cali_idx"] != 42 class TestConfigureAmsSlotPersistsKProfile: """Phase 13 P13-T-BE-2: configure_ams_slot persists K-profile to DB. Pre-Phase-13 the endpoint sent extrusion_cali_sel via MQTT but never recorded the choice in spool_k_profile / spoolman_k_profile, so the next RFID re-read had no stored profile to apply. """ @pytest.mark.asyncio @pytest.mark.integration async def test_writes_spoolman_kprofile_when_spoolman_assigned( self, async_client: AsyncClient, db_session, printer_factory, ): """SpoolmanSlotAssignment present → SpoolmanKProfile row created with cali_idx + k_value + name.""" from backend.app.models.settings import Settings from backend.app.models.spoolman_k_profile import SpoolmanKProfile from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory(model="H2D") db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=3, spoolman_spool_id=216, ) ) await db_session.commit() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.extrusion_cali_set.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "Bambu PLA Metal", "tray_color": "FF8800FF", "nozzle_temp_min": 220, "nozzle_temp_max": 240, "cali_idx": 5, "nozzle_diameter": "0.4", "k_value": 0.022, }, ) assert response.status_code == 200 kp_result = await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == 216)) kp = kp_result.scalar_one_or_none() assert kp is not None assert kp.cali_idx == 5 assert kp.k_value == pytest.approx(0.022) assert kp.extruder == 0 assert kp.nozzle_diameter == "0.4" assert kp.name == "Bambu PLA Metal" @pytest.mark.asyncio @pytest.mark.integration async def test_writes_spool_kprofile_when_local_assigned( self, async_client: AsyncClient, db_session, printer_factory, ): """Local SpoolAssignment present → SpoolKProfile row created.""" from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory(model="H2D") spool = Spool(material="PLA", color_name="Red", rgba="FF0000FF") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=3, ) ) await db_session.commit() spool_id = spool.id mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.extrusion_cali_set.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "PFUSdev01", "tray_type": "PLA", "tray_sub_brands": "Devil Design PLA", "tray_color": "FF0000FF", "nozzle_temp_min": 220, "nozzle_temp_max": 240, "cali_idx": 7, "nozzle_diameter": "0.4", "k_value": 0.028, }, ) assert response.status_code == 200 kp_result = await db_session.execute(select(SpoolKProfile).where(SpoolKProfile.spool_id == spool_id)) kp = kp_result.scalar_one_or_none() assert kp is not None assert kp.cali_idx == 7 assert kp.k_value == pytest.approx(0.028) assert kp.extruder == 0 assert kp.nozzle_diameter == "0.4" assert kp.name == "Devil Design PLA" @pytest.mark.asyncio @pytest.mark.integration async def test_no_assignment_no_persist( self, async_client: AsyncClient, db_session, printer_factory, ): """No SpoolAssignment AND no SpoolmanSlotAssignment → no DB write, MQTT still sent.""" from backend.app.models.spool_k_profile import SpoolKProfile from backend.app.models.spoolman_k_profile import SpoolmanKProfile printer = await printer_factory(model="H2D") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 5, "k_value": 0.020, }, ) assert response.status_code == 200 # MQTT sent (was successful), but no DB writes mock_client.extrusion_cali_sel.assert_called_once() local_count = (await db_session.execute(select(SpoolKProfile))).scalars().all() sm_count = (await db_session.execute(select(SpoolmanKProfile))).scalars().all() assert len(local_count) == 0 assert len(sm_count) == 0 @pytest.mark.asyncio @pytest.mark.integration async def test_negative_cali_idx_no_persist( self, async_client: AsyncClient, db_session, printer_factory, ): """cali_idx=-1 (no profile selected) → no DB write even when assignment exists.""" from backend.app.models.settings import Settings from backend.app.models.spoolman_k_profile import SpoolmanKProfile from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory(model="H2D") db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=3, spoolman_spool_id=216, ) ) await db_session.commit() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.extrusion_cali_set.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": -1, "k_value": 0.0, }, ) assert response.status_code == 200 sm_kps = ( (await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == 216))) .scalars() .all() ) assert len(sm_kps) == 0 # cali_idx=-1 means "no profile" — don't write @pytest.mark.asyncio @pytest.mark.integration async def test_zero_cali_idx_persists( self, async_client: AsyncClient, db_session, printer_factory, ): """cali_idx=0 is the first valid profile slot (NOT a sentinel for missing).""" from backend.app.models.settings import Settings from backend.app.models.spoolman_k_profile import SpoolmanKProfile from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory(model="H2D") db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=3, spoolman_spool_id=216, ) ) await db_session.commit() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 0, "k_value": 0.020, }, ) assert response.status_code == 200 kp = ( await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == 216)) ).scalar_one_or_none() assert kp is not None assert kp.cali_idx == 0 # explicitly testing 0 is valid @pytest.mark.asyncio @pytest.mark.integration async def test_upsert_idempotent( self, async_client: AsyncClient, db_session, printer_factory, ): """Repeated POSTs update the same row (UNIQUE on spool_id+printer+extruder+nozzle_diameter).""" from backend.app.models.settings import Settings from backend.app.models.spoolman_k_profile import SpoolmanKProfile from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory(model="H2D") db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=3, spoolman_spool_id=216, ) ) await db_session.commit() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state # First call with cali_idx=5 await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 5, "k_value": 0.020, }, ) # Second call with cali_idx=10 (same slot/spool/extruder/nozzle) await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Matte", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 10, "k_value": 0.025, }, ) # Should be exactly ONE row (updated), not two kps = ( (await db_session.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == 216))) .scalars() .all() ) assert len(kps) == 1 assert kps[0].cali_idx == 10 # updated to most recent assert kps[0].name == "PLA Matte" @pytest.mark.asyncio @pytest.mark.integration async def test_external_slot_extruder_inversion( self, async_client: AsyncClient, db_session, printer_factory, ): """ams_id=255 + tray_id=0 → kp.extruder=1 (ext-L); tray_id=1 → extruder=0 (ext-R).""" from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory(model="H2D") spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() # Note: SpoolmanSlotAssignment can't store ams_id=255 with tray_id=1 # under the ck_tray_id_range constraint (0-3 valid). External-slot # K-profile persistence is therefore tested via local SpoolAssignment. db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=255, tray_id=0, ) ) await db_session.commit() spool_id = spool.id mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} # truthy so external-inversion path runs mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/255/0/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 5, "k_value": 0.020, }, ) assert response.status_code == 200 kp = ( await db_session.execute(select(SpoolKProfile).where(SpoolKProfile.spool_id == spool_id)) ).scalar_one_or_none() assert kp is not None # tray_id=0 → extruder = 1 - 0 = 1 assert kp.extruder == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_dual_nozzle_extruder_persists( self, async_client: AsyncClient, db_session, printer_factory, ): """ams_extruder_map with extruder=1 → kp.extruder=1 persisted correctly.""" from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spool_k_profile import SpoolKProfile printer = await printer_factory(model="H2D") spool = Spool(material="PLA") db_session.add(spool) await db_session.flush() db_session.add( SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=2, tray_id=3, ) ) await db_session.commit() spool_id = spool.id mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"2": 1} # AMS 2 is on extruder 1 mock_state.raw_data = {"ams": {"ams": []}} with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/2/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 5, "k_value": 0.020, }, ) assert response.status_code == 200 kp = ( await db_session.execute(select(SpoolKProfile).where(SpoolKProfile.spool_id == spool_id)) ).scalar_one_or_none() assert kp is not None assert kp.extruder == 1 @pytest.mark.asyncio @pytest.mark.integration async def test_db_error_does_not_fail_endpoint( self, async_client: AsyncClient, db_session, printer_factory, ): """DB errors during K-profile persistence are best-effort — endpoint still returns 200. Verifies the try/except wrap added in P13-3b: if DB upsert fails (e.g. because the schema is out of sync, a constraint violation, or any other transient error), the MQTT command was already sent successfully so we shouldn't return 500 to the user. The error is logged and the endpoint returns success. """ from backend.app.models.settings import Settings from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment # A spoolman_slot_assignments row only exists in Spoolman mode, and # since #2812 the mode is asked for explicitly rather than inferred # from which table happens to hold rows. db_session.add(Settings(key="spoolman_enabled", value="true")) printer = await printer_factory(model="H2D") db_session.add( SpoolmanSlotAssignment( printer_id=printer.id, ams_id=0, tray_id=3, spoolman_spool_id=216, ) ) await db_session.commit() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True mock_client.request_status_update.return_value = True mock_state = MagicMock() mock_state.ams_extruder_map = {"0": 0} mock_state.raw_data = {"ams": {"ams": []}} # Force the K-profile persistence path to fail by patching the # SpoolmanKProfile model class with a sentinel that raises when # instantiated. The MQTT call has already happened by then, so the # endpoint must catch and log without returning 500. with ( patch("backend.app.api.routes.printers.printer_manager") as mock_pm, patch( "backend.app.models.spoolman_k_profile.SpoolmanKProfile", side_effect=RuntimeError("Simulated DB error"), ), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = mock_state response = await async_client.post( f"/api/v1/printers/{printer.id}/slots/0/3/configure", params={ "tray_info_idx": "GFL05", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_color": "FFFFFFFF", "nozzle_temp_min": 190, "nozzle_temp_max": 230, "cali_idx": 5, "k_value": 0.020, }, ) # Endpoint returns success — MQTT was sent, K-profile failed silently assert response.status_code == 200 # MQTT was indeed called mock_client.extrusion_cali_sel.assert_called_once() class TestPrinterAccessCodeVisibility: """Regression coverage: GET /printers and GET /printers/{id} must NOT return ``access_code`` to callers without PRINTERS_UPDATE authority. Holding ``access_code`` lets the caller talk to the printer's MQTT directly with serial+code, bypassing every PRINTERS_CONTROL / PRINTERS_FILES / PRINTERS_AMS_RFID check Bambuddy enforces. Trust matrix encoded here: - Auth disabled → access_code visible (single-trust mode) - JWT Admin → access_code visible - JWT Operator (has *_UPDATE) → access_code visible (VP-card UX) - JWT Viewer → access_code STRIPPED - API key with can_read_status → access_code STRIPPED """ @pytest.fixture async def auth_setup(self, async_client: AsyncClient): await async_client.post( "/api/v1/auth/setup", json={ "auth_enabled": True, "admin_username": "pcadmin", "admin_password": "AdminPass1!", }, ) async def _login(username, password): resp = await async_client.post( "/api/v1/auth/login", json={"username": username, "password": password}, ) return resp.json()["access_token"] admin_token = await _login("pcadmin", "AdminPass1!") groups = ( await async_client.get( "/api/v1/groups/", headers={"Authorization": f"Bearer {admin_token}"}, ) ).json() operators_group = next(g for g in groups if g["name"] == "Operators") viewers_group = next(g for g in groups if g["name"] == "Viewers") for username, password, group in ( ("pcoperator", "Operpass1!", operators_group["id"]), ("pcviewer", "Viewpass1!", viewers_group["id"]), ): await async_client.post( "/api/v1/users/", headers={"Authorization": f"Bearer {admin_token}"}, json={"username": username, "password": password, "group_ids": [group]}, ) operator_token = await _login("pcoperator", "Operpass1!") viewer_token = await _login("pcviewer", "Viewpass1!") return { "admin_token": admin_token, "operator_token": operator_token, "viewer_token": viewer_token, } async def _seed_printer_with_known_code(self, async_client: AsyncClient, admin_token: str) -> int: resp = await async_client.post( "/api/v1/printers/", headers={"Authorization": f"Bearer {admin_token}"}, json={ "name": "AC-Visibility", "serial_number": "00M09AVISIBILITY", "ip_address": "192.168.42.42", "access_code": "SECRET-CODE", "is_active": True, "model": "X1C", }, ) assert resp.status_code == 200, resp.text return resp.json()["id"] @pytest.mark.asyncio @pytest.mark.integration async def test_auth_disabled_includes_access_code(self, async_client: AsyncClient, printer_factory): """Single-trust mode: behaviour preserved, code is visible.""" printer = await printer_factory(name="AuthOff", access_code="LOCAL-CODE") list_resp = await async_client.get("/api/v1/printers/") detail_resp = await async_client.get(f"/api/v1/printers/{printer.id}") assert list_resp.status_code == 200 assert detail_resp.status_code == 200 match = next(p for p in list_resp.json() if p["id"] == printer.id) assert match["access_code"] == "LOCAL-CODE" assert detail_resp.json()["access_code"] == "LOCAL-CODE" @pytest.mark.asyncio @pytest.mark.integration async def test_admin_jwt_includes_access_code(self, async_client: AsyncClient, auth_setup): printer_id = await self._seed_printer_with_known_code(async_client, auth_setup["admin_token"]) headers = {"Authorization": f"Bearer {auth_setup['admin_token']}"} list_resp = await async_client.get("/api/v1/printers/", headers=headers) detail_resp = await async_client.get(f"/api/v1/printers/{printer_id}", headers=headers) assert list_resp.status_code == 200 assert detail_resp.status_code == 200 match = next(p for p in list_resp.json() if p["id"] == printer_id) assert match["access_code"] == "SECRET-CODE" assert detail_resp.json()["access_code"] == "SECRET-CODE" @pytest.mark.asyncio @pytest.mark.integration async def test_operator_jwt_includes_access_code(self, async_client: AsyncClient, auth_setup): """Operators hold PRINTERS_UPDATE (default role) — the VP-card UX surfaces the target printer's access_code so they can configure their slicer. The visibility predicate must keep working for them. """ printer_id = await self._seed_printer_with_known_code(async_client, auth_setup["admin_token"]) headers = {"Authorization": f"Bearer {auth_setup['operator_token']}"} list_resp = await async_client.get("/api/v1/printers/", headers=headers) detail_resp = await async_client.get(f"/api/v1/printers/{printer_id}", headers=headers) assert list_resp.status_code == 200 assert detail_resp.status_code == 200 match = next(p for p in list_resp.json() if p["id"] == printer_id) assert match["access_code"] == "SECRET-CODE" assert detail_resp.json()["access_code"] == "SECRET-CODE" @pytest.mark.asyncio @pytest.mark.integration async def test_viewer_jwt_excludes_access_code(self, async_client: AsyncClient, auth_setup): """The fix: Viewers hold PRINTERS_READ but not PRINTERS_UPDATE, and must NOT be able to read the printer's secret. """ printer_id = await self._seed_printer_with_known_code(async_client, auth_setup["admin_token"]) headers = {"Authorization": f"Bearer {auth_setup['viewer_token']}"} list_resp = await async_client.get("/api/v1/printers/", headers=headers) detail_resp = await async_client.get(f"/api/v1/printers/{printer_id}", headers=headers) assert list_resp.status_code == 200 assert detail_resp.status_code == 200 match = next(p for p in list_resp.json() if p["id"] == printer_id) # Field absent OR null — both are acceptable (no usable secret reaches the wire). assert "access_code" not in match or match["access_code"] is None body = detail_resp.json() assert "access_code" not in body or body["access_code"] is None # And the rest of the payload still arrives so the UI keeps working. assert match["name"] == "AC-Visibility" assert body["name"] == "AC-Visibility" @pytest.mark.asyncio @pytest.mark.integration async def test_api_key_excludes_access_code(self, async_client: AsyncClient, auth_setup, db_session): """API keys with can_read_status hold PRINTERS_READ but the predicate gates on PRINTERS_UPDATE (admin-only / API-key-unmapped). The key must NOT be able to exfiltrate access_code. """ from backend.app.core.auth import generate_api_key from backend.app.models.api_key import APIKey printer_id = await self._seed_printer_with_known_code(async_client, auth_setup["admin_token"]) full_key, key_hash, key_prefix = generate_api_key() api_key = APIKey( name="visibility-key", key_hash=key_hash, key_prefix=key_prefix, can_read_status=True, enabled=True, ) db_session.add(api_key) await db_session.commit() list_resp = await async_client.get("/api/v1/printers/", headers={"X-API-Key": full_key}) detail_resp = await async_client.get(f"/api/v1/printers/{printer_id}", headers={"X-API-Key": full_key}) assert list_resp.status_code == 200 assert detail_resp.status_code == 200 match = next(p for p in list_resp.json() if p["id"] == printer_id) assert "access_code" not in match or match["access_code"] is None body = detail_resp.json() assert "access_code" not in body or body["access_code"] is None class TestSetNozzleTemperatureAPI: """Integration tests for POST /printers/{id}/temperature/nozzle (#1661).""" @pytest.mark.asyncio @pytest.mark.integration async def test_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/temperature/nozzle?target=220") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=220") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_default_nozzle_index(self, async_client: AsyncClient, printer_factory): """Omitting nozzle defaults to 0 (right/default).""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_nozzle_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=220") assert response.status_code == 200 mock_client.set_nozzle_temperature.assert_called_once_with(220, 0) @pytest.mark.asyncio @pytest.mark.integration async def test_dual_nozzle_left(self, async_client: AsyncClient, printer_factory): """nozzle=1 reaches the client method as the second positional arg.""" printer = await printer_factory(name="P", model="H2D") mock_client = MagicMock() mock_client.set_nozzle_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=260&nozzle=1") assert response.status_code == 200 mock_client.set_nozzle_temperature.assert_called_once_with(260, 1) @pytest.mark.asyncio @pytest.mark.integration async def test_target_zero_allowed(self, async_client: AsyncClient, printer_factory): """target=0 turns the heater off; must NOT be rejected by Query bounds.""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_nozzle_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=0") assert response.status_code == 200 mock_client.set_nozzle_temperature.assert_called_once_with(0, 0) @pytest.mark.asyncio @pytest.mark.integration async def test_target_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=400") assert response.status_code == 422 # FastAPI bounds violation @pytest.mark.asyncio @pytest.mark.integration async def test_nozzle_index_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=220&nozzle=2") assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_client_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_nozzle_temperature.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/nozzle?target=220") assert response.status_code == 500 class TestSetBedTemperatureAPI: """Integration tests for POST /printers/{id}/temperature/bed (#1661).""" @pytest.mark.asyncio @pytest.mark.integration async def test_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/temperature/bed?target=60") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/bed?target=60") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_success(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_bed_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/bed?target=60") assert response.status_code == 200 mock_client.set_bed_temperature.assert_called_once_with(60) @pytest.mark.asyncio @pytest.mark.integration async def test_target_zero_allowed(self, async_client: AsyncClient, printer_factory): """target=0 turns the bed heater off.""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_bed_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/bed?target=0") assert response.status_code == 200 mock_client.set_bed_temperature.assert_called_once_with(0) @pytest.mark.asyncio @pytest.mark.integration async def test_target_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/bed?target=200") assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_client_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_bed_temperature.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/bed?target=60") assert response.status_code == 500 class TestSetChamberTemperatureAPI: """Integration tests for POST /printers/{id}/temperature/chamber. Gated on supports_chamber_heater(model). Sensor-only models that report chamber temp but have no heater (X1C, X1E, P2S) get a 400 at the route level rather than a silent no-op at the firmware level. """ @pytest.mark.asyncio @pytest.mark.integration async def test_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/temperature/chamber?target=45") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("sensor_only_model", ["X1C", "X1E", "P2S"]) async def test_sensor_only_model_rejected(self, async_client: AsyncClient, printer_factory, sensor_only_model): """Models with sensor but no heater must 400 before any client call.""" printer = await printer_factory(name="P", model=sensor_only_model) mock_client = MagicMock() mock_client.set_chamber_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=45") assert response.status_code == 400 # Client must NOT be called for sensor-only models. mock_client.set_chamber_temperature.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=45") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("heater_model", ["H2C", "H2D", "H2DPRO", "H2S", "X2D"]) async def test_success_per_heater_model(self, async_client: AsyncClient, printer_factory, heater_model): """All five heater-equipped models accept the command.""" printer = await printer_factory(name="P", model=heater_model) mock_client = MagicMock() mock_client.set_chamber_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=45") assert response.status_code == 200 mock_client.set_chamber_temperature.assert_called_once_with(45) @pytest.mark.asyncio @pytest.mark.integration async def test_target_zero_allowed(self, async_client: AsyncClient, printer_factory): """target=0 turns the chamber heater off.""" printer = await printer_factory(name="P", model="H2D") mock_client = MagicMock() mock_client.set_chamber_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=0") assert response.status_code == 200 mock_client.set_chamber_temperature.assert_called_once_with(0) @pytest.mark.asyncio @pytest.mark.integration async def test_target_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=100") assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_ceiling_is_65_not_60(self, async_client: AsyncClient, printer_factory): """The H2 series heats the chamber to 65 °C — 65 must be accepted and 66 rejected. The route used to cap at 60, which put the top of the H2D range out of reach.""" printer = await printer_factory(name="P", model="H2D") mock_client = MagicMock() mock_client.set_chamber_temperature.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client accepted = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=65") assert accepted.status_code == 200 mock_client.set_chamber_temperature.assert_called_once_with(65) rejected = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=66") assert rejected.status_code == 422 @pytest.mark.asyncio @pytest.mark.integration async def test_client_failure_returns_500(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") mock_client = MagicMock() mock_client.set_chamber_temperature.return_value = False with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/temperature/chamber?target=45") assert response.status_code == 500 class TestSetFanSpeedAPI: """Integration tests for POST /printers/{id}/fan-speed (#1661). The fan-id mapping (part->1, aux->2, chamber->3, aux2->10) is the critical correctness gate — wrong mapping would target the wrong physical fan. "aux2" (M106 P10) is the optional left auxiliary part cooling fan on P2S/X2D. """ @pytest.mark.asyncio @pytest.mark.integration async def test_invalid_fan_name_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=foo&speed=50") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=part&speed=50") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize( "fan_name,expected_fan_id", [("part", 1), ("aux", 2), ("chamber", 3), ("aux2", 10)], ) async def test_fan_id_mapping(self, async_client: AsyncClient, printer_factory, fan_name, expected_fan_id): """Verify each fan name maps to the correct hardware fan-id.""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_fan_speed.return_value = True # aux2 is presence-gated on the printer reporting airduct part 10, so # give the mock a reported speed. Set explicitly rather than leaning on # MagicMock's auto-attribute, which would satisfy the gate by accident. mock_client.state.left_aux_fan_speed = 0 with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=100") assert response.status_code == 200 called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args assert called_fan_id == expected_fan_id @pytest.mark.asyncio @pytest.mark.integration async def test_aux2_rejected_when_printer_has_no_left_aux_fan(self, async_client: AsyncClient, printer_factory): """A printer that never reports airduct part 10 must not be sent M106 P10. Without the gate the endpoint accepted aux2 for every model, so a POST against an A1 would fire a command for hardware that does not exist. """ printer = await printer_factory(name="P", model="A1") mock_client = MagicMock() mock_client.set_fan_speed.return_value = True mock_client.state.left_aux_fan_speed = None with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=aux2&speed=50") assert response.status_code == 400 assert "left auxiliary fan" in response.json()["detail"] mock_client.set_fan_speed.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_other_fans_unaffected_by_the_aux2_gate(self, async_client: AsyncClient, printer_factory): """The gate is aux2-only — a base P2S can still drive its built-in fans.""" printer = await printer_factory(name="P", model="P2S") mock_client = MagicMock() mock_client.set_fan_speed.return_value = True mock_client.state.left_aux_fan_speed = None with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client for fan_name in ("part", "aux", "chamber"): response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=50") assert response.status_code == 200, fan_name @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize( "speed_pct,expected_pwm", [(0, 0), (50, 128), (100, 255)], ) async def test_pwm_conversion(self, async_client: AsyncClient, printer_factory, speed_pct, expected_pwm): """0-100% must convert to 0-255 PWM (round-to-nearest).""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.set_fan_speed.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=part&speed={speed_pct}") assert response.status_code == 200 _called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args assert called_pwm == expected_pwm @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize( "model,expected_label", [ ("P2S", "Exhaust fan"), ("X2D", "Exhaust fan"), ("X1C", "Chamber fan"), ("P1S", "Chamber fan"), ("H2D", "Chamber fan"), ], ) async def test_chamber_fan_message_matches_model_label( self, async_client: AsyncClient, printer_factory, model, expected_label ): """The success toast must use the same name as the printer card badge. On P2S/X2D the big_fan2 fan is labelled "Exhaust"; everywhere else it stays "Chamber". A mismatch means the user clicks "Exhaust" and gets told "Chamber fan set to N%". """ printer = await printer_factory(name="P", model=model) mock_client = MagicMock() mock_client.set_fan_speed.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=chamber&speed=50") assert response.status_code == 200 assert response.json()["message"] == f"{expected_label} set to 50%" @pytest.mark.asyncio @pytest.mark.integration async def test_speed_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=part&speed=150") assert response.status_code == 422 class TestSelectExtruderAPI: """Integration tests for POST /printers/{id}/select-extruder (#1661).""" @pytest.mark.asyncio @pytest.mark.integration async def test_not_found(self, async_client: AsyncClient): response = await async_client.post("/api/v1/printers/99999/select-extruder?extruder=0") assert response.status_code == 404 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/select-extruder?extruder=0") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("extruder", [0, 1]) async def test_select_each_extruder(self, async_client: AsyncClient, printer_factory, extruder): printer = await printer_factory(name="P", model="H2D") mock_client = MagicMock() mock_client.select_extruder.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/select-extruder?extruder={extruder}") assert response.status_code == 200 mock_client.select_extruder.assert_called_once_with(extruder) @pytest.mark.asyncio @pytest.mark.integration async def test_extruder_index_out_of_range_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="H2D") response = await async_client.post(f"/api/v1/printers/{printer.id}/select-extruder?extruder=2") assert response.status_code == 422 class TestXYJogAPI: """Integration tests for POST /printers/{id}/xy-jog (#1661).""" @pytest.mark.asyncio @pytest.mark.integration async def test_zero_movement_rejected(self, async_client: AsyncClient, printer_factory): """x=0 AND y=0 (or omitted) must be rejected — no-op jog is a UI bug.""" printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/xy-jog") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("x,y", [(201, 0), (0, 201), (-300, 0), (0, -250)]) async def test_oversize_movement_rejected(self, async_client: AsyncClient, printer_factory, x, y): """Per-axis bound is 200mm; over-bound must be rejected.""" printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/xy-jog?x={x}&y={y}") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/xy-jog?x=10&y=0") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_success_x_only_emits_relative_gcode(self, async_client: AsyncClient, printer_factory): """X-only jog is a bare relative move (no M211), wraps in G91/G90, X only.""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.send_gcode.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/xy-jog?x=10&y=0") assert response.status_code == 200 sent = mock_client.send_gcode.call_args.args[0] # Never touch M211 — bare move, exactly like the touchscreen (#2579). assert "M211" not in sent assert sent.startswith("G91\n") assert sent.endswith("\nG90") assert "X10.00" in sent assert "Y" not in sent # y=0 must NOT be included @pytest.mark.asyncio @pytest.mark.integration async def test_success_both_axes(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.send_gcode.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/xy-jog?x=-5&y=7") assert response.status_code == 200 sent = mock_client.send_gcode.call_args.args[0] assert "X-5.00" in sent and "Y7.00" in sent class TestExtruderJogAPI: """Integration tests for POST /printers/{id}/extruder-jog (#1661). Note: Bambu firmware enforces the cold-extrude guard (min-temp refusal) so the route deliberately does not gate on temp. """ @pytest.mark.asyncio @pytest.mark.integration async def test_zero_distance_rejected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/extruder-jog?distance=0") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration @pytest.mark.parametrize("distance", [101, -200]) async def test_oversize_distance_rejected(self, async_client: AsyncClient, printer_factory, distance): """Per-axis bound is 100mm; over-bound must be rejected.""" printer = await printer_factory(name="P", model="X1C") response = await async_client.post(f"/api/v1/printers/{printer.id}/extruder-jog?distance={distance}") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_not_connected(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = None response = await async_client.post(f"/api/v1/printers/{printer.id}/extruder-jog?distance=5") assert response.status_code == 400 @pytest.mark.asyncio @pytest.mark.integration async def test_extrude_uses_relative_e_mode(self, async_client: AsyncClient, printer_factory): """Extruder jog must wrap with M83 (relative E) and restore M82 (absolute E).""" printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.send_gcode.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/extruder-jog?distance=5") assert response.status_code == 200 sent = mock_client.send_gcode.call_args.args[0] assert sent.startswith("M83\n") assert sent.endswith("\nM82") assert "E5.00" in sent @pytest.mark.asyncio @pytest.mark.integration async def test_retract_uses_negative_distance(self, async_client: AsyncClient, printer_factory): printer = await printer_factory(name="P", model="X1C") mock_client = MagicMock() mock_client.send_gcode.return_value = True with patch("backend.app.api.routes.printers.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client response = await async_client.post(f"/api/v1/printers/{printer.id}/extruder-jog?distance=-3.5") assert response.status_code == 200 sent = mock_client.send_gcode.call_args.args[0] assert "E-3.50" in sent class TestCoverWhenFileServiceIsWedged: """The cover endpoint must not report a wedged printer as "no cover". #2780: once a printer stops completing the FTPS handshake, every cover request walked all 16 candidate paths three times over and ended in a 404 that read as "this print has no thumbnail" — the opposite of the truth. """ @pytest.mark.asyncio @pytest.mark.integration async def test_returns_503_naming_the_file_service(self, async_client: AsyncClient, printer_factory, db_session): printer = await printer_factory(name="Wedged P2S") # Spell out the storage fields rather than leaving them to MagicMock: # every attribute of a bare mock is truthy, so the storage gate added # in #2780 would read this stub as "the print is on internal storage" # and answer 404 before the handshake check ever ran. This test is # about a printer whose file service stopped answering, so say that # its last print went to the card. state = MagicMock( subtask_name="job", state="RUNNING", gcode_file=None, current_project_url="ftp://job.gcode.3mf", sdcard=True, sdcard_reported=True, ) with ( patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch("backend.app.api.routes.printers.ftps_handshake_blocked", return_value=True), patch( "backend.app.api.routes.printers.download_file_try_paths_async", new=AsyncMock(return_value=False), ) as mock_download, ): response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert response.status_code == 503, response.text assert "TLS" in response.json()["detail"] # The whole point: no FTP fan-out against a printer that cannot answer. mock_download.assert_not_called() class TestCoverWhenThePrintIsOnInternalStorage: """The cover lives inside the 3MF, so it is only reachable if the 3MF is. #2780: on an H2C or P2S the sliced file routinely stays on internal storage, which FTPS does not serve. Walking all sixteen candidate paths to arrive at a bare 404 spends connections the printer is already short of and tells the user "this print has no cover", which is not what happened. """ @pytest.fixture(autouse=True) def _clear_404_cache(self): """The route remembers 404s per printer+print so repeat requests skip the fan-out. It is module-level, printer ids restart at 1 for each test, and both tests here use the same subtask name -- so without this the second test reads the first one's cached miss and never reaches the code it is checking. """ from backend.app.api.routes.printers import _cover_404_cache _cover_404_cache.clear() yield _cover_404_cache.clear() @pytest.mark.asyncio @pytest.mark.integration async def test_probes_the_named_file_instead_of_fanning_out( self, async_client: AsyncClient, printer_factory, db_session ): """#2856 narrowed this from a skip to a bounded look. The dispatch names the file, and an H2D with a card in serves that exact name from /cache while reporting `brtc://emmc` -- so ask, but ask once: five paths for one name, not sixteen, and no retries on top. """ printer = await printer_factory(name="H2D") state = MagicMock( subtask_name="job", state="RUNNING", gcode_file=None, current_project_url="brtc://emmc/job.gcode.3mf", sdcard=True, sdcard_reported=True, ) with ( patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch("backend.app.api.routes.printers.ftps_handshake_blocked", return_value=False), patch( "backend.app.api.routes.printers.download_file_try_paths_async", new=AsyncMock(return_value=False), ) as mock_download, ): response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert response.status_code == 404, response.text # A probe that found nothing leaves the printer's own account standing, # so the reason still reaches the caller (#2780). assert "internal_storage" in response.json()["detail"] assert mock_download.await_count == 1, "the probe is one look, not the retry loop" assert mock_download.await_args.args[2] == [ "/job.gcode.3mf", "/cache/job.gcode.3mf", "/model/job.gcode.3mf", "/data/job.gcode.3mf", "/data/Metadata/job.gcode.3mf", ] @pytest.mark.asyncio @pytest.mark.integration async def test_skips_the_fan_out_and_says_why_with_nothing_to_probe_for( self, async_client: AsyncClient, printer_factory, db_session ): """No dispatch to name a file and an empty slot: nothing to look for and nothing to look on, so the route still short-circuits with the reason the frontend explains (#2780).""" printer = await printer_factory(name="H2C") state = MagicMock( subtask_name="job", state="RUNNING", gcode_file=None, current_project_url=None, sdcard=False, sdcard_reported=True, ) with ( patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch("backend.app.api.routes.printers.ftps_handshake_blocked", return_value=False), patch( "backend.app.api.routes.printers.download_file_try_paths_async", new=AsyncMock(return_value=False), ) as mock_download, ): response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert response.status_code == 404, response.text assert "no_external_storage" in response.json()["detail"] mock_download.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_a_print_on_external_storage_still_fans_out( self, async_client: AsyncClient, printer_factory, db_session ): """The regression guard. Every install whose covers work today reaches the download exactly as before.""" printer = await printer_factory(name="X1C") state = MagicMock( subtask_name="job", state="RUNNING", gcode_file=None, current_project_url="ftp://job.gcode.3mf", sdcard=True, sdcard_reported=True, ) with ( patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch("backend.app.api.routes.printers.ftps_handshake_blocked", return_value=False), patch( "backend.app.api.routes.printers.download_file_try_paths_async", new=AsyncMock(return_value=False), ) as mock_download, ): await async_client.get(f"/api/v1/printers/{printer.id}/cover") mock_download.assert_called() class TestClearPlateOnAPoweredDownPrinter: """Releasing the plate-clear gate must not require a reachable printer. #2864: with Auto Power Off the end-of-print state is a dirty plate on a machine Bambuddy has just switched off. The endpoint answered 400 for anything not connected, so the operator who physically cleared that plate had no way to say so — not from the API, not from the UI — and everything gated on the flag stayed stuck until the printer was powered back on by hand. Nothing in the clear path talks to the printer: the flag is Bambuddy-side and persisted. """ @pytest.mark.asyncio @pytest.mark.integration async def test_clear_plate_succeeds_while_disconnected( self, async_client: AsyncClient, printer_factory, db_session ): printer = await printer_factory(name="Powered-off X1C") # What the smart-plug power-off leaves behind: the client is still # registered, but its state is blanked to unknown/disconnected. state = MagicMock(state="unknown", connected=False) with ( patch("backend.app.api.routes.printers.printer_manager.is_connected", return_value=False), patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch( "backend.app.api.routes.printers.printer_manager.is_awaiting_plate_clear", return_value=True, ), patch("backend.app.api.routes.printers.printer_manager.set_awaiting_plate_clear") as mock_set, ): response = await async_client.post(f"/api/v1/printers/{printer.id}/clear-plate") assert response.status_code == 200, response.text mock_set.assert_called_once_with(printer.id, False) @pytest.mark.asyncio @pytest.mark.integration async def test_clear_plate_succeeds_with_no_client_state_at_all( self, async_client: AsyncClient, printer_factory, db_session ): """A printer disconnected manually, or never connected since the last restart, has no client state — the persisted gate is still releasable.""" printer = await printer_factory(name="Never connected") with ( patch("backend.app.api.routes.printers.printer_manager.is_connected", return_value=False), patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=None), patch( "backend.app.api.routes.printers.printer_manager.is_awaiting_plate_clear", return_value=True, ), patch("backend.app.api.routes.printers.printer_manager.set_awaiting_plate_clear") as mock_set, ): response = await async_client.post(f"/api/v1/printers/{printer.id}/clear-plate") assert response.status_code == 200, response.text mock_set.assert_called_once_with(printer.id, False) @pytest.mark.asyncio @pytest.mark.integration async def test_clear_plate_still_rejects_when_there_is_nothing_to_clear( self, async_client: AsyncClient, printer_factory, db_session ): """Dropping the connection guard must not turn the endpoint into a no-op accept: an offline printer with a clean plate is still a 400.""" printer = await printer_factory(name="Powered-off, clean plate") with ( patch("backend.app.api.routes.printers.printer_manager.is_connected", return_value=False), patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=None), patch( "backend.app.api.routes.printers.printer_manager.is_awaiting_plate_clear", return_value=False, ), patch("backend.app.api.routes.printers.printer_manager.set_awaiting_plate_clear") as mock_set, ): response = await async_client.post(f"/api/v1/printers/{printer.id}/clear-plate") assert response.status_code == 400, response.text mock_set.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_plate_clear_state_still_reaches_mqtt_without_a_client( self, async_client: AsyncClient, printer_factory, db_session ): """Clearing the gate on a printer with no registered client must still update the retained MQTT topic (#2525) — otherwise it keeps telling Home Assistant the plate is dirty after it was acknowledged.""" from backend.app.services.printer_manager import printer_manager printer = await printer_factory(name="Disconnected P1S") info = await printer_manager._printer_info_from_db(printer.id) assert info is not None assert info.name == "Disconnected P1S" assert info.serial_number == printer.serial_number @pytest.mark.asyncio @pytest.mark.integration async def test_status_reports_the_gate_without_client_state( self, async_client: AsyncClient, printer_factory, db_session ): """Without this the schema default reported a clean plate for a printer the DB says is still gated, and the UI hid the control that releases it.""" printer = await printer_factory(name="No client state") with ( patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=None), patch( "backend.app.api.routes.printers.printer_manager.is_awaiting_plate_clear", return_value=True, ), ): response = await async_client.get(f"/api/v1/printers/{printer.id}/status") assert response.status_code == 200, response.text body = response.json() assert body["connected"] is False assert body["awaiting_plate_clear"] is True class TestPrintableObjectsReload: """Rebuilding the object list must ask the disk before the printer. The archive of a running print normally holds the very 3MF the printer is executing. Pulling it back over FTPS costs a full transfer from a machine that is mid-print — 15 MB on the print this was written for — and on a printer that kept the file on internal storage it cannot succeed at all. """ @pytest.mark.asyncio @pytest.mark.integration async def test_the_archive_on_disk_is_used_before_any_ftp( self, async_client: AsyncClient, printer_factory, archive_factory, db_session ): printer = await printer_factory(name="H2C-1") await archive_factory(printer.id, status="printing", subtask_id="330151809") client = MagicMock() client.state = MagicMock( printable_objects={}, skipped_objects=[], subtask_id="330151809", subtask_name="HULA", state="RUNNING", gcode_file="/data/Metadata/plate_1.gcode", printable_objects_bbox_all=None, ) ftp = AsyncMock(return_value=None) with ( patch("backend.app.api.routes.printers.printer_manager.get_client", return_value=client), patch( "backend.app.services.archive.extract_printable_objects_from_archive", return_value=({101: {"name": "left"}, 102: {"name": "right"}}, None), ) as from_disk, patch("backend.app.services.bambu_ftp.download_file_try_paths_async", new=ftp), ): response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 200, response.text assert response.json()["total"] == 2 from_disk.assert_called_once() ftp.assert_not_awaited() @pytest.mark.asyncio @pytest.mark.integration async def test_a_reload_that_the_disk_cannot_answer_still_asks_the_printer( self, async_client: AsyncClient, printer_factory, archive_factory, db_session ): """The fan-out is the fallback, not dead code: a print whose archive has no 3MF (or none at all) still has the printer to ask.""" printer = await printer_factory(name="H2C-1") await archive_factory(printer.id, status="printing", subtask_id="330151809") client = MagicMock() client.state = MagicMock( printable_objects={}, skipped_objects=[], subtask_id="330151809", subtask_name="HULA", state="RUNNING", gcode_file="/data/Metadata/plate_1.gcode", printable_objects_bbox_all=None, ) ftp = AsyncMock(return_value=None) with ( patch("backend.app.api.routes.printers.printer_manager.get_client", return_value=client), patch( "backend.app.services.archive.extract_printable_objects_from_archive", return_value=({}, None), ), patch("backend.app.services.bambu_ftp.download_file_try_paths_async", new=ftp), ): response = await async_client.get(f"/api/v1/printers/{printer.id}/print/objects") assert response.status_code == 200, response.text ftp.assert_awaited() class TestCoverUsesTheRunningPrintsArchive: """The cover must not re-fetch a 3MF Bambuddy already has on disk. The in-memory download cache dies with the process, so after a restart mid-print every cover, top view and skip-objects plate mask went back to the printer for the whole file — measured on the maintainer's H2C as three concurrent fan-outs, thirteen seconds and a 0-byte read, on a machine that was busy printing. The archive of the running print holds that exact file. """ @pytest.mark.asyncio @pytest.mark.integration async def test_the_archived_3mf_is_preferred_over_ftp( self, async_client: AsyncClient, printer_factory, archive_factory, db_session, tmp_path ): import zipfile from backend.app.core.config import settings as app_settings printer = await printer_factory(name="H2C-1") # A 3MF that is a real zip but carries no cover, so the endpoint reaches # its "no cover in this file" answer without any FTP work. threemf = tmp_path / "job.3mf" with zipfile.ZipFile(threemf, "w") as zf: zf.writestr("Metadata/slice_info.config", "") await archive_factory( printer.id, status="printing", subtask_id="330151809", file_path=str(threemf.relative_to(tmp_path)), ) state = MagicMock( subtask_name="job", subtask_id="330151809", state="RUNNING", gcode_file="/data/Metadata/plate_1.gcode", current_project_url="ftp://job.3mf", sdcard=True, sdcard_reported=True, ) ftp = AsyncMock(return_value=None) with ( patch.object(app_settings, "base_dir", tmp_path), patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch("backend.app.api.routes.printers.download_file_try_paths_async", new=ftp), ): response = await async_client.get(f"/api/v1/printers/{printer.id}/cover") # No cover inside this 3MF, so a 404 — but reached from the file on disk. assert response.status_code == 404, response.text ftp.assert_not_awaited() @pytest.mark.asyncio @pytest.mark.integration async def test_the_archived_file_is_never_deleted_by_the_cover_flow( self, async_client: AsyncClient, printer_factory, archive_factory, db_session, tmp_path ): """The cover flow unlinks the file it downloaded. Handed the archive's own copy, that would delete the print's 3MF.""" import zipfile from backend.app.core.config import settings as app_settings printer = await printer_factory(name="H2C-1") threemf = tmp_path / "job.3mf" with zipfile.ZipFile(threemf, "w") as zf: zf.writestr("Metadata/slice_info.config", "") await archive_factory( printer.id, status="printing", subtask_id="330151809", file_path=str(threemf.relative_to(tmp_path)), ) state = MagicMock( subtask_name="job", subtask_id="330151809", state="RUNNING", gcode_file="/data/Metadata/plate_1.gcode", current_project_url="ftp://job.3mf", sdcard=True, sdcard_reported=True, ) with ( patch.object(app_settings, "base_dir", tmp_path), patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state), patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1), patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None), patch( "backend.app.api.routes.printers.download_file_try_paths_async", new=AsyncMock(return_value=None), ), ): await async_client.get(f"/api/v1/printers/{printer.id}/cover") assert threemf.is_file(), "the cover flow deleted the running print's archived 3MF"