PacketStation/tests/server/test_prints.py

"""Phase 7 printed copies and exports through the API: every print kind with its stamp on every page, the print log
and the audit, one sheet, a document print and a foreign asset, nothing to print, the data book before complete,
the prints list, the ITP print and PDF, the exports, the signed-out call, the PM role, a removed packet, and the
append-only print log."""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import psycopg
import pytest
from closehelpers import (
    at_complete,
    flat,
    is_pdf,
    page_texts,
    pdf_pages,
    q1,
    qall,
    remove_by_sql,
    rich_complete,
)
from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pshelpers import ORIGIN_HEADERS, app_conn, needs_pg
from qchelpers import approve_all, at_qc, sign_all_blocks, sign_off, start_itp, upload_doc
from shophelpers import Shop, draft, ok, open_shop, refused

from packetstation.settings import Settings

pytestmark = needs_pg

WHAT = {
    "traveler": "Job traveler",
    "map-h": "Drawing with heat map",
    "map-w": "Drawing with weld map",
    "clean": "Drawing (no markup)",
    "void": "Superseded drawings (VOID)",
    "databook": "Data book",
}


@pytest.fixture
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
    yield from open_shop(pg_settings, fake_dir)


def get_pdf(c: TestClient, url: str) -> bytes:
    r = c.get(url)
    assert r.status_code == 200, (url, r.text[:500])
    assert r.headers["content-type"].startswith("application/pdf"), url
    assert r.headers.get("x-content-type-options") == "nosniff"
    assert r.headers.get("cache-control") == "no-store"
    assert r.headers["content-disposition"].startswith("inline;"), r.headers["content-disposition"]
    assert is_pdf(r.content)
    return r.content


def assert_print_stamp(data: bytes, what: str, by: str = "Clara Clerk (4001)") -> None:
    for i, t in enumerate(page_texts(data)):
        ft = flat(t)
        assert f"PRINTED COPY - {what} - printed by {by}" in ft, (i, what, ft[-400:])
        assert "verify revision before use" in ft, (i, what)


def last_print(settings: Settings, pid: int) -> Any:
    return q1(
        settings,
        "SELECT kind, what, pages, by_name, stamp_sig, stamp, asset_id FROM print_log WHERE packet_id = %s"
        " ORDER BY id DESC LIMIT 1",
        (pid,),
    )


def last_audit(settings: Settings, pid: int, action: str) -> Any:
    row = q1(
        settings,
        "SELECT detail, user_id FROM audit_events WHERE action = %s AND packet_id = %s ORDER BY id DESC LIMIT 1",
        (action, pid),
    )
    assert row is not None, (action, pid)
    return row


def test_every_print_kind(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    c = shop("clerk.clara")
    for kind, what in WHAT.items():
        data = get_pdf(c, f"/api/packets/{pid}/print/{kind}")
        assert_print_stamp(data, what)
        row = last_print(shop.settings, pid)
        assert (row[0], row[1], row[2], row[3]) == (kind, what, pdf_pages(data), "Clara Clerk"), row
        assert row[4] and len(row[4]) == 16 and row[4] in row[5]
        au = last_audit(shop.settings, pid, "print")
        assert au[0]["kind"] == kind and au[0]["what"] == what and au[0]["pages"] == pdf_pages(data)
        assert au[0]["stamp_sig"] == row[4]
    # page counts: two live sheets, one void sheet
    assert pdf_pages(get_pdf(c, f"/api/packets/{pid}/print/clean")) == 3
    assert pdf_pages(get_pdf(c, f"/api/packets/{pid}/print/void")) == 1
    # one sheet
    for kind in ("map-h", "map-w", "clean"):
        data = get_pdf(c, f"/api/packets/{pid}/print/{kind}?sheet=2")
        assert pdf_pages(data) == 1
        assert_print_stamp(data, f"{WHAT[kind]} (sheet 2)")
        assert last_print(shop.settings, pid)[1] == f"{WHAT[kind]} (sheet 2)"
    # the download form
    r = c.get(f"/api/packets/{pid}/print/traveler?download=1")
    assert r.status_code == 200 and r.headers["content-disposition"].startswith("attachment;")
    assert '-Lot001 Job traveler.pdf"' in r.headers["content-disposition"], r.headers["content-disposition"]


def test_document_prints(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    c = shop("clerk.clara")
    mtr = next(b for b in pkt["bom"] if b["item"] == 1)
    mtr_id = (mtr.get("mtr") or {}).get("asset_id") or mtr.get("mtr_asset_id")
    png = next(b for b in pkt["bom"] if b["item"] == 2)
    png_id = (png.get("mtr") or {}).get("asset_id") or png.get("mtr_asset_id")
    doc = next(d for d in pkt["docs"] if d.get("removed") is None)
    doc_id = doc.get("asset_id") or doc["url"].rsplit("/", 1)[-1]
    folder = pkt["qc_folder_docs"][0]
    folder_id = folder.get("asset_id") or folder["url"].rsplit("/", 1)[-1]
    for asset, what_starts in (
        (mtr_id, "MTR - item 1"),
        (png_id, "MTR - item 2"),
        (doc_id, "RT report W1-W2"),
        (folder_id, "WPS / PQR - Unit folder doc"),
    ):
        data = get_pdf(c, f"/api/packets/{pid}/print/doc?asset={asset}")
        row = last_print(shop.settings, pid)
        assert row[0] == "doc" and row[1].startswith(what_starts) and str(row[6]) == str(asset), row
        assert_print_stamp(data, row[1])
        assert last_audit(shop.settings, pid, "print")[0].get("asset_id") in (str(asset), None)
    # a PNG MTR prints on one letter page
    assert pdf_pages(get_pdf(c, f"/api/packets/{pid}/print/doc?asset={png_id}")) == 1
    # an asset that is not the packet's
    foreign = upload_doc(c, "other.pdf")["id"]
    refused(c.get(f"/api/packets/{pid}/print/doc?asset={foreign}"), 404, "not_found")
    r = c.get(f"/api/packets/{pid}/print/doc")
    assert r.status_code in (404, 422), r.text
    r = c.get(f"/api/packets/{pid}/print/nonsense")
    assert r.status_code in (404, 422), r.text


def test_nothing_to_print_and_wrong_status(shop: Shop) -> None:
    pkt = at_qc(shop)
    c = shop("clerk.clara")
    refused(c.get(f"/api/packets/{pkt['id']}/print/void"), 404, "nothing_to_print", "Nothing to print.")
    refused(c.get(f"/api/packets/{pkt['id']}/print/databook"), 409, "wrong_status")
    refused(
        c.get(f"/api/packets/{pkt['id']}/pdf/databook"),
        409,
        "wrong_status",
        "The data book is built once the packet is back from Q.C.",
    )
    assert q1(shop.settings, "SELECT count(*) FROM print_log WHERE packet_id = %s", (pkt["id"],))[0] == 0
    # the traveler prints in any status
    get_pdf(c, f"/api/packets/{pkt['id']}/print/traveler")


def test_prints_list_newest_first_and_exports(shop: Shop) -> None:
    pkt = at_complete(shop)
    pid = pkt["id"]
    c = shop("clerk.clara")
    assert ok(c.get(f"/api/packets/{pid}/prints")) == {"prints": []}
    get_pdf(c, f"/api/packets/{pid}/print/traveler")
    get_pdf(shop("pm.paula"), f"/api/packets/{pid}/pdf/traveler")
    data = get_pdf(shop("pm.paula"), f"/api/packets/{pid}/pdf/databook")
    assert "not yet filed" in flat(page_texts(data)[0])
    assert shop("pm.paula").get(f"/api/packets/{pid}/export/traceability.csv").status_code == 200
    out = ok(shop("qc.quinn").get(f"/api/packets/{pid}/prints"))["prints"]
    assert [p["kind"] for p in out] == ["csv", "databook-export", "traveler", "traveler"]
    assert out[0]["by"]["display_name"] == "Paula PM" and out[-1]["by"]["display_name"] == "Clara Clerk"
    assert out[1]["pages"] == pdf_pages(data) and out[1]["what"]
    assert set(out[0]) >= {"id", "at", "by", "kind", "what", "pages"}
    ids = [p["id"] for p in out]
    assert ids == sorted(ids, reverse=True)
    kinds = [
        r[0]["kind"]
        for r in qall(
            shop.settings,
            "SELECT detail FROM audit_events WHERE action = 'export' AND packet_id = %s ORDER BY id",
            (pid,),
        )
    ]
    assert kinds == ["traveler", "databook", "csv"]
    assert (
        q1(shop.settings, "SELECT count(*) FROM audit_events WHERE action = 'print' AND packet_id = %s", (pid,))[0] == 1
    )
    # prints and exports write no packet log rows
    actions = [r[0] for r in qall(shop.settings, "SELECT action FROM packet_log WHERE packet_id = %s", (pid,))]
    assert not any("print" in a.lower() or "export" in a.lower() for a in actions)


def test_itp_print_and_pdf(shop: Shop) -> None:
    pkt = at_complete(shop)
    unit_id = pkt["unit"]["id"]
    itp = start_itp(shop, unit_id)
    q = shop("qc.quinn")
    data = get_pdf(q, f"/api/itps/{itp['id']}/pdf")
    assert "OPEN, not signed off" in flat(page_texts(data)[0])
    row = q1(shop.settings, "SELECT kind, unit_id, itp_id, packet_id FROM print_log ORDER BY id DESC LIMIT 1")
    assert tuple(row) == ("itp-pdf", unit_id, itp["id"], None)
    r = q.get(f"/api/itps/{itp['id']}/pdf?download=1")
    assert r.status_code == 200 and r.headers["content-disposition"].startswith("attachment;")
    assert f"{itp['no']}.pdf" in r.headers["content-disposition"]
    data = get_pdf(shop("pm.paula"), f"/api/itps/{itp['id']}/print")
    for t in page_texts(data):
        assert "PRINTED COPY" in flat(t) and "verify revision before use" in flat(t)
    assert q1(shop.settings, "SELECT kind FROM print_log ORDER BY id DESC LIMIT 1")[0] == "itp"
    itp = ok(sign_off(q, sign_all_blocks(shop, approve_all(shop, itp))))
    data = get_pdf(q, f"/api/itps/{itp['id']}/pdf")
    assert "signed off" in flat(page_texts(data)[0]) and "OPEN, not signed off" not in flat(page_texts(data)[0])
    refused(q.get("/api/itps/999999/pdf"), 404, "not_found")


def test_signed_out_pm_removed_and_append_only(shop: Shop) -> None:
    gone = draft(shop)
    remove_by_sql(shop.settings, gone["id"])
    pkt = at_complete(shop)
    pid = pkt["id"]
    anon = TestClient(shop.main.app, headers=ORIGIN_HEADERS)
    for url in (
        f"/api/packets/{pid}/pdf/databook",
        f"/api/packets/{pid}/pdf/traveler",
        f"/api/packets/{pid}/export/traceability.csv",
        f"/api/packets/{pid}/print/traveler",
        f"/api/packets/{pid}/prints",
    ):
        assert anon.get(url).status_code == 401, url
    # the PM role reads every one
    pm = shop("pm.paula")
    for kind in ("traveler", "map-h", "map-w", "clean", "databook"):
        get_pdf(pm, f"/api/packets/{pid}/print/{kind}")
    # a missing packet and a removed packet are 404
    refused(pm.get("/api/packets/999999/print/traveler"), 404, "not_found")
    # a fresh session: the PDFs above can take longer than the idle lock (15 minutes over the tunnel)
    clara = shop.fresh("clerk.clara")
    for url in (
        f"/api/packets/{gone['id']}/print/traveler",
        f"/api/packets/{gone['id']}/pdf/traveler",
        f"/api/packets/{gone['id']}/export/traceability.csv",
    ):
        refused(clara.get(url), 404, "not_found")
    # the print log is append-only
    with app_conn(shop.settings) as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("UPDATE print_log SET what = 'x' WHERE packet_id = %s", (pid,))
    with app_conn(shop.settings) as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("DELETE FROM print_log WHERE packet_id = %s", (pid,))
    assert q1(shop.settings, "SELECT count(*) FROM print_log WHERE packet_id = %s", (pid,))[0] == 5