bambuddy/backend/tests/integration/test_pipeline_runs_api.py

"""Integration tests for Slicer Pipeline runs (#1425 PR B).

Slicing itself is a network call to the slicer sidecar — these tests
stub ``slice_and_persist`` so the orchestration logic is exercised without
needing a live sidecar in CI.
"""

from __future__ import annotations

from unittest.mock import AsyncMock, patch

import pytest
from httpx import AsyncClient


def _pipeline_payload(**overrides) -> dict:
    payload = {
        "name": "Production Batch",
        "description": None,
        "printer_preset": {"source": "local", "id": "1"},
        "process_preset": {"source": "local", "id": "2"},
        "filament_presets": [{"source": "local", "id": "3"}],
        "bed_type": None,
    }
    payload.update(overrides)
    return payload


@pytest.fixture
async def pipeline_factory(async_client: AsyncClient):
    """Create pipelines via the API + optionally set a target printer."""

    async def _make(target_printer_id: int | None = None, **overrides) -> dict:
        resp = await async_client.post("/api/v1/slicer-pipelines/", json=_pipeline_payload(**overrides))
        assert resp.status_code == 201, resp.text
        pipeline = resp.json()
        if target_printer_id is not None:
            put_resp = await async_client.put(
                f"/api/v1/slicer-pipelines/{pipeline['id']}",
                json={"target_kind": "specific_printer", "target_printer_id": target_printer_id},
            )
            assert put_resp.status_code == 200, put_resp.text
            pipeline = put_resp.json()
        return pipeline

    return _make


@pytest.fixture
async def printer_factory(db_session):
    """Insert a Printer row for tests that need a target_printer_id."""
    from backend.app.models.printer import Printer

    counter = [0]

    async def _make(**overrides) -> Printer:
        counter[0] += 1
        defaults = {
            "name": f"X1C #{counter[0]}",
            "serial_number": f"SERIAL{counter[0]:04d}",
            "ip_address": "192.0.2.1",
            "access_code": "ABCD1234",
            "model": "Bambu Lab X1 Carbon",
            "is_active": True,
        }
        defaults.update(overrides)
        printer = Printer(**defaults)
        db_session.add(printer)
        await db_session.commit()
        await db_session.refresh(printer)
        return printer

    return _make


@pytest.fixture
async def library_file_factory(db_session):
    """Insert a LibraryFile row for tests that need a source_library_file_id."""
    from pathlib import Path

    from backend.app.core.config import settings as app_settings
    from backend.app.models.library import LibraryFile

    counter = [0]

    async def _make(**overrides) -> LibraryFile:
        counter[0] += 1
        # Materialise an empty file on disk so the orchestration's path-exists
        # guard passes when tests reach it.
        rel = f"test_pipeline_run_{counter[0]}.3mf"
        abs_path = Path(app_settings.base_dir) / rel
        abs_path.parent.mkdir(parents=True, exist_ok=True)
        abs_path.write_bytes(b"")
        defaults = {
            "filename": f"cube_{counter[0]}.3mf",
            "file_path": rel,
            "file_type": "3mf",
            "file_size": 0,
            "file_hash": f"hash_{counter[0]}",
            "source_type": "uploaded",
        }
        defaults.update(overrides)
        row = LibraryFile(**defaults)
        db_session.add(row)
        await db_session.commit()
        await db_session.refresh(row)
        return row

    return _make


class TestSlicerPipelineTarget:
    """PUT /slicer-pipelines/{id} accepts the new target fields."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_writes_target(self, async_client: AsyncClient, pipeline_factory, printer_factory):
        printer = await printer_factory()
        pipeline = await pipeline_factory()
        resp = await async_client.put(
            f"/api/v1/slicer-pipelines/{pipeline['id']}",
            json={"target_kind": "specific_printer", "target_printer_id": printer.id},
        )
        assert resp.status_code == 200, resp.text
        updated = resp.json()
        assert updated["target_kind"] == "specific_printer"
        assert updated["target_printer_id"] == printer.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_target_printer_id_zero_clears(
        self, async_client: AsyncClient, pipeline_factory, printer_factory
    ):
        """Empty-select dropdown sends target_printer_id=0 → backend treats
        as 'clear' rather than referencing printer #0 (which doesn't exist)."""
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        resp = await async_client.put(
            f"/api/v1/slicer-pipelines/{pipeline['id']}",
            json={"target_printer_id": 0},
        )
        assert resp.status_code == 200, resp.text
        assert resp.json()["target_printer_id"] is None


class TestCheckEligibility:
    """POST /slicer-pipelines/{id}/check-eligibility surfaces structured issues."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_target_set(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        library_file_factory,
    ):
        pipeline = await pipeline_factory()  # no target set
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
            json={"source_library_file_id": src.id},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["ok"] is False
        kinds = [i["kind"] for i in body["issues"]]
        # PR A defaults target_kind to 'printer_class' so a freshly-saved
        # pipeline with no target_model_class surfaces ``class_not_set``; the
        # PR B UI path that hadn't pinned a target_printer_id would surface
        # ``printer_not_set``. Both signal the same thing to the operator;
        # accept either.
        assert kinds == ["class_not_set"] or kinds == ["printer_not_set"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_printer_disabled(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        printer = await printer_factory(is_active=False)
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        with patch("backend.app.api.routes.pipeline_runs._load_printer_status", new=AsyncMock(return_value=None)):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
                json={"source_library_file_id": src.id},
            )
        assert resp.status_code == 200
        body = resp.json()
        kinds = [i["kind"] for i in body["issues"]]
        assert "printer_disabled" in kinds
        # printer_offline also fires because get_status returns None — both
        # issues are expected and both block.
        assert "printer_offline" in kinds
        assert body["ok"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_online_match_clears_issues(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """Patch printer_manager so AMS slot 0 carries the same canonical
        type the pipeline's local-tier filament preset declares."""
        from backend.app.models.local_preset import LocalPreset

        preset = LocalPreset(
            name="My PLA",
            preset_type="filament",
            source="manual",
            setting="{}",
            filament_type="PLA",
            default_filament_colour="#FFFFFF",
        )
        db_session.add(preset)
        await db_session.commit()
        await db_session.refresh(preset)

        printer = await printer_factory()
        pipeline = await pipeline_factory(
            target_printer_id=printer.id,
            filament_presets=[{"source": "local", "id": str(preset.id)}],
        )
        src = await library_file_factory()

        # The printer reports the generic material in tray_type and the product
        # name in tray_sub_brands, so this is the shape a real AMS sends.
        live_status = {
            "connected": True,
            "raw_data": {"ams": [{"tray": [{"tray_type": "PLA", "tray_color": "FFFFFFFF"}]}]},
        }
        with patch(
            "backend.app.api.routes.pipeline_runs._load_printer_status",
            new=AsyncMock(return_value=live_status),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
                json={"source_library_file_id": src.id},
            )
        assert resp.status_code == 200
        body = resp.json()
        assert body["ok"] is True
        assert body["issues"] == []
        assert body["target_printer_name"] == printer.name

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_product_name_in_tray_type_is_reported_as_a_mismatch(
        self,
        async_client: AsyncClient,
        db_session,
        printer_factory,
        pipeline_factory,
        library_file_factory,
    ):
        """Eligibility answers with the dispatch matcher's type rules, not its own.

        It used to alias "PLA Basic" to "PLA" and pass this; the matcher never
        did, so the run cleared the pre-flight and then failed to map the slot.
        Flagging it here is the honest answer even though it is the stricter one.
        """
        from backend.app.models.local_preset import LocalPreset

        preset = LocalPreset(
            name="My PLA",
            preset_type="filament",
            source="manual",
            setting="{}",
            filament_type="PLA",
            default_filament_colour="#FFFFFF",
        )
        db_session.add(preset)
        await db_session.commit()
        await db_session.refresh(preset)

        printer = await printer_factory()
        pipeline = await pipeline_factory(
            target_printer_id=printer.id,
            filament_presets=[{"source": "local", "id": str(preset.id)}],
        )
        src = await library_file_factory()

        live_status = {
            "connected": True,
            "raw_data": {"ams": [{"tray": [{"tray_type": "PLA Basic", "tray_color": "FFFFFFFF"}]}]},
        }
        with patch(
            "backend.app.api.routes.pipeline_runs._load_printer_status",
            new=AsyncMock(return_value=live_status),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
                json={"source_library_file_id": src.id},
            )
        assert resp.status_code == 200
        body = resp.json()
        assert [i["kind"] for i in body["issues"]] == ["filament_type_mismatch"]


class TestRunPipeline:
    """POST /slicer-pipelines/{id}/run orchestrates slice + enqueue."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_with_issues_and_no_force_returns_409(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        library_file_factory,
    ):
        pipeline = await pipeline_factory()  # no target set
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
            json={"source_library_file_id": src.id},
        )
        assert resp.status_code == 409
        # Eligibility report rides in detail.
        detail = resp.json()["detail"]
        assert detail["ok"] is False
        # printer_not_set or class_not_set — depends on the PR A default
        # target_kind. Both mean "no target chosen yet".
        kinds = [i["kind"] for i in detail["issues"]]
        assert "printer_not_set" in kinds or "class_not_set" in kinds

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_force_with_no_target_still_400(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        library_file_factory,
    ):
        """``force=True`` bypasses the 409 but the run endpoint still needs a
        target to enqueue against — the second guard returns 400."""
        pipeline = await pipeline_factory()
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
            json={"source_library_file_id": src.id, "force": True},
        )
        assert resp.status_code == 400

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_creates_run_and_job(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()

        live_status = {"connected": True, "raw_data": {"ams": []}}
        # AMS empty → eligibility surfaces filament_unverified (non-blocking)
        # for the standard-tier filament refs the default factory uses; report
        # is ok=True so no force needed.
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 9001

        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id},
            )
        assert resp.status_code == 202, resp.text
        body = resp.json()
        assert body["pipeline_id"] == pipeline["id"]
        assert body["source_library_file_id"] == src.id
        assert body["copies"] == 1
        assert body["status"] == "queued"
        assert len(body["jobs"]) == 1
        assert body["jobs"][0]["copy_index"] == 0
        assert body["eligibility_overridden"] is False
        # slice_job_id rides on the response so the frontend can call
        # trackJob and render the progress toast.
        assert body["slice_job_id"] == 9001


class TestRunListAndGet:
    """Run history surfaces."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_runs_empty(
        self,
        async_client: AsyncClient,
        pipeline_factory,
    ):
        pipeline = await pipeline_factory()
        resp = await async_client.get(f"/api/v1/slicer-pipelines/{pipeline['id']}/runs")
        assert resp.status_code == 200
        assert resp.json() == {"runs": [], "total": 0}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_run_404(
        self,
        async_client: AsyncClient,
    ):
        resp = await async_client.get("/api/v1/pipeline-runs/99999")
        assert resp.status_code == 404


class TestCancelRun:
    """Cancellation marks the run + linked queue entry."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cancel_unknown_run_404(self, async_client: AsyncClient):
        resp = await async_client.post("/api/v1/pipeline-runs/99999/cancel")
        assert resp.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cancel_marks_queued_run(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        live_status = {"connected": True, "raw_data": {"ams": []}}
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 9001

        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            run_resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id},
            )
        run_id = run_resp.json()["id"]
        cancel_resp = await async_client.post(f"/api/v1/pipeline-runs/{run_id}/cancel")
        assert cancel_resp.status_code == 200
        assert cancel_resp.json()["status"] == "cancelled"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_accepts_archive_source(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        db_session,
    ):
        """``source_archive_id`` is accepted in place of source_library_file_id."""
        from pathlib import Path

        from backend.app.core.config import settings as app_settings
        from backend.app.models.archive import PrintArchive

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)

        rel = "test_pipeline_archive_source.3mf"
        (Path(app_settings.base_dir) / rel).write_bytes(b"")
        archive = PrintArchive(
            printer_id=printer.id,
            filename="Archive Source.3mf",
            file_path=rel,
            file_size=0,
            source_3mf_path=rel,
        )
        db_session.add(archive)
        await db_session.commit()
        await db_session.refresh(archive)

        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 7777

        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_archive_id": archive.id},
            )
        assert resp.status_code == 202, resp.text
        body = resp.json()
        assert body["source_library_file_id"] is None
        assert body["source_archive_id"] == archive.id
        assert body["slice_job_id"] == 7777

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_rejects_no_source(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
    ):
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        resp = await async_client.post(f"/api/v1/slicer-pipelines/{pipeline['id']}/run", json={})
        assert resp.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_rejects_both_sources(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
            json={"source_library_file_id": src.id, "source_archive_id": 99},
        )
        assert resp.status_code == 422


class TestPipelineC:
    """PR C — multi-copy, class targeting, fanout strategies, retry-failed,
    dashboard list, max-copies cap."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_copies_cap_enforced(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        # Default cap is 50; over-request returns 422 even with valid eligibility.
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
            json={"source_library_file_id": src.id, "copies": 9999},
        )
        assert resp.status_code == 422  # schema gate (le=1000)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_run_copies_3_creates_3_jobs(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 5555

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()

        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id, "copies": 3},
            )
        assert resp.status_code == 202, resp.text
        body = resp.json()
        assert body["copies"] == 3
        assert len(body["jobs"]) == 3
        assert [j["copy_index"] for j in body["jobs"]] == [0, 1, 2]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_class_eligibility_per_printer_breakdown(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        """target_kind='printer_class' surfaces per-printer reports."""
        await printer_factory(model="X1C")
        await printer_factory(model="X1C")
        await printer_factory(model="P1S")  # noise — different model
        pipeline = await pipeline_factory()
        # Wire class targeting via PUT.
        put_resp = await async_client.put(
            f"/api/v1/slicer-pipelines/{pipeline['id']}",
            json={
                "target_kind": "printer_class",
                "target_printer_id": 0,
                "target_model_class": "X1C",
                "fanout_strategy": "max_parallel",
            },
        )
        assert put_resp.status_code == 200, put_resp.text
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
            json={"source_library_file_id": src.id},
        )
        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["target_kind"] == "printer_class"
        assert body["target_model_class"] == "X1C"
        # Two X1Cs were created — both should appear in the per-printer breakdown.
        assert len(body["printer_reports"]) == 2
        assert all(r["printer_name"].startswith("X1C") for r in body["printer_reports"])
        # AMS empty + no live state → both are offline, so ok=False.
        assert body["ok"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_class_eligibility_matches_the_model_regardless_of_case(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        """As the queue and the scheduler match a model (Q-U8)."""
        await printer_factory(model="X1C")
        pipeline = await pipeline_factory()
        await async_client.put(
            f"/api/v1/slicer-pipelines/{pipeline['id']}",
            json={"target_kind": "printer_class", "target_printer_id": 0, "target_model_class": "x1c"},
        )
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
            json={"source_library_file_id": src.id},
        )
        body = resp.json()
        assert len(body["printer_reports"]) == 1
        assert not any(i["kind"] == "no_class_matches" for i in body["issues"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_class_eligibility_no_matching_printers(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
    ):
        await printer_factory(model="P1S")  # only a P1S in the install
        pipeline = await pipeline_factory()
        await async_client.put(
            f"/api/v1/slicer-pipelines/{pipeline['id']}",
            json={
                "target_kind": "printer_class",
                "target_printer_id": 0,
                "target_model_class": "X1C",
            },
        )
        src = await library_file_factory()
        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/check-eligibility",
            json={"source_library_file_id": src.id},
        )
        body = resp.json()
        assert body["ok"] is False
        assert any(i["kind"] == "no_class_matches" for i in body["issues"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_all_runs_dashboard_endpoint(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from backend.app.models.pipeline_run import PipelineRun

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        for i in range(3):
            run = PipelineRun(
                pipeline_id=pipeline["id"],
                source_library_file_id=src.id,
                copies=1,
                status="completed" if i % 2 == 0 else "failed",
            )
            db_session.add(run)
        await db_session.commit()

        resp = await async_client.get("/api/v1/pipeline-runs?limit=10")
        assert resp.status_code == 200
        body = resp.json()
        assert body["total"] == 3
        assert len(body["runs"]) == 3
        # Newest first.
        assert body["runs"][0]["id"] > body["runs"][-1]["id"]

        # Filter by status.
        resp = await async_client.get("/api/v1/pipeline-runs?status=failed")
        body = resp.json()
        assert all(r["status"] == "failed" for r in body["runs"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_retry_failed_creates_child_run(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from dataclasses import dataclass

        from backend.app.models.pipeline_run import PipelineJob, PipelineRun

        @dataclass
        class _FakeSliceJob:
            id: int = 6666

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        # Build a parent run with 3 jobs: 1 completed, 2 failed → retry
        # should request copies=2.
        parent = PipelineRun(
            pipeline_id=pipeline["id"],
            source_library_file_id=src.id,
            copies=3,
            status="partial_failure",
        )
        db_session.add(parent)
        await db_session.flush()
        for idx, status in enumerate(["completed", "failed", "failed"]):
            db_session.add(PipelineJob(pipeline_run_id=parent.id, copy_index=idx, status=status))
        await db_session.commit()
        await db_session.refresh(parent)

        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(f"/api/v1/pipeline-runs/{parent.id}/retry-failed")
        assert resp.status_code == 202, resp.text
        body = resp.json()
        assert body["copies"] == 2  # only the 2 failed copies
        assert body["parent_run_id"] == parent.id


class TestPolishFollowUp:
    """Polish-pass fixes: dashboard target filters, clear endpoint, and the
    deleted-queue-entry → cancelled rollup behaviour."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_dashboard_filters_by_target_printer(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from backend.app.models.pipeline_run import PipelineRun

        printer_a = await printer_factory()
        printer_b = await printer_factory()
        pipe_a = await pipeline_factory(target_printer_id=printer_a.id)
        pipe_b = await pipeline_factory(target_printer_id=printer_b.id)
        src = await library_file_factory()
        for pipe in (pipe_a, pipe_a, pipe_b):
            db_session.add(
                PipelineRun(
                    pipeline_id=pipe["id"],
                    source_library_file_id=src.id,
                    copies=1,
                    status="completed",
                )
            )
        await db_session.commit()

        resp = await async_client.get(f"/api/v1/pipeline-runs?target_printer_id={printer_a.id}")
        assert resp.status_code == 200
        body = resp.json()
        assert body["total"] == 2
        assert all(r["target_printer_id"] == printer_a.id for r in body["runs"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_dashboard_filters_by_target_model_class(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from backend.app.models.pipeline_run import PipelineRun

        await printer_factory(model="X1C")
        await printer_factory(model="P1S")
        # Two pipelines, one class-targeting X1C, one P1S.
        pipe_x = await pipeline_factory()
        await async_client.put(
            f"/api/v1/slicer-pipelines/{pipe_x['id']}",
            json={"target_kind": "printer_class", "target_printer_id": 0, "target_model_class": "X1C"},
        )
        pipe_p = await pipeline_factory()
        await async_client.put(
            f"/api/v1/slicer-pipelines/{pipe_p['id']}",
            json={"target_kind": "printer_class", "target_printer_id": 0, "target_model_class": "P1S"},
        )
        src = await library_file_factory()
        for pipe in (pipe_x, pipe_p, pipe_p):
            db_session.add(
                PipelineRun(
                    pipeline_id=pipe["id"],
                    source_library_file_id=src.id,
                    copies=1,
                    status="completed",
                )
            )
        await db_session.commit()

        resp = await async_client.get("/api/v1/pipeline-runs?target_model_class=P1S")
        assert resp.status_code == 200
        body = resp.json()
        assert body["total"] == 2
        assert all(r["target_model_class"] == "P1S" for r in body["runs"])

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_clear_endpoint_deletes_terminal_runs_only(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from backend.app.models.pipeline_run import PipelineRun

        printer = await printer_factory()
        pipe = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        for status in ("completed", "failed", "cancelled", "partial_failure", "dispatching", "in_progress"):
            db_session.add(
                PipelineRun(
                    pipeline_id=pipe["id"],
                    source_library_file_id=src.id,
                    copies=1,
                    status=status,
                )
            )
        await db_session.commit()

        resp = await async_client.post("/api/v1/pipeline-runs/clear")
        assert resp.status_code == 200, resp.text
        assert resp.json()["deleted"] == 4  # 4 terminal statuses cleared

        # The in-flight rows survive.
        survivors = (await async_client.get("/api/v1/pipeline-runs")).json()
        assert survivors["total"] == 2
        assert {r["status"] for r in survivors["runs"]} == {"dispatching", "in_progress"}

    @staticmethod
    async def _dispatched_run(db_session, pipeline_factory, printer_factory, library_file_factory):
        """A run in 'dispatching' whose one job is queued as a real queue entry."""
        from datetime import datetime, timezone

        from backend.app.models.pipeline_run import PipelineJob, PipelineRun
        from backend.app.models.print_queue import PrintQueueItem

        printer = await printer_factory()
        pipe = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        item = PrintQueueItem(printer_id=printer.id, status="pending", library_file_id=src.id)
        run = PipelineRun(pipeline_id=pipe["id"], source_library_file_id=src.id, copies=1, status="dispatching")
        db_session.add_all([item, run])
        await db_session.flush()
        db_session.add(
            PipelineJob(
                pipeline_run_id=run.id,
                copy_index=0,
                queue_entry_id=item.id,
                assigned_printer_id=printer.id,
                status="queued",
                dispatched_at=datetime.now(timezone.utc),
            )
        )
        await db_session.commit()
        return run.id, item.id, src.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleted_queue_entry_rolls_up_as_cancelled(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """When the queue entry that a PipelineJob is linked to gets deleted
        from the print-queue page, the job's live status should roll up to
        ``cancelled`` so the run doesn't sit forever showing ``queued`` /
        ``dispatching``. The delete leaves the job without its queue entry."""
        run_id, item_id, _ = await self._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )

        assert (await async_client.delete(f"/api/v1/queue/{item_id}")).status_code == 200
        resp = await async_client.get(f"/api/v1/pipeline-runs/{run_id}")

        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["jobs"][0]["status"] == "cancelled"
        assert body["status"] == "cancelled"
        assert body["copies_cancelled"] == 1 and body["copies_in_progress"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_job_never_dispatched_is_not_called_cancelled(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """No queue entry yet because it was never dispatched: still waiting."""
        from backend.app.models.pipeline_run import PipelineJob, PipelineRun

        printer = await printer_factory()
        pipe = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        run = PipelineRun(pipeline_id=pipe["id"], source_library_file_id=src.id, copies=1, status="slicing")
        db_session.add(run)
        await db_session.flush()
        db_session.add(PipelineJob(pipeline_run_id=run.id, copy_index=0, status="pending"))
        await db_session.commit()

        body = (await async_client.get(f"/api/v1/pipeline-runs/{run.id}")).json()

        assert body["jobs"][0]["status"] == "pending"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_retry_counts_a_deleted_queue_entry_as_cancelled(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """What the run view shows as cancelled, retry starts again."""
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 6667

        run_id, item_id, _ = await self._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )
        assert (await async_client.delete(f"/api/v1/queue/{item_id}")).status_code == 200

        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(f"/api/v1/pipeline-runs/{run_id}/retry-failed")

        assert resp.status_code == 202, resp.text
        assert resp.json()["copies"] == 1


class TestCancelTerminal:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cancel_terminal_run_is_idempotent(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from backend.app.models.pipeline_run import PipelineRun

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        run = PipelineRun(
            pipeline_id=pipeline["id"],
            source_library_file_id=src.id,
            copies=1,
            status="completed",
        )
        db_session.add(run)
        await db_session.commit()
        await db_session.refresh(run)
        resp = await async_client.post(f"/api/v1/pipeline-runs/{run.id}/cancel")
        assert resp.status_code == 200
        assert resp.json()["status"] == "completed"  # unchanged


class TestRunViaApiKey:
    """An API key may run a pipeline (#1425 follow-up).

    Every pipeline endpoint used to answer 403 for API keys — the three
    permissions were parked as administrative in PR A, before the run dispatch
    existed to decide about. ``pipelines:run`` now needs the key's Manage Queue
    *and* Manage Library scopes together, because a run slices into the library
    and then queues prints.
    """

    async def _admin_and_key(self, async_client: AsyncClient, db_session, **flags):
        """Enable auth, then mint a key owned by the admin. The owner matters:
        a key never out-ranks its owner, and only an owned key can stand in for
        a user when a cloud preset has to be resolved."""
        from sqlalchemy import select

        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey
        from backend.app.models.user import User

        await async_client.post(
            "/api/v1/auth/setup",
            json={"auth_enabled": True, "admin_username": "pipeadmin", "admin_password": "AdminPass1!"},
        )
        admin = (await db_session.execute(select(User).where(User.username == "pipeadmin"))).scalar_one()

        full_key, key_hash, key_prefix = generate_api_key()
        db_session.add(
            APIKey(
                name="pipeline-runner",
                key_hash=key_hash,
                key_prefix=key_prefix,
                user_id=admin.id,
                enabled=True,
                **{"can_read_status": False, "can_queue": False, "can_manage_library": False, **flags},
            )
        )
        await db_session.commit()
        return admin, full_key

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_scoped_key_runs_the_pipeline(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 7777

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        # Auth goes on only now: the factories above post as an anonymous
        # caller, which is how every other test in this file works.
        _, key = await self._admin_and_key(
            async_client, db_session, can_read_status=True, can_queue=True, can_manage_library=True
        )

        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id, "copies": 1, "force": True},
                headers={"X-API-Key": key},
            )

        assert resp.status_code == 202, resp.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_key_without_manage_library_is_refused(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """Queueing prints is only half of what a run does. The refusal names
        the flag that is missing rather than calling the whole thing
        administrative."""
        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        _, key = await self._admin_and_key(async_client, db_session, can_read_status=True, can_queue=True)

        resp = await async_client.post(
            f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
            json={"source_library_file_id": src.id, "copies": 1, "force": True},
            headers={"X-API-Key": key},
        )

        assert resp.status_code == 403
        assert "can_manage_library" in resp.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_cloud_scoped_key_slices_as_its_owner(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """A pipeline can be built on Bambu/Orca Cloud presets, and resolving
        those reads a cloud token off a user record. The permission gate hands
        an API-keyed request ``current_user=None``, so without falling back to
        the key's owner such a pipeline would have nobody to resolve against
        and would fail at slice time — the same fallback the direct slice route
        makes."""
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 7778

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        admin, key = await self._admin_and_key(
            async_client,
            db_session,
            can_read_status=True,
            can_queue=True,
            can_manage_library=True,
            can_access_cloud=True,
        )

        enqueue = AsyncMock(return_value=_FakeSliceJob())
        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch("backend.app.services.slice_dispatch.slice_dispatch.enqueue", new=enqueue),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id, "copies": 1, "force": True},
                headers={"X-API-Key": key},
            )

        assert resp.status_code == 202, resp.text
        assert enqueue.await_args.kwargs["owner_id"] == admin.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_key_without_cloud_scope_stays_anonymous(
        self,
        async_client: AsyncClient,
        pipeline_factory,
        printer_factory,
        library_file_factory,
        db_session,
    ):
        """The fallback is the cloud scope's own opt-in, not a general identity
        for API keys: a key without it slices unattributed, exactly as before."""
        from dataclasses import dataclass

        @dataclass
        class _FakeSliceJob:
            id: int = 7779

        printer = await printer_factory()
        pipeline = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        # Same owner and same three scopes as the test above — only the cloud
        # opt-in differs.
        _, full_key = await self._admin_and_key(
            async_client, db_session, can_read_status=True, can_queue=True, can_manage_library=True
        )

        enqueue = AsyncMock(return_value=_FakeSliceJob())
        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch(
                "backend.app.api.routes.pipeline_runs._load_printer_status",
                new=AsyncMock(return_value=live_status),
            ),
            patch("backend.app.services.slice_dispatch.slice_dispatch.enqueue", new=enqueue),
        ):
            resp = await async_client.post(
                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
                json={"source_library_file_id": src.id, "copies": 1, "force": True},
                headers={"X-API-Key": full_key},
            )

        assert resp.status_code == 202, resp.text
        assert enqueue.await_args.kwargs["owner_id"] is None


class TestJobKeepsItsFinalStatus:
    """A dispatched job stores its queue entry's final status, so deleting the entry
    later cannot turn a printed copy into one that looks pending (or retryable)."""

    @staticmethod
    async def _job(db_session, run_id: int):
        from sqlalchemy import select

        from backend.app.models.pipeline_run import PipelineJob

        db_session.expire_all()
        return (await db_session.execute(select(PipelineJob).where(PipelineJob.pipeline_run_id == run_id))).scalar_one()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_printed_copy_stays_completed_after_its_entry_is_deleted(
        self, async_client: AsyncClient, pipeline_factory, printer_factory, library_file_factory, db_session
    ):
        from backend.app.models.print_queue import PrintQueueItem

        run_id, item_id, _ = await TestPolishFollowUp._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )
        item = await db_session.get(PrintQueueItem, item_id)
        item.status = "completed"
        await db_session.commit()

        assert (await async_client.delete(f"/api/v1/queue/{item_id}")).status_code == 200
        body = (await async_client.get(f"/api/v1/pipeline-runs/{run_id}")).json()
        retry = await async_client.post(f"/api/v1/pipeline-runs/{run_id}/retry-failed")

        assert (body["jobs"][0]["status"], body["status"]) == ("completed", "completed")
        assert retry.status_code == 400, retry.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    @pytest.mark.parametrize(
        ("bulk", "expected"), [("cancel", "cancelled"), ("delete", "cancelled"), ("fail", "failed")]
    )
    async def test_bulk_statements_store_the_final_status(
        self, pipeline_factory, printer_factory, library_file_factory, db_session, bulk, expected
    ):
        from sqlalchemy import delete, update

        from backend.app.models.print_queue import PrintQueueItem

        run_id, item_id, _ = await TestPolishFollowUp._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )
        if bulk == "delete":
            statement = delete(PrintQueueItem).where(PrintQueueItem.id == item_id)
        elif bulk == "cancel":
            # As the library trash cancels the jobs of a deleted file: on the table itself.
            statement = PrintQueueItem.__table__.update().where(PrintQueueItem.id == item_id).values(status="cancelled")
        else:
            statement = update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(status="failed")
        await db_session.execute(statement)
        await db_session.commit()

        assert (await self._job(db_session, run_id)).status == expected

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_job_whose_entry_vanished_earlier_is_unknown_not_retried(
        self, async_client: AsyncClient, pipeline_factory, printer_factory, library_file_factory, db_session
    ):
        from backend.app.models.pipeline_run import PipelineJob, PipelineRun

        printer = await printer_factory()
        pipe = await pipeline_factory(target_printer_id=printer.id)
        src = await library_file_factory()
        run = PipelineRun(pipeline_id=pipe["id"], source_library_file_id=src.id, copies=1, status="dispatching")
        db_session.add(run)
        await db_session.flush()
        # Dispatched before jobs stored a final status; the entry is gone since.
        db_session.add(PipelineJob(pipeline_run_id=run.id, copy_index=0, queue_entry_id=None, status="queued"))
        await db_session.commit()

        body = (await async_client.get(f"/api/v1/pipeline-runs/{run.id}")).json()
        retry = await async_client.post(f"/api/v1/pipeline-runs/{run.id}/retry-failed")

        assert (body["jobs"][0]["status"], body["status"]) == ("unknown", "completed")
        assert retry.status_code == 400, retry.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_startup_stores_the_status_of_entries_that_ended_before(
        self, pipeline_factory, printer_factory, library_file_factory, db_session
    ):
        from backend.app.core.database import _backfill_pipeline_job_final_status

        run_id, item_id, _ = await TestPolishFollowUp._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )
        connection = await db_session.connection()
        # Ended before jobs stored it: written past the session hooks.
        await connection.exec_driver_sql(f"UPDATE print_queue SET status = 'aborted' WHERE id = {int(item_id)}")
        await db_session.commit()

        await _backfill_pipeline_job_final_status(await db_session.connection())
        await _backfill_pipeline_job_final_status(await db_session.connection())  # a restart changes nothing
        await db_session.commit()

        assert (await self._job(db_session, run_id)).status == "failed"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cancelling_a_run_leaves_an_unknown_copy_unknown(
        self, async_client: AsyncClient, pipeline_factory, printer_factory, library_file_factory, db_session
    ):
        """Cancelled would make retry print a copy that may have printed already."""
        from dataclasses import dataclass

        from backend.app.models.pipeline_run import PipelineJob

        @dataclass
        class _FakeSliceJob:
            id: int = 6668

        run_id, _, _ = await TestPolishFollowUp._dispatched_run(
            db_session, pipeline_factory, printer_factory, library_file_factory
        )
        # A second copy, dispatched before jobs stored a final status; its entry is gone.
        db_session.add(PipelineJob(pipeline_run_id=run_id, copy_index=1, queue_entry_id=None, status="queued"))
        await db_session.commit()

        cancelled = await async_client.post(f"/api/v1/pipeline-runs/{run_id}/cancel")
        live_status = {"connected": True, "raw_data": {"ams": []}}
        with (
            patch("backend.app.api.routes.pipeline_runs._load_printer_status", new=AsyncMock(return_value=live_status)),
            patch(
                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
                new=AsyncMock(return_value=_FakeSliceJob()),
            ),
        ):
            retry = await async_client.post(f"/api/v1/pipeline-runs/{run_id}/retry-failed")

        assert cancelled.status_code == 200, cancelled.text
        assert sorted(job["status"] for job in cancelled.json()["jobs"]) == ["cancelled", "unknown"]
        assert retry.status_code == 202, retry.text
        assert retry.json()["copies"] == 1