bambuddy/backend/tests/unit/test_scheduler_cleanup_library.py

import json
import logging
from contextlib import ExitStack
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from sqlalchemy import event, select
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

import backend.app.models  # noqa: F401 - populate Base.metadata
import backend.app.services.print_scheduler as scheduler_module
from backend.app.core.database import Base
from backend.app.models.archive import PrintArchive
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
from backend.app.utils.archive_paths import archive_photos_dir
from backend.app.utils.library_paths import library_photos_dir
from backend.tests._fixtures.background_tasks import discarding_spawn_patch


@pytest.fixture
async def queue_factory(tmp_path):
    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)
    case_counter = 0

    async def make_case(
        *,
        cleanup=True,
        is_external=False,
        thumbnail_path=None,
        siblings=(),
        photos=(),
        source_type=None,
        slicer_ams_mapping=None,
    ):
        nonlocal case_counter
        case_counter += 1

        base_dir = tmp_path / f"case-{case_counter}"
        base_dir.mkdir()
        archive_dir = base_dir / "archive"
        source_path = base_dir / "library" / f"source-{case_counter}.3mf"
        source_path.parent.mkdir()
        source_path.write_bytes(b"library source")

        thumbnail_actual_path = None
        thumbnail_db_path = None
        if thumbnail_path == "relative":
            thumbnail_db_path = f"thumbs/preview-{case_counter}.png"
            thumbnail_actual_path = base_dir / thumbnail_db_path
        elif thumbnail_path == "absolute":
            thumbnail_actual_path = tmp_path / f"absolute-preview-{case_counter}.png"
            thumbnail_db_path = str(thumbnail_actual_path)
        elif thumbnail_path is not None:
            thumbnail_actual_path = Path(thumbnail_path)
            thumbnail_db_path = str(thumbnail_path)

        if thumbnail_actual_path:
            thumbnail_actual_path.parent.mkdir(parents=True, exist_ok=True)
            thumbnail_actual_path.write_bytes(b"thumbnail")

        async with session_maker() as db:
            printer = Printer(
                name=f"Printer {case_counter}",
                serial_number=f"SERIAL-{case_counter}",
                ip_address="127.0.0.1",
                access_code="access-code",
                model="X1C",
            )
            library_file = LibraryFile(
                filename=f"source-{case_counter}.3mf",
                file_path=str(source_path),
                file_type="3mf",
                file_size=source_path.stat().st_size,
                file_hash=None,
                thumbnail_path=thumbnail_db_path,
                file_metadata=None,
                is_external=is_external,
                photos=list(photos) or None,
                source_type=source_type,
            )
            db.add_all([printer, library_file])
            await db.flush()

            # Photos of the printed result (#3077). Written through the real
            # helper so the test cannot drift from the layout the code uses.
            photos_dir = None
            if photos:
                with patch.object(scheduler_module.settings, "archive_dir", archive_dir):
                    photos_dir = library_photos_dir(library_file.id)
                photos_dir.mkdir(parents=True, exist_ok=True)
                for name in photos:
                    (photos_dir / name).write_bytes(f"photo {name}".encode())

            item = PrintQueueItem(
                printer_id=printer.id,
                library_file_id=library_file.id,
                status="pending",
                cleanup_library_after_dispatch=cleanup,
                bed_levelling="on",
                flow_cali="off",
                vibration_cali=True,
                layer_inspect=False,
                timelapse=False,
                use_ams=True,
                nozzle_offset_cali="on",
                # A callable gets the printer's id (only known here).
                slicer_ams_mapping=(
                    slicer_ams_mapping(printer.id) if callable(slicer_ams_mapping) else slicer_ams_mapping
                ),
            )
            db.add(item)
            await db.flush()

            # The other copies of a quantity>1 dispatch (#2819). Each entry is a
            # dict of overrides: `status`, `own_archive` for a copy that already
            # holds one, and `extra_variant` for a cross-model copy that keeps a
            # candidate this cleanup does not consume.
            sibling_ids = []
            other_file = None
            for spec in siblings:
                sibling = PrintQueueItem(
                    printer_id=printer.id,
                    library_file_id=None if spec.get("variants") else library_file.id,
                    status=spec.get("status", "pending"),
                    cleanup_library_after_dispatch=cleanup,
                )
                if spec.get("own_archive"):
                    own = PrintArchive(
                        printer_id=printer.id,
                        filename="already-dispatched.3mf",
                        file_path="archives/already-dispatched.3mf",
                        file_size=1,
                        status="printing",
                    )
                    db.add(own)
                    await db.flush()
                    sibling.archive_id = own.id
                db.add(sibling)
                await db.flush()
                if spec.get("variants"):
                    db.add(
                        PrintQueueVariant(
                            queue_item_id=sibling.id,
                            library_file_id=library_file.id,
                            target_model="X1C",
                            position=0,
                        )
                    )
                    if spec.get("extra_variant"):
                        if other_file is None:
                            other_path = base_dir / "library" / f"other-{case_counter}.3mf"
                            other_path.write_bytes(b"other source")
                            other_file = LibraryFile(
                                filename=f"other-{case_counter}.3mf",
                                file_path=str(other_path),
                                file_type="3mf",
                                file_size=other_path.stat().st_size,
                            )
                            db.add(other_file)
                            await db.flush()
                        db.add(
                            PrintQueueVariant(
                                queue_item_id=sibling.id,
                                library_file_id=other_file.id,
                                target_model="P1S",
                                position=1,
                            )
                        )
                sibling_ids.append(sibling.id)

            await db.commit()

            return SimpleNamespace(
                session_maker=session_maker,
                base_dir=base_dir,
                archive_dir=archive_dir,
                source_path=source_path,
                photos_dir=photos_dir,
                photo_names=list(photos),
                thumbnail_path=thumbnail_actual_path,
                printer_id=printer.id,
                library_file_id=library_file.id,
                queue_item_id=item.id,
                sibling_ids=sibling_ids,
                other_library_file_id=other_file.id if other_file is not None else None,
                archive_path=None,
                archive_kwargs=None,
                upload=AsyncMock(return_value=True),
                start_print=MagicMock(return_value=True),
            )

    try:
        yield make_case
    finally:
        await engine.dispose()


async def _dispatch_library_item(
    ctx,
    *,
    archive_failure=False,
    unlink_side_effect=None,
    cleanup_commit_failure=False,
    photo_commit_failure=False,
):
    scheduler = PrintScheduler()

    async def archive_print(
        self,
        *,
        printer_id,
        source_file,
        original_filename,
        created_by_id=None,
        project_id=None,
        cost_center_id=None,
        plate_id=None,
        library_file_id=None,
        **extra,
    ):
        ctx.archive_kwargs = extra
        if archive_failure:
            raise RuntimeError("archive copy failed")

        archive_rel_path = Path("archives") / f"archive-{ctx.queue_item_id}.3mf"
        ctx.archive_path = ctx.base_dir / archive_rel_path
        ctx.archive_path.parent.mkdir(parents=True, exist_ok=True)
        ctx.archive_path.write_bytes(Path(source_file).read_bytes())

        archive = PrintArchive(
            printer_id=printer_id,
            filename=original_filename,
            file_path=str(archive_rel_path),
            file_size=ctx.archive_path.stat().st_size,
            content_hash=None,
            thumbnail_path=None,
            timelapse_path=None,
            print_time_seconds=120,
            status="completed",
            project_id=project_id,
            library_file_id=library_file_id,
            created_by_id=created_by_id,
        )
        self.db.add(archive)
        await self.db.flush()
        return archive

    patches = [
        patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
        patch.object(scheduler_module.settings, "archive_dir", ctx.archive_dir),
        patch("backend.app.services.archive.ArchiveService.archive_print", new=archive_print),
        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.print_scheduler.printer_manager.start_print", ctx.start_print),
        patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
        patch(
            "backend.app.services.print_scheduler.get_ftp_retry_settings", AsyncMock(return_value=(False, 0, 0, 1.0))
        ),
        patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
        patch("backend.app.services.print_scheduler.upload_file_async", ctx.upload),
        patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
        discarding_spawn_patch(),
        patch("backend.app.services.notification_service.notification_service.on_queue_job_started", AsyncMock()),
        patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
        patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
        patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
        patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
    ]
    if unlink_side_effect:
        patches.append(patch.object(type(ctx.source_path), "unlink", unlink_side_effect))

    with ExitStack() as stack:
        for patcher in patches:
            stack.enter_context(patcher)

        async with ctx.session_maker() as db:
            if cleanup_commit_failure:
                _arm_commit_failure_on_library_delete(db)
            if photo_commit_failure:
                _arm_commit_failure_on_photo_move(db, stack)
            item = await db.get(PrintQueueItem, ctx.queue_item_id)
            await scheduler._start_print(db, item)


def _fail_inside_the_next_flush(db, statement):
    """Make the next flush on this session fail, once, the way SQLite does.

    Raising *instead of* calling `db.commit()` does not reproduce a failed
    commit and is not the dangerous case: the session stays ACTIVE, nothing is
    expired, and every loaded instance still reads out of `__dict__`. What a
    busy writer actually gives you is a statement error raised inside the
    flush — SQLite takes the write lock at the first DML statement, not at
    COMMIT, so "database is locked" surfaces there (#1853). SQLAlchemy rolls
    that back internally through `safe_reraise` before re-raising, which
    expires every loaded instance and leaves the session in pending-rollback
    state: the next ORM attribute read raises PendingRollbackError, *before*
    the handler's own rollback can run. That is the state a handler on this
    path has to survive, so it is the state these tests have to produce.
    """
    sync_session = db.sync_session
    fired = False

    def after_flush(session, flush_context):
        nonlocal fired
        if fired:
            return
        fired = True
        raise OperationalError(statement, {}, Exception("database is locked"))

    event.listen(sync_session, "after_flush", after_flush)


def _arm_commit_failure_on_library_delete(db):
    """Make the one commit that removes the library row fail, once.

    Stands in for the "database is locked" cascades the commit's own comment
    cites (#1853). Armed by the delete rather than by a call count so it
    cannot drift onto a different commit.
    """
    original_delete = db.delete
    original_commit = db.commit
    armed = False

    async def delete(obj):
        nonlocal armed
        if isinstance(obj, LibraryFile):
            armed = True
        return await original_delete(obj)

    async def commit():
        nonlocal armed
        if armed:
            armed = False
            _fail_inside_the_next_flush(db, "DELETE FROM library_files WHERE library_files.id = ?")
        return await original_commit()

    db.delete = delete
    db.commit = commit


def _arm_commit_failure_on_photo_move(db, stack):
    """Make the commit that records the carried photos fail, once.

    The second commit of this path (#3077): the archive and the delete are
    already committed, the pictures are already on disk under the archive,
    and only `archive.photos` is pending. Armed by the move itself so it
    cannot drift onto the delete's commit.
    """
    original_commit = db.commit
    original_move = scheduler_module.move_library_photos
    armed = False

    def move_library_photos(file_id, photos, destination):
        nonlocal armed
        carried = original_move(file_id, photos, destination)
        armed = bool(carried)
        return carried

    async def commit():
        nonlocal armed
        if armed:
            armed = False
            _fail_inside_the_next_flush(db, "UPDATE print_archives SET photos=? WHERE print_archives.id = ?")
        return await original_commit()

    stack.enter_context(patch.object(scheduler_module, "move_library_photos", move_library_photos))
    db.commit = commit


async def _queue_snapshot(ctx):
    async with ctx.session_maker() as db:
        item = await db.get(PrintQueueItem, ctx.queue_item_id)
        library_file = await db.get(LibraryFile, ctx.library_file_id)
        archive = await db.get(PrintArchive, item.archive_id) if item.archive_id else None
        return item, library_file, archive


@pytest.mark.asyncio
async def test_cleanup_unlinks_library_file_and_removes_db_row(queue_factory):
    ctx = await queue_factory(cleanup=True)

    await _dispatch_library_item(ctx)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "printing"
    assert item.library_file_id is None
    assert item.archive_id == archive.id
    assert library_file is None
    assert not ctx.source_path.exists()


@pytest.mark.asyncio
async def test_external_library_file_skips_cleanup(queue_factory):
    ctx = await queue_factory(cleanup=True, is_external=True)

    await _dispatch_library_item(ctx)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "printing"
    assert item.library_file_id == ctx.library_file_id
    assert item.archive_id == archive.id
    assert library_file is not None
    assert ctx.source_path.exists()


@pytest.mark.asyncio
async def test_cleanup_moves_the_photos_into_the_archive(queue_factory):
    """Photos follow the consumed file into the archive that replaces it (#3077).

    The row is hard-deleted here, so leaving the photo directory alone
    orphaned it under an id nothing points at any more — and the pictures
    of a print that still has a record disappeared from the UI.
    """
    ctx = await queue_factory(cleanup=True, photos=["a1b2c3d4.jpg", "e5f6a7b8.png"])

    await _dispatch_library_item(ctx)

    _, library_file, archive = await _queue_snapshot(ctx)
    assert library_file is None
    assert not ctx.photos_dir.exists()
    assert archive.photos == ctx.photo_names
    with patch.object(scheduler_module.settings, "base_dir", ctx.base_dir):
        destination = archive_photos_dir(archive)
    for name in ctx.photo_names:
        assert (destination / name).read_bytes() == f"photo {name}".encode()


@pytest.mark.asyncio
async def test_external_library_file_keeps_its_photos(queue_factory):
    ctx = await queue_factory(cleanup=True, is_external=True, photos=["a1b2c3d4.jpg"])

    await _dispatch_library_item(ctx)

    _, library_file, archive = await _queue_snapshot(ctx)
    assert library_file is not None
    assert (ctx.photos_dir / "a1b2c3d4.jpg").is_file()
    assert archive.photos is None


@pytest.mark.asyncio
async def test_archive_creation_failure_keeps_the_photos(queue_factory):
    ctx = await queue_factory(cleanup=True, photos=["a1b2c3d4.jpg"])

    await _dispatch_library_item(ctx, archive_failure=True)

    _, library_file, archive = await _queue_snapshot(ctx)
    assert archive is None
    assert library_file is not None
    assert (ctx.photos_dir / "a1b2c3d4.jpg").is_file()


@pytest.mark.asyncio
async def test_cleanup_commit_failure_keeps_the_photos_with_the_library_file(queue_factory):
    """The photos move after the delete commits, not before it (#3077).

    The commit that removes the library row can fail; the except branch rolls
    it back and the file is in the library again. Photos moved ahead of that
    commit would be gone from under it — the row would name a directory that
    no longer exists, and the pictures would sit under an archive that was
    rolled back too.
    """
    ctx = await queue_factory(cleanup=True, photos=["a1b2c3d4.jpg"])

    await _dispatch_library_item(ctx, cleanup_commit_failure=True)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "failed"
    assert archive is None
    assert library_file is not None
    assert library_file.photos == ["a1b2c3d4.jpg"]
    assert (ctx.photos_dir / "a1b2c3d4.jpg").read_bytes() == b"photo a1b2c3d4.jpg"
    assert not (ctx.base_dir / "archives" / "photos").exists()


@pytest.mark.asyncio
async def test_photo_commit_failure_still_dispatches_the_print(queue_factory, caplog):
    """A failed photos commit must not take the dispatch down with it (#3077).

    The archive and the delete are committed by then, so the print goes ahead
    and the pictures sit unnamed under the archive.

    The commit fails inside the flush, which is where a locked SQLite fails —
    see `_fail_inside_the_next_flush`. That expires every loaded instance
    twice over: once by SQLAlchemy's internal rollback, before the handler
    runs at all, and again at the handler's own `rollback()`. So the handler
    may not read an ORM attribute on either side of that rollback. Before it,
    a read raises PendingRollbackError; after it, MissingGreenlet — the nozzle
    guard's `archive.nozzle_diameter` and the upload's `printer.name` are the
    ones that used to die.
    """
    ctx = await queue_factory(cleanup=True, photos=["a1b2c3d4.jpg"])

    with caplog.at_level(logging.WARNING, logger="backend.app.services.print_scheduler"):
        await _dispatch_library_item(ctx, photo_commit_failure=True)

    item, library_file, archive = await _queue_snapshot(ctx)
    # The handler absorbed it rather than the failure being skipped: it names
    # the queue item and the archive from ints it held before the commit.
    assert any(
        f"Queue item {ctx.queue_item_id}: failed to carry library photos into archive {item.archive_id}"
        in record.message
        for record in caplog.records
    )
    assert item.status == "printing"
    assert item.archive_id == archive.id
    assert library_file is None
    assert not archive.photos
    ctx.upload.assert_awaited()
    ctx.start_print.assert_called()
    with patch.object(scheduler_module.settings, "base_dir", ctx.base_dir):
        destination = archive_photos_dir(archive)
    assert (destination / "a1b2c3d4.jpg").read_bytes() == b"photo a1b2c3d4.jpg"


@pytest.mark.asyncio
async def test_archive_creation_failure_skips_cleanup_and_dispatch(queue_factory):
    ctx = await queue_factory(cleanup=True, thumbnail_path="relative")

    await _dispatch_library_item(ctx, archive_failure=True)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "failed"
    assert item.error_message == "Failed to create archive from library file"
    assert item.archive_id is None
    assert archive is None
    assert library_file is not None
    assert ctx.source_path.exists()
    assert ctx.thumbnail_path.exists()
    ctx.upload.assert_not_awaited()
    ctx.start_print.assert_not_called()


@pytest.mark.parametrize("thumbnail_path", ["absolute", "relative"])
@pytest.mark.asyncio
async def test_cleanup_resolves_absolute_and_relative_thumbnail_paths(queue_factory, thumbnail_path):
    ctx = await queue_factory(cleanup=True, thumbnail_path=thumbnail_path)

    await _dispatch_library_item(ctx)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "printing"
    assert item.archive_id == archive.id
    assert library_file is None
    assert not ctx.source_path.exists()
    assert not ctx.thumbnail_path.exists()


@pytest.mark.asyncio
async def test_archive_copy_survives_library_cleanup(queue_factory):
    ctx = await queue_factory(cleanup=True)

    await _dispatch_library_item(ctx)

    assert not ctx.source_path.exists()
    assert ctx.archive_path.exists()
    assert ctx.archive_path.read_bytes() == b"library source"
    uploaded_path = ctx.upload.await_args.args[2]
    assert uploaded_path == ctx.archive_path


async def _sibling_snapshot(ctx):
    async with ctx.session_maker() as db:
        return [await db.get(PrintQueueItem, sid) for sid in ctx.sibling_ids]


async def _variant_files(ctx, sibling_id):
    async with ctx.session_maker() as db:
        rows = await db.execute(
            select(PrintQueueVariant.library_file_id).where(PrintQueueVariant.queue_item_id == sibling_id)
        )
        return sorted(rows.scalars().all())


# ---------------------------------------------------------------------------
# Sibling copies of the same library row (#2819)
#
# `quantity > 1` on the printer-card upload-and-print flow puts the cleanup flag
# on every copy, and batch clones inherit `library_file_id`. Consuming the row
# for the first copy used to leave the others pointing at it, which failed with
# "Library file not found" on SQLite and deleted the rows outright on
# PostgreSQL, where the FK cascade is enforced.
#
# These run on SQLite, so they cover the orphan half directly. The cascade half
# was verified by hand against a real PostgreSQL 16, building this same fixture
# on both backends and comparing every row: without the fix the copies were gone
# after the delete -- including the finished ones a batch order counts its
# progress from, and a copy already printing from its own archive. With it, the
# two backends agree row for row. `print_archives.library_file_id` is SET NULL,
# so it is cleared by the same delete, which is why looking the archive up by
# the consumed library id -- the obvious alternative fix -- cannot work there.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_pending_copies_are_repointed_at_the_archive(queue_factory):
    ctx = await queue_factory(cleanup=True, siblings=({}, {}))

    await _dispatch_library_item(ctx)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert library_file is None
    for sibling in await _sibling_snapshot(ctx):
        # Still queued -- the point is that they can now run, not that they run now.
        assert sibling.status == "pending"
        assert sibling.archive_id == archive.id
        assert sibling.library_file_id is None
        # Their file is already consumed; leaving this armed would delete
        # whatever library row they were next given.
        assert sibling.cleanup_library_after_dispatch is False


@pytest.mark.asyncio
async def test_copy_that_already_has_its_own_archive_keeps_it(queue_factory):
    ctx = await queue_factory(cleanup=True, siblings=({"own_archive": True, "status": "printing"},))

    await _dispatch_library_item(ctx)

    _, _, archive = await _queue_snapshot(ctx)
    (sibling,) = await _sibling_snapshot(ctx)
    # It is mid-print from its own archive and does not need the library file.
    # Re-pointing it would swap the file under a job already running.
    assert sibling.archive_id != archive.id
    assert sibling.status == "printing"
    # Cleared all the same: on PostgreSQL a row still naming the file goes with
    # it, and this one is a job that is currently printing.
    assert sibling.library_file_id is None


@pytest.mark.parametrize("status", ["completed", "failed", "cancelled", "aborted"])
@pytest.mark.asyncio
async def test_finished_copies_keep_their_outcome_and_survive_the_delete(queue_factory, status):
    ctx = await queue_factory(cleanup=True, siblings=({"status": status},))

    await _dispatch_library_item(ctx)

    (sibling,) = await _sibling_snapshot(ctx)
    # A finished row is a record of what happened, not a spare part -- it keeps
    # its outcome and is not handed the archive.
    assert sibling.status == status
    assert sibling.archive_id is None
    # But the reference has to go: it is the only thing tying the row to the
    # cascade that would otherwise delete it, and a batch order counts its
    # progress from rows exactly like this one.
    assert sibling.library_file_id is None


@pytest.mark.asyncio
async def test_skipped_copy_is_repointed_because_it_can_come_back(queue_factory):
    ctx = await queue_factory(cleanup=True, siblings=({"status": "skipped"},))

    await _dispatch_library_item(ctx)

    _, _, archive = await _queue_snapshot(ctx)
    (sibling,) = await _sibling_snapshot(ctx)
    # Clearing the printer's previous-success gate puts skipped items back to
    # pending, so this one is only waiting -- not finished.
    assert sibling.status == "skipped"
    assert sibling.archive_id == archive.id
    assert sibling.library_file_id is None


@pytest.mark.asyncio
async def test_copies_are_untouched_when_the_dispatch_does_not_consume_the_file(queue_factory):
    ctx = await queue_factory(cleanup=False, siblings=({},))

    await _dispatch_library_item(ctx)

    item, library_file, _ = await _queue_snapshot(ctx)
    assert library_file is not None
    (sibling,) = await _sibling_snapshot(ctx)
    assert sibling.library_file_id == ctx.library_file_id
    assert sibling.archive_id is None


@pytest.mark.asyncio
async def test_cross_model_copy_keeps_its_other_candidate_instead_of_the_archive(queue_factory):
    ctx = await queue_factory(cleanup=True, siblings=({"variants": True, "extra_variant": True},))

    await _dispatch_library_item(ctx)

    (sibling,) = await _sibling_snapshot(ctx)
    # It still has somewhere to go, and that candidate carries its own target
    # model -- pointing it at this archive would print a file the matcher never
    # chose.
    assert sibling.archive_id is None
    assert sibling.library_file_id is None
    assert await _variant_files(ctx, sibling.id) == [ctx.other_library_file_id]


@pytest.mark.asyncio
async def test_copy_whose_only_candidate_was_consumed_is_repointed(queue_factory):
    ctx = await queue_factory(cleanup=True, siblings=({"variants": True},))

    await _dispatch_library_item(ctx)

    _, _, archive = await _queue_snapshot(ctx)
    (sibling,) = await _sibling_snapshot(ctx)
    # Its one candidate is gone. Without the re-point the resolver would hold it
    # pending forever with nothing left to dispatch.
    assert sibling.archive_id == archive.id
    assert sibling.library_file_id is None
    assert await _variant_files(ctx, sibling.id) == []


@pytest.mark.asyncio
async def test_oserror_during_unlink_logs_orphan_path_and_does_not_crash_dispatch(queue_factory, caplog):
    ctx = await queue_factory(cleanup=True, thumbnail_path="relative")
    original_unlink = type(ctx.source_path).unlink

    def unlink_with_source_failure(path, *args, **kwargs):
        if Path(path) == ctx.source_path:
            raise OSError("permission denied")
        return original_unlink(path, *args, **kwargs)

    with caplog.at_level("WARNING", logger="backend.app.services.print_scheduler"):
        await _dispatch_library_item(ctx, unlink_side_effect=unlink_with_source_failure)

    item, library_file, archive = await _queue_snapshot(ctx)
    assert item.status == "printing"
    assert item.archive_id == archive.id
    assert item.library_file_id is None
    assert library_file is None
    assert ctx.source_path.exists()
    assert not ctx.thumbnail_path.exists()
    assert ctx.archive_path.exists()
    assert "TRANSIENT_LIBRARY_FILE_ORPHAN" in caplog.text
    assert str(ctx.source_path) in caplog.text
    assert "permission denied" in caplog.text


@pytest.mark.asyncio
async def test_lost_k_profiles_are_restored_before_the_print_starts(queue_factory):
    """A slot reset to the default K must be fixed before the job goes out (#3219)."""
    ctx = await queue_factory(cleanup=False)
    order = []
    reapply = AsyncMock(side_effect=lambda *a, **k: order.append("reapply") or 0)
    ctx.start_print.side_effect = lambda *a, **k: order.append("start") or True

    with patch.object(scheduler_module.kprofile_drift, "reapply_lost_kprofiles", reapply):
        await _dispatch_library_item(ctx)

    assert order == ["reapply", "start"]
    # No stored mapping: every loaded tray is a candidate, and no retry wait.
    reapply.assert_awaited_once_with(ctx.printer_id, None, throttle=False)


@pytest.mark.asyncio
async def test_a_failing_k_profile_check_does_not_stop_the_print(queue_factory):
    ctx = await queue_factory(cleanup=False)
    reapply = AsyncMock(side_effect=RuntimeError("boom"))

    with patch.object(scheduler_module.kprofile_drift, "reapply_lost_kprofiles", reapply):
        await _dispatch_library_item(ctx)

    ctx.start_print.assert_called_once()


# A virtual printer's upload is a File Manager file (source_type
# "virtual_printer") that its queue items print. The archive is made at
# dispatch, so that is where the upload's slicer AMS pick (#2700) and the
# virtual printer's archive-name setting reach it.


@pytest.mark.asyncio
@pytest.mark.parametrize("moved", [False, True], ids=["same-printer", "moved-to-another"])
async def test_a_vp_upload_hands_its_slicer_pick_to_the_archive(queue_factory, moved):
    # The pick stays tagged with the printer the slicer resolved it on, also
    # when the job was moved to another printer before it printed: its tray
    # numbers mean nothing there, so a reprint there must not reuse it.
    resolved_on = 4242 if moved else None
    ctx = await queue_factory(
        cleanup=False,
        source_type="virtual_printer",
        slicer_ams_mapping=lambda printer_id: json.dumps(
            {"mapping": [4, -1, 12], "printer_id": resolved_on or printer_id}
        ),
    )

    await _dispatch_library_item(ctx)

    assert ctx.archive_kwargs["slicer_ams_mapping"] == [4, -1, 12]
    assert ctx.archive_kwargs["slicer_ams_mapping_printer_id"] == (resolved_on or ctx.printer_id)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("source_type", "setting", "expected"),
    [
        ("virtual_printer", "filename", {"prefer_filename_for_name": True}),
        ("virtual_printer", "metadata", {}),
        (None, "filename", {}),
    ],
    ids=["vp-filename", "vp-metadata", "other-file"],
)
async def test_the_vp_name_setting_names_the_archive_of_its_uploads_only(queue_factory, source_type, setting, expected):
    ctx = await queue_factory(cleanup=False, source_type=source_type)

    with patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value=setting)):
        await _dispatch_library_item(ctx)

    assert ctx.archive_kwargs == expected