bambuddy/backend/tests/integration/test_library_api.py

"""Integration tests for Library API endpoints."""

import io
import tempfile
import zipfile
from pathlib import Path

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from backend.app.core.config import settings as app_settings
from backend.app.models.print_queue import PrintQueueItem


async def _read_queue_item(db_session, item_id: int) -> PrintQueueItem:
    """Re-read a queue row the route just committed through its own session."""
    db_session.expire_all()
    return (await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()


class TestLibraryFoldersAPI:
    """Integration tests for library folders endpoints."""

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

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

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

            defaults = {
                "name": f"Test Folder {counter}",
            }
            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.mark.asyncio
    @pytest.mark.integration
    async def test_list_folders_empty(self, async_client: AsyncClient, db_session):
        """Verify empty folder list returns empty array."""
        response = await async_client.get("/api/v1/library/folders")
        assert response.status_code == 200
        assert response.json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_folder_tree_exposes_latest_activity_at_from_files(
        self, async_client: AsyncClient, folder_factory, db_session
    ):
        """#1770: folder list returns latest_activity_at = MAX(folder.updated_at,
        MAX(immediate-child file.updated_at)) so the frontend can sort by
        recent activity. Adding a file with a later updated_at must bubble it.
        """
        from datetime import datetime, timedelta

        from backend.app.models.library import LibraryFile

        folder = await folder_factory(name="Active Folder")
        # File whose updated_at is well after the folder's. Activity should
        # surface this timestamp, not the folder's stale one.
        future = datetime.utcnow() + timedelta(hours=24)
        db_session.add(
            LibraryFile(
                folder_id=folder.id,
                filename="model.3mf",
                file_path="library/model.3mf",
                file_type="3mf",
                file_size=123,
                updated_at=future,
            )
        )
        await db_session.commit()

        response = await async_client.get("/api/v1/library/folders")
        assert response.status_code == 200
        items = response.json()
        assert len(items) == 1
        item = items[0]
        assert item["id"] == folder.id
        assert item["latest_activity_at"] is not None
        # latest_activity_at should be at least the future stamp we set.
        assert item["latest_activity_at"] >= future.isoformat()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_folder_tree_latest_activity_at_falls_back_to_folder_updated_at(
        self, async_client: AsyncClient, folder_factory, db_session
    ):
        """#1770: a folder with no files reports its own updated_at, not null —
        otherwise the activity sort would dump every empty folder to one end."""
        await folder_factory(name="Empty Folder")
        response = await async_client.get("/api/v1/library/folders")
        assert response.status_code == 200
        items = response.json()
        assert len(items) == 1
        item = items[0]
        # latest_activity_at == folder.updated_at when there are no files
        assert item["latest_activity_at"] is not None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_folder(self, async_client: AsyncClient, db_session):
        """Verify folder can be created."""
        data = {"name": "New Folder"}
        response = await async_client.post("/api/v1/library/folders", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["name"] == "New Folder"
        assert result["id"] is not None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_nested_folder(self, async_client: AsyncClient, folder_factory, db_session):
        """Verify nested folder can be created."""
        parent = await folder_factory(name="Parent")
        data = {"name": "Child", "parent_id": parent.id}
        response = await async_client.post("/api/v1/library/folders", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["name"] == "Child"
        assert result["parent_id"] == parent.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_folder(self, async_client: AsyncClient, folder_factory, db_session):
        """Verify single folder can be retrieved."""
        folder = await folder_factory(name="Test Folder")
        response = await async_client.get(f"/api/v1/library/folders/{folder.id}")
        assert response.status_code == 200
        result = response.json()
        assert result["id"] == folder.id
        assert result["name"] == "Test Folder"

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

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update_folder(self, async_client: AsyncClient, folder_factory, db_session):
        """Verify folder can be updated."""
        folder = await folder_factory(name="Old Name")
        data = {"name": "New Name"}
        response = await async_client.put(f"/api/v1/library/folders/{folder.id}", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["name"] == "New Name"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_folder(self, async_client: AsyncClient, folder_factory, db_session):
        """Verify folder can be deleted."""
        folder = await folder_factory()
        response = await async_client.delete(f"/api/v1/library/folders/{folder.id}")
        assert response.status_code == 200
        result = response.json()
        assert result.get("message") or result.get("success", True)


class TestLibraryFilesAPI:
    """Integration tests for library files endpoints."""

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

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

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

            defaults = {"name": f"Test Folder {counter}"}
            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):
        """Factory to create test files."""
        _counter = [0]

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

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

            defaults = {
                "filename": f"test_file_{counter}.3mf",
                "file_path": f"/test/path/test_file_{counter}.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_list_files_empty(self, async_client: AsyncClient, db_session):
        """Verify empty file list returns empty array."""
        response = await async_client.get("/api/v1/library/files")
        assert response.status_code == 200
        assert response.json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_in_folder(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
        """Verify files can be filtered by folder."""
        folder = await folder_factory()
        file1 = await file_factory(folder_id=folder.id)
        await file_factory()  # File in root (no folder)

        response = await async_client.get(f"/api/v1/library/files?folder_id={folder.id}")
        assert response.status_code == 200
        result = response.json()
        assert len(result) == 1
        assert result[0]["id"] == file1.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_by_project_id(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
        """#932: project_id filter returns files across all folders linked to the project.

        Replaces the prior N+1 pattern where the frontend fired one request per
        linked folder. A single JOIN query must return every file in folders whose
        project_id matches, while excluding files from unlinked folders.
        """
        from backend.app.models.project import Project

        project = Project(name="Test Project for Files", color="#00ff00")
        db_session.add(project)
        await db_session.commit()
        await db_session.refresh(project)

        folder_a = await folder_factory(name="Folder A", project_id=project.id)
        folder_b = await folder_factory(name="Folder B", project_id=project.id)
        other_folder = await folder_factory(name="Unlinked")

        linked_a = await file_factory(folder_id=folder_a.id, filename="a.3mf")
        linked_b = await file_factory(folder_id=folder_b.id, filename="b.3mf")
        await file_factory(folder_id=other_folder.id, filename="unlinked.3mf")
        await file_factory(filename="root.3mf")  # no folder → not part of any project

        response = await async_client.get(f"/api/v1/library/files?project_id={project.id}")
        assert response.status_code == 200
        result = response.json()
        ids = {f["id"] for f in result}
        assert ids == {linked_a.id, linked_b.id}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_folder_id_takes_precedence_over_project_id(
        self, async_client: AsyncClient, folder_factory, file_factory, db_session
    ):
        """When both folder_id and project_id are passed, folder_id wins.

        Documented precedence in list_files(): folder_id > project_id > include_root.
        This guards the behavior so a future refactor can't silently flip it.
        """
        from backend.app.models.project import Project

        project = Project(name="Precedence Project")
        db_session.add(project)
        await db_session.commit()
        await db_session.refresh(project)

        folder_linked = await folder_factory(name="Linked", project_id=project.id)
        folder_other = await folder_factory(name="Other")

        await file_factory(folder_id=folder_linked.id, filename="linked.3mf")
        other_file = await file_factory(folder_id=folder_other.id, filename="other.3mf")

        # folder_id points at a folder that is NOT in the project — must return
        # that folder's contents and ignore project_id entirely.
        response = await async_client.get(f"/api/v1/library/files?folder_id={folder_other.id}&project_id={project.id}")
        assert response.status_code == 200
        result = response.json()
        assert len(result) == 1
        assert result[0]["id"] == other_file.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_internal_only(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
        """#1621: `internal_only=true` restricts the listing to files in managed
        storage (`is_external=False`) so a linked NAS with hundreds of files
        doesn't drown the user's own uploads in the "All Files" sidebar view."""
        internal_folder = await folder_factory(name="My uploads")
        external_folder = await folder_factory(name="NAS", is_external=True, external_path="/mnt/nas")

        internal_file = await file_factory(folder_id=internal_folder.id, filename="mine.3mf", is_external=False)
        await file_factory(folder_id=external_folder.id, filename="nas.3mf", is_external=True)
        root_file = await file_factory(filename="root.3mf", is_external=False)  # Root-uploaded is always internal.

        response = await async_client.get("/api/v1/library/files?include_root=false&internal_only=true")
        assert response.status_code == 200
        ids = {f["id"] for f in response.json()}
        assert ids == {internal_file.id, root_file.id}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_external_only(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
        """#1621 symmetric: `external_only=true` returns the combined view
        across every linked external folder so users with several mounts can
        see all external content in one place without clicking each folder."""
        internal_folder = await folder_factory(name="My uploads")
        nas_a = await folder_factory(name="NAS A", is_external=True, external_path="/mnt/a")
        nas_b = await folder_factory(name="NAS B", is_external=True, external_path="/mnt/b")

        await file_factory(folder_id=internal_folder.id, filename="mine.3mf", is_external=False)
        ext_a = await file_factory(folder_id=nas_a.id, filename="a.3mf", is_external=True)
        ext_b = await file_factory(folder_id=nas_b.id, filename="b.3mf", is_external=True)

        response = await async_client.get("/api/v1/library/files?include_root=false&external_only=true")
        assert response.status_code == 200
        ids = {f["id"] for f in response.json()}
        assert ids == {ext_a.id, ext_b.id}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_internal_and_external_mutually_exclusive(self, async_client: AsyncClient, db_session):
        """Both flags together is a caller bug — fail loud (400) rather than
        silently picking one, so a frontend regression is caught immediately."""
        response = await async_client.get("/api/v1/library/files?internal_only=true&external_only=true")
        assert response.status_code == 400
        assert "mutually exclusive" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_file(self, async_client: AsyncClient, file_factory, db_session):
        """Verify single file can be retrieved."""
        lib_file = await file_factory(filename="test.3mf")
        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
        assert response.status_code == 200
        result = response.json()
        assert result["id"] == lib_file.id
        assert result["filename"] == "test.3mf"

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

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_file(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file can be deleted."""
        lib_file = await file_factory()
        response = await async_client.delete(f"/api/v1/library/files/{lib_file.id}")
        assert response.status_code == 200
        result = response.json()
        assert result.get("message") or result.get("success", True)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_rename_file(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file can be renamed."""
        lib_file = await file_factory(filename="old_name.3mf")
        data = {"filename": "new_name.3mf"}
        response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["filename"] == "new_name.3mf"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_rename_file_invalid_path_separator(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file rename fails with a forward slash (FAT32-illegal, #1540)."""
        lib_file = await file_factory(filename="test.3mf")
        data = {"filename": "path/to/file.3mf"}
        response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
        assert response.status_code == 400
        assert "invalid character" in response.json()["detail"].lower()
        assert "/" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_rename_file_invalid_backslash(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file rename fails with a backslash (FAT32-illegal, #1540)."""
        lib_file = await file_factory(filename="test.3mf")
        data = {"filename": "path\\to\\file.3mf"}
        response = await async_client.put(f"/api/v1/library/files/{lib_file.id}", json=data)
        assert response.status_code == 400
        assert "invalid character" in response.json()["detail"].lower()
        assert "\\" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_stats(self, async_client: AsyncClient, folder_factory, file_factory, db_session):
        """Verify library stats endpoint returns counts."""
        await folder_factory()
        await folder_factory()
        await file_factory()

        response = await async_client.get("/api/v1/library/stats")
        assert response.status_code == 200
        result = response.json()
        assert result["total_folders"] == 2
        assert result["total_files"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_file_list_includes_user_tracking_fields(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file list response includes user tracking fields (Issue #206)."""
        lib_file = await file_factory(filename="test.3mf")
        response = await async_client.get("/api/v1/library/files?include_root=false")
        assert response.status_code == 200
        result = response.json()
        assert len(result) >= 1
        # Find our test file
        test_file = next((f for f in result if f["id"] == lib_file.id), None)
        assert test_file is not None
        # User tracking fields should be present (even if null)
        assert "created_by_id" in test_file
        assert "created_by_username" in test_file

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_file_detail_includes_user_tracking_fields(self, async_client: AsyncClient, file_factory, db_session):
        """Verify file detail response includes user tracking fields (Issue #206)."""
        lib_file = await file_factory(filename="test_detail.3mf")
        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
        assert response.status_code == 200
        result = response.json()
        # User tracking fields should be present (even if null)
        assert "created_by_id" in result
        assert "created_by_username" in result

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_file_with_user_tracking(self, async_client: AsyncClient, db_session):
        """Verify file created with user shows username in response (Issue #206)."""
        from backend.app.models.library import LibraryFile
        from backend.app.models.user import User

        # Create a test user
        user = User(username="testuploader", password_hash="fakehash", role="user")
        db_session.add(user)
        await db_session.flush()

        # Create a file with created_by_id set
        lib_file = LibraryFile(
            filename="user_uploaded.3mf",
            file_path="/test/user_uploaded.3mf",
            file_size=2048,
            file_type="3mf",
            created_by_id=user.id,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)

        # Verify file detail shows username
        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}")
        assert response.status_code == 200
        result = response.json()
        assert result["created_by_id"] == user.id
        assert result["created_by_username"] == "testuploader"

        # Verify file list also shows username
        response = await async_client.get("/api/v1/library/files?include_root=false")
        assert response.status_code == 200
        files = response.json()
        test_file = next((f for f in files if f["id"] == lib_file.id), None)
        assert test_file is not None
        assert test_file["created_by_id"] == user.id
        assert test_file["created_by_username"] == "testuploader"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_recursive_includes_subfolders(
        self, async_client: AsyncClient, folder_factory, file_factory
    ):
        """#1268: ?recursive=true with folder_id must include every descendant.

        Tree:
            toys/             ← f_toys, direct file "robot_top.3mf"
              cars/           ← child of toys, file "robot_car.3mf"
                race/         ← grandchild, file "robot_race.3mf"
            other/            ← unrelated, file "robot_other.3mf" (must NOT appear)
        """
        toys = await folder_factory(name="toys")
        cars = await folder_factory(name="cars", parent_id=toys.id)
        race = await folder_factory(name="race", parent_id=cars.id)
        other = await folder_factory(name="other")

        top = await file_factory(folder_id=toys.id, filename="robot_top.3mf")
        mid = await file_factory(folder_id=cars.id, filename="robot_car.3mf")
        deep = await file_factory(folder_id=race.id, filename="robot_race.3mf")
        await file_factory(folder_id=other.id, filename="robot_other.3mf")

        # Non-recursive: only the file directly under toys.
        r = await async_client.get(f"/api/v1/library/files?folder_id={toys.id}")
        assert r.status_code == 200
        assert {f["id"] for f in r.json()} == {top.id}

        # Recursive: toys + cars + race files, but NOT other/.
        r = await async_client.get(f"/api/v1/library/files?folder_id={toys.id}&recursive=true")
        assert r.status_code == 200
        assert {f["id"] for f in r.json()} == {top.id, mid.id, deep.id}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_recursive_without_folder_id_is_noop(
        self, async_client: AsyncClient, folder_factory, file_factory
    ):
        """recursive=true is meaningful only with folder_id — without it the
        existing include_root branch handles scoping. Just confirming the new
        param doesn't shadow that path."""
        folder = await folder_factory()
        f_in = await file_factory(folder_id=folder.id)
        f_root = await file_factory()

        r = await async_client.get("/api/v1/library/files?include_root=false&recursive=true")
        assert r.status_code == 200
        assert {f["id"] for f in r.json()} == {f_in.id, f_root.id}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_folder_readme_returns_first_markdown(
        self, async_client: AsyncClient, folder_factory, file_factory
    ):
        """#1268: /folders/{id}/readme reads on-disk content of the first .md."""
        folder = await folder_factory()
        with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
            f.write("# Robot\n\nA cute little robot.")
            md_path = f.name
        try:
            await file_factory(
                folder_id=folder.id,
                filename="README.md",
                file_path=md_path,
                file_type="md",
                file_size=Path(md_path).stat().st_size,
            )
            r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
            assert r.status_code == 200
            body = r.json()
            assert body["filename"] == "README.md"
            assert body["content"] == "# Robot\n\nA cute little robot."
            assert body["truncated"] is False
        finally:
            import os

            os.unlink(md_path)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_folder_readme_prefers_readme_over_other_md(
        self, async_client: AsyncClient, folder_factory, file_factory
    ):
        """When the folder has multiple .md files, README.md / description.md
        wins regardless of insertion order or filename case."""
        folder = await folder_factory()
        with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
            f.write("notes notes notes")
            notes_path = f.name
        with tempfile.NamedTemporaryFile(suffix=".md", delete=False, mode="w", encoding="utf-8") as f:
            f.write("the real one")
            readme_path = f.name
        try:
            # notes.md inserted FIRST — naive ordering would pick this one.
            await file_factory(
                folder_id=folder.id,
                filename="notes.md",
                file_path=notes_path,
                file_type="md",
            )
            await file_factory(
                folder_id=folder.id,
                filename="readme.md",  # lowercase to confirm case-insensitive match
                file_path=readme_path,
                file_type="md",
            )
            r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
            assert r.status_code == 200
            assert r.json()["filename"] == "readme.md"
            assert r.json()["content"] == "the real one"
        finally:
            import os

            os.unlink(notes_path)
            os.unlink(readme_path)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_folder_readme_404_when_no_markdown(
        self, async_client: AsyncClient, folder_factory, file_factory
    ):
        """No .md in the folder → 404 so the FE can hide the side panel."""
        folder = await folder_factory()
        await file_factory(folder_id=folder.id, filename="model.3mf", file_type="3mf")
        r = await async_client.get(f"/api/v1/library/folders/{folder.id}/readme")
        assert r.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_folder_readme_404_when_folder_missing(self, async_client: AsyncClient):
        r = await async_client.get("/api/v1/library/folders/999999/readme")
        assert r.status_code == 404


class TestLibraryAddToQueueAPI:
    """Integration tests for /api/v1/library/files/add-to-queue endpoint."""

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

        async def _create_printer(**kwargs):
            from backend.app.models.printer import Printer

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

            defaults = {
                "name": f"Test Printer {counter}",
                "ip_address": f"192.168.1.{100 + counter}",
                "serial_number": f"TESTSERIAL{counter:04d}",
                "access_code": "12345678",
                "model": "X1C",
            }
            defaults.update(kwargs)

            printer = Printer(**defaults)
            db_session.add(printer)
            await db_session.commit()
            await db_session.refresh(printer)
            return printer

        return _create_printer

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

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

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

            defaults = {
                "filename": f"test_file_{counter}.gcode.3mf",
                "file_path": f"/test/path/test_file_{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_library_file

    @pytest.fixture
    async def on_disk_file_factory(self, library_file_factory):
        """A library file whose bytes exist, so the route gets past its disk check."""
        written: list[Path] = []

        async def _create(**kwargs):
            counter = len(written) + 1
            rel_path = kwargs.pop("file_path", f"archive/library/files/queue_probe_{counter}.gcode.3mf")
            abs_path = Path(app_settings.base_dir) / rel_path
            abs_path.parent.mkdir(parents=True, exist_ok=True)
            abs_path.write_bytes(b"probe")
            written.append(abs_path)
            kwargs.setdefault("filename", f"queue_probe_{counter}.gcode.3mf")
            return await library_file_factory(file_path=rel_path, **kwargs)

        yield _create

        for path in written:
            path.unlink(missing_ok=True)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_add_to_queue_file_not_found(self, async_client: AsyncClient, printer_factory, db_session):
        """Nothing queued is not a success (#3112).

        This used to assert 200: the caller got an OK for a call that created
        nothing, with the reason in a body it had no cause to read. The reason
        is still reported, now where a failed call puts it.
        """
        await printer_factory()

        data = {"file_ids": [9999]}
        response = await async_client.post("/api/v1/library/files/add-to-queue", json=data)
        assert response.status_code == 400
        errors = response.json()["detail"]["errors"]
        assert len(errors) == 1
        assert errors[0]["file_id"] == 9999

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_add_non_sliced_file_to_queue_fails(
        self, async_client: AsyncClient, printer_factory, library_file_factory, db_session
    ):
        """Verify non-sliced file cannot be added to queue."""
        await printer_factory()
        lib_file = await library_file_factory(
            filename="model.stl",
            file_path="/test/path/model.stl",
            file_type="stl",
        )

        data = {"file_ids": [lib_file.id]}
        response = await async_client.post("/api/v1/library/files/add-to-queue", json=data)
        assert response.status_code == 400
        errors = response.json()["detail"]["errors"]
        assert len(errors) == 1
        assert "sliced" in errors[0]["error"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_partial_success_still_returns_200(
        self, async_client: AsyncClient, printer_factory, library_file_factory, on_disk_file_factory, db_session
    ):
        """Items really were created, so the call succeeded (#3112).

        The per-file errors ride along with them, which is the whole point of a
        bulk endpoint. Only a call that produced nothing is a failed call.
        """
        await printer_factory()
        good = await on_disk_file_factory()
        bad = await library_file_factory(filename="model.stl", file_path="/test/path/model.stl", file_type="stl")

        response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [good.id, bad.id]})
        assert response.status_code == 200
        result = response.json()
        assert [a["file_id"] for a in result["added"]] == [good.id]
        assert [e["file_id"] for e in result["errors"]] == [bad.id]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_target_model_is_inferred_so_the_item_can_be_dispatched(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        """#3112: an item with no printer and no target model is inert.

        The scheduler dispatches on `item.printer_id` or on
        `item.target_model or item.variants`; a row with neither matches no
        branch and waits forever. With an active X1C present, a file that says
        it was sliced for one is aimed at it.
        """
        await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})

        response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
        assert response.status_code == 200
        item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
        assert item.printer_id is None
        assert item.target_model == "X1C"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_printer_of_that_model_leaves_the_item_unassigned(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        """Aiming an item at hardware nobody owns would only look like progress.

        Having no H2D is the user's situation, not their mistake, so the file
        is still queued -- as the unassigned row it has always been.
        """
        await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "H2D"})

        response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
        assert response.status_code == 200
        item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
        assert item.printer_id is None
        assert item.target_model is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_explicit_printer_wins_over_the_files_own_model(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        printer = await printer_factory(model="X1C")
        # Read before the queue row is re-read: that expires the session, and
        # a lazy refresh of this row would then happen outside the greenlet.
        printer_id = printer.id
        lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue",
            json={"file_ids": [lib_file.id], "printer_id": printer_id},
        )
        assert response.status_code == 200
        item = await _read_queue_item(db_session, response.json()["added"][0]["queue_item_id"])
        assert item.printer_id == printer_id
        assert item.target_model is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_incompatible_target_model_is_refused_per_file(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        """The same cross-model gate POST /queue/ applies (#2578).

        The scheduler hands model-based items to hardware with no human in the
        loop, so a file sliced for one model must not be aimed at another.
        """
        await printer_factory(model="A1")
        lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "X1C"})

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue",
            json={"file_ids": [lib_file.id], "target_model": "A1"},
        )
        assert response.status_code == 400
        assert "cannot be dispatched" in response.json()["detail"]["errors"][0]["error"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_target_model_without_an_active_printer_is_refused(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory(file_metadata={"sliced_for_model": "H2D"})

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue",
            json={"file_ids": [lib_file.id], "target_model": "H2D"},
        )
        assert response.status_code == 400
        assert "No active printers" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_printer_and_target_model_together_are_refused(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        printer = await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory()

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue",
            json={"file_ids": [lib_file.id], "printer_id": printer.id, "target_model": "X1C"},
        )
        assert response.status_code == 400
        assert "both" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unknown_printer_is_refused(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory()

        response = await async_client.post(
            "/api/v1/library/files/add-to-queue",
            json={"file_ids": [lib_file.id], "printer_id": 999999},
        )
        assert response.status_code == 400
        assert response.json()["detail"] == "Printer not found"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_filename_the_printer_cannot_store_is_refused(
        self, async_client: AsyncClient, printer_factory, on_disk_file_factory, db_session
    ):
        """The Bambu SD card is FAT32; an illegal character 553s at upload.

        POST /queue/ has rejected these at queue time since #1540. This route
        did not, so the mistake surfaced as a print that failed later.
        """
        await printer_factory(model="X1C")
        lib_file = await on_disk_file_factory(filename="bad:name?.gcode.3mf")

        response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
        assert response.status_code == 400
        assert response.json()["detail"]["errors"][0]["file_id"] == lib_file.id


class TestLibraryZipExtractAPI:
    """Integration tests for ZIP extraction endpoint."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_invalid_file_type(self, async_client: AsyncClient, db_session):
        """Verify non-ZIP files are rejected."""
        # Create a fake file that's not a ZIP
        files = {"file": ("test.txt", b"This is not a zip file", "text/plain")}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
        assert response.status_code == 400
        assert "ZIP" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_basic(self, async_client: AsyncClient, db_session):
        """Verify basic ZIP extraction works."""
        import io

        # Create a simple ZIP file in memory
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("test1.txt", "Content of file 1")
            zf.writestr("test2.txt", "Content of file 2")
        zip_buffer.seek(0)

        files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 2
        assert len(result["files"]) == 2
        assert len(result["errors"]) == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_with_folders(self, async_client: AsyncClient, db_session):
        """Verify ZIP extraction preserves folder structure."""
        import io

        # Create a ZIP file with folder structure
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("folder1/file1.txt", "Content 1")
            zf.writestr("folder1/subfolder/file2.txt", "Content 2")
            zf.writestr("folder2/file3.txt", "Content 3")
        zip_buffer.seek(0)

        files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
        params = {"preserve_structure": "true"}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 3
        assert result["folders_created"] >= 3  # folder1, folder1/subfolder, folder2

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_flat(self, async_client: AsyncClient, db_session):
        """Verify ZIP extraction can extract flat (no folders)."""
        import io

        # Create a ZIP file with folder structure
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("folder/file1.txt", "Content 1")
            zf.writestr("folder/file2.txt", "Content 2")
        zip_buffer.seek(0)

        files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
        params = {"preserve_structure": "false"}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 2
        assert result["folders_created"] == 0  # No folders created when flat

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_skips_macos_files(self, async_client: AsyncClient, db_session):
        """Verify ZIP extraction skips __MACOSX and hidden files."""
        import io

        # Create a ZIP file with macOS junk files
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("real_file.txt", "Real content")
            zf.writestr("__MACOSX/._real_file.txt", "macOS metadata")
            zf.writestr(".hidden_file", "Hidden content")
        zip_buffer.seek(0)

        files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 1  # Only real_file.txt
        assert result["files"][0]["filename"] == "real_file.txt"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_create_folder_from_zip(self, async_client: AsyncClient, db_session):
        """Verify ZIP extraction creates a folder from the ZIP filename."""
        import io

        # Create a ZIP file with some files
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("file1.txt", "Content 1")
            zf.writestr("file2.txt", "Content 2")
        zip_buffer.seek(0)

        files = {"file": ("MyProject.zip", zip_buffer.read(), "application/zip")}
        params = {"create_folder_from_zip": "true", "preserve_structure": "false"}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 2
        assert result["folders_created"] == 1  # MyProject folder created

        # Verify the files are in a folder
        assert result["files"][0]["folder_id"] is not None
        assert result["files"][1]["folder_id"] is not None
        # Both files should be in the same folder
        assert result["files"][0]["folder_id"] == result["files"][1]["folder_id"]

        # Verify the folder was created with the right name
        folder_response = await async_client.get(f"/api/v1/library/folders/{result['files'][0]['folder_id']}")
        assert folder_response.status_code == 200
        folder = folder_response.json()
        assert folder["name"] == "MyProject"


class TestLibraryStlThumbnailAPI:
    """Integration tests for STL thumbnail generation endpoints."""

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

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

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

            defaults = {
                "filename": f"test_model_{counter}.stl",
                "file_path": f"/test/path/test_model_{counter}.stl",
                "file_size": 1024,
                "file_type": "stl",
            }
            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_batch_generate_thumbnails_empty(self, async_client: AsyncClient, db_session):
        """Verify batch thumbnail generation with no files."""
        data = {"all_missing": True}
        response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["processed"] == 0
        assert result["succeeded"] == 0
        assert result["failed"] == 0
        assert result["results"] == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_batch_generate_thumbnails_no_criteria(self, async_client: AsyncClient, db_session):
        """Verify batch thumbnail generation with no criteria returns empty."""
        data = {}
        response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["processed"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_batch_generate_thumbnails_file_not_on_disk(
        self, async_client: AsyncClient, file_factory, db_session
    ):
        """Verify batch thumbnail generation handles missing files gracefully."""
        # Create a file in DB but not on disk
        stl_file = await file_factory(
            filename="missing.stl",
            file_path="/nonexistent/path/missing.stl",
            thumbnail_path=None,
        )

        data = {"file_ids": [stl_file.id]}
        response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
        assert response.status_code == 200
        result = response.json()
        assert result["processed"] == 1
        assert result["succeeded"] == 0
        assert result["failed"] == 1
        assert result["results"][0]["success"] is False
        assert "not found" in result["results"][0]["error"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_batch_generate_thumbnails_with_real_stl(self, async_client: AsyncClient, db_session):
        """Verify batch thumbnail generation with a real STL file."""
        from backend.app.models.library import LibraryFile

        # Create a simple ASCII STL cube
        stl_content = """solid cube
facet normal 0 0 -1
  outer loop
    vertex 0 0 0
    vertex 1 0 0
    vertex 1 1 0
  endloop
endfacet
facet normal 0 0 1
  outer loop
    vertex 0 0 1
    vertex 1 1 1
    vertex 1 0 1
  endloop
endfacet
endsolid cube"""

        with tempfile.NamedTemporaryFile(suffix=".stl", delete=False, mode="w") as f:
            f.write(stl_content)
            stl_path = f.name

        try:
            # Create file in DB pointing to real STL
            lib_file = LibraryFile(
                filename="test_cube.stl",
                file_path=stl_path,
                file_size=len(stl_content),
                file_type="stl",
                thumbnail_path=None,
            )
            db_session.add(lib_file)
            await db_session.commit()
            await db_session.refresh(lib_file)

            data = {"file_ids": [lib_file.id]}
            response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
            assert response.status_code == 200
            result = response.json()
            assert result["processed"] == 1
            # Result depends on whether trimesh/matplotlib are installed
            # Either succeeds or fails gracefully
            assert result["succeeded"] + result["failed"] == 1
        finally:
            import os

            if os.path.exists(stl_path):
                os.unlink(stl_path)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_upload_file_with_stl_thumbnail_param(self, async_client: AsyncClient, db_session):
        """Verify file upload accepts generate_stl_thumbnails parameter."""
        # Create a simple STL file
        stl_content = b"solid test\nendsolid test"

        files = {"file": ("test.stl", stl_content, "application/octet-stream")}
        params = {"generate_stl_thumbnails": "false"}
        response = await async_client.post("/api/v1/library/files", files=files, params=params)
        assert response.status_code == 200
        result = response.json()
        assert result["filename"] == "test.stl"
        assert result["file_type"] == "stl"
        # No thumbnail should be generated when disabled
        assert result["thumbnail_path"] is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_extract_zip_with_stl_thumbnail_param(self, async_client: AsyncClient, db_session):
        """Verify ZIP extraction accepts generate_stl_thumbnails parameter."""
        # Create a ZIP file containing an STL
        stl_content = b"solid test\nendsolid test"
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("model.stl", stl_content)
        zip_buffer.seek(0)

        files = {"file": ("test.zip", zip_buffer.read(), "application/zip")}
        params = {"generate_stl_thumbnails": "false"}
        response = await async_client.post("/api/v1/library/files/extract-zip", files=files, params=params)
        assert response.status_code == 200
        result = response.json()
        assert result["extracted"] == 1
        assert result["files"][0]["filename"] == "model.stl"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_batch_generate_thumbnails_by_folder(self, async_client: AsyncClient, file_factory, db_session):
        """Verify batch thumbnail generation can filter by folder."""
        from backend.app.models.library import LibraryFolder

        # Create a folder
        folder = LibraryFolder(name="STL Folder")
        db_session.add(folder)
        await db_session.commit()
        await db_session.refresh(folder)

        # Create STL file in folder (no thumbnail)
        stl_in_folder = await file_factory(
            filename="in_folder.stl",
            folder_id=folder.id,
            thumbnail_path=None,
        )

        # Create STL file at root (no thumbnail)
        _stl_at_root = await file_factory(
            filename="at_root.stl",
            folder_id=None,
            thumbnail_path=None,
        )

        # Request thumbnails only for files in folder
        data = {"folder_id": folder.id, "all_missing": True}
        response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
        assert response.status_code == 200
        result = response.json()
        # Should only process the file in the folder
        assert result["processed"] == 1
        assert result["results"][0]["file_id"] == stl_in_folder.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_batch_generate_thumbnails_all_missing(self, async_client: AsyncClient, file_factory, db_session):
        """Verify batch thumbnail generation finds all STL files missing thumbnails."""
        # Create files with and without thumbnails
        _stl_with_thumb = await file_factory(
            filename="with_thumb.stl",
            thumbnail_path="/some/path/thumb.png",
        )
        stl_without_thumb1 = await file_factory(
            filename="without_thumb1.stl",
            thumbnail_path=None,
        )
        stl_without_thumb2 = await file_factory(
            filename="without_thumb2.stl",
            thumbnail_path=None,
        )

        data = {"all_missing": True}
        response = await async_client.post("/api/v1/library/generate-stl-thumbnails", json=data)
        assert response.status_code == 200
        result = response.json()
        # Should only process files without thumbnails
        assert result["processed"] == 2
        file_ids = {r["file_id"] for r in result["results"]}
        assert stl_without_thumb1.id in file_ids
        assert stl_without_thumb2.id in file_ids


class TestLibraryPathHelpers:
    """Tests for path handling utilities used for backup portability."""

    def test_to_relative_path_converts_absolute(self):
        """Verify absolute paths are converted to relative paths."""
        from backend.app.api.routes.library import to_relative_path
        from backend.app.core.config import settings

        base_dir = str(settings.base_dir)
        abs_path = f"{base_dir}/archive/library/files/test.3mf"
        rel_path = to_relative_path(abs_path)

        assert not rel_path.startswith("/")
        assert rel_path == "archive/library/files/test.3mf"

    def test_to_relative_path_handles_path_object(self):
        """Verify Path objects are handled correctly."""
        from pathlib import Path

        from backend.app.api.routes.library import to_relative_path
        from backend.app.core.config import settings

        abs_path = Path(settings.base_dir) / "archive" / "test.3mf"
        rel_path = to_relative_path(abs_path)

        assert not rel_path.startswith("/")
        assert rel_path == "archive/test.3mf"

    def test_to_relative_path_returns_empty_for_empty_input(self):
        """Verify empty input returns empty string."""
        from backend.app.api.routes.library import to_relative_path

        assert to_relative_path("") == ""
        assert to_relative_path(None) == ""

    def test_to_absolute_path_converts_relative(self):
        """Verify relative paths are converted to absolute paths."""
        from backend.app.api.routes.library import to_absolute_path
        from backend.app.core.config import settings

        rel_path = "archive/library/files/test.3mf"
        abs_path = to_absolute_path(rel_path)

        assert abs_path is not None
        assert abs_path.is_absolute()
        assert str(abs_path) == f"{settings.base_dir}/archive/library/files/test.3mf"

    def test_to_absolute_path_handles_already_absolute(self):
        """Verify already absolute paths are returned as-is (for backwards compatibility)."""
        from backend.app.api.routes.library import to_absolute_path

        abs_path_str = "/data/archive/test.3mf"
        result = to_absolute_path(abs_path_str)

        assert result is not None
        assert str(result) == abs_path_str

    def test_to_absolute_path_returns_none_for_empty(self):
        """Verify None/empty input returns None."""
        from backend.app.api.routes.library import to_absolute_path

        assert to_absolute_path(None) is None
        assert to_absolute_path("") is None


class TestLibraryPermissions:
    """Tests for library permission enforcement."""

    @pytest.fixture
    async def auth_setup(self, db_session):
        """Set up auth with users of different permission levels."""
        from backend.app.core.auth import create_access_token, get_password_hash
        from backend.app.models.group import Group
        from backend.app.models.settings import Settings
        from backend.app.models.user import User

        # Enable auth
        settings = Settings(key="auth_enabled", value="true")
        db_session.add(settings)
        await db_session.commit()

        # Groups are auto-seeded during db init, but we need to commit them
        await db_session.commit()

        # Get groups
        from sqlalchemy import select

        admin_group = (await db_session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
        operator_group = (await db_session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()
        viewer_group = (await db_session.execute(select(Group).where(Group.name == "Viewers"))).scalar_one()

        password_hash = get_password_hash("password")

        # Create users
        admin_user = User(username="admin_lib", password_hash=password_hash, role="admin", is_active=True)
        admin_user.groups.append(admin_group)

        operator_user = User(username="operator_lib", password_hash=password_hash, is_active=True)
        operator_user.groups.append(operator_group)

        viewer_user = User(username="viewer_lib", password_hash=password_hash, is_active=True)
        viewer_user.groups.append(viewer_group)

        db_session.add_all([admin_user, operator_user, viewer_user])
        await db_session.commit()

        # Create tokens
        admin_token = create_access_token(data={"sub": admin_user.username})
        operator_token = create_access_token(data={"sub": operator_user.username})
        viewer_token = create_access_token(data={"sub": viewer_user.username})

        return {
            "admin_user": admin_user,
            "operator_user": operator_user,
            "viewer_user": viewer_user,
            "admin_token": admin_token,
            "operator_token": operator_token,
            "viewer_token": viewer_token,
        }

    @pytest.fixture
    async def test_file(self, db_session, auth_setup):
        """Create a test file owned by the operator user."""
        from backend.app.models.library import LibraryFile

        operator_user = auth_setup["operator_user"]
        lib_file = LibraryFile(
            filename="test.txt",
            file_path="data/archive/library/files/test.txt",
            file_type="txt",
            file_size=100,
            created_by_id=operator_user.id,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)
        return lib_file

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_requires_library_read(self, async_client: AsyncClient, db_session, auth_setup):
        """Verify list_files requires library:read permission."""
        viewer_token = auth_setup["viewer_token"]

        # Viewers have library:read, should succeed
        response = await async_client.get("/api/v1/library/files", headers={"Authorization": f"Bearer {viewer_token}"})
        assert response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_files_denied_without_permission(self, async_client: AsyncClient, db_session):
        """Verify list_files denied without auth when auth is enabled."""
        from backend.app.models.settings import Settings

        # Enable auth
        settings = Settings(key="auth_enabled", value="true")
        db_session.add(settings)
        await db_session.commit()

        # Request without token should fail
        response = await async_client.get("/api/v1/library/files")
        assert response.status_code == 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_file_own_by_owner(self, async_client: AsyncClient, db_session, auth_setup, test_file):
        """Verify operator can delete their own files."""
        from pathlib import Path

        # Create actual file on disk so delete doesn't fail
        from backend.app.core.config import settings as app_settings

        file_path = Path(app_settings.base_dir) / test_file.file_path
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text("test content")

        operator_token = auth_setup["operator_token"]

        response = await async_client.delete(
            f"/api/v1/library/files/{test_file.id}", headers={"Authorization": f"Bearer {operator_token}"}
        )
        assert response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_file_own_denied_for_others_file(self, async_client: AsyncClient, db_session, auth_setup):
        """Verify operator cannot delete files owned by others."""
        # Create another operator user with a file
        from sqlalchemy import select

        from backend.app.core.auth import create_access_token
        from backend.app.models.group import Group
        from backend.app.models.library import LibraryFile
        from backend.app.models.user import User

        operator_group = (await db_session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()

        from backend.app.core.auth import get_password_hash as get_pw_hash

        other_user = User(username="other_op", password_hash=get_pw_hash("password"), is_active=True)
        other_user.groups.append(operator_group)
        db_session.add(other_user)
        await db_session.commit()
        await db_session.refresh(other_user)

        # Create file owned by other user
        other_file = LibraryFile(
            filename="other.txt",
            file_path="data/archive/library/files/other.txt",
            file_type="txt",
            file_size=100,
            created_by_id=other_user.id,
        )
        db_session.add(other_file)
        await db_session.commit()
        await db_session.refresh(other_file)

        # Original operator should not be able to delete it
        operator_token = auth_setup["operator_token"]
        response = await async_client.delete(
            f"/api/v1/library/files/{other_file.id}", headers={"Authorization": f"Bearer {operator_token}"}
        )
        assert response.status_code == 403
        assert "your own files" in response.json()["detail"].lower()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_file_admin_can_delete_any(self, async_client: AsyncClient, db_session, auth_setup):
        """Verify admin can delete any file."""
        from pathlib import Path

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

        # Create file owned by operator
        operator_user = auth_setup["operator_user"]
        lib_file = LibraryFile(
            filename="admin_can_delete.txt",
            file_path="data/archive/library/files/admin_can_delete.txt",
            file_type="txt",
            file_size=100,
            created_by_id=operator_user.id,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)

        # Create actual file on disk
        file_path = Path(app_settings.base_dir) / lib_file.file_path
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text("test content")

        # Admin should be able to delete it
        admin_token = auth_setup["admin_token"]
        response = await async_client.delete(
            f"/api/v1/library/files/{lib_file.id}", headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_viewer_cannot_delete_files(self, async_client: AsyncClient, db_session, auth_setup, test_file):
        """Verify viewer cannot delete any files."""
        viewer_token = auth_setup["viewer_token"]

        response = await async_client.delete(
            f"/api/v1/library/files/{test_file.id}", headers={"Authorization": f"Bearer {viewer_token}"}
        )
        # Viewers don't have delete_own or delete_all permissions
        assert response.status_code == 403

    # ---------- #1832: API-key curation under can_manage_library ----------
    #
    # require_ownership_permission gates API keys on `all_perm`, but the
    # library deliberately split UPDATE_OWN/DELETE_OWN (allowed under
    # can_manage_library) from UPDATE_ALL/DELETE_ALL (previously denied).
    # That made the entire curation surface (DELETE, PUT rename, POST move)
    # unreachable for API keys, including for files the key's owner uploaded.
    # The fix folds UPDATE_ALL/DELETE_ALL into can_manage_library so the
    # checker passes; LIBRARY_PURGE stays admin-only.

    @pytest.fixture
    async def manage_library_key(self, db_session, auth_setup):
        """Mint an API key owned by the admin user with can_manage_library."""
        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey

        admin = auth_setup["admin_user"]
        full_key, key_hash, key_prefix = generate_api_key()
        row = APIKey(
            name="lib-curation",
            key_hash=key_hash,
            key_prefix=key_prefix,
            user_id=admin.id,
            can_manage_library=True,
        )
        db_session.add(row)
        await db_session.commit()
        return full_key

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_apikey_with_manage_library_can_delete_file(
        self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
    ):
        """Pre-#1832 this 403'd with "administrative operations" because
        LIBRARY_DELETE_ALL wasn't in _APIKEY_SCOPE_BY_PERMISSION."""
        from pathlib import Path

        from backend.app.core.config import settings as app_settings

        # Materialise the file on disk so the delete handler doesn't 500 on
        # the path it tries to unlink.
        file_path = Path(app_settings.base_dir) / test_file.file_path
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text("test content")

        response = await async_client.delete(
            f"/api/v1/library/files/{test_file.id}",
            headers={"X-API-Key": manage_library_key},
        )
        assert response.status_code == 200, response.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_apikey_with_manage_library_can_rename_file(
        self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
    ):
        """PUT /library/files/{id} is gated on LIBRARY_UPDATE_ALL/OWN. Same
        #1832 path as delete."""
        response = await async_client.put(
            f"/api/v1/library/files/{test_file.id}",
            headers={"X-API-Key": manage_library_key},
            json={"filename": "renamed.txt"},
        )
        assert response.status_code == 200, response.text
        assert response.json()["filename"] == "renamed.txt"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_apikey_with_manage_library_can_move_file(
        self, async_client: AsyncClient, db_session, auth_setup, test_file, manage_library_key
    ):
        """POST /library/files/move (bulk) is gated on LIBRARY_UPDATE_ALL/OWN
        — same checker, same #1832 path."""
        # Create a target folder the move can land in.
        from backend.app.models.library import LibraryFolder

        folder = LibraryFolder(name="target")
        db_session.add(folder)
        await db_session.commit()
        await db_session.refresh(folder)

        response = await async_client.post(
            "/api/v1/library/files/move",
            headers={"X-API-Key": manage_library_key},
            json={"file_ids": [test_file.id], "folder_id": folder.id},
        )
        assert response.status_code == 200, response.text

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_apikey_without_manage_library_still_blocked(
        self, async_client: AsyncClient, db_session, auth_setup, test_file
    ):
        """Regression guard: a key WITHOUT can_manage_library must still get
        403 — the fix widens the allowed-permission set, it doesn't bypass
        the per-key scope check."""
        from backend.app.core.auth import generate_api_key
        from backend.app.models.api_key import APIKey

        admin = auth_setup["admin_user"]
        full_key, key_hash, key_prefix = generate_api_key()
        row = APIKey(
            name="read-only",
            key_hash=key_hash,
            key_prefix=key_prefix,
            user_id=admin.id,
            can_read_status=True,
            can_manage_library=False,
        )
        db_session.add(row)
        await db_session.commit()

        response = await async_client.delete(
            f"/api/v1/library/files/{test_file.id}",
            headers={"X-API-Key": full_key},
        )
        assert response.status_code == 403

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_apikey_with_manage_library_still_cannot_purge(
        self, async_client: AsyncClient, db_session, auth_setup, manage_library_key
    ):
        """LIBRARY_PURGE deliberately stays in _APIKEY_DENIED_PERMISSIONS as
        a genuinely destructive op that bypasses the soft-delete window.
        can_manage_library does NOT grant it."""
        response = await async_client.post(
            "/api/v1/library/purge",
            headers={"X-API-Key": manage_library_key},
            json={"days_in_trash": 30},
        )
        assert response.status_code == 403


class TestPrintFileUploadValidation:
    """#1401: pre-flight rejection of unprintable uploads at the library +
    archive routes. Smoke tests the shared ``validate_print_file_upload``
    helper through both surfaces a user can reach with a drag-drop."""

    def _valid_3mf_bytes(self, name: str = "Metadata/plate_1.gcode") -> bytes:
        """Build a minimal-but-real zip with the gcode-3mf magic in it so
        the validator's ``startswith(b"PK\\x03\\x04")`` check passes."""
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(name, "; G-code\nG28\n")
        return buf.getvalue()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_rejects_raw_gcode_upload(self, async_client: AsyncClient, db_session):
        """``Foo.gcode`` direct uploads are blocked at the library route —
        the dispatcher would otherwise append ``.3mf`` and ship raw gcode
        to the printer as a fake 3MF."""
        files = {"file": ("plate_1.gcode", b"; raw gcode\nG28\n", "application/octet-stream")}
        response = await async_client.post("/api/v1/library/files", files=files)
        assert response.status_code == 400
        # Error message must name the actual remedy, not just say "invalid".
        assert "gcode.3mf" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_rejects_non_zip_3mf_upload(self, async_client: AsyncClient, db_session):
        """A ``.3mf`` upload whose body isn't a zip is rejected — covers
        raw gcode renamed to .3mf, corrupted downloads, etc."""
        files = {"file": ("model.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
        response = await async_client.post("/api/v1/library/files", files=files)
        assert response.status_code == 400
        assert "ZIP container" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_rejects_non_zip_gcode_3mf_upload(self, async_client: AsyncClient, db_session):
        """The compound-extension ``.gcode.3mf`` case is gated by the same
        zip-magic check — splitext returns just ``.3mf``, but the suffix
        match covers both."""
        files = {"file": ("plate_1.gcode.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
        response = await async_client.post("/api/v1/library/files", files=files)
        assert response.status_code == 400
        assert "ZIP container" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_accepts_valid_gcode_3mf_upload(self, async_client: AsyncClient, db_session):
        """A real ``.gcode.3mf`` zip uploads successfully — the existing
        happy path is not regressed by the new validation."""
        files = {
            "file": (
                "plate_1.gcode.3mf",
                self._valid_3mf_bytes(),
                "application/zip",
            )
        }
        response = await async_client.post("/api/v1/library/files", files=files)
        assert response.status_code == 200
        result = response.json()
        assert result["filename"] == "plate_1.gcode.3mf"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_upload_classifies_gcode_3mf_as_compound(self, async_client: AsyncClient, db_session):
        """#1600 follow-up: upload path used to strip to the trailing
        extension and store ``file_type='3mf'`` for sliced outputs, while
        the external-folder scan stored ``file_type='gcode.3mf'``. Now
        every ingest path goes through ``classify_file_type`` and
        produces the canonical compound name."""
        files = {
            "file": (
                "sliced.gcode.3mf",
                self._valid_3mf_bytes(),
                "application/zip",
            )
        }
        response = await async_client.post("/api/v1/library/files", files=files)
        assert response.status_code == 200
        assert response.json()["file_type"] == "gcode.3mf"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_get_gcode_endpoint_accepts_compound_file_type(self, async_client: AsyncClient, db_session):
        """#1600 follow-up: pre-fix, ``GET /files/{id}/gcode`` only handled
        ``file_type`` of ``gcode`` or ``3mf`` and 400'd on a row whose
        ``file_type`` was ``gcode.3mf`` — exactly the rows the external-
        folder scan was creating. The gate now treats both as 3MF and
        unzips the embedded gcode the same way."""
        from backend.app.models.library import LibraryFile

        # Persist a real `.gcode.3mf` zip under file_type='gcode.3mf' so
        # the endpoint hits the new branch.
        with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
            tmp.write(self._valid_3mf_bytes(name="Metadata/plate_1.gcode"))
            tmp_path = tmp.name

        lib_file = LibraryFile(
            filename="sliced.gcode.3mf",
            file_path=tmp_path,
            file_type="gcode.3mf",
            file_size=Path(tmp_path).stat().st_size,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)

        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")
        assert response.status_code == 200
        assert b"G28" in response.content

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_get_gcode_recovers_legacy_gcode_type_for_3mf(self, async_client: AsyncClient, db_session):
        """#1709 regression guard. Before the fix, ``slice_and_persist``
        wrote a `.gcode.3mf` ZIP container to disk but stored the row with
        ``file_type='gcode'`` — the preview endpoint then streamed the
        ZIP body as ``text/plain`` and the embedded G-code viewer saw
        ``PK\\x03\\x04...`` instead of the toolpath. New sliced rows now
        store ``file_type='gcode.3mf'``; rows already written under the
        bug self-heal because the endpoint also detects the ZIP via the
        ``.gcode.3mf`` filename suffix when the column is still legacy."""
        from backend.app.models.library import LibraryFile

        with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
            tmp.write(self._valid_3mf_bytes(name="Metadata/plate_1.gcode"))
            tmp_path = tmp.name

        lib_file = LibraryFile(
            filename="legacy-sliced.gcode.3mf",
            file_path=tmp_path,
            file_type="gcode",
            file_size=Path(tmp_path).stat().st_size,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)

        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")
        assert response.status_code == 200
        assert response.headers["content-type"].startswith("text/plain")
        assert b"G28" in response.content
        # The whole point of #1709: must NOT be ZIP bytes shoved at the viewer.
        assert not response.content.startswith(b"PK")

    async def _multi_plate_file(self, db_session):
        """A two-plate `.gcode.3mf` written plate 2 first, as Bambu Studio does.

        The member order is copied from the file this was reported on — taking
        the first `.gcode` in the zip opened plate 2.
        """
        from backend.app.models.library import LibraryFile

        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("Metadata/plate_2.gcode", "; plate two\nG28\n")
            zf.writestr("Metadata/plate_1.gcode", "; plate one\nG28\n")
        with tempfile.NamedTemporaryFile(suffix=".gcode.3mf", delete=False) as tmp:
            tmp.write(buf.getvalue())
            tmp_path = tmp.name

        lib_file = LibraryFile(
            filename="two-plates.gcode.3mf",
            file_path=tmp_path,
            file_type="gcode.3mf",
            file_size=Path(tmp_path).stat().st_size,
        )
        db_session.add(lib_file)
        await db_session.commit()
        await db_session.refresh(lib_file)
        return lib_file

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_gcode_serves_the_requested_plate(self, async_client: AsyncClient, db_session):
        """The viewer has always sent ``?plate=``; this route took no such
        parameter, and FastAPI drops unknown query parameters without a word —
        so picking a plate did nothing at all."""
        lib_file = await self._multi_plate_file(db_session)

        first = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=1")
        second = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=2")

        assert first.status_code == 200
        assert b"plate one" in first.content
        assert second.status_code == 200
        assert b"plate two" in second.content

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_gcode_without_a_plate_serves_the_first_plate(self, async_client: AsyncClient, db_session):
        """Not the first member in the zip — that is plate 2 in this file, and
        opening a multi-plate file from the File Manager passes no plate."""
        lib_file = await self._multi_plate_file(db_session)

        response = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode")

        assert response.status_code == 200
        assert b"plate one" in response.content

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_gcode_rejects_a_plate_the_file_does_not_hold(self, async_client: AsyncClient, db_session):
        """404 rather than quietly rendering some other plate."""
        lib_file = await self._multi_plate_file(db_session)

        missing = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=3")
        zeroth = await async_client.get(f"/api/v1/library/files/{lib_file.id}/gcode?plate=0")

        assert missing.status_code == 404
        assert zeroth.status_code == 400

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_library_still_accepts_non_print_extensions(self, async_client: AsyncClient, db_session):
        """STL / image / other non-print uploads bypass the validator
        entirely — Bambuddy is also a library, not just a print dispatcher."""
        files = {"file": ("model.stl", b"solid test\nendsolid test", "application/octet-stream")}
        response = await async_client.post(
            "/api/v1/library/files", files=files, params={"generate_stl_thumbnails": "false"}
        )
        assert response.status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_archive_upload_rejects_non_zip(self, async_client: AsyncClient, db_session):
        """``POST /archives/upload`` shares the same validator — covers the
        manual archive-upload entry point too."""
        files = {"file": ("model.3mf", b"; raw gcode\nG28\n", "application/octet-stream")}
        response = await async_client.post("/api/v1/archives/upload", files=files)
        assert response.status_code == 400
        assert "ZIP container" in response.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_archive_bulk_upload_collects_per_file_errors(self, async_client: AsyncClient, db_session):
        """The bulk-archive route reports validation failures per file and
        continues processing the remaining items — one bad upload in a
        10-file drag-drop must not abort the whole batch."""
        good = self._valid_3mf_bytes()
        bad = b"; raw gcode\nG28\n"
        # httpx multipart with a list-of-tuples preserves order + same field name.
        files = [
            ("files", ("good.3mf", good, "application/zip")),
            ("files", ("bad.3mf", bad, "application/octet-stream")),
        ]
        response = await async_client.post("/api/v1/archives/upload-bulk", files=files)
        assert response.status_code == 200
        body = response.json()
        # The bulk route's archive_print may still reject the "good" file
        # downstream (no printer match, etc.) — we don't care about that
        # here; what matters is the bad file lands in `errors` with the
        # validator's message and the route didn't 500.
        assert body["failed"] >= 1
        bad_errors = [e for e in body["errors"] if e["filename"] == "bad.3mf"]
        assert bad_errors, body
        assert "ZIP container" in bad_errors[0]["error"]