PacketStation/tests/server/test_tracecsv.py

"""Phase 7 traceability CSV: the formula guard of csv_cell, and the export endpoint (content type, file name, CRLF
line ends, no BOM, the header rows, a BOM row with a short and its receipts, the weld rows, the Q.C. rows, the
audit row)."""

from __future__ import annotations

import csv
import io
from collections.abc import Iterator

import pytest
from closehelpers import HEATS, RECEIPT_HEAT, at_complete, q1, qall, rich_complete
from fakedir import FakeDirectory
from pshelpers import app_conn, needs_pg
from shophelpers import Shop, open_shop

from packetstation.pdf.tracecsv import csv_cell
from packetstation.settings import Settings


@pytest.mark.parametrize(
    ("value", "want"),
    [
        ("=1+1", "'=1+1"),
        ("+31", "'+31"),
        ("-2", "'-2"),
        ("@SUM(A1)", "'@SUM(A1)"),
        ("\tTAB", "'\tTAB"),
        ("\rCR", '"\'\rCR"'),
        ('say "hi"', '"say ""hi"""'),
        ("a,b", '"a,b"'),
        ("two\nlines", '"two\nlines"'),
        ('=HYPERLINK("x")', '"\'=HYPERLINK(""x"")"'),
        ("HT-10", "HT-10"),
        ("", ""),
        (None, ""),
        (3, "3"),
    ],
)
def test_csv_cell(value: object, want: str) -> None:
    assert csv_cell(value) == want


# ---------------------------------------------------------------- the endpoint


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


def rows_of(text: str) -> list[list[str]]:
    return list(csv.reader(io.StringIO(text, newline="")))


@needs_pg
def test_traceability_csv(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    project = pkt["unit"]["project"]["number"]
    r = shop("pm.paula").get(f"/api/packets/{pid}/export/traceability.csv")
    assert r.status_code == 200, r.text
    assert r.headers["content-type"].startswith("text/csv")
    cd = r.headers["content-disposition"]
    assert "_D-5001_traceability.csv" in cd and project in cd, cd
    assert r.headers.get("cache-control") == "no-store"
    raw = r.content
    assert not raw.startswith(b"\xef\xbb\xbf")
    assert b"\r\n" in raw and b"\n" not in raw.replace(b"\r\n", b"")
    rows = rows_of(raw.decode("utf-8"))
    assert rows[0] == ["PESCO weld & heat traceability record"]
    assert rows[1] == ["Project", project, "Lot", "001", "Operation", "WELD-PART"], rows[1]
    flat_rows = [",".join(x) for x in rows]
    assert any(x.startswith("Drawing,D-5001,Rev,B") for x in flat_rows), flat_rows[:6]
    assert any(x.startswith("WPS,WPS-11,Station,Stall 1") for x in flat_rows), flat_rows[:6]
    assert any(x.startswith("Filler,ER70S-6") for x in flat_rows), flat_rows[:6]
    head = rows.index(
        [
            "BOM item",
            "Qty",
            "Issued",
            "Location",
            "Part ID",
            "Description",
            "Length",
            "Material",
            "Heat no.",
            "MTR on file",
            "Open short qty",
            "Short reason",
            "Short received",
        ]
    )
    bom = {r[0]: r for r in rows[head + 1 : head + 4]}
    assert bom["1"][4] == "PIPE-4" and bom["1"][8] == HEATS[1] and bom["1"][9] == "Yes"
    assert bom["2"][9] == "Yes"
    short = bom["3"]
    assert HEATS[3] in short[8] and RECEIPT_HEAT in short[8]
    assert short[10] == "1" and short[11] == "On order"
    assert short[12].startswith(f"1 pc {RECEIPT_HEAT} "), short
    weld_head = next(i for i, x in enumerate(rows) if x[:2] == ["Weld", "Work order"])
    assert rows[weld_head] == [
        "Weld",
        "Work order",
        "Piece",
        "Joins items",
        "Heats joined",
        "Type",
        "Size",
        "Process",
        "WPS",
        "Stamp",
        "Logged",
    ]
    welds = rows[weld_head + 1 : weld_head + 3]
    assert [w[0] for w in welds] == ["W1", "W2"]
    assert welds[0][3] == "1 + 2" and welds[0][4] == f"{HEATS[1]} + {HEATS[2]}" and welds[0][5] == "Fillet"
    assert any(x[:2] == ["Checklist item", "Done"] for x in rows)
    qc = next(x for x in rows if x and x[0] == "Q.C. inspection")
    assert qc[:4] == ["Q.C. inspection", "accepted", "Inspector", "Quinn QC"], qc
    assert "Quinn QC" in qc and "Q1" in qc
    assert any(x and x[0] == "NDE" for x in rows)
    assert any(x and x[0].startswith("Q.C.: ") for x in rows)
    vt = [x for x in rows if x and x[0].startswith("VT W")]
    assert [x[0] for x in vt] == ["VT W1", "VT W2"] and all(x[1] == "accept" for x in vt)
    au = q1(
        shop.settings,
        "SELECT detail FROM audit_events WHERE action = 'export' AND packet_id = %s ORDER BY id DESC",
        (pid,),
    )
    assert au is not None and au[0]["kind"] == "csv"
    log = qall(shop.settings, "SELECT kind, by_name FROM print_log WHERE packet_id = %s", (pid,))
    assert ("csv", "Paula PM") in [tuple(x) for x in log]


@needs_pg
def test_formula_values_are_guarded_in_the_export(shop: Shop) -> None:
    pkt = at_complete(shop)
    with app_conn(shop.settings) as conn:
        conn.execute("UPDATE bom_lines SET description = '=cmd|calc' WHERE packet_id = %s AND item = 1", (pkt["id"],))
    r = shop("clerk.clara").get(f"/api/packets/{pkt['id']}/export/traceability.csv")
    assert r.status_code == 200
    assert "'=cmd|calc" in r.text and ",=cmd" not in r.text