PacketStation/tests/server/test_pdf_databook.py

"""Phase 7 data book builder, called directly on a packet built through the API: two drawing sheets and a void
copy of sheet 1, a heat table, two welds, an MTR PDF and an MTR PNG, a short with a receipt, a packet document, a
Q.C. folder document, the closed stage and the Q.C. accept.

The index (seven sections in order, the expected rows), the page count, the cover, the index page, the void
sheet, the traceability page, the warning rows, the metadata, the stamp on every page and its signature, and the
stamp per mode. Nothing here reads VISUAL or M-Files."""

from __future__ import annotations

import re
from collections.abc import Iterator
from datetime import UTC, datetime
from typing import Any

import pytest
from closehelpers import (
    HEATS,
    RECEIPT_HEAT,
    builder_conn,
    delete_asset_file,
    flat,
    page_texts,
    pdf_reader,
    rich_complete,
    who,
)
from fakedir import FakeDirectory
from pshelpers import TEST_SEAL_KEY, needs_pg
from qchelpers import g
from shophelpers import Shop, open_shop, run

from packetstation.pdf import databook, stamps
from packetstation.pdf.style import clean
from packetstation.settings import Settings

pytestmark = needs_pg

NOW = datetime(2026, 10, 17, 21, 4, tzinfo=UTC)
SEAL = {"hash": "ab12cd34ef56" + "0" * 52, "key_id": "k1234567", "sealed_at": NOW}
SECTIONS = [1, 2, 3, 4, 5, 6, 7]
SIG_RE = re.compile(r"stamp ([0-9a-f]{16})")


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


def build(shop: Shop, pid: int, mode: str = "preview", **kw: Any) -> Any:
    with builder_conn(shop.settings) as conn:
        return databook.build(
            conn,
            shop.settings,
            pid,
            who=who(shop),
            now=NOW,
            seal=SEAL if mode == "filed" else None,
            mode=mode,
            **kw,
        )


def index(built: Any) -> list[dict[str, Any]]:
    keys = ("section", "doc", "ref", "rev", "page", "warn", "count")
    return [{k: g(r, k) for k in keys} for r in built.index]


def docs_in(rows: list[dict[str, Any]], section: int) -> list[str]:
    return [str(r["doc"]) for r in rows if r["section"] == section]


def test_data_book_preview(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    built = build(shop, pid)
    data: bytes = built.bytes
    assert data[:5] == b"%PDF-"
    reader = pdf_reader(data)
    assert len(reader.pages) == built.pages
    rows = index(built)

    # the seven sections, in order, each present
    seen = [r["section"] for r in rows]
    assert seen == sorted(seen) and sorted(set(seen)) == SECTIONS, seen

    s1 = docs_in(rows, 1)
    assert sum("approved drawing" in d for d in s1) == 2, s1
    assert any(d.startswith("SUPERSEDED - VOID by CO-1") for d in s1), s1
    drawing_rows = [r for r in rows if r["section"] == 1 and "approved drawing" in str(r["doc"])]
    assert all(r["ref"] == "D-5001" and r["rev"] == "RB" for r in drawing_rows), drawing_rows

    s2 = docs_in(rows, 2)
    assert s2[0].startswith("Material traceability record"), s2
    assert any(d.startswith("Heat map") and "heat number table" in d for d in s2), s2

    s3 = docs_in(rows, 3)
    assert any(d.startswith("Weld map") and "W1, W2" in d for d in s3), s3
    assert any(d.startswith("Weld log, welding record and checklist") for d in s3), s3

    s4 = docs_in(rows, 4)
    assert len(s4) == 1 and s4[0].startswith("Q.C. inspection") and "ACCEPTED" in s4[0], s4

    s5 = docs_in(rows, 5)
    assert any("WPS / PQR" in d and "Unit folder doc" in d for d in s5), s5
    assert any("RT report W1-W2" in d for d in s5), s5
    itp_warn = [r for r in rows if r["section"] == 5 and r["doc"] == "No signed ITP filed for this unit"]
    assert len(itp_warn) == 1 and itp_warn[0]["warn"] is True

    s6 = docs_in(rows, 6)
    assert len(s6) == 2 and all(d.startswith("MTR") for d in s6), s6
    assert "item 1" in s6[0] and "item 2" in s6[1]
    mtr_rows = [r for r in rows if r["section"] == 6]
    assert mtr_rows[0]["ref"] == HEATS[1]

    assert docs_in(rows, 7) == ["Packet record history"]

    # pages: the parts plus the front matter (cover and index); page numbers point into the book
    parts = sum(int(r["count"] or 0) for r in rows)
    front = built.pages - parts
    assert front == 2, (built.pages, parts)  # the cover and one index page
    numbered = [r for r in rows if r["page"] is not None]
    assert all(front < int(r["page"]) <= built.pages for r in numbered), numbered
    assert all(r["page"] is None and not r["count"] for r in rows if r["warn"] and "could not" in str(r["doc"]))

    texts = page_texts(data)
    assert "MANUFACTURING" in texts[0] and "DATA BOOK" in flat(texts[0])
    assert pkt["unit"]["project"]["number"] in flat(texts[0])
    index_text = flat(" ".join(texts[1:front]))
    assert "INDEX" in index_text and "TABLE OF CONTENTS" in index_text
    for r in rows:
        words = flat(clean(str(r["doc"])))[:30]
        assert words in index_text, (words, index_text[:2000])

    # the void sheet carries the void text, the traceability page the heats
    void_page = next(r for r in rows if str(r["doc"]).startswith("SUPERSEDED"))["page"]
    assert "VOID - superseded by CO-1" in flat(texts[int(void_page) - 1])
    trace_page = next(r for r in rows if str(r["doc"]).startswith("Material traceability"))["page"]
    trace = flat(" ".join(texts[int(trace_page) - 1 : int(trace_page) + 1]))
    for heat in (HEATS[1], HEATS[2], RECEIPT_HEAT):
        assert heat in trace, heat
    assert "open items" in trace.lower() and "item 3" in trace

    # metadata
    meta = reader.metadata
    assert meta.title == f"{pkt['unit']['project']['number']} Lot 001 Manufacturing Data Book - Spool 7 outlet"
    assert meta.producer == "Packet Station"
    assert meta.get("/Author") == "PESCO - Process Equipment & Service Company, Inc."

    # the stamp on every page, preview mode, and its signature verifies
    sigs = set()
    for i, t in enumerate(texts):
        ft = flat(t)
        assert "Controlled copy - exported by Clara Clerk (4001)" in ft, (i, ft[-300:])
        assert "not yet filed" in ft, i
        m = SIG_RE.search(ft)
        assert m, (i, ft[-300:])
        sigs.add(m.group(1))
    assert len(sigs) == 1
    sig = sigs.pop()
    user = who(shop)["id"]
    key = TEST_SEAL_KEY.encode()
    assert stamps.verify_stamp(key, pid, "databook", "", user, NOW.isoformat(), sig)
    assert not stamps.verify_stamp(key, pid, "traveler", "", user, NOW.isoformat(), sig)


def test_data_book_filed_mode_and_warnings(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    filed = build(shop, pid, mode="filed")
    texts = page_texts(filed.bytes)
    assert all("seal ab12cd34ef56" in flat(t) for t in texts)
    assert not any("not yet filed" in flat(t) for t in texts)
    m = SIG_RE.search(flat(texts[0]))
    assert m and filed.stamp_sig == m.group(1)
    assert stamps.verify_stamp(TEST_SEAL_KEY.encode(), pid, "databook", SEAL["hash"], who(shop)["id"], NOW, m.group(1))
    meta = pdf_reader(filed.bytes).metadata
    assert "seal" in (meta.creator or "") and "4001" in (meta.creator or "")

    # no MTRs: the MTR warning row; a deleted packet document file: could not be read
    run(shop.settings, "UPDATE bom_lines SET mtr_asset_id = NULL WHERE packet_id = %s", (pid,))
    doc = next(d for d in pkt["docs"] if d.get("removed") is None)
    asset_id = doc.get("asset_id") or doc["url"].rsplit("/", 1)[-1]
    delete_asset_file(shop.settings, asset_id)
    built = build(shop, pid)
    rows = index(built)
    mtr = [r for r in rows if r["section"] == 6]
    assert [r["doc"] for r in mtr] == ["No MTRs attached in Packet Station — add from the warehouse files"]
    assert mtr[0]["warn"] is True and mtr[0]["page"] is None
    bad = [r for r in rows if r["section"] == 5 and str(r["doc"]).endswith("could not be read")]
    assert len(bad) == 1 and "RT report W1-W2" in bad[0]["doc"] and bad[0]["warn"] is True
    # the index still fits on one page: the cover, the index and the parts
    assert built.pages == 2 + sum(int(r["count"] or 0) for r in rows)


def test_data_book_with_an_itp_document_has_no_itp_warning(shop: Shop) -> None:
    pkt = rich_complete(shop, itp_doc=True, mtrs=False)
    rows = index(build(shop, pkt["id"]))
    s5 = [r for r in rows if r["section"] == 5]
    assert s5[0]["doc"].startswith("ITP"), s5
    assert not any(r["doc"] == "No signed ITP filed for this unit" for r in s5)