"""Every path that creates queue jobs behaves like the print dialog (C4). Only ``POST /queue/`` announced a new job ("job added" notification and the MQTT relay event); File Manager's "Add to queue", the webhook, a virtual printer, a pipeline run and an order's "dispatch remaining" said nothing. The default print options (timelapse, bed levelling, ...) reached only the print dialog and the virtual printer. The copies an order dispatches lost their cost centre, so they were billed to none and never budget-checked. And a ``scheduled_time`` sent with a UTC offset lost the offset, so the job started hours late. """ from datetime import datetime from pathlib import Path from unittest.mock import AsyncMock, patch from uuid import uuid4 import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from backend.app.core.auth import generate_api_key, get_password_hash from backend.app.core.config import settings as app_settings from backend.app.models.api_key import APIKey from backend.app.models.archive import PrintArchive from backend.app.models.finance import CostCenter, CostCenterMember from backend.app.models.group import Group from backend.app.models.print_queue import PrintQueueItem from backend.app.models.printer import Printer from backend.app.models.settings import Settings from backend.app.models.user import User from backend.app.services.mqtt_relay import mqtt_relay from backend.app.services.notification_service import notification_service pytestmark = [pytest.mark.asyncio, pytest.mark.integration] async def _set(db_session, key: str, value: str) -> None: row = await db_session.scalar(select(Settings).where(Settings.key == key)) if row is None: db_session.add(Settings(key=key, value=value)) else: row.value = value await db_session.commit() async def _items(test_engine, ids=None) -> list[PrintQueueItem]: maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) async with maker() as fresh: query = select(PrintQueueItem).order_by(PrintQueueItem.id) if ids is not None: query = query.where(PrintQueueItem.id.in_(ids)) return list((await fresh.execute(query)).scalars().all()) @pytest.fixture def announced(): """The notification and relay calls, recorded.""" notify = AsyncMock() relay = AsyncMock() with ( patch.object(notification_service, "on_queue_job_added", notify), patch.object(mqtt_relay, "on_queue_job_added", relay), ): yield notify, relay @pytest.fixture async def library_files(db_session): from backend.app.models.library import LibraryFile made = [] async def make(stem: str): name = f"{stem}_{uuid4().hex[:8]}.gcode.3mf" rel_path = f"archive/library/files/{name}" abs_path = Path(app_settings.base_dir) / rel_path abs_path.parent.mkdir(parents=True, exist_ok=True) abs_path.write_bytes(b"probe") made.append(abs_path) lib_file = LibraryFile(filename=name, file_path=rel_path, file_size=5, file_type="3mf") db_session.add(lib_file) await db_session.commit() await db_session.refresh(lib_file) return lib_file yield make for path in made: path.unlink(missing_ok=True) async def _printer(db_session, name="C4 Printer", model="X1C") -> Printer: printer = Printer( name=name, ip_address=f"192.168.44.{len(name)}", serial_number=f"C4{uuid4().hex[:12].upper()}", access_code="12345678", model=model, ) db_session.add(printer) await db_session.commit() await db_session.refresh(printer) return printer async def _archive(db_session, filename="widget.gcode.3mf", **extra) -> PrintArchive: archive = PrintArchive( filename=filename, print_name="widget", file_path=f"/tmp/c4_{uuid4().hex}.3mf", # nosec B108 file_size=1, content_hash=uuid4().hex, status="completed", **extra, ) db_session.add(archive) await db_session.commit() await db_session.refresh(archive) return archive async def _webhook_key(db_session) -> dict[str, str]: full_key, key_hash, key_prefix = generate_api_key() db_session.add( APIKey(name="c4", key_hash=key_hash, key_prefix=key_prefix, user_id=None, can_queue=True, can_read_status=True) ) await db_session.commit() return {"X-API-Key": full_key} class TestEveryPathAnnouncesItsJobs: async def test_a_file_manager_bulk_add_sends_one_summary( self, async_client: AsyncClient, db_session, library_files, announced ): notify, relay = announced printer = await _printer(db_session) files = [await library_files("bracket"), await library_files("hinge")] response = await async_client.post( "/api/v1/library/files/add-to-queue", json={"file_ids": [f.id for f in files], "printer_id": printer.id}, ) assert response.status_code == 200, response.text notify.assert_awaited_once() kwargs = notify.await_args.kwargs assert kwargs["job_name"].startswith("2 jobs (bracket_") assert "hinge_" in kwargs["job_name"] assert kwargs["target"] == printer.name assert kwargs["printer_id"] == printer.id # The relay is a machine event: one per job. assert sorted(c.kwargs["job_id"] for c in relay.await_args_list) == sorted( a["queue_item_id"] for a in response.json()["added"] ) async def test_the_webhook_announces_its_job(self, async_client: AsyncClient, db_session, announced): notify, relay = announced printer = await _printer(db_session) archive = await _archive(db_session) response = await async_client.post( "/api/v1/webhook/queue/add", json={"printer_id": printer.id, "archive_id": archive.id}, headers=await _webhook_key(db_session), ) assert response.status_code == 200, response.text notify.assert_awaited_once() assert notify.await_args.kwargs["job_name"] == "widget" assert relay.await_args.kwargs["job_id"] == response.json()["id"] async def test_an_orders_dispatch_announces_the_copies(self, async_client: AsyncClient, db_session, announced): notify, _ = announced printer = await _printer(db_session) archive = await _archive(db_session) order = ( await async_client.post( "/api/v1/queue/batches", json={"name": "o", "archive_id": archive.id, "plates": [{"plate_id": None, "quantity_target": 3}]}, ) ).json() first = await async_client.post( "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "batch_id": order["id"]} ) assert first.status_code == 200, first.text notify.reset_mock() response = await async_client.post(f"/api/v1/queue/batches/{order['id']}/dispatch", json={}) assert response.status_code == 200, response.text notify.assert_awaited_once() assert notify.await_args.kwargs["job_name"] == "widget ×2" async def test_a_failing_notification_does_not_fail_the_add( self, async_client: AsyncClient, db_session, library_files ): lib_file = await library_files("plain") with patch.object(notification_service, "on_queue_job_added", AsyncMock(side_effect=RuntimeError("smtp"))): response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]}) assert response.status_code == 200 assert len(response.json()["added"]) == 1 async def test_the_print_dialog_still_announces_once(self, async_client: AsyncClient, db_session, announced): notify, relay = announced printer = await _printer(db_session) archive = await _archive(db_session) response = await async_client.post( "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "quantity": 3} ) assert response.status_code == 200, response.text notify.assert_awaited_once() assert notify.await_args.kwargs["job_name"] == "widget ×3" assert relay.await_count == 3 class TestTheSummary: async def test_jobs_for_different_printers_are_announced_per_printer(self, db_session, test_engine, announced): """A provider scoped to one printer must not hear of another's jobs.""" from backend.app.services.queue_announcements import announce_queue_jobs_added notify, _ = announced a, b = await _printer(db_session, "Alpha"), await _printer(db_session, "Bravo") archive = await _archive(db_session) items = [ PrintQueueItem(printer_id=a.id, archive_id=archive.id, position=1), PrintQueueItem(printer_id=b.id, archive_id=archive.id, position=2), PrintQueueItem(printer_id=None, target_model="P1S", archive_id=archive.id, position=3), ] db_session.add_all(items) await db_session.commit() await announce_queue_jobs_added(db_session, [i.id for i in items]) sent = sorted((c.kwargs["printer_id"] or 0, c.kwargs["target"]) for c in notify.await_args_list) assert sent == sorted([(a.id, "Alpha"), (b.id, "Bravo"), (0, "Any P1S")]) async def test_many_different_files_are_named_briefly(self, db_session, announced): from backend.app.services.queue_announcements import announce_queue_jobs_added notify, _ = announced items = [] for n in range(5): archive = await _archive(db_session, filename=f"part{n}.3mf") items.append(PrintQueueItem(printer_id=None, archive_id=archive.id, position=n)) db_session.add_all(items) await db_session.commit() await announce_queue_jobs_added(db_session, [i.id for i in items]) assert notify.await_args.kwargs["job_name"] == "5 jobs (part0, part1, part2, …)" assert notify.await_args.kwargs["target"] == "Unassigned" async def test_nothing_added_says_nothing(self, db_session, announced): from backend.app.services.queue_announcements import announce_queue_jobs_added notify, relay = announced await announce_queue_jobs_added(db_session, []) notify.assert_not_awaited() relay.assert_not_awaited() class TestTheDefaultPrintOptions: async def test_unset_they_are_the_settings_defaults(self, db_session): from backend.app.services.print_defaults import default_print_options assert await default_print_options(db_session) == { "bed_levelling": "auto", "flow_cali": "auto", "nozzle_offset_cali": "auto", "vibration_cali": True, "layer_inspect": False, "timelapse": False, } async def test_set_they_are_read_including_legacy_values(self, db_session): from backend.app.services.print_defaults import default_print_options for key, value in { "default_bed_levelling": "false", # legacy boolean row "default_flow_cali": "on", "default_nozzle_offset_cali": "off", "default_vibration_cali": "false", "default_layer_inspect": "true", "default_timelapse": "true", }.items(): await _set(db_session, key, value) assert await default_print_options(db_session) == { "bed_levelling": "off", "flow_cali": "on", "nozzle_offset_cali": "off", "vibration_cali": False, "layer_inspect": True, "timelapse": True, } async def test_a_file_manager_add_uses_them( self, async_client: AsyncClient, db_session, test_engine, library_files ): await _set(db_session, "default_timelapse", "true") await _set(db_session, "default_bed_levelling", "off") lib_file = await library_files("defaults") response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]}) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [response.json()["added"][0]["queue_item_id"]]) assert (item.timelapse, item.bed_levelling) == (True, "off") async def test_the_webhook_uses_them(self, async_client: AsyncClient, db_session, test_engine): await _set(db_session, "default_timelapse", "true") await _set(db_session, "default_layer_inspect", "true") printer = await _printer(db_session) archive = await _archive(db_session) response = await async_client.post( "/api/v1/webhook/queue/add", json={"printer_id": printer.id, "archive_id": archive.id}, headers=await _webhook_key(db_session), ) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [response.json()["id"]]) assert (item.timelapse, item.layer_inspect) == (True, True) class TestAPipelineRun: async def _run(self, db_session, test_engine, tmp_path): from backend.app.api.routes.pipeline_runs import _make_orchestration_callable from backend.app.core.printer_scope import ALL_PRINTERS from backend.app.models.library import LibraryFile from backend.app.models.pipeline_run import PipelineJob, PipelineRun from backend.app.models.slicer_pipeline import SlicerPipeline from backend.app.schemas.slicer import SliceResponse printer = await _printer(db_session, "Pipe") source = LibraryFile(filename="pipe.3mf", file_path="library/pipe.3mf", file_type="3mf", file_size=1) pipeline = SlicerPipeline( name="p", printer_preset_source="local", printer_preset_id="1", process_preset_source="local", process_preset_id="2", filament_presets_json='[{"source": "local", "id": "3"}]', target_kind="specific_printer", target_printer_id=printer.id, ) db_session.add_all([source, pipeline]) await db_session.commit() run = PipelineRun(pipeline_id=pipeline.id, source_library_file_id=source.id, copies=2, created_by=None) db_session.add(run) await db_session.commit() db_session.add_all([PipelineJob(pipeline_run_id=run.id, copy_index=i) for i in range(2)]) await db_session.commit() model = tmp_path / "pipe.3mf" model.write_bytes(b"model") async def sliced(session, **kwargs): out = LibraryFile(filename="pipe.gcode.3mf", file_path="library/pipe.gcode.3mf", file_type="3mf") out.file_size = 1 session.add(out) await session.flush() return SliceResponse( library_file_id=out.id, name="pipe", print_time_seconds=1, filament_used_g=1.0, filament_used_mm=1.0 ) orchestrate = _make_orchestration_callable( run_id=run.id, pipeline_id=pipeline.id, src_kind="library_file", src_id=source.id, src_filename="pipe.3mf", src_path=Path(model), creator_user_id=None, copies=2, printer_scope=ALL_PRINTERS, ) with ( patch( "backend.app.api.routes.pipeline_runs.async_session", async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False), ), patch("backend.app.api.routes.library.slice_and_persist", new=sliced), patch("backend.app.api.routes.pipeline_runs._publish_run_event", new=AsyncMock()), ): await orchestrate(1) return await _items(test_engine) async def test_announces_its_copies_and_uses_the_defaults(self, db_session, test_engine, tmp_path, announced): notify, _ = announced await _set(db_session, "default_timelapse", "true") items = await self._run(db_session, test_engine, tmp_path) assert len(items) == 2 assert all(item.timelapse for item in items) notify.assert_awaited_once() assert notify.await_args.kwargs["job_name"] == "pipe ×2" class TestAVirtualPrinter: async def test_announces_what_it_queued(self, tmp_path): from backend.tests.unit.services.test_vp_queue_intake_d2 import _enqueue, _session, _vp added: list = [] factory, _ = _session(added) inst = _vp(tmp_path, factory, target_printer_id=7) announce = AsyncMock() await _enqueue(inst, tmp_path, {"timelapse": True}, announce=announce) announce.assert_awaited_once() assert announce.await_args.args[1] == [item.id for item in added] async def test_where_the_slicer_says_nothing_it_uses_the_defaults(self, tmp_path): """Read from the same helper as every other path now.""" from backend.tests.unit.services.test_vp_queue_intake_d2 import _enqueue, _session, _vp added: list = [] factory, _ = _session(added) inst = _vp(tmp_path, factory, target_printer_id=7) await _enqueue( inst, tmp_path, {"flow_cali": True}, {"default_timelapse": "true", "default_layer_inspect": "true", "default_vibration_cali": "false"}, ) assert (added[0].timelapse, added[0].layer_inspect, added[0].vibration_cali) == (True, True, False) class TestAnOrdersCopiesKeepTheirCostCentre: @pytest.fixture async def billed(self, db_session): await _set(db_session, "auth_enabled", "true") await _set(db_session, "advanced_auth_enabled", "false") group = Group( name="c4-dispatchers", description="t", permissions=[ "queue:create", "queue:read_all", "queue:update_all", "archives:read_all", "archives:reprint_all", ], is_system=False, ) db_session.add(group) await db_session.flush() admin = User(username="c4admin", password_hash=get_password_hash("Pass1word!"), role="admin", is_active=True) clerk = User(username="c4clerk", password_hash=get_password_hash("Pass1word!"), is_active=True, groups=[group]) center = CostCenter(name="c4-center", is_active=True, is_private=False) db_session.add_all([admin, clerk, center]) await db_session.commit() printer = await _printer(db_session) archive = await _archive(db_session, cost=1.25, filament_used_grams=50.0) await _set(db_session, "billing_enabled", "true") return {"admin": admin, "clerk": clerk, "center": center, "printer": printer, "archive": archive} async def _login(self, client, username): response = await client.post("/api/v1/auth/login", json={"username": username, "password": "Pass1word!"}) assert response.status_code == 200, response.text return {"Authorization": f"Bearer {response.json()['access_token']}"} async def _order(self, client, headers, world): order = await client.post( "/api/v1/queue/batches", json={"name": "o", "archive_id": world["archive"].id, "plates": [{"plate_id": None, "quantity_target": 3}]}, headers=headers, ) assert order.status_code == 200, order.text first = await client.post( "/api/v1/queue/", json={ "printer_id": world["printer"].id, "archive_id": world["archive"].id, "batch_id": order.json()["id"], "cost_center_id": world["center"].id, }, headers=headers, ) assert first.status_code == 200, first.text return order.json()["id"] async def test_the_copies_are_billed_to_it(self, async_client: AsyncClient, test_engine, billed): admin = await self._login(async_client, "c4admin") batch_id = await self._order(async_client, admin, billed) response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin) assert response.status_code == 200, response.text items = [i for i in await _items(test_engine) if i.batch_id == batch_id] assert len(items) == 3 assert {i.cost_center_id for i in items} == {billed["center"].id} async def test_a_dispatcher_who_may_not_use_it_is_refused( self, async_client: AsyncClient, db_session, test_engine, billed ): admin = await self._login(async_client, "c4admin") batch_id = await self._order(async_client, admin, billed) clerk = await self._login(async_client, "c4clerk") response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=clerk) assert response.status_code == 403, response.text assert len([i for i in await _items(test_engine) if i.batch_id == batch_id]) == 1 db_session.add(CostCenterMember(cost_center_id=billed["center"].id, user_id=billed["clerk"].id, can_print=True)) await db_session.commit() response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=clerk) assert response.status_code == 200, response.text async def test_the_budget_counts_every_copy(self, async_client: AsyncClient, db_session, test_engine, billed): center = await db_session.get(CostCenter, billed["center"].id) admin = await self._login(async_client, "c4admin") batch_id = await self._order(async_client, admin, billed) (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id] # Room for this one and one more, not for the two copies still owed. center.total_budget = 2 * template.estimated_cost + 0.01 await db_session.commit() response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin) assert response.status_code == 400, response.text assert "budget" in response.json()["detail"] async def test_copies_without_one_are_refused_while_billing_is_on( self, async_client: AsyncClient, db_session, test_engine, billed ): """Queued before billing was switched on: refused here, as POST /queue/ would, rather than failing later at the start.""" admin = await self._login(async_client, "c4admin") batch_id = await self._order(async_client, admin, billed) (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id] row = await db_session.get(PrintQueueItem, template.id) row.cost_center_id = None await db_session.commit() response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin) assert response.status_code == 400, response.text assert "Cost center is required" in response.json()["detail"] async def test_a_budget_that_holds_every_copy_dispatches( self, async_client: AsyncClient, db_session, test_engine, billed ): """The copies are not counted against themselves.""" center = await db_session.get(CostCenter, billed["center"].id) admin = await self._login(async_client, "c4admin") batch_id = await self._order(async_client, admin, billed) (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id] center.total_budget = 3 * template.estimated_cost + 0.01 await db_session.commit() response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin) assert response.status_code == 200, response.text copies = [i for i in await _items(test_engine) if i.batch_id == batch_id and i.id != template.id] assert [c.estimated_cost for c in copies] == [template.estimated_cost] * 2 class TestAScheduledTimeWithAnOffset: async def test_the_queue_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine): printer = await _printer(db_session) archive = await _archive(db_session) response = await async_client.post( "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "2026-10-15T10:00:00+02:00"}, ) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [response.json()["id"]]) assert item.scheduled_time == datetime(2026, 10, 15, 8, 0) assert response.json()["scheduled_time"] == "2026-10-15T08:00:00Z" async def test_an_edit_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine): printer = await _printer(db_session) archive = await _archive(db_session) created = await async_client.post("/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id}) response = await async_client.patch( f"/api/v1/queue/{created.json()['id']}", json={"scheduled_time": "2026-10-15T10:00:00-05:00"} ) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [created.json()["id"]]) assert item.scheduled_time == datetime(2026, 10, 15, 15, 0) async def test_a_bulk_edit_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine): printer = await _printer(db_session) archive = await _archive(db_session) created = await async_client.post("/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id}) response = await async_client.patch( "/api/v1/queue/bulk", json={"item_ids": [created.json()["id"]], "scheduled_time": "2026-10-15T10:00:00+02:00"}, ) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [created.json()["id"]]) assert item.scheduled_time == datetime(2026, 10, 15, 8, 0) async def test_the_webhook_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine): printer = await _printer(db_session) archive = await _archive(db_session) response = await async_client.post( "/api/v1/webhook/queue/add", json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "2026-10-15T10:00:00+02:00"}, headers=await _webhook_key(db_session), ) assert response.status_code == 200, response.text (item,) = await _items(test_engine, [response.json()["id"]]) assert item.scheduled_time == datetime(2026, 10, 15, 8, 0) async def test_the_postgres_bind_hook_converts_rather_than_drops_the_offset(): """The backstop for any other aware value: naive columns hold UTC.""" from datetime import timedelta, timezone from backend.app.core import database with patch.object(database, "is_sqlite", return_value=False): engine = database._create_engine("postgresql+asyncpg://u:p@127.0.0.1:1/never") try: (hook,) = [ fn for fn in engine.sync_engine.dispatch.before_cursor_execute if "strip" in getattr(fn, "__name__", "") ] aware = datetime(2026, 10, 15, 10, 0, tzinfo=timezone(timedelta(hours=2))) _, params = hook(None, None, "", {"t": aware, "rows": [(aware,)]}, None, False) assert params == {"t": datetime(2026, 10, 15, 8, 0), "rows": [(datetime(2026, 10, 15, 8, 0),)]} finally: await engine.dispose() async def test_a_time_at_the_edge_of_the_calendar_is_refused_not_a_crash(async_client: AsyncClient, db_session): printer = await _printer(db_session) archive = await _archive(db_session) queued = await async_client.post( "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "0001-01-01T00:00:00+05:00"}, ) webhook = await async_client.post( "/api/v1/webhook/queue/add", json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "0001-01-01T00:00:00+05:00"}, headers=await _webhook_key(db_session), ) assert (queued.status_code, webhook.status_code) == (422, 400) async def test_a_failed_notification_log_leaves_the_session_usable(db_session): from backend.app.services.queue_announcements import announce_queue_jobs_added archive = await _archive(db_session) item = PrintQueueItem(printer_id=None, archive_id=archive.id, position=1) db_session.add(item) await db_session.commit() async def log_then_fail(**kwargs): db_session.add(Settings(key="dup", value="1")) db_session.add(Settings(key="dup", value="2")) await kwargs["db"].commit() with patch.object(notification_service, "on_queue_job_added", side_effect=log_then_fail): await announce_queue_jobs_added(db_session, [item.id]) db_session.add(Settings(key="after", value="1")) await db_session.commit()