bambuddy/backend/tests/integration/test_projects_api.py

"""Integration tests for Projects API endpoints."""

import pytest
from httpx import AsyncClient


class TestProjectsAPI:
    """Integration tests for /api/v1/projects endpoints."""

    @pytest.fixture
    async def project_factory(self, db_session):
        """Factory to create test projects."""
        _counter = [0]

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            _counter[0] += 1
            counter = _counter[0]

            defaults = {
                "name": f"Test Project {counter}",
                "description": "Test project description",
                "color": "#FF0000",
            }
            defaults.update(kwargs)

            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_projects_empty(self, async_client: AsyncClient):
        """Verify empty list when no projects exist."""
        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        assert isinstance(response.json(), list)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_projects_with_data(self, async_client: AsyncClient, project_factory, db_session):
        """Verify list returns existing projects."""
        await project_factory(name="My Project")
        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        data = response.json()
        assert any(p["name"] == "My Project" for p in data)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_project(self, async_client: AsyncClient):
        """Verify project can be created."""
        data = {
            "name": "New Project",
            "description": "A new project",
            "color": "#00FF00",
        }
        response = await async_client.post("/api/v1/projects/", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["name"] == "New Project"
        assert result["color"] == "#00FF00"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_project(self, async_client: AsyncClient, project_factory, db_session):
        """Verify single project can be retrieved."""
        project = await project_factory(name="Get Test Project")
        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        assert response.json()["name"] == "Get Test Project"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_project_not_found(self, async_client: AsyncClient):
        """Verify 404 for non-existent project."""
        response = await async_client.get("/api/v1/projects/9999")
        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_project(self, async_client: AsyncClient, project_factory, db_session):
        """Verify project can be updated."""
        project = await project_factory(name="Original")
        response = await async_client.patch(
            f"/api/v1/projects/{project.id}", json={"name": "Updated", "description": "Updated description"}
        )
        assert response.status_code == 200
        result = response.json()
        assert result["name"] == "Updated"
        assert result["description"] == "Updated description"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_project(self, async_client: AsyncClient, project_factory, db_session):
        """Verify project can be deleted."""
        project = await project_factory()
        response = await async_client.delete(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        data = response.json()
        assert data["message"] == "Project deleted"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_project_not_found(self, async_client: AsyncClient):
        """Verify 404 for deleting non-existent project."""
        response = await async_client.delete("/api/v1/projects/9999")
        assert response.status_code == 404


class TestProjectUrlAndCoverImage:
    """Tests for #1155 — url field + cover image upload/get/delete."""

    @pytest.fixture
    async def project_factory(self, db_session):
        async def _create(**kwargs):
            from backend.app.models.project import Project

            defaults = {"name": "URL/Cover Project", "color": "#00ff00"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_project_accepts_https_url(self, async_client: AsyncClient):
        response = await async_client.post(
            "/api/v1/projects/",
            json={"name": "With URL", "url": "https://makerworld.com/models/12345"},
        )
        assert response.status_code == 200
        body = response.json()
        assert body["url"] == "https://makerworld.com/models/12345"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_project_rejects_javascript_url(self, async_client: AsyncClient):
        # `<a href>` rendering would execute javascript: URLs — schema must reject.
        response = await async_client.post(
            "/api/v1/projects/",
            json={"name": "Hostile", "url": "javascript:alert(1)"},
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_project_rejects_data_url(self, async_client: AsyncClient):
        response = await async_client.post(
            "/api/v1/projects/",
            json={"name": "Hostile", "url": "data:text/html,<script>alert(1)</script>"},
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_patch_project_clears_url_when_explicitly_null(self, async_client: AsyncClient, project_factory):
        project = await project_factory(url="https://example.com")
        response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"url": None})
        assert response.status_code == 200
        assert response.json()["url"] is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_upload_cover_image_then_serve_then_delete(self, async_client: AsyncClient, project_factory):
        project = await project_factory()

        # 1x1 PNG (smallest valid PNG bytes)
        png_bytes = bytes.fromhex(
            "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4"
            "890000000d49444154789c63f80f00000100010000000000000049454e44ae42"
            "6082"
        )
        upload = await async_client.post(
            f"/api/v1/projects/{project.id}/cover-image",
            files={"file": ("cover.png", png_bytes, "image/png")},
        )
        assert upload.status_code == 200, upload.text
        body = upload.json()
        assert body["status"] == "success"
        assert body["filename"].endswith(".png")
        cover_filename = body["filename"]

        # GET should serve the bytes back
        served = await async_client.get(f"/api/v1/projects/{project.id}/cover-image")
        assert served.status_code == 200
        assert served.headers["content-type"] == "image/png"
        assert served.content == png_bytes

        # Project response should reflect the cover_image_filename field
        view = await async_client.get(f"/api/v1/projects/{project.id}")
        assert view.json()["cover_image_filename"] == cover_filename

        # DELETE should clear the field
        deleted = await async_client.delete(f"/api/v1/projects/{project.id}/cover-image")
        assert deleted.status_code == 200
        view2 = await async_client.get(f"/api/v1/projects/{project.id}")
        assert view2.json()["cover_image_filename"] is None
        # And subsequent GET should 404
        served2 = await async_client.get(f"/api/v1/projects/{project.id}/cover-image")
        assert served2.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_upload_cover_image_rejects_non_image(self, async_client: AsyncClient, project_factory):
        project = await project_factory()
        response = await async_client.post(
            f"/api/v1/projects/{project.id}/cover-image",
            files={"file": ("evil.exe", b"MZ\x00\x00", "application/octet-stream")},
        )
        assert response.status_code == 400

    @pytest.mark.integration
    def test_cover_image_get_uses_query_token_gate(self):
        """Regression guard: GET /projects/{id}/cover-image MUST be gated by a
        dependency that accepts ``?token=…`` in the query string rather than by
        a header-only bearer gate, because browsers can't attach an
        ``Authorization`` header to ``<img src>`` requests. Swapping to a
        header-only gate would silently 401 every cover image when auth is
        enabled.

        The token type changed in #3025 -- the route took the camera-stream
        token until then, which made ``camera:view`` a prerequisite for seeing
        a project cover -- so this pins the media gate. What it is really
        asserting is unchanged: the credential has to fit in a URL."""
        from fastapi.routing import APIRoute

        from backend.app.api.routes.projects import router

        # Find the GET cover-image route. The router exposes path/methods/
        # dependencies via APIRoute objects.

        cover_get = None
        for route in router.routes:
            if isinstance(route, APIRoute) and route.path.endswith("/cover-image") and "GET" in route.methods:
                cover_get = route
                break

        assert cover_get is not None, "GET cover-image route missing"

        # The route's dependant tree includes a Depends(require_media_token_permission(...))
        # — its `call` is the inner check function returned by that factory.
        # Walk the dependant tree and assert one of the dependencies came from
        # the media-token factory, NOT from require_permission_if_auth_enabled.
        from backend.app.core.auth import require_media_token_permission
        from backend.app.core.permissions import Permission

        # The factory returns a fresh closure each call; the most reliable
        # signature is the qualified name of the function in the closure chain.
        expected_qualname = require_media_token_permission(Permission.PROJECTS_READ).__qualname__

        gate_qualnames = [dep.call.__qualname__ for dep in cover_get.dependant.dependencies if dep.call]
        assert expected_qualname in gate_qualnames, (
            f"GET cover-image route is not gated by a media-token dependency. Found: {gate_qualnames}"
        )


class TestProjectPartsTracking:
    """Tests for project parts tracking feature."""

    @pytest.fixture
    async def project_factory(self, db_session):
        """Factory to create test projects."""

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {
                "name": "Parts Test Project",
                "description": "Test project",
                "color": "#FF0000",
            }
            defaults.update(kwargs)

            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def archive_factory(self, db_session):
        """Factory to create a test archive plus a matching PrintLogEntry.

        Project stats aggregate from ``print_log_entries`` (#1593), so a
        test that only writes archives wouldn't exercise the production
        path — production always writes one log entry per run. The
        factory mirrors that: every archive whose status is anything other
        than ``"archived"`` (file shelved without printing) gets a log
        entry whose status matches the archive.
        """

        async def _create_archive(**kwargs):
            from backend.app.models.archive import PrintArchive
            from backend.app.models.print_log import PrintLogEntry

            defaults = {
                "filename": "test.3mf",
                "file_path": "test/test.3mf",
                "file_size": 1000,
                "print_name": "Test Print",
                "status": "completed",
                "quantity": 1,
            }
            defaults.update(kwargs)

            archive = PrintArchive(**defaults)
            db_session.add(archive)
            await db_session.commit()
            await db_session.refresh(archive)

            if archive.status != "archived":
                db_session.add(
                    PrintLogEntry(
                        archive_id=archive.id,
                        print_name=archive.print_name,
                        status=archive.status,
                    )
                )
                await db_session.commit()
            return archive

        return _create_archive

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_project_with_target_parts_count(self, async_client: AsyncClient):
        """Verify project can be created with target_parts_count."""
        data = {
            "name": "Parts Project",
            "target_count": 10,  # 10 plates
            "target_parts_count": 50,  # 50 parts total
        }
        response = await async_client.post("/api/v1/projects/", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["target_count"] == 10
        assert result["target_parts_count"] == 50

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_project_target_parts_count(self, async_client: AsyncClient, project_factory, db_session):
        """Verify target_parts_count can be updated."""
        project = await project_factory()
        response = await async_client.patch(
            f"/api/v1/projects/{project.id}",
            json={"target_parts_count": 100},
        )
        assert response.status_code == 200
        assert response.json()["target_parts_count"] == 100

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_project_parts_progress_calculation(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Verify parts progress is calculated from archive quantities."""
        # Create project with target of 20 parts
        project = await project_factory(target_parts_count=20)

        # Create archives with different quantities
        await archive_factory(project_id=project.id, quantity=3, status="completed")  # 3 parts
        await archive_factory(project_id=project.id, quantity=5, status="completed")  # 5 parts
        await archive_factory(project_id=project.id, quantity=2, status="completed")  # 2 parts
        # Total: 10 parts completed out of 20 = 50%

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        data = response.json()

        # Check stats
        assert data["stats"]["completed_prints"] == 10  # Sum of quantities
        assert data["stats"]["parts_progress_percent"] == 50.0  # 10/20 = 50%
        assert data["stats"]["remaining_parts"] == 10  # 20 - 10 = 10

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_project_list_shows_parts_count(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Verify project list returns correct completed_count (parts sum)."""
        project = await project_factory(name="List Parts Project", target_parts_count=100)

        # Create archives with quantities
        await archive_factory(project_id=project.id, quantity=4, status="completed")
        await archive_factory(project_id=project.id, quantity=6, status="completed")
        # Total: 10 parts, 2 plates

        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        data = response.json()

        # Find our project
        our_project = next((p for p in data if p["name"] == "List Parts Project"), None)
        assert our_project is not None
        assert our_project["archive_count"] == 2  # 2 plates
        assert our_project["completed_count"] == 10  # 10 parts (sum of quantities)
        assert our_project["target_parts_count"] == 100

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_plates_vs_parts_progress(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Verify plates and parts progress are calculated separately."""
        # Project needs 5 plates producing 25 parts total (5 parts per plate)
        project = await project_factory(target_count=5, target_parts_count=25)

        # Complete 2 plates, each with 5 parts
        await archive_factory(project_id=project.id, quantity=5, status="completed")
        await archive_factory(project_id=project.id, quantity=5, status="completed")
        # Plates: 2/5 = 40%, Parts: 10/25 = 40%

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        data = response.json()

        assert data["stats"]["total_archives"] == 2  # 2 plates
        assert data["stats"]["completed_prints"] == 10  # 10 parts
        assert data["stats"]["progress_percent"] == 40.0  # plates: 2/5
        assert data["stats"]["parts_progress_percent"] == 40.0  # parts: 10/25


class TestProjectArchivedStatusNotCounted:
    """Tests for bug #630: archived files added to a project should not count as printed."""

    @pytest.fixture
    async def project_factory(self, db_session):
        """Factory to create test projects."""

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {
                "name": "Archived Status Test",
                "description": "Test project",
                "color": "#FF0000",
            }
            defaults.update(kwargs)

            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def archive_factory(self, db_session):
        """Factory to create a test archive plus a matching PrintLogEntry —
        see TestProjectPartsTracking.archive_factory for rationale (#1593)."""

        async def _create_archive(**kwargs):
            from backend.app.models.archive import PrintArchive
            from backend.app.models.print_log import PrintLogEntry

            defaults = {
                "filename": "test.3mf",
                "file_path": "test/test.3mf",
                "file_size": 1000,
                "print_name": "Test Print",
                "status": "completed",
                "quantity": 1,
            }
            defaults.update(kwargs)

            archive = PrintArchive(**defaults)
            db_session.add(archive)
            await db_session.commit()
            await db_session.refresh(archive)

            if archive.status != "archived":
                db_session.add(
                    PrintLogEntry(
                        archive_id=archive.id,
                        print_name=archive.print_name,
                        status=archive.status,
                    )
                )
                await db_session.commit()
            return archive

        return _create_archive

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_archived_files_not_counted_as_completed(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Archived files added to a project should not count in completed_prints stats."""
        project = await project_factory(target_parts_count=20)

        # 2 actually printed (completed), 3 just archived (not printed yet)
        await archive_factory(project_id=project.id, quantity=2, status="completed")
        await archive_factory(project_id=project.id, quantity=3, status="archived")
        await archive_factory(project_id=project.id, quantity=5, status="archived")

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        data = response.json()

        # Only the completed archive should count
        assert data["stats"]["completed_prints"] == 2
        assert data["stats"]["parts_progress_percent"] == 10.0  # 2/20 = 10%
        assert data["stats"]["remaining_parts"] == 18

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_archived_files_not_counted_in_project_list(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Project list endpoint should not count archived files as completed."""
        project = await project_factory(name="List Archived Test", target_parts_count=50)

        await archive_factory(project_id=project.id, quantity=4, status="completed")
        await archive_factory(project_id=project.id, quantity=6, status="archived")

        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        data = response.json()

        our_project = next((p for p in data if p["name"] == "List Archived Test"), None)
        assert our_project is not None
        assert our_project["completed_count"] == 4  # Only completed, not archived
        # Post-#1593: archive_count is "print runs", not "files attached". An
        # ``archived``-status file (shelved without printing) has no
        # PrintLogEntry and doesn't count — only the actual printed run does.
        assert our_project["archive_count"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_only_completed_status_counts(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Only 'completed' status should count in stats, not archived/failed/etc."""
        project = await project_factory(target_parts_count=100)

        await archive_factory(project_id=project.id, quantity=10, status="completed")
        await archive_factory(project_id=project.id, quantity=5, status="archived")
        await archive_factory(project_id=project.id, quantity=3, status="failed")
        await archive_factory(project_id=project.id, quantity=2, status="aborted")

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        data = response.json()

        assert data["stats"]["completed_prints"] == 10  # Only "completed"
        assert data["stats"]["failed_prints"] == 2  # failed + aborted (count of runs)
        # Post-#1593: total_archives counts runs from print_log_entries, not
        # files. The ``archived`` row is a shelved file with no run, so it
        # contributes 0; the other three (completed, failed, aborted) each
        # produced a run.
        assert data["stats"]["total_archives"] == 3
        # total_items sums quantity per run: 10 (completed) + 3 (failed) + 2 (aborted) = 15
        assert data["stats"]["total_items"] == 15


class TestProjectStatsPerRun:
    """Project stats aggregate per-run from ``print_log_entries`` so
    reprints and multi-plate prints count every run (#1593). Pre-fix the
    stats counted ``print_archives`` (one row per file), so 3 reprints of
    one file showed as 1 job with plate-1-only filament/time/cost.
    """

    @pytest.fixture
    async def project_factory(self, db_session):
        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {"name": "Per-Run Stats Project", "color": "#FF0000"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def archive_with_runs(self, db_session):
        """Build a single archive + N PrintLogEntry rows.

        Models the reporter's case: one source file (archive) is reprinted
        N times, each run with its own duration / filament / cost.
        """

        async def _create(*, project_id: int, runs: list[dict], archive_status: str = "completed", quantity: int = 1):
            from backend.app.models.archive import PrintArchive
            from backend.app.models.print_log import PrintLogEntry

            archive = PrintArchive(
                filename="reprinted.3mf",
                file_path="test/reprinted.3mf",
                file_size=1000,
                print_name="Reprinted Print",
                status=archive_status,
                quantity=quantity,
                project_id=project_id,
            )
            db_session.add(archive)
            await db_session.commit()
            await db_session.refresh(archive)

            for run in runs:
                db_session.add(
                    PrintLogEntry(
                        archive_id=archive.id,
                        print_name=archive.print_name,
                        status=run.get("status", "completed"),
                        duration_seconds=run.get("duration_seconds"),
                        filament_used_grams=run.get("filament_used_grams"),
                        cost=run.get("cost"),
                        energy_kwh=run.get("energy_kwh"),
                        energy_cost=run.get("energy_cost"),
                    )
                )
            await db_session.commit()
            return archive

        return _create

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_three_reprints_count_as_three_jobs_with_summed_totals(
        self, async_client: AsyncClient, project_factory, archive_with_runs
    ):
        """Reporter's case: 3 runs of one multi-plate file should report
        3 jobs and summed time / filament / cost — pre-fix it reported 1
        job with plate-1-only totals."""
        project = await project_factory()
        await archive_with_runs(
            project_id=project.id,
            runs=[
                {"duration_seconds": 7140, "filament_used_grams": 19.2, "cost": 0.40},
                {"duration_seconds": 6000, "filament_used_grams": 20.0, "cost": 0.40},
                {"duration_seconds": 6300, "filament_used_grams": 18.8, "cost": 0.40},
            ],
        )

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        stats = response.json()["stats"]

        assert stats["total_archives"] == 3, "3 runs must show as 3 jobs"
        assert stats["completed_prints"] == 3, "Each run with quantity=1 contributes 1 part"
        assert stats["total_filament_grams"] == round(19.2 + 20.0 + 18.8, 2)
        assert stats["total_print_time_hours"] == round((7140 + 6000 + 6300) / 3600, 2)
        # Cost rounds at 2 decimals — 3 * 0.40 = 1.20
        assert stats["estimated_cost"] == 1.20

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_orphan_log_entries_do_not_bleed_into_projects(
        self, async_client: AsyncClient, project_factory, db_session
    ):
        """Log rows whose ``archive_id`` is NULL (archive deleted via
        ON DELETE SET NULL) must not leak into any project — the inner
        join filters them out by construction."""
        from backend.app.models.print_log import PrintLogEntry

        project = await project_factory()

        # Orphan log entries — no archive_id.
        for _ in range(5):
            db_session.add(
                PrintLogEntry(
                    archive_id=None,
                    print_name="Orphan Run",
                    status="completed",
                    duration_seconds=3600,
                    filament_used_grams=20.0,
                    cost=0.5,
                )
            )
        await db_session.commit()

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        stats = response.json()["stats"]

        # None of the orphan rows are attributable to this project.
        assert stats["total_archives"] == 0
        assert stats["completed_prints"] == 0
        assert stats["total_filament_grams"] == 0
        assert stats["total_print_time_hours"] == 0
        assert stats["estimated_cost"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_mixed_run_outcomes_split_completed_and_failed(
        self, async_client: AsyncClient, project_factory, archive_with_runs
    ):
        """A multi-run archive with mixed outcomes splits cleanly between
        completed_prints (per-quantity) and failed_prints (per-run)."""
        project = await project_factory()
        await archive_with_runs(
            project_id=project.id,
            quantity=2,
            runs=[
                {"status": "completed", "filament_used_grams": 30.0},
                {"status": "completed", "filament_used_grams": 30.0},
                {"status": "failed", "filament_used_grams": 5.0},
                {"status": "aborted", "filament_used_grams": 2.0},
            ],
        )

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        stats = response.json()["stats"]

        assert stats["total_archives"] == 4
        # 2 completed runs × quantity=2 each = 4 parts
        assert stats["completed_prints"] == 4
        # 2 failure runs (failed + aborted) count as 2, not 2*quantity
        assert stats["failed_prints"] == 2
        # All 4 runs contribute filament: 30 + 30 + 5 + 2 = 67
        assert stats["total_filament_grams"] == 67.0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_quick_stats_in_list_view_agree_with_per_project_stats(
        self, async_client: AsyncClient, project_factory, archive_with_runs
    ):
        """The /projects list view's quick stats must agree with
        /projects/{id}'s detailed stats — both come from the same per-run
        aggregation."""
        project = await project_factory(name="Quick-Stats Alignment")
        await archive_with_runs(
            project_id=project.id,
            quantity=1,
            runs=[
                {"status": "completed"},
                {"status": "completed"},
                {"status": "failed"},
            ],
        )

        list_resp = await async_client.get("/api/v1/projects/")
        ours = next(p for p in list_resp.json() if p["name"] == "Quick-Stats Alignment")
        assert ours["archive_count"] == 3
        assert ours["completed_count"] == 2
        assert ours["failed_count"] == 1


class TestProjectArchivesAPI:
    """Tests for project-archive relationships."""

    @pytest.fixture
    async def project_factory(self, db_session):
        """Factory to create test projects."""

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {
                "name": "Archive Test Project",
                "description": "Test project",
                "color": "#0000FF",
            }
            defaults.update(kwargs)

            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_project_with_archives(self, async_client: AsyncClient, project_factory, db_session):
        """Verify project can be retrieved with archive count."""
        project = await project_factory()
        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        # Project should have an archive count (may be 0)
        data = response.json()
        assert "name" in data

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_archives_in_project_returns_archives_with_creator(
        self, async_client: AsyncClient, project_factory, db_session
    ):
        """``GET /projects/{id}/archives`` must eagerly load both the project AND
        the creator User. Without selectinload(created_by) the response
        converter triggers a lazy attribute load on a closed async session
        and the request 500s with MissingGreenlet — exactly what was reported
        the moment a user with auth enabled (so archives carry created_by_id)
        opened a project view.
        """
        from backend.app.models.archive import PrintArchive
        from backend.app.models.user import User

        # Seed: a user (the eventual creator) and a project owning two archives,
        # one with created_by_id set, one without.
        creator = User(
            username="archive-creator",
            password_hash="x",
            role="user",
            is_active=True,
        )
        db_session.add(creator)
        await db_session.commit()
        await db_session.refresh(creator)

        project = await project_factory(name="Project Archives Smoke")

        attributed = PrintArchive(
            filename="attributed.3mf",
            file_path="x/attributed.3mf",
            file_size=2048,
            print_name="Attributed Print",
            status="completed",
            quantity=1,
            project_id=project.id,
            created_by_id=creator.id,
        )
        anonymous = PrintArchive(
            filename="anon.3mf",
            file_path="x/anon.3mf",
            file_size=2048,
            print_name="Anonymous Print",
            status="completed",
            quantity=1,
            project_id=project.id,
            created_by_id=None,
        )
        db_session.add_all([attributed, anonymous])
        await db_session.commit()

        response = await async_client.get(f"/api/v1/projects/{project.id}/archives?limit=100&offset=0")
        assert response.status_code == 200, f"Expected 200, got {response.status_code} body={response.text}"

        rows = response.json()
        assert len(rows) == 2

        # Both archive shapes serialise — the attributed one surfaces the
        # creator username (proving the eager-load worked) and the anonymous
        # one stays None without exploding.
        by_filename = {r["filename"]: r for r in rows}
        assert by_filename["attributed.3mf"]["created_by_username"] == "archive-creator"
        assert by_filename["attributed.3mf"]["created_by_id"] == creator.id
        assert by_filename["anon.3mf"]["created_by_username"] is None
        assert by_filename["anon.3mf"]["created_by_id"] is None


class TestProjectExportImport:
    """Tests for project export/import functionality."""

    @pytest.fixture
    async def project_factory(self, db_session):
        """Factory to create test projects."""
        _counter = [0]

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            _counter[0] += 1
            counter = _counter[0]

            defaults = {
                "name": f"Export Test Project {counter}",
                "description": "Test project for export",
                "color": "#00FF00",
            }
            defaults.update(kwargs)

            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def bom_item_factory(self, db_session):
        """Factory to create test BOM items."""

        async def _create_bom_item(project_id: int, **kwargs):
            from backend.app.models.project_bom import ProjectBOMItem

            defaults = {
                "project_id": project_id,
                "name": "Test Part",
                "quantity_needed": 1,
                "quantity_acquired": 0,
                "sort_order": 0,
            }
            defaults.update(kwargs)

            item = ProjectBOMItem(**defaults)
            db_session.add(item)
            await db_session.commit()
            await db_session.refresh(item)
            return item

        return _create_bom_item

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_export_project(self, async_client: AsyncClient, project_factory, bom_item_factory, db_session):
        """Verify project export includes BOM items."""
        project = await project_factory(
            name="Export Me",
            description="A test project",
            target_count=10,
            target_parts_count=50,
            budget=100.0,
        )

        # Add BOM items
        await bom_item_factory(project.id, name="M3x8 Screws", quantity_needed=20, unit_price=0.10)
        await bom_item_factory(project.id, name="Heat Inserts", quantity_needed=10, unit_price=0.25)

        # Test JSON format export
        response = await async_client.get(f"/api/v1/projects/{project.id}/export?format=json")
        assert response.status_code == 200

        data = response.json()
        assert data["name"] == "Export Me"
        assert data["description"] == "A test project"
        assert data["target_count"] == 10
        assert data["target_parts_count"] == 50
        assert data["budget"] == 100.0
        assert len(data["bom_items"]) == 2

        # Check BOM items
        bom_names = [item["name"] for item in data["bom_items"]]
        assert "M3x8 Screws" in bom_names
        assert "Heat Inserts" in bom_names

        # Test ZIP format export (default)
        zip_response = await async_client.get(f"/api/v1/projects/{project.id}/export")
        assert zip_response.status_code == 200
        assert zip_response.headers["content-type"] == "application/zip"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_project(self, async_client: AsyncClient):
        """Verify project can be imported with BOM items."""
        import_data = {
            "name": "Imported Project",
            "description": "Imported from JSON",
            "color": "#FF00FF",
            "target_count": 5,
            "target_parts_count": 25,
            "budget": 50.0,
            "bom_items": [
                {
                    "name": "PTFE Tubes",
                    "quantity_needed": 4,
                    "quantity_acquired": 0,
                    "unit_price": 2.50,
                    "sourcing_url": "https://example.com",
                    "stl_filename": None,
                    "remarks": "Need 4mm ID",
                },
            ],
        }

        response = await async_client.post("/api/v1/projects/import", json=import_data)
        assert response.status_code == 200

        data = response.json()
        assert data["name"] == "Imported Project"
        assert data["description"] == "Imported from JSON"
        assert data["target_count"] == 5
        assert data["target_parts_count"] == 25
        assert data["budget"] == 50.0
        assert data["id"] > 0  # Has a valid ID
        # BOM stats should show 1 item imported
        assert data["stats"]["bom_total_items"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_export_project_with_linked_folder(self, async_client: AsyncClient, project_factory, db_session):
        """Verify project export includes linked folders."""
        from backend.app.models.library import LibraryFolder

        project = await project_factory(name="Project With Folder")

        # Create a linked folder
        folder = LibraryFolder(name="Project Files", project_id=project.id)
        db_session.add(folder)
        await db_session.commit()

        response = await async_client.get(f"/api/v1/projects/{project.id}/export?format=json")
        assert response.status_code == 200

        data = response.json()
        assert data["name"] == "Project With Folder"
        assert len(data["linked_folders"]) == 1
        assert data["linked_folders"][0]["name"] == "Project Files"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_project_with_linked_folder(self, async_client: AsyncClient):
        """Verify project import accepts linked folders data."""
        import_data = {
            "name": "Imported With Folders",
            "linked_folders": [
                {"name": "STL Files"},
                {"name": "Documentation"},
            ],
        }

        # Import should succeed with linked_folders
        response = await async_client.post("/api/v1/projects/import", json=import_data)
        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "Imported With Folders"
        assert data["id"] > 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_project_from_json_file(self, async_client: AsyncClient):
        """Verify project can be imported from JSON file upload."""
        import io
        import json

        project_data = {
            "name": "File Uploaded Project",
            "description": "Imported from JSON file",
            "color": "#123456",
        }

        # Create a file-like object
        file_content = json.dumps(project_data).encode()
        files = {"file": ("project.json", io.BytesIO(file_content), "application/json")}

        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "File Uploaded Project"
        assert data["description"] == "Imported from JSON file"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_project_from_zip_file(self, async_client: AsyncClient):
        """Verify project can be imported from ZIP file with files."""
        import io
        import json
        import zipfile

        project_data = {
            "name": "ZIP Imported Project",
            "description": "Imported from ZIP",
            "linked_folders": [{"name": "TestFolder", "files": [{"filename": "test.txt"}]}],
        }

        # Create a ZIP file in memory
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("project.json", json.dumps(project_data))
            zf.writestr("files/TestFolder/test.txt", "Hello World")

        zip_buffer.seek(0)
        files = {"file": ("project.zip", zip_buffer, "application/zip")}

        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "ZIP Imported Project"
        assert data["description"] == "Imported from ZIP"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_export_zip_contains_files(self, async_client: AsyncClient, project_factory, db_session):
        """Verify ZIP export contains actual files from linked folders."""
        import io
        import json
        import zipfile
        from pathlib import Path

        from backend.app.api.routes.library import get_library_dir
        from backend.app.models.library import LibraryFile, LibraryFolder

        project = await project_factory(name="Project With Files")

        # Create a linked folder with is_external fields
        folder = LibraryFolder(
            name="TestExportFolder",
            project_id=project.id,
            is_external=False,
            external_readonly=False,
            external_show_hidden=False,
        )
        db_session.add(folder)
        await db_session.flush()

        # Create a test file on disk
        library_dir = get_library_dir()
        folder_path = library_dir / "TestExportFolder"
        folder_path.mkdir(parents=True, exist_ok=True)
        test_file_path = folder_path / "test_export.txt"
        test_file_path.write_text("Export test content")

        # Create library file record
        lib_file = LibraryFile(
            folder_id=folder.id,
            filename="test_export.txt",
            file_path="TestExportFolder/test_export.txt",
            file_type="other",
            file_size=19,
            is_external=False,
        )
        db_session.add(lib_file)
        await db_session.commit()

        # Export as ZIP
        response = await async_client.get(f"/api/v1/projects/{project.id}/export")
        assert response.status_code == 200
        assert response.headers["content-type"] == "application/zip"

        # Verify ZIP contents
        zip_buffer = io.BytesIO(response.content)
        with zipfile.ZipFile(zip_buffer, "r") as zf:
            assert "project.json" in zf.namelist()
            assert "files/TestExportFolder/test_export.txt" in zf.namelist()

            # Verify file content
            file_content = zf.read("files/TestExportFolder/test_export.txt").decode()
            assert file_content == "Export test content"

            # Verify project.json
            project_data = json.loads(zf.read("project.json"))
            assert project_data["name"] == "Project With Files"

        # Cleanup
        test_file_path.unlink(missing_ok=True)
        folder_path.rmdir()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_invalid_file_type(self, async_client: AsyncClient):
        """Verify import rejects invalid file types."""
        import io

        files = {"file": ("project.txt", io.BytesIO(b"invalid"), "text/plain")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400
        assert "must be .zip or .json" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_zip_missing_project_json(self, async_client: AsyncClient):
        """Verify import rejects ZIP without project.json."""
        import io
        import zipfile

        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w") as zf:
            zf.writestr("other.txt", "no project.json here")

        zip_buffer.seek(0)
        files = {"file": ("project.zip", zip_buffer, "application/zip")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400
        assert "project.json" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_invalid_json(self, async_client: AsyncClient):
        """Verify import rejects invalid JSON content."""
        import io

        files = {"file": ("project.json", io.BytesIO(b"not valid json"), "application/json")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400
        assert "Invalid JSON" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_rejects_absolute_path_in_folder_name(self, async_client: AsyncClient, tmp_path):
        """Absolute paths in `linked_folders[*].name` must not escape library_dir.

        Verbatim shape from the upstream advisory: attacker sets folder name to
        an absolute path, expecting Python's ``Path("/lib") / "/anywhere"`` to
        collapse to ``Path("/anywhere")`` and let the next file write land
        outside the library directory.
        """
        import io
        import json
        import zipfile

        target_outside = tmp_path / "outside" / "owned"
        # Build a ZIP whose folder name points outside library_dir entirely.
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(
                "project.json",
                json.dumps(
                    {
                        "name": "innocent",
                        "linked_folders": [{"name": str(target_outside)}],
                    }
                ),
            )
            zf.writestr(f"files/{target_outside}/evil.pth", b"import os; os.system('echo pwned > /tmp/owned')\n")

        zip_buffer.seek(0)
        files = {"file": ("evil.zip", zip_buffer, "application/zip")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400, response.text
        assert not target_outside.exists(), "Attacker payload landed outside library_dir"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_rejects_dotdot_in_folder_name(self, async_client: AsyncClient):
        """`..` segments in folder name must be rejected."""
        import io
        import json
        import zipfile

        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(
                "project.json",
                json.dumps(
                    {
                        "name": "innocent",
                        "linked_folders": [{"name": "../../../etc"}],
                    }
                ),
            )
            zf.writestr("files/../../../etc/x.txt", b"x")

        zip_buffer.seek(0)
        files = {"file": ("evil.zip", zip_buffer, "application/zip")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400, response.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_rejects_dotdot_in_relative_path(self, async_client: AsyncClient):
        """`..` segments in the per-entry path (Vector B in the advisory) must
        be rejected even when the folder name itself is fine."""
        import io
        import json
        import zipfile

        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(
                "project.json",
                json.dumps(
                    {
                        "name": "innocent",
                        "linked_folders": [{"name": "ok"}],
                    }
                ),
            )
            # Folder name is benign, but the file path inside attempts to
            # escape via ``..``.
            zf.writestr("files/ok/../../../etc/x.txt", b"x")

        zip_buffer.seek(0)
        files = {"file": ("evil.zip", zip_buffer, "application/zip")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 400, response.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_legit_nested_zip_still_works(self, async_client: AsyncClient):
        """A legitimate ZIP with a nested file path inside the folder must
        continue to import cleanly. Guards against the fix being over-strict."""
        import io
        import json
        import zipfile

        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(
                "project.json",
                json.dumps(
                    {
                        "name": "nested-ok",
                        "linked_folders": [{"name": "OkFolder"}],
                    }
                ),
            )
            zf.writestr("files/OkFolder/sub/dir/inside.txt", b"hello")

        zip_buffer.seek(0)
        files = {"file": ("nested.zip", zip_buffer, "application/zip")}
        response = await async_client.post("/api/v1/projects/import/file", files=files)
        assert response.status_code == 200, response.text
        data = response.json()
        assert data["name"] == "nested-ok"


class TestProjectListEditableFields:
    """Tests for #2536 — the project list payload must carry every field the
    shared edit dialog renders. The dialog is opened from both the project list
    and the project detail page and seeds itself from whichever project object it
    is handed, so a field missing from the list payload shows up blank there and
    is saved back over the stored value."""

    @pytest.fixture
    async def project_factory(self, db_session):
        async def _create(**kwargs):
            from backend.app.models.project import Project

            defaults = {"name": "Editable Fields Project", "color": "#123456"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_carries_the_fields_the_edit_dialog_renders(self, async_client: AsyncClient, project_factory):
        """The list view is where the reporter saw an empty tags field."""
        from datetime import datetime

        await project_factory(
            name="Tagged Project",
            tags="prototype,client-work",
            due_date=datetime(2026, 8, 1, 12, 0, 0),
            priority="high",
            target_parts_count=7,
        )

        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        item = next(p for p in response.json() if p["name"] == "Tagged Project")

        assert item["tags"] == "prototype,client-work"
        assert item["due_date"].startswith("2026-08-01")
        assert item["priority"] == "high"
        assert item["target_parts_count"] == 7

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_template_list_carries_them_too(self, async_client: AsyncClient, project_factory):
        """Templates feed the same dialog, so they need the same payload."""
        await project_factory(
            name="Tagged Template",
            is_template=True,
            tags="reusable",
            priority="urgent",
            target_parts_count=3,
        )

        response = await async_client.get("/api/v1/projects/templates")
        assert response.status_code == 200
        item = next(p for p in response.json() if p["name"] == "Tagged Template")

        assert item["tags"] == "reusable"
        assert item["priority"] == "urgent"
        assert item["target_parts_count"] == 3

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_priority_survives_an_edit_that_does_not_touch_it(self, async_client: AsyncClient, project_factory):
        """A save from the list view used to submit the default priority over a
        stored 'high' — the dialog never received the real one."""
        project = await project_factory(name="Important", priority="high", tags="keep-me")

        response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"name": "Still Important"})
        assert response.status_code == 200

        result = response.json()
        assert result["priority"] == "high"
        assert result["tags"] == "keep-me"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_explicit_null_clears_tags_and_due_date(self, async_client: AsyncClient, project_factory):
        """Emptying the field in the dialog has to actually remove the value."""
        from datetime import datetime

        project = await project_factory(name="Clearable", tags="obsolete", due_date=datetime(2026, 8, 1, 12, 0, 0))

        response = await async_client.patch(f"/api/v1/projects/{project.id}", json={"tags": None, "due_date": None})
        assert response.status_code == 200

        result = response.json()
        assert result["tags"] is None
        assert result["due_date"] is None


class TestProjectFileProgress:
    """Per-file print progress inside a project (#1897).

    Covers GET /projects/{id}/file-progress (attribution: library_file_id →
    content hash → filename, completed runs only, project-scoped), the
    target_sets field round-trip, and the add-to-queue project inheritance
    that feeds the attribution chain.
    """

    @pytest.fixture
    async def project_factory(self, db_session):
        _counter = [0]

        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            _counter[0] += 1
            defaults = {"name": f"Progress Project {_counter[0]}"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def folder_factory(self, db_session):
        _counter = [0]

        async def _create_folder(**kwargs):
            from backend.app.models.library import LibraryFolder

            _counter[0] += 1
            defaults = {"name": f"ProgressFolder {_counter[0]}"}
            defaults.update(kwargs)
            folder = LibraryFolder(**defaults)
            db_session.add(folder)
            await db_session.commit()
            await db_session.refresh(folder)
            return folder

        return _create_folder

    @pytest.fixture
    async def file_factory(self, db_session):
        _counter = [0]

        async def _create_file(**kwargs):
            from backend.app.models.library import LibraryFile

            _counter[0] += 1
            counter = _counter[0]
            defaults = {
                "filename": f"plate_{counter}.gcode.3mf",
                "file_path": f"library/plate_{counter}.gcode.3mf",
                "file_size": 1024,
                "file_type": "3mf",
            }
            defaults.update(kwargs)
            lib_file = LibraryFile(**defaults)
            db_session.add(lib_file)
            await db_session.commit()
            await db_session.refresh(lib_file)
            return lib_file

        return _create_file

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_counts_by_library_file_id(
        self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory
    ):
        """Runs stamped with library_file_id count toward that file even when
        the archive's filename differs (rename after dispatch)."""
        project = await project_factory()
        folder = await folder_factory(project_id=project.id)
        file_a = await file_factory(folder_id=folder.id)
        file_b = await file_factory(folder_id=folder.id)
        printer = await printer_factory()

        for _ in range(2):
            await archive_factory(
                printer.id,
                project_id=project.id,
                library_file_id=file_a.id,
                filename="renamed_on_dispatch.gcode.3mf",
            )
        await archive_factory(printer.id, project_id=project.id, library_file_id=file_b.id)

        response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress")
        assert response.status_code == 200
        counts = {row["file_id"]: row["completed_count"] for row in response.json()}
        assert counts == {file_a.id: 2, file_b.id: 1}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_hash_and_filename_fallback(
        self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory
    ):
        """Historical archives without library_file_id match by content hash,
        then by filename."""
        project = await project_factory()
        folder = await folder_factory(project_id=project.id)
        hashed_file = await file_factory(folder_id=folder.id, file_hash="a" * 64)
        named_file = await file_factory(folder_id=folder.id, filename="unique_name.gcode.3mf")
        printer = await printer_factory()

        # Hash match despite a different filename
        await archive_factory(
            printer.id, project_id=project.id, content_hash="a" * 64, filename="printer_copy.gcode.3mf"
        )
        # Filename match with no hash on either side
        await archive_factory(printer.id, project_id=project.id, filename="unique_name.gcode.3mf")

        response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress")
        counts = {row["file_id"]: row["completed_count"] for row in response.json()}
        assert counts == {hashed_file.id: 1, named_file.id: 1}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_only_completed_runs_count(
        self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory
    ):
        """Failed runs and never-printed archives do not advance the count."""
        project = await project_factory()
        folder = await folder_factory(project_id=project.id)
        lib_file = await file_factory(folder_id=folder.id)
        printer = await printer_factory()

        await archive_factory(printer.id, project_id=project.id, library_file_id=lib_file.id)
        await archive_factory(
            printer.id, project_id=project.id, library_file_id=lib_file.id, status="failed", run_status="failed"
        )
        await archive_factory(printer.id, project_id=project.id, library_file_id=lib_file.id, with_run=False)

        response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress")
        counts = {row["file_id"]: row["completed_count"] for row in response.json()}
        assert counts == {lib_file.id: 1}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_scoped_to_project(
        self, async_client: AsyncClient, project_factory, folder_factory, file_factory, printer_factory, archive_factory
    ):
        """Runs of the same file outside the project (no project / another
        project) are excluded."""
        project = await project_factory()
        other_project = await project_factory()
        folder = await folder_factory(project_id=project.id)
        lib_file = await file_factory(folder_id=folder.id)
        printer = await printer_factory()

        await archive_factory(printer.id, project_id=None, library_file_id=lib_file.id)
        await archive_factory(printer.id, project_id=other_project.id, library_file_id=lib_file.id)

        response = await async_client.get(f"/api/v1/projects/{project.id}/file-progress")
        assert response.json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unknown_project_404(self, async_client: AsyncClient):
        response = await async_client.get("/api/v1/projects/999999/file-progress")
        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_target_sets_roundtrip(self, async_client: AsyncClient):
        """target_sets survives create, update, and explicit-null clearing."""
        create = await async_client.post("/api/v1/projects/", json={"name": "Sets Project", "target_sets": 10})
        assert create.status_code == 200
        project = create.json()
        assert project["target_sets"] == 10

        update = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"target_sets": 4})
        assert update.status_code == 200, update.json()
        assert update.json()["target_sets"] == 4

        cleared = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"target_sets": None})
        assert cleared.json()["target_sets"] is None

        untouched = await async_client.patch(f"/api/v1/projects/{project['id']}", json={"name": "Renamed"})
        assert untouched.json()["target_sets"] is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_add_to_queue_inherits_folder_project(
        self, async_client: AsyncClient, project_factory, folder_factory, file_factory, db_session, tmp_path
    ):
        """Queueing a file from a project-linked folder attributes the queue
        item (and thus the later archive) to that project; a root file stays
        unattributed."""
        from sqlalchemy import select

        from backend.app.models.print_queue import PrintQueueItem

        project = await project_factory()
        folder = await folder_factory(project_id=project.id)

        on_disk = tmp_path / "linked.gcode.3mf"
        on_disk.write_bytes(b"fake sliced content")
        linked_file = await file_factory(folder_id=folder.id, file_path=str(on_disk))

        root_disk = tmp_path / "root.gcode.3mf"
        root_disk.write_bytes(b"fake sliced content")
        root_file = await file_factory(folder_id=None, file_path=str(root_disk))

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue", json={"file_ids": [linked_file.id, root_file.id]}
        )
        assert response.status_code == 200
        assert len(response.json()["added"]) == 2

        result = await db_session.execute(
            select(PrintQueueItem.library_file_id, PrintQueueItem.project_id).where(
                PrintQueueItem.library_file_id.in_([linked_file.id, root_file.id])
            )
        )
        projects_by_file = dict(result.all())
        assert projects_by_file[linked_file.id] == project.id
        assert projects_by_file[root_file.id] is None


class TestSoftDeletedArchivesLeaveTheProject:
    """Deleting a print removes it from its project, everywhere (#2731).

    The default archive delete is soft (#1343): the files go, the row stays so
    global Quick Stats keeps counting its filament / time / cost. Nothing in the
    projects module filtered on that, so a deleted print stayed listed on the
    project with a thumbnail pointing at a file that no longer existed — and
    could not be unassigned, because the only unassign UI lives on the Archives
    page, which correctly hides it.

    Unlike Quick Stats, project *counts* exclude it too. A project is a piece of
    work with a definite membership, not a lifetime total, so a project that
    lists one print must not claim two.
    """

    @pytest.fixture
    async def project_factory(self, db_session):
        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {"name": "Deleted Archive Project", "color": "#FF0000"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def archive_factory(self, db_session):
        """Archive + matching PrintLogEntry, as production always writes both."""

        async def _create_archive(**kwargs):
            from backend.app.models.archive import PrintArchive
            from backend.app.models.print_log import PrintLogEntry

            defaults = {
                "filename": "test.3mf",
                "file_path": "test/test.3mf",
                "file_size": 1000,
                "print_name": "Test Print",
                "status": "completed",
                "quantity": 1,
                "thumbnail_path": "test/thumb.png",
            }
            defaults.update(kwargs)
            archive = PrintArchive(**defaults)
            db_session.add(archive)
            await db_session.commit()
            await db_session.refresh(archive)

            db_session.add(
                PrintLogEntry(
                    archive_id=archive.id,
                    print_name=archive.print_name,
                    status=archive.status,
                    filament_used_grams=10.0,
                )
            )
            await db_session.commit()
            return archive

        return _create_archive

    @staticmethod
    async def _soft_delete(db_session, archive) -> int:
        """Soft-delete *archive* and return its id.

        The commit expires the instance, so reading an attribute off it
        afterwards is lazy IO outside the greenlet context (MissingGreenlet).
        Callers take the id from here instead.
        """
        from datetime import datetime, timezone

        archive_id = archive.id
        archive.deleted_at = datetime.now(timezone.utc)
        await db_session.commit()
        return archive_id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleted_archive_is_not_listed_on_the_project(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """The reported symptom: a card with a broken preview image."""
        project = await project_factory()
        await archive_factory(project_id=project.id, print_name="Kept")
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        await self._soft_delete(db_session, gone)

        response = await async_client.get(f"/api/v1/projects/{project.id}/archives")
        assert response.status_code == 200
        assert [a["print_name"] for a in response.json()] == ["Kept"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleted_archive_is_not_a_preview_on_the_project_card(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """The overview page renders these as thumbnails too, so it broke there
        as well — not just on the detail page."""
        project = await project_factory()
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        await self._soft_delete(db_session, gone)

        response = await async_client.get("/api/v1/projects/")
        assert response.status_code == 200
        row = next(p for p in response.json() if p["id"] == project.id)
        assert row["archives"] == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_project_counts_exclude_the_deleted_archive(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """The list shows one print, so the count must say one."""
        project = await project_factory()
        await archive_factory(project_id=project.id, print_name="Kept")
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        await self._soft_delete(db_session, gone)

        response = await async_client.get("/api/v1/projects/")
        row = next(p for p in response.json() if p["id"] == project.id)
        assert row["archive_count"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_project_stats_exclude_the_deleted_archive(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """Deliberate divergence from #1343: the contribution leaves the project
        even though it stays in global Quick Stats."""
        project = await project_factory()
        await archive_factory(project_id=project.id, print_name="Kept")
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        await self._soft_delete(db_session, gone)

        response = await async_client.get(f"/api/v1/projects/{project.id}")
        assert response.status_code == 200
        stats = response.json()["stats"]
        assert stats["total_archives"] == 1
        assert stats["total_filament_grams"] == pytest.approx(10.0)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleted_archive_is_not_in_the_project_timeline(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """A timeline entry for it links to an archive that 404s when clicked."""
        project = await project_factory()
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        await self._soft_delete(db_session, gone)

        response = await async_client.get(f"/api/v1/projects/{project.id}/timeline")
        assert response.status_code == 200
        assert not any(e.get("description") == "Deleted" for e in response.json())

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_live_archive_is_untouched_by_all_of_this(
        self, async_client: AsyncClient, project_factory, archive_factory
    ):
        """The filter must not cost a project its actual prints."""
        project = await project_factory()
        await archive_factory(project_id=project.id, print_name="Kept")

        listing = await async_client.get(f"/api/v1/projects/{project.id}/archives")
        assert [a["print_name"] for a in listing.json()] == ["Kept"]

        stats = await async_client.get(f"/api/v1/projects/{project.id}")
        assert stats.json()["stats"]["total_archives"] == 1

        row = next(p for p in (await async_client.get("/api/v1/projects/")).json() if p["id"] == project.id)
        assert row["archive_count"] == 1
        assert len(row["archives"]) == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassigning_an_already_orphaned_link_still_works(
        self, async_client: AsyncClient, project_factory, archive_factory, db_session
    ):
        """The listings hide it, but the API must still be able to clear the
        link — that is the repair path for rows written before this fix."""
        from sqlalchemy import select

        from backend.app.models.archive import PrintArchive

        project = await project_factory()
        gone = await archive_factory(project_id=project.id, print_name="Deleted")
        gone_id = await self._soft_delete(db_session, gone)

        response = await async_client.post(
            f"/api/v1/projects/{project.id}/remove-archives", json={"archive_ids": [gone_id]}
        )
        assert response.status_code == 200

        db_session.expire_all()
        result = await db_session.execute(select(PrintArchive.project_id).where(PrintArchive.id == gone_id))
        assert result.scalar_one() is None


class TestSubProjectRollup:
    """Tests for #1264 — nesting projects and rolling their figures up.

    The parent/child columns predate this; what these cover is the roll-up,
    the cycle guard that a roll-up needs to terminate, and what a delete does
    to the branch hanging off it.
    """

    @pytest.fixture
    async def project_factory(self, db_session):
        async def _create_project(**kwargs):
            from backend.app.models.project import Project

            defaults = {"name": "Rollup Project", "color": "#FF0000"}
            defaults.update(kwargs)
            project = Project(**defaults)
            db_session.add(project)
            await db_session.commit()
            await db_session.refresh(project)
            return project

        return _create_project

    @pytest.fixture
    async def run_factory(self, db_session):
        """One completed run against a project, with figures worth summing."""

        async def _create_run(project_id, *, grams=100.0, cost=5.0, seconds=3600, status="completed", quantity=1):
            from backend.app.models.archive import PrintArchive
            from backend.app.models.print_log import PrintLogEntry

            archive = PrintArchive(
                filename="test.3mf",
                file_path="test/test.3mf",
                file_size=1000,
                print_name="Run",
                status=status,
                quantity=quantity,
                project_id=project_id,
            )
            db_session.add(archive)
            await db_session.commit()
            await db_session.refresh(archive)

            db_session.add(
                PrintLogEntry(
                    archive_id=archive.id,
                    print_name=archive.print_name,
                    status=status,
                    duration_seconds=seconds,
                    filament_used_grams=grams,
                    cost=cost,
                )
            )
            await db_session.commit()
            return archive

        return _create_run

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_master_project_rolls_up_every_sub_project(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """The whole point of the feature: one number for the programme."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        tail = await project_factory(name="Tail", parent_id=master.id)

        await run_factory(master.id, grams=10.0, cost=1.0, seconds=3600)
        await run_factory(wing.id, grams=20.0, cost=2.0, seconds=7200)
        await run_factory(tail.id, grams=30.0, cost=3.0, seconds=1800)

        body = (await async_client.get(f"/api/v1/projects/{master.id}")).json()

        assert body["descendant_count"] == 2
        assert body["rollup_stats"]["total_archives"] == 3
        assert body["rollup_stats"]["total_filament_grams"] == 60.0
        assert body["rollup_stats"]["estimated_cost"] == 6.0
        assert body["rollup_stats"]["total_print_time_hours"] == 3.5

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_masters_own_stats_still_mean_its_own_prints(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """``stats`` keeps its existing meaning — anyone who nested projects
        over the API before this shipped must not see their figures restated."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        await run_factory(master.id, grams=10.0)
        await run_factory(wing.id, grams=20.0)

        body = (await async_client.get(f"/api/v1/projects/{master.id}")).json()

        assert body["stats"]["total_archives"] == 1
        assert body["stats"]["total_filament_grams"] == 10.0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_roll_up_reaches_past_the_first_generation(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """Nesting is arbitrary depth, so a grandchild has to count too."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        spar = await project_factory(name="Spar", parent_id=wing.id)
        await run_factory(spar.id, grams=50.0)

        body = (await async_client.get(f"/api/v1/projects/{master.id}")).json()

        assert body["descendant_count"] == 2
        assert body["rollup_stats"]["total_filament_grams"] == 50.0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_childless_project_reports_no_roll_up_at_all(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """Null, not a copy of ``stats`` — the page uses the absence to stay
        quiet rather than printing the same figures twice."""
        lonely = await project_factory(name="Solo")
        await run_factory(lonely.id)

        body = (await async_client.get(f"/api/v1/projects/{lonely.id}")).json()

        assert body["rollup_stats"] is None
        assert body["descendant_count"] == 0
        assert body["children"] == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_each_listed_child_carries_its_own_branch_total(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """Otherwise the listed rows do not add up to the master's total and
        the page contradicts itself."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        spar = await project_factory(name="Spar", parent_id=wing.id)
        await run_factory(wing.id, grams=20.0, cost=2.0)
        await run_factory(spar.id, grams=30.0, cost=3.0)

        body = (await async_client.get(f"/api/v1/projects/{master.id}")).json()

        assert len(body["children"]) == 1
        row = body["children"][0]
        assert row["name"] == "Wing"
        assert row["descendant_count"] == 1
        assert row["total_archives"] == 2
        assert row["total_filament_grams"] == 50.0
        assert row["total_cost"] == 5.0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_roll_up_progress_measures_against_the_summed_targets(
        self, async_client: AsyncClient, project_factory, run_factory
    ):
        """A target on each part of the tree is a target for the whole."""
        master = await project_factory(name="Airframe", target_count=2)
        wing = await project_factory(name="Wing", parent_id=master.id, target_count=2)
        await run_factory(master.id)
        await run_factory(wing.id)
        await run_factory(wing.id)

        body = (await async_client.get(f"/api/v1/projects/{master.id}")).json()

        assert body["stats"]["progress_percent"] == 50.0  # 1 of its own 2
        assert body["rollup_stats"]["progress_percent"] == 75.0  # 3 of the tree's 4
        assert body["rollup_stats"]["remaining_prints"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_project_cannot_be_moved_under_its_own_sub_project(
        self, async_client: AsyncClient, project_factory
    ):
        """Rejecting only the direct self-parent left A -> B -> A reachable in
        two calls, and a cycle has no root to roll anything up to."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)

        response = await async_client.patch(f"/api/v1/projects/{master.id}", json={"parent_id": wing.id})

        assert response.status_code == 400
        assert "sub-projects" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_guard_reaches_a_distant_descendant_too(self, async_client: AsyncClient, project_factory):
        """A three-deep loop is no more legal than a two-deep one."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        spar = await project_factory(name="Spar", parent_id=wing.id)

        response = await async_client.patch(f"/api/v1/projects/{master.id}", json={"parent_id": spar.id})

        assert response.status_code == 400

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_unrelated_project_is_still_a_legal_parent(self, async_client: AsyncClient, project_factory):
        """The guard must not refuse ordinary nesting."""
        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        other = await project_factory(name="Ground Station")

        response = await async_client.patch(f"/api/v1/projects/{other.id}", json={"parent_id": wing.id})

        assert response.status_code == 200
        assert response.json()["parent_id"] == wing.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_cycle_already_in_the_database_does_not_hang_the_roll_up(
        self, async_client: AsyncClient, project_factory, db_session
    ):
        """Databases written before the guard was widened can hold A -> B -> A.
        Reading one has to terminate, not spin."""
        from sqlalchemy import update as sa_update

        from backend.app.models.project import Project

        first = await project_factory(name="First")
        second = await project_factory(name="Second", parent_id=first.id)
        # Straight to the table: the API now refuses to write this.
        await db_session.execute(sa_update(Project).where(Project.id == first.id).values(parent_id=second.id))
        await db_session.commit()

        response = await async_client.get(f"/api/v1/projects/{first.id}")

        assert response.status_code == 200
        assert response.json()["descendant_count"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleting_a_middle_layer_promotes_its_children(
        self, async_client: AsyncClient, project_factory, db_session
    ):
        """Collapse the tree by one rather than scattering the branch."""
        from sqlalchemy import select

        from backend.app.models.project import Project

        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        spar = await project_factory(name="Spar", parent_id=wing.id)
        # Read the ids out before expiring: an expired instance refreshes itself
        # on attribute access, which is a lazy load in a sync frame.
        master_id, spar_id = master.id, spar.id

        response = await async_client.delete(f"/api/v1/projects/{wing.id}")
        assert response.status_code == 200

        db_session.expire_all()
        parent_id = (await db_session.execute(select(Project.parent_id).where(Project.id == spar_id))).scalar_one()
        assert parent_id == master_id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleting_a_top_level_project_frees_its_children(
        self, async_client: AsyncClient, project_factory, db_session
    ):
        """Nothing to promote to, so the child becomes top-level — and the
        delete has to succeed at all, which the bare FK would have refused."""
        from sqlalchemy import select

        from backend.app.models.project import Project

        master = await project_factory(name="Airframe")
        wing = await project_factory(name="Wing", parent_id=master.id)
        wing_id = wing.id  # See the sibling test: expiring invalidates the instance.

        response = await async_client.delete(f"/api/v1/projects/{master.id}")
        assert response.status_code == 200

        db_session.expire_all()
        parent_id = (await db_session.execute(select(Project.parent_id).where(Project.id == wing_id))).scalar_one()
        assert parent_id is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_grid_can_tell_a_sub_project_from_a_top_level_one(
        self, async_client: AsyncClient, project_factory
    ):
        """Without these the list view shows eight sub-projects as eight
        unrelated ones."""
        master = await project_factory(name="Airframe")
        await project_factory(name="Wing", parent_id=master.id)

        rows = {p["name"]: p for p in (await async_client.get("/api/v1/projects/")).json()}

        assert rows["Airframe"]["parent_id"] is None
        assert rows["Airframe"]["child_count"] == 1
        assert rows["Wing"]["parent_id"] == master.id
        assert rows["Wing"]["child_count"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_filtered_listing_still_admits_to_its_hidden_children(
        self, async_client: AsyncClient, project_factory
    ):
        """A parent that claimed no children would invite deleting it as if
        nothing hung off it."""
        master = await project_factory(name="Airframe", status="active")
        await project_factory(name="Wing", parent_id=master.id, status="completed")

        rows = {p["name"]: p for p in (await async_client.get("/api/v1/projects/?status=active")).json()}

        assert "Wing" not in rows
        assert rows["Airframe"]["child_count"] == 1