diff --git a/backend/app/models/print_queue.py b/backend/app/models/print_queue.py index 706660a94..552586604 100644 --- a/backend/app/models/print_queue.py +++ b/backend/app/models/print_queue.py @@ -165,6 +165,82 @@ class PrintQueueItem(Base): project: Mapped["Project | None"] = relationship(back_populates="queue_items") batch: Mapped["PrintBatch | None"] = relationship(back_populates="queue_items") created_by: Mapped["User | None"] = relationship() + variants: Mapped[list["PrintQueueVariant"]] = relationship( + back_populates="queue_item", + cascade="all, delete-orphan", + order_by="PrintQueueVariant.position", + ) + + +class PrintQueueVariant(Base): + """One candidate file for a queue item that may print on several models (#671). + + A user with an H2S and an H2C slices the same job twice and does not care + which machine runs it. Each slice becomes a variant; the scheduler walks them + in ``position`` order and takes the first whose model has an idle printer. + + **This is a snapshot, not a pointer.** The candidate list is copied from the + library's variant group when the item is queued, and every per-file setting + the dispatcher needs is copied with it. Two reasons: + + - Editing the library group afterwards must not silently change a job that is + already waiting in the queue. + - The per-file settings genuinely differ between candidates and are choices + the user made for *this* job, not properties of the file. An H2C slice is + dual-nozzle and will not have the same slot count, AMS mapping or nozzle + mapping as the H2S slice of the same model. + + On a match the winning variant's fields are written onto the queue row before + the dispatch commit, so everything downstream — upload, archive creation, + print history, reprint — sees an ordinary single-file item and needs no + knowledge that variants exist. + + Variants reference library files only. An archive records a print that already + happened, of one specific file, so it is never a candidate for "which of these + should we run". + """ + + __tablename__ = "print_queue_variants" + + id: Mapped[int] = mapped_column(primary_key=True) + queue_item_id: Mapped[int] = mapped_column( + ForeignKey("print_queue.id", ondelete="CASCADE"), nullable=False, index=True + ) + # User's priority order. When two printers are idle in the same scheduler + # pass, the lowest position wins — so the choice is reproducible instead of + # depending on which match the matcher happened to find first. + position: Mapped[int] = mapped_column(Integer, default=0) + + # CASCADE: deleting the file drops this candidate but leaves the item and its + # other candidates alone. Losing the *last* candidate is handled by the + # resolver, which holds the item pending with an explicit waiting_reason + # rather than letting it sit there looking dispatchable forever. + library_file_id: Mapped[int] = mapped_column(ForeignKey("library_files.id", ondelete="CASCADE"), nullable=False) + # Normalized short name ("H2S"), taken from the file's own sliced_for_model + # at creation, or picked by the user for a legacy file that declares none. + target_model: Mapped[str] = mapped_column(String(50), nullable=False) + + # Per-file dispatch settings, same semantics as the identically named columns + # on PrintQueueItem — see there for the formats. + plate_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + ams_mapping: Mapped[str | None] = mapped_column(Text, nullable=True) + nozzle_mapping: Mapped[str | None] = mapped_column(Text, nullable=True) + filament_overrides: Mapped[str | None] = mapped_column(Text, nullable=True) + required_filament_types: Mapped[str | None] = mapped_column(Text, nullable=True) + print_time_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True) + + # How many times this candidate has been dispatched and bounced back to + # pending by the start-watchdog. The resolver tries least-attempted first, so + # a printer that accepts the file and never starts (#1678) hands the job to + # the other machine on the next lap instead of burning the item's whole + # DISPATCH_MAX_ATTEMPTS budget against the same wedged printer — which is the + # entire reason the user queued an alternative. + attempt_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + + queue_item: Mapped["PrintQueueItem"] = relationship(back_populates="variants") + library_file: Mapped["LibraryFile"] = relationship() from backend.app.models.archive import PrintArchive # noqa: E402 diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index ebe44cfbc..96cfa8800 100644 --- a/backend/app/services/print_scheduler.py +++ b/backend/app/services/print_scheduler.py @@ -4,6 +4,7 @@ import asyncio import json import logging import time +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -17,7 +18,7 @@ from backend.app.core.tasks import spawn_background_task from backend.app.core.websocket import ws_manager from backend.app.models.archive import PrintArchive from backend.app.models.library import LibraryFile -from backend.app.models.print_queue import PrintQueueItem +from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant from backend.app.models.printer import Printer from backend.app.models.settings import Settings from backend.app.models.smart_plug import SmartPlug @@ -156,6 +157,123 @@ def _canonical_filament_type(ftype: str) -> str: return _FILAMENT_EQUIV_MAP.get(upper, upper) +@dataclass(slots=True) +class _ModelCandidate: + """One (file, printer model) pair the model-based matcher may try. + + Model-based assignment used to have exactly one of these per item, held + directly in the item's own columns. Cross-model queue items (#671) have + several, held in ``print_queue_variants``. Both shapes are normalised into + this so the matching, the cross-model gate and the waiting-reason handling + are written once and an item without variants provably takes the same path + it took before variants existed. + + ``variant`` is None for the item's own columns and set for a real variant + row, which is what :meth:`PrintScheduler._resolve_variant` writes onto the + item once that candidate wins. + """ + + target_model: str | None + sliced_for: str | None + required_filament_types: str | None + filament_overrides: str | None + variant: "PrintQueueVariant | None" = None + + +def _sliced_for_model(archive, library_file) -> str | None: + """Model a 3MF declares it was sliced for, from whichever source holds it.""" + if archive is not None: + return archive.sliced_for_model + if library_file is not None and library_file.file_metadata: + return library_file.file_metadata.get("sliced_for_model") + return None + + +def _candidates_for(item: PrintQueueItem) -> list[_ModelCandidate]: + """Candidate files for ``item``, best first. + + An item with no variant rows yields exactly one candidate built from its own + columns — the pre-#671 behaviour, unchanged. + + Variants come back least-attempted first, ties broken by the user's + ``position``. On the first pass every count is zero, so this is purely the + user's priority order. After a start-watchdog bounce the printer that failed + drops behind, so the next lap tries the other machine rather than spending the + item's whole retry budget on the one that is wedged. Once every candidate has + been tried equally often they cycle again, which keeps the item-level + ``DISPATCH_MAX_ATTEMPTS`` bound from #2555 intact — a job with alternatives + still gives up, it just does not give up without trying them. + """ + if not item.variants: + return [ + _ModelCandidate( + target_model=item.target_model, + sliced_for=_sliced_for_model(item.archive, item.library_file), + required_filament_types=item.required_filament_types, + filament_overrides=item.filament_overrides, + ) + ] + + ordered = sorted(item.variants, key=lambda v: (v.attempt_count or 0, v.position, v.id)) + return [ + _ModelCandidate( + target_model=v.target_model, + sliced_for=_sliced_for_model(None, v.library_file), + required_filament_types=v.required_filament_types, + filament_overrides=v.filament_overrides, + variant=v, + ) + for v in ordered + ] + + +def _collapse_waiting_reasons(per_model: list[tuple[str | None, str]]) -> str | None: + """Fold one waiting reason per candidate into a single line for the item. + + A cross-model item produces a reason per candidate, and pasting them + together unlabelled reads as gibberish ("No idle printer; PETG not loaded" + — on which machine?). Each reason is prefixed with its model, except in the + single-candidate case where the item already displays its target model and + the prefix would be noise. + + Identical reasons collapse rather than repeat, so three idle-less models + read as one clause. + + When *every* candidate is merely busy the parts are joined with the ``" | "`` + separator :meth:`PrintScheduler._is_busy_only` already parses, and left + unprefixed. That case must keep testing busy-only: a fleet that is simply + printing needs no user action, and labelling the clauses would turn each pass + over a two-model item into a "job waiting" notification. + """ + reasons = [(model, reason) for model, reason in per_model if reason] + if not reasons: + return None + if len(reasons) == 1: + return reasons[0][1] + + distinct = list(dict.fromkeys(reason for _model, reason in reasons)) + if len(distinct) == 1: + return distinct[0] + + if all(PrintScheduler._is_busy_only(reason) for _model, reason in reasons): + return " | ".join(distinct) + + return "; ".join(f"{model or 'unassigned'}: {reason}" for model, reason in reasons) + + +def _candidate_model_label(candidates: list[_ModelCandidate]) -> str | None: + """Human label for the models an item is waiting on ("H2S or H2C"). + + Notifications take a single target model. For a cross-model item the item's + own ``target_model`` is whichever variant happens to be first, which reads as + a lie once it is the H2C that actually runs — so name all of them. + """ + models = list(dict.fromkeys(c.target_model for c in candidates if c.target_model)) + if not models: + return None + return " or ".join(models) + + def _mapping_is_all_unresolved(mapping: list | None) -> bool: """True if ``mapping`` is a non-empty list whose every entry is the unresolved sentinel (-1 / None) — i.e. no required slot ever matched a tray. @@ -404,6 +522,10 @@ class PrintScheduler: .options( selectinload(PrintQueueItem.archive), selectinload(PrintQueueItem.library_file), + # Cross-model candidates (#671), plus each candidate's file + # for the same cross-model gate. Lazy-loading either would + # raise in async. + selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file), ) .order_by( PrintQueueItem.printer_id, @@ -422,6 +544,10 @@ class PrintScheduler: .options( selectinload(PrintQueueItem.archive), selectinload(PrintQueueItem.library_file), + # Cross-model candidates (#671), plus each candidate's file + # for the same cross-model gate. Lazy-loading either would + # raise in async. + selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file), ) .order_by(PrintQueueItem.printer_id, PrintQueueItem.position) ) @@ -684,60 +810,94 @@ class PrintScheduler: other.been_jumped = True await db.commit() - elif item.target_model: - # Model-based assignment - find any idle printer of matching model - # Parse required filament types if present - required_types = None - if item.required_filament_types: - try: - required_types = json.loads(item.required_filament_types) - except json.JSONDecodeError: - pass # Ignore malformed filament types; treat as no constraint + elif item.target_model or item.variants: + # Model-based assignment - find any idle printer of matching model. + # A plain model-based item has exactly one candidate, built from + # its own columns. A cross-model item (#671) has one per sliced + # variant and takes the first that matches, walking them in the + # user's priority order so the pick is reproducible when more + # than one printer is free in the same pass. + candidates = _candidates_for(item) + printer_id = None + chosen: _ModelCandidate | None = None + per_model_reasons: list[tuple[str | None, str]] = [] - # Parse filament overrides if present - filament_overrides = None - if item.filament_overrides: - try: - filament_overrides = json.loads(item.filament_overrides) - except json.JSONDecodeError: - pass - - # If overrides exist, use override types for validation instead - effective_types = required_types - if filament_overrides: - override_types = sorted({o["type"] for o in filament_overrides if "type" in o}) - if override_types: - # Merge: keep original types for non-overridden slots, add override types - effective_types = sorted(set(required_types or []) | set(override_types)) - - # Cross-model safety gate (#2578): never hand a 3MF sliced - # for an incompatible model to a printer, no matter how the - # row got into the DB (old rows, direct API writes). Held - # as pending with an actionable waiting_reason — the user - # fixes it by editing the item's target model. - sliced_for = None - if item.archive: - sliced_for = item.archive.sliced_for_model - elif item.library_file and item.library_file.file_metadata: - sliced_for = item.library_file.file_metadata.get("sliced_for_model") - - if not is_gcode_compatible(sliced_for, item.target_model): - printer_id = None - waiting_reason = ( - f"File was sliced for {sliced_for}, which is not compatible with " - f"{item.target_model} — edit the item and fix its target model" + if not item.variants and not item.archive_id and not item.library_file_id: + # Every candidate file was deleted out from under this item + # (variant rows go with their library file). Dispatching would + # fail deep in the upload with "No archive_id or library_file_id"; + # hold it here with something the user can act on instead. + candidates = [] + per_model_reasons.append( + ( + item.target_model, + "Every file for this job has been deleted — add a file back or remove the item", + ) ) - skip_reasons["sliced_model_mismatch"] = skip_reasons.get("sliced_model_mismatch", 0) + 1 - else: - printer_id, waiting_reason = await self._find_idle_printer_for_model( + + for candidate in candidates: + # Parse required filament types if present + required_types = None + if candidate.required_filament_types: + try: + required_types = json.loads(candidate.required_filament_types) + except json.JSONDecodeError: + pass # Ignore malformed filament types; treat as no constraint + + # Parse filament overrides if present + filament_overrides = None + if candidate.filament_overrides: + try: + filament_overrides = json.loads(candidate.filament_overrides) + except json.JSONDecodeError: + pass + + # If overrides exist, use override types for validation instead + effective_types = required_types + if filament_overrides: + override_types = sorted({o["type"] for o in filament_overrides if "type" in o}) + if override_types: + # Merge: keep original types for non-overridden slots, add override types + effective_types = sorted(set(required_types or []) | set(override_types)) + + # Cross-model safety gate (#2578): never hand a 3MF sliced + # for an incompatible model to a printer, no matter how the + # row got into the DB (old rows, direct API writes). Held + # as pending with an actionable waiting_reason — the user + # fixes it by editing the item's target model. + if not is_gcode_compatible(candidate.sliced_for, candidate.target_model): + per_model_reasons.append( + ( + candidate.target_model, + f"File was sliced for {candidate.sliced_for}, which is not compatible with " + f"{candidate.target_model} — edit the item and fix its target model", + ) + ) + skip_reasons["sliced_model_mismatch"] = skip_reasons.get("sliced_model_mismatch", 0) + 1 + continue + + match_id, match_reason = await self._find_idle_printer_for_model( db, - item.target_model, + candidate.target_model, busy_printers, effective_types, item.target_location, filament_overrides=filament_overrides, require_plate_clear=require_plate_clear, ) + if match_id: + printer_id = match_id + chosen = candidate + break + per_model_reasons.append((candidate.target_model, match_reason or "")) + + waiting_reason = None if printer_id else _collapse_waiting_reasons(per_model_reasons) + + # Fold the winning variant's file and settings onto the item + # before anything else looks at them — the guards below and + # every step of the dispatch read the item's own columns. + if chosen is not None: + self._resolve_variant(item, chosen) # Update waiting_reason if changed and send notification when first waiting if item.waiting_reason != waiting_reason: @@ -751,7 +911,7 @@ class PrintScheduler: job_name = await self._get_job_name(db, item) await notification_service.on_queue_job_waiting( job_name=job_name, - target_model=item.target_model, + target_model=_candidate_model_label(candidates) or item.target_model, waiting_reason=waiting_reason, db=db, ) @@ -1387,6 +1547,46 @@ class PrintScheduler: matches += 1 return matches + def _resolve_variant(self, item: PrintQueueItem, candidate: _ModelCandidate) -> None: + """Fold the winning candidate's file and settings onto the queue row (#671). + + This is the whole trick that keeps cross-model items cheap: the many-to-many + never escapes the selection loop. By the time the pass commits, the row + looks exactly like an ordinary single-file model-based item, so the upload, + archive creation, expected-print registration, print history and reprint + paths need no knowledge that variants exist. + + No-ops for a non-variant candidate, which is already the item's own columns. + + Safe to run and re-run: the item's file columns are only ever *read* when it + has no variants, so an item that gets resolved and then skipped (library-row + conflict, previous-print gate) is simply resolved again on the next pass. + """ + variant = candidate.variant + if variant is None: + return + + item.library_file_id = variant.library_file_id + item.library_file = variant.library_file + # The dispatcher checks archive_id first and would print that instead of + # the file we just picked. Creation refuses to combine the two, so this + # only ever fires on a hand-written row — clear it rather than silently + # dispatch something the matcher never considered. + item.archive_id = None + item.archive = None + + item.target_model = variant.target_model + item.plate_id = variant.plate_id + item.ams_mapping = variant.ams_mapping + item.nozzle_mapping = variant.nozzle_mapping + item.filament_overrides = variant.filament_overrides + item.required_filament_types = variant.required_filament_types + if variant.print_time_seconds is not None: + # The row carried the shortest candidate's estimate so SJF could order + # it before a printer was known; now that one is chosen, record what is + # actually going to run so history and the ETA agree with reality. + item.print_time_seconds = variant.print_time_seconds + async def _ensure_ams_mapping(self, db: AsyncSession, printer_id: int, item: PrintQueueItem) -> None: """Ensure the queue item carries a usable AMS mapping before dispatch. @@ -2975,6 +3175,22 @@ class PrintScheduler: library_file = result.scalar_one_or_none() if library_file: return library_file.filename.replace(".gcode.3mf", "").replace(".3mf", "") + # A cross-model item (#671) holds no file of its own until a printer is + # picked, so name it after its first candidate — otherwise every waiting + # notification for one reads "Job #12". Queried rather than read off + # item.variants because callers outside the selection loop have not + # eager-loaded them, and a lazy load raises in async. + first_variant_name = ( + await db.execute( + select(LibraryFile.filename) + .join(PrintQueueVariant, PrintQueueVariant.library_file_id == LibraryFile.id) + .where(PrintQueueVariant.queue_item_id == item.id) + .order_by(PrintQueueVariant.position, PrintQueueVariant.id) + .limit(1) + ) + ).scalar_one_or_none() + if first_variant_name: + return first_variant_name.replace(".gcode.3mf", "").replace(".3mf", "") return f"Job #{item.id}" async def _get_printer(self, db: AsyncSession, printer_id: int) -> Printer | None: @@ -3996,6 +4212,17 @@ class PrintScheduler: return "already_moved_on" item.dispatch_attempts = (item.dispatch_attempts or 0) + 1 item.started_at = None + # Charge the attempt to the candidate that was actually dispatched, so + # a cross-model item (#671) reaches for its other file next lap instead + # of retrying the printer that just failed to start. Matched by file + # because that is what the resolver copied onto the row. + if item.library_file_id is not None: + await db.execute( + update(PrintQueueVariant) + .where(PrintQueueVariant.queue_item_id == item.id) + .where(PrintQueueVariant.library_file_id == item.library_file_id) + .values(attempt_count=PrintQueueVariant.attempt_count + 1) + ) if command_rejected: # No retry budget for this one: the printer refused to verify the # command, and re-uploading the same 3MF to the same printer will diff --git a/backend/tests/unit/test_scheduler_cross_model_variants.py b/backend/tests/unit/test_scheduler_cross_model_variants.py new file mode 100644 index 000000000..bdd52785d --- /dev/null +++ b/backend/tests/unit/test_scheduler_cross_model_variants.py @@ -0,0 +1,456 @@ +"""Cross-model queue items — one job, several sliced files (#671). + +The reporter has an H2S and an H2C and does not care which one runs the job. +He slices it twice; both slices become variants of a single queue item, and the +scheduler takes the first whose model has an idle printer. + +The design constraint that shapes everything here: the many-to-many must never +escape the selection loop. Once a candidate wins, its file and settings are +folded onto the queue row, so the upload, archive creation, print history and +reprint paths keep seeing an ordinary single-file item. These tests assert both +halves — that the right candidate is picked, and that the row afterwards looks +like it was queued for that file all along. +""" + +from contextlib import ExitStack +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +import backend.app.models # noqa: F401 - populate Base.metadata +from backend.app.core.database import Base +from backend.app.models.library import LibraryFile +from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant +from backend.app.models.printer import Printer +from backend.app.services.print_scheduler import ( + PrintScheduler, + _candidate_model_label, + _candidates_for, + _collapse_waiting_reasons, +) + +# --------------------------------------------------------------------------- +# Candidate ordering — pure +# --------------------------------------------------------------------------- + + +def _fake_variant(*, vid, position, model, attempts=0): + return SimpleNamespace( + id=vid, + position=position, + target_model=model, + attempt_count=attempts, + library_file=SimpleNamespace(file_metadata={"sliced_for_model": model}), + required_filament_types=None, + filament_overrides=None, + ) + + +def _fake_item(variants): + return SimpleNamespace( + variants=variants, + target_model=None, + archive=None, + library_file=None, + required_filament_types=None, + filament_overrides=None, + ) + + +def test_no_variants_yields_the_items_own_columns(): + """The pre-#671 path must be provably unchanged: one candidate, built from + the item itself.""" + item = SimpleNamespace( + variants=[], + target_model="H2D", + archive=None, + library_file=SimpleNamespace(file_metadata={"sliced_for_model": "H2D"}), + required_filament_types='["PLA"]', + filament_overrides=None, + ) + candidates = _candidates_for(item) + assert len(candidates) == 1 + assert candidates[0].target_model == "H2D" + assert candidates[0].sliced_for == "H2D" + assert candidates[0].required_filament_types == '["PLA"]' + assert candidates[0].variant is None + + +def test_variants_come_back_in_user_priority_order(): + item = _fake_item( + [ + _fake_variant(vid=2, position=1, model="H2C"), + _fake_variant(vid=1, position=0, model="H2S"), + ] + ) + assert [c.target_model for c in _candidates_for(item)] == ["H2S", "H2C"] + + +def test_least_attempted_candidate_is_tried_first(): + """A printer that accepts the file and never starts must not eat the item's + whole retry budget — the alternative gets the next lap.""" + item = _fake_item( + [ + _fake_variant(vid=1, position=0, model="H2S", attempts=1), + _fake_variant(vid=2, position=1, model="H2C", attempts=0), + ] + ) + assert [c.target_model for c in _candidates_for(item)] == ["H2C", "H2S"] + + +def test_equal_attempts_fall_back_to_priority(): + """Once every candidate has failed equally often they cycle in the user's + order, so the item still reaches its DISPATCH_MAX_ATTEMPTS ceiling.""" + item = _fake_item( + [ + _fake_variant(vid=1, position=0, model="H2S", attempts=2), + _fake_variant(vid=2, position=1, model="H2C", attempts=2), + ] + ) + assert [c.target_model for c in _candidates_for(item)] == ["H2S", "H2C"] + + +# --------------------------------------------------------------------------- +# Waiting reasons — pure +# --------------------------------------------------------------------------- + + +def test_single_candidate_reason_is_unprefixed(): + """One candidate means the card already shows the model; prefixing it would + just be noise.""" + assert _collapse_waiting_reasons([("H2D", "Busy: H2D-1 (Printing)")]) == "Busy: H2D-1 (Printing)" + + +def test_identical_reasons_collapse_to_one_clause(): + collapsed = _collapse_waiting_reasons([("H2S", "Busy: shared-1 (Printing)"), ("H2C", "Busy: shared-1 (Printing)")]) + assert collapsed == "Busy: shared-1 (Printing)" + + +def test_all_busy_stays_busy_only_so_no_notification_fires(): + """Two models busy on differently-named printers still has to read as + busy-only. Labelling the clauses would make every pass over a cross-model + item look like it needs the user, when it just needs a printer to finish.""" + scheduler = PrintScheduler() + collapsed = _collapse_waiting_reasons([("H2S", "Busy: H2S-1 (Printing)"), ("H2C", "Busy: H2C-1 (Printing)")]) + assert collapsed == "Busy: H2S-1 (Printing) | Busy: H2C-1 (Printing)" + assert scheduler._is_busy_only(collapsed) + + +def test_differing_reasons_are_labelled_by_model(): + collapsed = _collapse_waiting_reasons([("H2S", "No PETG loaded"), ("H2C", "Busy: H2C-1 (Printing)")]) + assert collapsed == "H2S: No PETG loaded; H2C: Busy: H2C-1 (Printing)" + assert not PrintScheduler._is_busy_only(collapsed), "a real blocker must still notify" + + +def test_empty_reasons_are_dropped(): + assert _collapse_waiting_reasons([("H2S", "")]) is None + assert _collapse_waiting_reasons([]) is None + + +def test_model_label_names_every_candidate(): + item = _fake_item( + [ + _fake_variant(vid=1, position=0, model="H2S"), + _fake_variant(vid=2, position=1, model="H2C"), + ] + ) + assert _candidate_model_label(_candidates_for(item)) == "H2S or H2C" + + +# --------------------------------------------------------------------------- +# Scheduler behaviour +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def queue_db(): + """In-memory DB with one H2S and one H2C.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + session_maker = async_sessionmaker(engine, expire_on_commit=False) + + async with session_maker() as db: + db.add_all( + [ + Printer( + id=1, + name="H2S-1", + serial_number="H2S0001", + ip_address="10.0.0.1", + access_code="x", + model="H2S", + is_active=True, + ), + Printer( + id=2, + name="H2C-1", + serial_number="H2C0001", + ip_address="10.0.0.2", + access_code="x", + model="H2C", + is_active=True, + ), + ] + ) + await db.commit() + + try: + yield SimpleNamespace(session_maker=session_maker) + finally: + await engine.dispose() + + +async def _add_variant_item(ctx, specs): + """Seed one pending queue item with a variant per (model, overrides) spec.""" + async with ctx.session_maker() as db: + item = PrintQueueItem( + status="pending", + position=1, + target_model=specs[0]["model"], + ) + db.add(item) + await db.flush() + + for position, spec in enumerate(specs): + lib = LibraryFile( + filename=f"job_{spec['model']}.gcode.3mf", + file_path=f"/library/job_{spec['model']}.gcode.3mf", + file_size=10, + file_type="gcode.3mf", + file_metadata={"sliced_for_model": spec.get("sliced_for", spec["model"])}, + ) + db.add(lib) + await db.flush() + db.add( + PrintQueueVariant( + queue_item_id=item.id, + position=position, + library_file_id=lib.id, + target_model=spec["model"], + plate_id=spec.get("plate_id"), + ams_mapping=spec.get("ams_mapping"), + nozzle_mapping=spec.get("nozzle_mapping"), + print_time_seconds=spec.get("print_time_seconds"), + attempt_count=spec.get("attempts", 0), + ) + ) + await db.commit() + return item.id + + +async def _run_check_queue(ctx, scheduler, finder, waiting_notification=None): + patches = [ + patch("backend.app.services.print_scheduler.async_session", ctx.session_maker), + patch("backend.app.core.database.async_session", ctx.session_maker), + patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)), + patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)), + patch( + "backend.app.services.notification_service.notification_service.on_queue_job_waiting", + waiting_notification or AsyncMock(), + ), + patch( + "backend.app.services.notification_service.notification_service.on_queue_job_assigned", + AsyncMock(), + ), + patch.object(scheduler, "_find_idle_printer_for_model", finder), + patch.object(scheduler, "_check_auto_drying", AsyncMock()), + # Selection is what's under test — keep AMS recomputation and the + # filament-deficit probe out of the way, and never actually dispatch. + patch.object(scheduler, "_ensure_ams_mapping", AsyncMock()), + patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)), + patch.object(scheduler, "_launch_uploads", MagicMock()), + ] + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + return await scheduler.check_queue() + + +async def _get_item(ctx, item_id): + async with ctx.session_maker() as db: + return (await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one() + + +def _finder_for(available: dict[str, int]): + """Matcher that offers a printer only for the listed models.""" + + async def _find(db, model, exclude_ids, *args, **kwargs): + if model in available: + return available[model], None + return None, f"No idle {model} printer" + + return AsyncMock(side_effect=_find) + + +@pytest.mark.asyncio +async def test_first_matching_variant_wins_and_is_folded_onto_the_row(queue_db): + """The H2C is free, the H2S is not — the item runs the H2C slice, and every + downstream consumer sees a plain single-file item pointing at it.""" + item_id = await _add_variant_item( + queue_db, + [ + {"model": "H2S", "plate_id": 1, "ams_mapping": "[1]", "print_time_seconds": 900}, + { + "model": "H2C", + "plate_id": 3, + "ams_mapping": "[4, 5]", + "nozzle_mapping": "[0, 1]", + "print_time_seconds": 1200, + }, + ], + ) + scheduler = PrintScheduler() + + await _run_check_queue(queue_db, scheduler, _finder_for({"H2C": 2})) + + item = await _get_item(queue_db, item_id) + assert item.printer_id == 2, "assigned to the H2C" + assert item.target_model == "H2C" + assert item.plate_id == 3 + assert item.ams_mapping == "[4, 5]" + assert item.nozzle_mapping == "[0, 1]" + assert item.print_time_seconds == 1200, "the estimate now describes what will actually run" + assert item.waiting_reason is None + assert item.archive_id is None + + async with queue_db.session_maker() as db: + chosen = ( + await db.execute(select(PrintQueueVariant).where(PrintQueueVariant.target_model == "H2C")) + ).scalar_one() + assert item.library_file_id == chosen.library_file_id + + +@pytest.mark.asyncio +async def test_priority_order_decides_when_both_are_free(queue_db): + """Both printers idle in the same pass: the user's first choice runs, so the + outcome is reproducible rather than whichever match came back first.""" + item_id = await _add_variant_item(queue_db, [{"model": "H2S"}, {"model": "H2C"}]) + scheduler = PrintScheduler() + + await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1, "H2C": 2})) + + item = await _get_item(queue_db, item_id) + assert item.printer_id == 1 + assert item.target_model == "H2S" + + +@pytest.mark.asyncio +async def test_cross_model_gate_is_applied_per_candidate(queue_db): + """A variant whose file disagrees with its own model is skipped, and the + other one still runs — the gate must not condemn the whole item.""" + item_id = await _add_variant_item( + queue_db, + [ + {"model": "H2S", "sliced_for": "X1C"}, + {"model": "H2C"}, + ], + ) + scheduler = PrintScheduler() + finder = _finder_for({"H2S": 1, "H2C": 2}) + + await _run_check_queue(queue_db, scheduler, finder) + + assert [c.args[1] for c in finder.await_args_list] == ["H2C"], "the mismatched variant never reaches the matcher" + item = await _get_item(queue_db, item_id) + assert item.printer_id == 2 + assert item.target_model == "H2C" + + +@pytest.mark.asyncio +async def test_no_match_reports_every_model_it_tried(queue_db): + """Nothing is free: the user must be able to tell which machines were + considered, not just that "a printer" was unavailable.""" + item_id = await _add_variant_item(queue_db, [{"model": "H2S"}, {"model": "H2C"}]) + scheduler = PrintScheduler() + waiting = AsyncMock() + + await _run_check_queue(queue_db, scheduler, _finder_for({}), waiting) + + item = await _get_item(queue_db, item_id) + assert item.printer_id is None + assert item.status == "pending" + assert "H2S: No idle H2S printer" in item.waiting_reason + assert "H2C: No idle H2C printer" in item.waiting_reason + assert waiting.await_args.kwargs["target_model"] == "H2S or H2C" + # The item holds no file of its own yet — the alert still has to name the job. + assert waiting.await_args.kwargs["job_name"] == "job_H2S" + + +@pytest.mark.asyncio +async def test_item_with_no_files_left_is_held_with_an_actionable_reason(queue_db): + """Deleting a library file takes its variant with it. An item stripped of + every candidate used to sail into dispatch and die there on "No archive_id + or library_file_id"; hold it where the user can see why.""" + async with queue_db.session_maker() as db: + db.add(PrintQueueItem(status="pending", position=1, target_model="H2S")) + await db.commit() + scheduler = PrintScheduler() + finder = _finder_for({"H2S": 1}) + + await _run_check_queue(queue_db, scheduler, finder) + + finder.assert_not_awaited() + async with queue_db.session_maker() as db: + item = (await db.execute(select(PrintQueueItem))).scalar_one() + assert item.status == "pending" + assert item.printer_id is None + assert "has been deleted" in item.waiting_reason + + +@pytest.mark.asyncio +async def test_plain_model_based_item_is_untouched(queue_db): + """Regression guard: an item with no variants takes exactly the path it took + before variants existed.""" + async with queue_db.session_maker() as db: + lib = LibraryFile( + filename="job.gcode.3mf", + file_path="/library/job.gcode.3mf", + file_size=10, + file_type="gcode.3mf", + file_metadata={"sliced_for_model": "H2S"}, + ) + db.add(lib) + await db.flush() + db.add( + PrintQueueItem( + status="pending", + position=1, + target_model="H2S", + library_file_id=lib.id, + plate_id=2, + ) + ) + await db.commit() + scheduler = PrintScheduler() + + await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1})) + + async with queue_db.session_maker() as db: + item = (await db.execute(select(PrintQueueItem))).scalar_one() + assert item.printer_id == 1 + assert item.target_model == "H2S" + assert item.plate_id == 2, "nothing overwrote the item's own settings" + + +@pytest.mark.asyncio +async def test_failed_candidate_steps_aside_for_the_alternative(queue_db): + """The H2S burned an attempt on the last lap. Both are free now — the H2C + goes first, which is the entire point of queueing an alternative.""" + item_id = await _add_variant_item( + queue_db, + [ + {"model": "H2S", "attempts": 1}, + {"model": "H2C", "attempts": 0}, + ], + ) + scheduler = PrintScheduler() + + await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1, "H2C": 2})) + + item = await _get_item(queue_db, item_id) + assert item.printer_id == 2 + assert item.target_model == "H2C"