bambuddy/backend/tests/integration/test_archive_scope_per_run.py

"""A caller limited to some printers sees an archive by its prints (#1727, archive redesign).

Plate 1 of one archive printed on printer A, then plate 2 on B: the archive
names B. A user of A sees the archive by plate 1 and only plate 1 of it: its
outcome, video and finish photos. What changes the whole archive is refused
to them, as it would change B's print too.
"""

from __future__ import annotations

from datetime import datetime, timedelta
from types import SimpleNamespace

import pytest

from backend.app.core.config import settings as app_settings
from backend.app.models.archive import PrintArchive
from backend.app.models.print_log import RUN_OPEN, PrintLogEntry
from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user

pytestmark = [pytest.mark.asyncio, pytest.mark.integration]

T1 = datetime(2026, 10, 30, 9)
T2 = T1 + timedelta(hours=1)
DIR = "archives/scoped"
PERMISSIONS = ["archives:read_all", "archives:update_all", "archives:delete_all", "printers:read", "projects:read"]


@pytest.fixture
def files(tmp_path, monkeypatch):
    monkeypatch.setattr(app_settings, "base_dir", tmp_path)
    monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
    (tmp_path / DIR / "photos").mkdir(parents=True)
    for name in ("a.mp4", "b.mp4", "photos/finish_a.jpg", "photos/finish_b.jpg", "photos/0a1b2c3d.jpg"):
        (tmp_path / DIR / name).write_bytes(b"x")
    return tmp_path


@pytest.fixture
async def plates(async_client, db_session, printer_factory, files):
    """Plate 1 completed on A with its video and photo, then plate 2 failed
    on B with its own; one photo uploaded by hand. Users scoped to A, to B
    and to a printer C that printed nothing."""
    a, b, c = await printer_factory(name="A"), await printer_factory(name="B"), await printer_factory(name="C")
    archive = PrintArchive(
        printer_id=b.id,
        filename="two.gcode.3mf",
        file_path=f"{DIR}/two.gcode.3mf",
        file_size=1,
        print_name="two",
        status="failed",
        started_at=T2,
        timelapse_path=f"{DIR}/b.mp4",
        photos=["finish_a.jpg", "finish_b.jpg", "0a1b2c3d.jpg"],
    )
    db_session.add(archive)
    await db_session.flush()
    first = PrintLogEntry(
        archive_id=archive.id,
        printer_id=a.id,
        plate_id=1,
        status="completed",
        started_at=T1,
        photos=["finish_a.jpg"],
        timelapse_path=f"{DIR}/a.mp4",
    )
    second = PrintLogEntry(
        archive_id=archive.id,
        printer_id=b.id,
        plate_id=2,
        status="failed",
        started_at=T2,
        photos=["finish_b.jpg"],
        timelapse_path=f"{DIR}/b.mp4",
    )
    db_session.add_all([first, second])
    await db_session.commit()
    admin = await _admin_token(async_client)
    users = {}
    for name, printer in (("of_a", a), ("of_b", b), ("of_c", c)):
        jwt, _ = await _user(async_client, admin, name, PERMISSIONS, printer_ids=[printer.id])
        users[name] = _auth(jwt)
    return SimpleNamespace(
        archive=archive.id, a=a.id, b=b.id, c=c.id, first=first.id, second=second.id, admin=_auth(admin), **users
    )


def _url(plates, path=""):
    return f"/api/v1/archives/{plates.archive}{path}"


async def _row(db_session, model, row_id):
    db_session.expire_all()
    return await db_session.get(model, row_id)


class TestSeeing:
    async def test_a_user_of_a_sees_the_archive_by_its_plate(self, async_client, plates):
        listed = (await async_client.get("/api/v1/archives/", headers=plates.of_a)).json()

        assert [a["id"] for a in listed] == [plates.archive]
        shown = listed[0]
        assert (shown["printer_id"], shown["status"], shown["run_id"]) == (plates.a, "completed", plates.first)
        assert shown["timelapse_path"] == f"{DIR}/a.mp4"
        assert shown["photos"] == ["finish_a.jpg"]
        assert shown["run_count"] == 1

    async def test_the_archive_page_shows_their_plate(self, async_client, plates):
        response = await async_client.get(_url(plates), headers=plates.of_a)

        assert response.status_code == 200, response.text
        body = response.json()
        assert (body["printer_id"], body["status"], body["run_id"]) == (plates.a, "completed", plates.first)

    async def test_the_other_plate_is_not_theirs(self, async_client, plates):
        response = await async_client.get(_url(plates, f"?run_id={plates.second}"), headers=plates.of_a)

        assert response.status_code == 404

    async def test_its_print_log_lists_their_plate_only(self, async_client, plates):
        runs = (await async_client.get(_url(plates, "/runs"), headers=plates.of_a)).json()

        assert [r["id"] for r in runs["items"]] == [plates.first]

    async def test_a_user_of_b_sees_its_own_and_the_uploads(self, async_client, plates):
        body = (await async_client.get(_url(plates), headers=plates.of_b)).json()

        assert (body["printer_id"], body["status"]) == (plates.b, "failed")
        assert body["timelapse_path"] == f"{DIR}/b.mp4"
        assert body["photos"] == ["finish_b.jpg", "0a1b2c3d.jpg"]

    async def test_a_printer_it_never_printed_on_sees_nothing(self, async_client, plates):
        assert (await async_client.get("/api/v1/archives/", headers=plates.of_c)).json() == []
        assert (await async_client.get(_url(plates), headers=plates.of_c)).status_code == 404

    async def test_a_plate_printing_now_counts(self, async_client, db_session, plates):
        db_session.add(
            PrintLogEntry(
                archive_id=plates.archive, printer_id=plates.c, status=RUN_OPEN, started_at=T2 + timedelta(hours=1)
            )
        )
        await db_session.commit()

        listed = (await async_client.get("/api/v1/archives/", headers=plates.of_c)).json()
        assert [a["id"] for a in listed] == [plates.archive]
        assert (await async_client.get(_url(plates), headers=plates.of_c)).status_code == 200

    async def test_search_finds_it_by_their_plate(self, async_client, plates):
        response = await async_client.get("/api/v1/archives/search?q=two", headers=plates.of_a)

        assert response.status_code == 200, response.text
        assert [a["id"] for a in response.json()] == [plates.archive]
        assert response.json()[0]["photos"] == ["finish_a.jpg"]

    async def test_the_admin_sees_it_whole(self, async_client, plates):
        body = (await async_client.get(_url(plates), headers=plates.admin)).json()

        assert (body["printer_id"], body["run_count"]) == (plates.b, 2)
        assert body["photos"] == ["finish_a.jpg", "finish_b.jpg", "0a1b2c3d.jpg"]


class TestMedia:
    async def test_the_other_plates_video_is_not_theirs(self, async_client, plates):
        assert (await async_client.get(_url(plates, "/timelapse"), headers=plates.of_a)).status_code == 404
        assert (await async_client.get(_url(plates, "/timelapse"), headers=plates.of_b)).status_code == 200

    async def test_the_other_plates_photo_is_not_theirs(self, async_client, plates):
        assert (await async_client.get(_url(plates, "/photos/finish_b.jpg"), headers=plates.of_a)).status_code == 404
        assert (await async_client.get(_url(plates, "/photos/finish_a.jpg"), headers=plates.of_a)).status_code == 200
        assert (await async_client.get(_url(plates, "/photos/0a1b2c3d.jpg"), headers=plates.of_a)).status_code == 200

    async def test_nor_theirs_to_delete(self, async_client, db_session, plates):
        response = await async_client.delete(_url(plates, "/photos/finish_b.jpg"), headers=plates.of_a)

        assert response.status_code == 404
        assert "finish_b.jpg" in (await _row(db_session, PrintArchive, plates.archive)).photos

    async def test_a_finish_photo_on_no_runs_list_is_the_archives_latest(self, async_client, db_session, plates):
        """Logged before runs kept their own photos: the archive's, here B's."""
        archive = await _row(db_session, PrintArchive, plates.archive)
        archive.photos = [*archive.photos, "finish_old.jpg"]
        await db_session.commit()

        assert "finish_old.jpg" not in (await async_client.get(_url(plates), headers=plates.of_a)).json()["photos"]
        assert "finish_old.jpg" in (await async_client.get(_url(plates), headers=plates.of_b)).json()["photos"]

    async def test_a_plate_logged_before_runs_kept_media_shows_none_of_the_others(
        self, async_client, db_session, files, plates
    ):
        """Its run has no video or photo list of its own: the archive's are
        B's latest, and only the photo uploaded by hand is everyone's."""
        first = await _row(db_session, PrintLogEntry, plates.first)
        first.photos, first.timelapse_path = None, None
        await db_session.commit()

        body = (await async_client.get(_url(plates), headers=plates.of_a)).json()
        assert (body["run_id"], body["timelapse_path"]) == (plates.first, None)
        assert body["photos"] == ["0a1b2c3d.jpg"]
        assert (await async_client.get(_url(plates, "/photos/finish_a.jpg"), headers=plates.of_a)).status_code == 404


class TestWholeArchive:
    async def test_deleting_it_is_refused(self, async_client, db_session, plates):
        response = await async_client.delete(_url(plates), headers=plates.of_a)

        assert response.status_code == 403, response.text
        assert (await _row(db_session, PrintArchive, plates.archive)).deleted_at is None

    async def test_also_to_the_user_of_its_own_printer(self, async_client, plates):
        """B is the archive's printer; plate 1 printed on A."""
        assert (await async_client.delete(_url(plates), headers=plates.of_b)).status_code == 403

    async def test_deleting_its_video_is_refused(self, async_client, db_session, plates):
        response = await async_client.delete(_url(plates, "/timelapse"), headers=plates.of_b)

        assert response.status_code == 403
        assert (await _row(db_session, PrintArchive, plates.archive)).timelapse_path == f"{DIR}/b.mp4"

    async def test_its_notes_are_theirs_to_edit(self, async_client, db_session, plates):
        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"notes": "nice", "tags": "x"})

        assert response.status_code == 200, response.text
        assert (await _row(db_session, PrintArchive, plates.archive)).notes == "nice"

    async def test_its_cost_is_not(self, async_client, db_session, plates):
        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"cost": 9.0})

        assert response.status_code == 403
        assert (await _row(db_session, PrintArchive, plates.archive)).cost is None

    async def test_nor_its_outcome_without_their_run(self, async_client, db_session, plates):
        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"user_verdict": "good"})

        assert response.status_code == 403

    async def test_their_plates_outcome_is(self, async_client, db_session, plates):
        response = await async_client.patch(
            _url(plates), headers=plates.of_a, json={"user_verdict": "good", "run_id": plates.first}
        )

        assert response.status_code == 200, response.text
        assert (await _row(db_session, PrintLogEntry, plates.first)).user_verdict == "good"

    async def test_the_admin_may_delete_it(self, async_client, db_session, plates):
        response = await async_client.delete(_url(plates), headers=plates.admin)

        assert response.status_code == 200, response.text

    @pytest.mark.parametrize(
        ("method", "path", "kwargs"),
        [
            ("post", "/timelapse/select?filename=v.mp4", {}),
            ("post", "/timelapse/upload", {"files": {"file": ("v.mp4", b"x", "video/mp4")}}),
            ("post", "/timelapse/process", {"data": {"speed": "2"}}),
            ("delete", "/timelapse/job", {}),
            ("post", "/source", {"files": {"file": ("s.3mf", b"x", "application/octet-stream")}}),
            ("delete", "/source", {}),
            ("post", "/f3d", {"files": {"file": ("d.f3d", b"x", "application/octet-stream")}}),
            ("delete", "/f3d", {}),
        ],
    )
    async def test_its_file_and_video_are_not_theirs_to_change(self, async_client, plates, method, path, kwargs):
        """B's user: the archive's own printer, but plate 1 printed on A."""
        response = await getattr(async_client, method)(_url(plates, path), headers=plates.of_b, **kwargs)

        assert response.status_code == 403, response.text

    async def test_scanning_for_its_video_is_not_theirs(self, async_client, db_session, plates):
        """A scan attaches a video to the archive: only without one."""
        archive = await _row(db_session, PrintArchive, plates.archive)
        archive.timelapse_path = None
        await db_session.commit()

        response = await async_client.post(_url(plates, "/timelapse/scan"), headers=plates.of_b)

        assert response.status_code == 403, response.text


class TestTheOtherPlatesVideo:
    @pytest.mark.parametrize(
        ("method", "path"),
        [
            ("get", "/timelapse/info"),
            ("get", "/timelapse/thumbnails"),
            ("post", "/media-download-token"),
        ],
    )
    async def test_is_missing_for_them(self, async_client, plates, method, path):
        response = await getattr(async_client, method)(_url(plates, path), headers=plates.of_a)

        assert response.status_code == 404, response.text

    async def test_is_not_offered_as_attached(self, async_client, plates):
        response = await async_client.get(_url(plates, "/printer-media"), headers=plates.of_a)

        assert response.status_code == 200, response.text
        assert response.json()["local_timelapse"] is None


class TestReviewFixes:
    async def test_a_null_run_id_is_no_run(self, async_client, db_session, plates):
        response = await async_client.patch(
            _url(plates), headers=plates.of_a, json={"run_id": None, "status": "completed", "failure_reason": "x"}
        )

        assert response.status_code == 403, response.text
        assert (await _row(db_session, PrintArchive, plates.archive)).status == "failed"

    async def test_its_project_page_is_its_file(self, async_client, plates):
        response = await async_client.patch(_url(plates, "/project-page"), headers=plates.of_a, json={"title": "x"})

        assert response.status_code == 403, response.text

    async def test_compare_and_similar_leave_it_out(self, async_client, db_session, plates):
        """Both show each archive's latest print as it is: B's."""
        other = PrintArchive(
            printer_id=plates.a, filename="two.gcode.3mf", file_path=f"{DIR}/o.3mf", file_size=1, print_name="two"
        )
        db_session.add(other)
        await db_session.commit()
        other_id = other.id

        compare = await async_client.get(
            f"/api/v1/archives/compare?archive_ids={plates.archive},{other_id}", headers=plates.of_a
        )
        assert compare.status_code == 404, compare.text
        similar = (await async_client.get(f"/api/v1/archives/{other_id}/similar", headers=plates.of_a)).json()
        assert plates.archive not in [s["archive"]["id"] for s in similar]

    async def test_the_project_previews_leave_it_out(self, async_client, db_session, plates):
        from backend.app.models.project import Project

        project = Project(name="P")
        db_session.add(project)
        await db_session.flush()
        (await db_session.get(PrintArchive, plates.archive)).project_id = project.id
        await db_session.commit()

        projects = (await async_client.get("/api/v1/projects/", headers=plates.of_a)).json()
        assert [p.get("archives", []) for p in projects] == [[]]

    async def test_its_photo_responses_show_theirs(self, async_client, plates):
        uploaded = await async_client.post(
            _url(plates, "/photos"), headers=plates.of_a, files={"file": ("p.jpg", b"\xff\xd8\xff", "image/jpeg")}
        )

        assert uploaded.status_code == 200, uploaded.text
        assert "finish_b.jpg" not in uploaded.json()["photos"]
        deleted = await async_client.delete(_url(plates, "/photos/0a1b2c3d.jpg"), headers=plates.of_a)
        assert deleted.status_code == 200, deleted.text
        assert "finish_b.jpg" not in deleted.json()["photos"]

    async def test_printer_media_does_not_name_the_other_printer(self, async_client, plates):
        response = await async_client.get(_url(plates, "/printer-media"), headers=plates.of_a)

        assert response.json()["printer_id"] is None

    async def test_its_energy_and_time_are_their_prints(self, async_client, db_session, plates):
        archive = await _row(db_session, PrintArchive, plates.archive)
        archive.energy_kwh, archive.wear_cost = 5.0, 2.0
        archive.status, archive.completed_at, archive.print_time_seconds = "completed", T2 + timedelta(hours=1), 3600
        first = await db_session.get(PrintLogEntry, plates.first)
        first.energy_kwh, first.wear_cost = 1.0, 0.5
        await db_session.commit()

        body = (await async_client.get(_url(plates), headers=plates.of_a)).json()
        assert (body["energy_kwh"], body["wear_cost"]) == (1.0, 0.5)
        assert body["actual_time_seconds"] is None
        whole = (await async_client.get(_url(plates), headers=plates.admin)).json()
        assert (whole["energy_kwh"], whole["wear_cost"]) == (5.0, 2.0)


async def test_queueing_it_is_theirs(db_session, plates, test_engine):
    """It prints on their printer (archive redesign)."""
    from unittest.mock import patch

    from fastapi import HTTPException
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    from backend.app.api.routes.print_queue import _assert_can_queue_archive
    from backend.app.core.printer_scope import PrinterScope

    archive = await db_session.get(PrintArchive, plates.archive)
    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    with patch("backend.app.core.database.async_session", maker):
        await _assert_can_queue_archive(archive, None, PrinterScope(frozenset({plates.a})))
        with pytest.raises(HTTPException) as refused:
            await _assert_can_queue_archive(archive, None, PrinterScope(frozenset({plates.c})))
    assert refused.value.status_code == 404


class TestLiveUpdates:
    async def test_reach_the_printers_its_prints_ran_on(self, db_session, plates, test_engine):
        from unittest.mock import patch

        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

        from backend.app.core import websocket

        db_session.add(
            PrintLogEntry(
                archive_id=plates.archive, printer_id=plates.c, status=RUN_OPEN, started_at=T2 + timedelta(hours=1)
            )
        )
        await db_session.commit()
        maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
        with patch("backend.app.core.database.async_session", maker):
            assert await websocket._archive_printer_ids(plates.archive) == (plates.b, {plates.a, plates.b, plates.c})


async def test_the_project_list_shows_their_plate(async_client, db_session, plates):
    from backend.app.models.project import Project

    project = Project(name="P")
    db_session.add(project)
    await db_session.flush()
    archive = await db_session.get(PrintArchive, plates.archive)
    archive.project_id = project.id
    project_id = project.id
    await db_session.commit()

    response = await async_client.get(f"/api/v1/projects/{project_id}/archives", headers=plates.of_a)

    assert response.status_code == 200, response.text
    assert [(a["id"], a["printer_id"], a["photos"]) for a in response.json()] == [
        (plates.archive, plates.a, ["finish_a.jpg"])
    ]