PacketStation/tests/server/test_pm_export.py

"""Phase 8: the Project Management exports. The CSV per table and for every table (the header rows, the formula
guard, the BOM and CRLF), the production status report PDF (pages, cover, the controlled copy stamp on every page),
and one print_log row plus one audit row `report exported` for each export.

The PM world (pmhelpers.py) is built once for the module; one draft's title starts with `=`.
"""

from __future__ import annotations

import csv
import io
import re
from collections.abc import Iterator
from typing import Any

import pytest
from closehelpers import flat, page_texts, pdf_pages, q1, qall
from pmhelpers import (
    FORMULA_TITLE,
    HEADS,
    PERIODS,
    PmWorld,
    anon,
    module_shop,
    pm_world,
    titles,
)
from pshelpers import needs_pg
from shophelpers import run

pytestmark = needs_pg

TABLE_KEYS = tuple(HEADS)  # projects, aging_*, hours_*, quality_*, jobs


@pytest.fixture(scope="module")
def world(tmp_path_factory: pytest.TempPathFactory) -> Iterator[PmWorld]:
    with module_shop(tmp_path_factory.mktemp("pm_export")) as shop:
        w = pm_world(shop)
        run(shop.settings, "UPDATE packets SET title = %s WHERE id = %s", (FORMULA_TITLE, w.k["k_draft"]))
        yield w


def export(w: PmWorld, table: str, sam: str = "pm.paula", **params: Any) -> Any:
    return w.shop(sam).get("/api/pm/export.csv", params={"table": table, **params})


def parse(body: bytes) -> list[list[str]]:
    assert body.startswith(b"\xef\xbb\xbf"), body[:10]  # the BOM, for Excel
    assert b"\r\n" in body and body.count(b"\n") == body.count(b"\r\n"), "every line ends CRLF"
    return list(csv.reader(io.StringIO(body.decode("utf-8-sig"), newline="")))


def filename(r: Any) -> str:
    cd = r.headers["content-disposition"]
    m = re.search(r'filename\*?=(?:UTF-8\'\')?"?([^";]+)"?', cd)
    assert m, cd
    return m.group(1)


def print_rows(w: PmWorld, kind: str) -> list[Any]:
    return qall(
        w.settings,
        "SELECT what, pages, by_name, packet_id, unit_id FROM print_log WHERE kind = %s ORDER BY id",
        (kind,),
    )


def audit_rows(w: PmWorld) -> list[dict[str, Any]]:
    return [
        r[0] for r in qall(w.settings, "SELECT detail FROM audit_events WHERE action = 'report exported' ORDER BY id")
    ]


def header_ok(rows: list[list[str]], name: str, label: str = "All projects", period: str = "90") -> None:
    assert rows[0] == [f"PESCO Packet Station — {name}"], rows[0]
    assert rows[1] == ["Scope", label]
    assert rows[2] == ["History period", PERIODS[period]]
    assert rows[3][0] == "Exported" and rows[3][2] == "by" and rows[3][3] == "Paula PM" and rows[3][1]
    assert rows[4] == [] or rows[4] == [""]


def table_at(rows: list[list[str]], title: str) -> tuple[list[str], list[list[str]]]:
    """The head and the body rows of the table titled `title` in a parsed CSV."""
    i = rows.index([title])
    head = rows[i + 1]
    body = []
    for r in rows[i + 2 :]:
        if r in ([], [""]):
            break
        body.append(r)
    return head, body


# ================================================================ CSV


@pytest.mark.parametrize("key", TABLE_KEYS)
def test_csv_per_table(world: PmWorld, key: str) -> None:
    r = export(world, key)
    assert r.status_code == 200, r.text
    assert r.headers["content-type"].startswith("text/csv")
    assert "attachment" in r.headers["content-disposition"]
    name = filename(r)
    assert name.startswith("PacketStation_") and name.endswith(f"_{world.today.isoformat()}.csv"), name
    assert re.fullmatch(r"PacketStation_[A-Za-z0-9_]+_\d{4}-\d{2}-\d{2}\.csv", name), name
    rows = parse(r.content)
    title = titles()[key]
    assert rows[0][0].startswith("PESCO Packet Station — ")
    head, body = table_at(rows, title)
    assert head == HEADS[key]
    assert all(len(b) == len(head) for b in body), key


def test_csv_jobs_and_the_formula_guard(world: PmWorld) -> None:
    r = export(world, "jobs")
    assert r.status_code == 200, r.text
    assert filename(r) == f"PacketStation_Jobs_{world.today.isoformat()}.csv"
    rows = parse(r.content)
    header_ok(rows, "Jobs")
    head, body = table_at(rows, titles()["jobs"])
    assert head == HEADS["jobs"]
    assert len(body) == 14  # 12 open and 2 filed
    col = head.index("Title")
    assert "'" + FORMULA_TITLE in [b[col] for b in body]
    assert FORMULA_TITLE not in [b[col] for b in body]
    every = [c for b in rows for c in b]
    assert not any(c[:1] in ("=", "+", "-", "@", "\t") for c in every if c), "the formula guard on every cell"


def test_csv_all(world: PmWorld) -> None:
    r = export(world, "all")
    assert r.status_code == 200, r.text
    assert filename(r) == f"PacketStation_All_reports_{world.today.isoformat()}.csv"
    rows = parse(r.content)
    header_ok(rows, "All reports")
    for key in TABLE_KEYS:
        head, _ = table_at(rows, titles()[key])
        assert head == HEADS[key], key
    _, units = table_at(rows, titles()["projects"])
    assert len(units) >= 2 + 8  # every project row and every unit (expand all)


def test_csv_follows_scope_and_period(world: PmWorld) -> None:
    pa = world.projects["L670"]
    r = export(world, "quality_sentback", project_id=pa, period="365")
    assert r.status_code == 200, r.text
    rows = parse(r.content)
    assert rows[1] == ["Scope", "Project L670"]
    assert rows[2] == ["History period", PERIODS["365"]]
    _, body = table_at(rows, titles("365")["quality_sentback"])
    assert len(body) == 4


def test_csv_unknown_table(world: PmWorld) -> None:
    r = export(world, "salaries")
    assert r.status_code == 404 and r.json()["code"] == "not_found", r.text


def test_csv_is_logged(world: PmWorld) -> None:
    before_print = len(print_rows(world, "report-csv"))
    before_audit = len(audit_rows(world))
    assert export(world, "jobs").status_code == 200
    rows = print_rows(world, "report-csv")
    assert len(rows) == before_print + 1
    what, pages, by_name, packet_id, unit_id = rows[-1]
    assert what.endswith(" · CSV · All projects") and by_name == "Paula PM"
    assert packet_id is None and unit_id is None and pages is None
    audit = audit_rows(world)
    assert len(audit) == before_audit + 1
    detail = audit[-1]
    assert set(detail) >= {"name", "format", "scope", "period"}
    assert str(detail["format"]).lower() == "csv" and detail["scope"] == "All projects"
    assert detail["period"] in ("90", PERIODS["90"])


def test_export_needs_a_session(world: PmWorld) -> None:
    c = anon(world.shop)
    assert c.get("/api/pm/export.csv", params={"table": "all"}).status_code == 401
    assert c.get("/api/pm/report.pdf").status_code == 401
    assert world.shop("clerk.clara").get("/api/pm/export.csv", params={"table": "jobs"}).status_code == 403
    assert world.shop("clerk.clara").get("/api/pm/report.pdf").status_code == 403


# ================================================================ PDF


def test_report_pdf(world: PmWorld) -> None:
    before = len(print_rows(world, "report-pdf"))
    before_audit = len(audit_rows(world))
    r = world.shop("pm.paula").get("/api/pm/report.pdf")
    assert r.status_code == 200, r.text
    assert r.headers["content-type"].startswith("application/pdf")
    assert r.headers["content-disposition"].startswith("inline")
    assert filename(r) == f"PESCO_Production_Status_{world.today.isoformat()}.pdf"
    data = r.content
    assert data[:5] == b"%PDF-"
    pages = pdf_pages(data)
    assert pages > 5
    texts = [flat(t) for t in page_texts(data)]
    assert "PRODUCTION STATUS REPORT" in texts[0]
    assert "Packet Station · All projects" in texts[0]
    for i, t in enumerate(texts):
        assert "Controlled copy" in t, f"page {i + 1} has no stamp"
    body = texts[1:]
    for i, t in enumerate(body):
        assert "Production status report" in t, i
        assert "CONFIDENTIAL & PROPRIETARY" in t, i
    every = " ".join(texts)
    for part in ("SUMMARY", "PROJECTS", "FLOW", "LABOR", "ISSUES"):
        assert part in every, part
    assert "L670" in every
    # logged once with the page count
    rows = print_rows(world, "report-pdf")
    assert len(rows) == before + 1
    what, logged_pages, by_name, packet_id, _ = rows[-1]
    assert what == "Production status report · PDF · All projects"
    assert logged_pages == pages and by_name == "Paula PM" and packet_id is None
    audit = audit_rows(world)
    assert len(audit) == before_audit + 1 and str(audit[-1]["format"]).lower() == "pdf"


def test_report_pdf_download_and_scope(world: PmWorld) -> None:
    r = world.shop("pm.paula").get(
        "/api/pm/report.pdf", params={"download": 1, "project_id": world.projects["L670"], "period": "all"}
    )
    assert r.status_code == 200, r.text
    assert r.headers["content-disposition"].startswith("attachment")
    first = flat(page_texts(r.content)[0])
    assert "Packet Station · Project L670" in first
    row = q1(world.settings, "SELECT what FROM print_log WHERE kind = 'report-pdf' ORDER BY id DESC LIMIT 1")
    assert row[0] == "Production status report · PDF · Project L670"