bambuddy/backend/tests/integration/test_verdict_per_run.py

"""Each print of an archive has its own outcome question (#1898, archive redesign P1b).

Plates of one archive printed on two printers are two runs. Each asks for its
own verdict with its own one-tap token; an answer lands on the run it was
asked about, and the archive shows its latest run's.
"""

from __future__ import annotations

from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from backend.app.models.archive import PrintArchive
from backend.app.models.print_log import PrintLogEntry

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

T1 = datetime(2026, 10, 30, 9)
T2 = T1 + timedelta(hours=1)


@pytest.fixture
async def two_plates(db_session, printer_factory):
    """One archive, plate 1 printed on A, then plate 2 on B; both asked for
    their outcome and neither is answered."""
    a, b = await printer_factory(name="A"), await printer_factory(name="B")
    archive = PrintArchive(
        printer_id=b.id,
        filename="two.gcode.3mf",
        file_path="archives/test/two.gcode.3mf",
        file_size=1,
        print_name="two",
        status="completed",
        started_at=T2,
        confirm_requested=True,
    )
    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, confirm_requested=True
    )
    second = PrintLogEntry(
        archive_id=archive.id, printer_id=b.id, plate_id=2, status="completed", started_at=T2, confirm_requested=True
    )
    db_session.add_all([first, second])
    await db_session.commit()
    return SimpleNamespace(archive=archive.id, a=a.id, b=b.id, first=first.id, second=second.id)


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


async def _prompt(db_session, plates, run_id):
    """The completion's outcome prompt for ``run_id``; returns (ws payload, provider kwargs)."""
    from backend.app.main import dispatch_outcome_confirmation

    with (
        patch("backend.app.main.ws_manager") as ws,
        patch("backend.app.main.notification_service") as notify,
    ):
        ws.send_print_confirm_request = AsyncMock()
        notify.on_print_confirm_request = AsyncMock()
        sent = await dispatch_outcome_confirmation(db_session, plates.a, "A", {}, plates.archive, run_id=run_id)
    assert sent
    return ws.send_print_confirm_request.await_args.args[1], notify.on_print_confirm_request.await_args.kwargs


async def test_each_plates_prompt_has_its_own_token(db_session, two_plates):
    first_ws, first_links = await _prompt(db_session, two_plates, two_plates.first)
    await _prompt(db_session, two_plates, two_plates.second)

    first_token = (await _row(db_session, PrintLogEntry, two_plates.first)).confirm_token
    second_token = (await _row(db_session, PrintLogEntry, two_plates.second)).confirm_token
    assert first_token and second_token and first_token != second_token
    assert (await _row(db_session, PrintArchive, two_plates.archive)).confirm_token is None
    assert first_ws["run_id"] == two_plates.first
    assert first_links["run_id"] == two_plates.first
    assert first_links["good_url"].endswith(f"/confirm/{first_token}/good")
    assert first_links["confirm_url"].endswith(f"?confirm={two_plates.archive}&run={two_plates.first}")


async def test_an_answered_plate_is_not_asked_again(db_session, two_plates):
    from backend.app.main import dispatch_outcome_confirmation

    first = await _row(db_session, PrintLogEntry, two_plates.first)
    first.user_verdict = "good"
    await db_session.commit()

    with patch("backend.app.main.ws_manager"), patch("backend.app.main.notification_service"):
        assert not await dispatch_outcome_confirmation(
            db_session, two_plates.a, "A", {}, two_plates.archive, run_id=two_plates.first
        )


async def test_a_plates_link_answers_that_plate(async_client, db_session, two_plates):
    await _prompt(db_session, two_plates, two_plates.first)
    token = (await _row(db_session, PrintLogEntry, two_plates.first)).confirm_token

    response = await async_client.post(f"/api/v1/archives/confirm/{token}/reject")
    assert response.status_code == 200, response.text

    first = await _row(db_session, PrintLogEntry, two_plates.first)
    assert (first.user_verdict, first.user_verdict_source) == ("reject", "link")
    assert first.confirm_token_used_at is not None
    assert (await _row(db_session, PrintLogEntry, two_plates.second)).user_verdict is None
    # The archive shows plate 2, its latest, still unanswered.
    assert (await _row(db_session, PrintArchive, two_plates.archive)).user_verdict is None

    again = await async_client.get(f"/api/v1/archives/confirm/{token}/good")
    assert "Already answered" in again.text and "Rejected" in again.text
    assert f"run={two_plates.first}" in again.text


async def test_a_link_sent_before_runs_had_tokens_answers_the_latest(async_client, db_session, two_plates):
    archive = await _row(db_session, PrintArchive, two_plates.archive)
    archive.confirm_token = "archive-token"
    await db_session.commit()

    assert (await async_client.post("/api/v1/archives/confirm/archive-token/good")).status_code == 200

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    assert (archive.user_verdict, archive.confirm_token_used_at is not None) == ("good", True)
    assert (await _row(db_session, PrintLogEntry, two_plates.second)).user_verdict == "good"
    assert (await _row(db_session, PrintLogEntry, two_plates.first)).user_verdict is None


async def test_the_dialog_answers_the_plate_it_asks_about(async_client, db_session, two_plates):
    response = await async_client.patch(
        f"/api/v1/archives/{two_plates.archive}",
        json={
            "user_verdict": "reject",
            "failure_reason": "spaghetti",
            "user_verdict_source": "dialog",
            "run_id": two_plates.first,
        },
    )
    assert response.status_code == 200, response.text

    first = await _row(db_session, PrintLogEntry, two_plates.first)
    assert (first.user_verdict, first.user_verdict_source, first.failure_reason) == ("reject", "dialog", "spaghetti")
    second = await _row(db_session, PrintLogEntry, two_plates.second)
    assert (second.user_verdict, second.failure_reason) == (None, None)
    archive = await _row(db_session, PrintArchive, two_plates.archive)
    assert (archive.user_verdict, archive.failure_reason) == (None, None)


async def test_an_answer_for_the_latest_plate_shows_on_the_archive(async_client, db_session, two_plates):
    response = await async_client.patch(
        f"/api/v1/archives/{two_plates.archive}",
        json={"user_verdict": "good", "user_verdict_source": "printer_card", "run_id": two_plates.second},
    )
    assert response.status_code == 200, response.text

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    assert (archive.user_verdict, archive.user_verdict_source) == ("good", "printer_card")
    assert (await _row(db_session, PrintLogEntry, two_plates.first)).user_verdict is None


async def test_an_answer_without_a_run_goes_to_the_latest(async_client, db_session, two_plates):
    response = await async_client.patch(f"/api/v1/archives/{two_plates.archive}", json={"user_verdict": "good"})
    assert response.status_code == 200, response.text

    assert (await _row(db_session, PrintLogEntry, two_plates.second)).user_verdict == "good"
    assert (await _row(db_session, PrintLogEntry, two_plates.first)).user_verdict is None
    assert (await _row(db_session, PrintArchive, two_plates.archive)).user_verdict == "good"


async def test_a_run_of_another_archive_or_other_fields_are_refused(async_client, db_session, two_plates):
    other = PrintArchive(filename="o.3mf", file_path="archives/test/o.3mf", file_size=1, status="completed")
    db_session.add(other)
    await db_session.flush()
    stranger = PrintLogEntry(archive_id=other.id, status="completed")
    db_session.add(stranger)
    await db_session.commit()

    url = f"/api/v1/archives/{two_plates.archive}"
    assert (await async_client.patch(url, json={"user_verdict": "good", "run_id": stranger.id})).status_code == 404
    assert (await async_client.patch(url, json={"notes": "x", "run_id": two_plates.first})).status_code == 400


async def test_the_archive_can_be_read_as_one_of_its_runs(async_client, db_session, two_plates):
    first = await _row(db_session, PrintLogEntry, two_plates.first)
    first.user_verdict = "reject"
    await db_session.commit()

    response = await async_client.get(f"/api/v1/archives/{two_plates.archive}?run_id={two_plates.first}")
    assert response.status_code == 200, response.text
    body = response.json()
    assert (body["run_id"], body["printer_id"], body["plate_id"], body["user_verdict"]) == (
        two_plates.first,
        two_plates.a,
        1,
        "reject",
    )
    plain = (await async_client.get(f"/api/v1/archives/{two_plates.archive}")).json()
    assert (plain["run_id"], plain["user_verdict"]) == (None, None)
    missing = await async_client.get(f"/api/v1/archives/{two_plates.archive}?run_id=999999")
    assert missing.status_code == 404


async def test_each_printer_card_shows_its_own_plate(async_client, db_session, two_plates):
    first = await _row(db_session, PrintLogEntry, two_plates.first)
    first.user_verdict = "good"
    await db_session.commit()

    response = await async_client.get("/api/v1/archives/last-per-printer")
    assert response.status_code == 200, response.text
    cards = {card["printer_id"]: card for card in response.json()}
    assert {pid: (c["id"], c["run_id"], c["plate_id"], c["user_verdict"]) for pid, c in cards.items()} == {
        two_plates.a: (two_plates.archive, two_plates.first, 1, "good"),
        two_plates.b: (two_plates.archive, two_plates.second, 2, None),
    }


async def test_clearing_a_printers_plate_answers_its_own_plate(db_session, two_plates):
    from backend.app.services.print_confirmation import resolve_pending_confirmation_as_good

    assert await resolve_pending_confirmation_as_good(db_session, two_plates.a) == two_plates.archive
    await db_session.commit()

    first = await _row(db_session, PrintLogEntry, two_plates.first)
    assert (first.user_verdict, first.user_verdict_source) == ("good", "plate_clear")
    assert (await _row(db_session, PrintLogEntry, two_plates.second)).user_verdict is None
    assert (await _row(db_session, PrintArchive, two_plates.archive)).user_verdict is None


async def test_a_new_plate_ending_shows_its_open_question_on_the_archive(db_session, two_plates):
    """The archive's verdict was the previous print's: the latest run's (none
    yet) replaces it once the new run ends."""
    from backend.app.services.print_log import mirror_latest_run

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    first = await db_session.get(PrintLogEntry, two_plates.first)
    archive.user_verdict = "good"
    archive.user_verdict_source = "dialog"
    first.user_verdict = "good"
    await db_session.commit()

    await mirror_latest_run(db_session, two_plates.archive)
    await db_session.commit()

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    assert (archive.user_verdict, archive.user_verdict_source) == (None, None)


async def test_a_run_without_provenance_keeps_the_archives(db_session, two_plates):
    """Runs logged before they kept how the verdict arrived: the same verdict
    on the archive keeps its own provenance."""
    from backend.app.services.print_log import mirror_latest_run

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    second = await db_session.get(PrintLogEntry, two_plates.second)
    archive.user_verdict = "good"
    archive.user_verdict_source = "plate_clear"
    second.user_verdict = "good"
    await db_session.commit()

    await mirror_latest_run(db_session, two_plates.archive)
    await db_session.commit()

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    assert (archive.user_verdict, archive.user_verdict_source) == ("good", "plate_clear")


async def test_a_run_logged_before_runs_owned_their_verdict_leaves_the_archives(db_session, two_plates):
    """Its verdict was never copied onto it: the archive's is kept."""
    from backend.app.services.print_log import mirror_latest_run

    archive = await _row(db_session, PrintArchive, two_plates.archive)
    second = await db_session.get(PrintLogEntry, two_plates.second)
    archive.user_verdict = "reject"
    second.confirm_requested = None
    await db_session.commit()

    await mirror_latest_run(db_session, two_plates.archive)
    await db_session.commit()

    assert (await _row(db_session, PrintArchive, two_plates.archive)).user_verdict == "reject"


async def test_a_run_on_a_printer_outside_the_callers_scope_is_not_theirs(async_client, db_session, two_plates):
    """Scoped to printer B, the archive's: plate 1 printed on A is neither
    read nor judged through the archive, and A's card is not theirs."""
    from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user

    admin = await _admin_token(async_client)
    jwt, _ = await _user(
        async_client,
        admin,
        "scoped",
        ["archives:read_all", "archives:update_all", "printers:read"],
        printer_ids=[two_plates.b],
    )
    url = f"/api/v1/archives/{two_plates.archive}"

    assert (await async_client.get(f"{url}?run_id={two_plates.first}", headers=_auth(jwt))).status_code == 404
    patched = await async_client.patch(
        url, headers=_auth(jwt), json={"user_verdict": "reject", "run_id": two_plates.first}
    )
    assert patched.status_code == 404
    assert (await _row(db_session, PrintLogEntry, two_plates.first)).user_verdict is None
    # Their own printer's plate is theirs.
    own = await async_client.patch(url, headers=_auth(jwt), json={"user_verdict": "good", "run_id": two_plates.second})
    assert own.status_code == 200, own.text

    cards = (await async_client.get("/api/v1/archives/last-per-printer", headers=_auth(jwt))).json()
    assert [(c["printer_id"], c["run_id"]) for c in cards] == [(two_plates.b, two_plates.second)]


async def test_a_card_is_not_shown_for_an_archive_outside_the_callers_scope(async_client, db_session, two_plates):
    """Scoped to printer A, whose plate it printed, while the archive names
    B: as before runs, the archive is not theirs to see."""
    from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user

    admin = await _admin_token(async_client)
    jwt, _ = await _user(
        async_client, admin, "scoped-a", ["archives:read_all", "printers:read"], printer_ids=[two_plates.a]
    )

    cards = (await async_client.get("/api/v1/archives/last-per-printer", headers=_auth(jwt))).json()
    assert cards == []