PacketStation/tests/server/test_close_file.py

"""Phase 7 close and file through the API: the VISUAL close and the filing as every role, the feed's open work
orders (the note suffix, the 409 and the confirm), the data book filed with the seal, the reopen that keeps the
broken seal and its data book, the refile with a second seal, a failed data book build that files nothing, and a
hand packet with no feed state.

The feed state is written by SQL in the shape feed/apply.py writes it. Nothing here reads VISUAL or M-Files."""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import pytest
from closehelpers import (
    at_complete,
    file_packet,
    filed,
    flat,
    leg,
    pdf_text,
    q1,
    qall,
    set_feed_legs,
    vclose,
)
from fakedir import FakeDirectory
from pshelpers import needs_pg
from qchelpers import at_qc
from shophelpers import SIGN, Shop, find_log, fresh, log_rows, ok, open_shop, refused

from packetstation.policy import Action, check, principal_from_roles
from packetstation.settings import Settings

pytestmark = needs_pg

MSG_VCLOSED = "Close the job in VISUAL first."
CLERICAL = ("clerk.clara", "clerk.leo")
NOT_CLERICAL = ("admin.ann", "wh.wade", "qc.quinn", "pm.paula", "weld.andy", "lead.lou", "sup.wes")


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


def oracle(sam: str, status: str, action: Action, routed_to: str = "databook", vclosed: bool = False) -> str:
    """The expected code for SET_VCLOSED and FILE, from the phase 2 rule table: Clerical only, status complete;
    FILE on the data book route needs the VISUAL close; an admin outside the role needs an IT reason."""
    if status != "complete":
        return "wrong_status"
    if action is Action.SET_VCLOSED and routed_to != "databook":
        return "wrong_status"
    if sam not in CLERICAL:
        return "it_reason_required" if sam == "admin.ann" else "forbidden"
    if action is Action.FILE and routed_to == "databook" and not vclosed:
        return "precondition"
    return ""


# ---------------------------------------------------------------- the pure rule, every role and status


ROLES = {
    "clerk.clara": ["clerical"],
    "clerk.leo": ["clerical_supervisor"],
    "admin.ann": ["admin"],
    "wh.wade": ["warehouse"],
    "qc.quinn": ["qc"],
    "pm.paula": ["pm"],
    "weld.andy": ["shop:03:worker"],
    "lead.lou": ["shop:03:lead"],
    "sup.wes": ["shop:03:supervisor"],
}


def test_rule_table_matches_the_oracle_for_every_role_and_status() -> None:
    import dataclasses

    from test_policy_matrix import state

    from packetstation.policy import STATUSES

    for sam, roles in ROLES.items():
        p = principal_from_roles("00000000-0000-0000-0000-0000000000b0", roles)
        for status in STATUSES:
            for action in (Action.SET_VCLOSED, Action.FILE):
                for route in ("databook", "shipping"):
                    for vclosed in (False, True):
                        s = dataclasses.replace(state(status), routed_to=route, vclosed=vclosed)
                        d = check(p, s, action)
                        want = oracle(sam, status, action, route, vclosed)
                        if want == "precondition":
                            assert not d.ok and d.why == MSG_VCLOSED, (sam, status, action, route, d)
                        elif want == "wrong_status" and action is Action.SET_VCLOSED and status == "complete":
                            # SET_VCLOSED on the Shipping route: refused, whatever code the engine picks
                            assert not d.ok, (sam, status, action, route)
                        else:
                            assert d.ok is (want == ""), (sam, status, action, route, vclosed, d.code)
                            assert d.code == want, (sam, status, action, route, vclosed, d.code)


# ---------------------------------------------------------------- the API, every role


def test_vclosed_and_file_as_every_role_at_complete(shop: Shop) -> None:
    pkt = at_complete(shop)
    pid = pkt["id"]
    for sam in NOT_CLERICAL:
        r = shop(sam).post(f"/api/packets/{pid}/vclosed", json={"closed": True})
        want = oracle(sam, "complete", Action.SET_VCLOSED)
        assert r.status_code in (400, 403) and r.json()["code"] == want, (sam, r.text)
        r = file_packet(shop, fresh(shop("clerk.clara"), pid), by=sam)
        assert r.json()["code"] in (oracle(sam, "complete", Action.FILE), "signature_required"), (sam, r.text)
        assert r.status_code in (400, 403), (sam, r.text)
    # Clerical: the file waits for the VISUAL close on the data book route
    refused(file_packet(shop, pkt), 403, "precondition", MSG_VCLOSED)
    pkt = vclose(shop, pkt, by="clerk.leo")
    assert pkt["vclosed"]["by"]["display_name"] == "Leo Lead"
    out = ok(file_packet(shop, pkt, by="clerk.clara"))
    assert out["status"] == "filed"


def test_vclosed_and_file_refused_outside_complete(shop: Shop) -> None:
    pkt = at_qc(shop)
    c = shop("clerk.clara")
    refused(c.post(f"/api/packets/{pkt['id']}/vclosed", json={"closed": True}), 403, "wrong_status")
    refused(file_packet(shop, pkt), 403, "wrong_status")
    done = filed(shop)
    refused(c.post(f"/api/packets/{done['id']}/vclosed", json={"closed": False}), 403, "filed")
    r = file_packet(shop, done)
    assert r.status_code == 403 and r.json()["code"] in ("filed", "wrong_status"), r.text


def test_shipping_route_files_without_the_visual_close(shop: Shop) -> None:
    pkt = at_complete(shop, route="shipping")
    r = shop("clerk.clara").post(f"/api/packets/{pkt['id']}/vclosed", json={"closed": True})
    assert r.status_code in (403, 409) and r.json()["code"] in ("wrong_status", "precondition"), r.text
    out = ok(file_packet(shop, pkt))
    assert out["status"] == "filed" and out["vclosed"] is None
    assert out["seal"]["databook"]["pages"] > 0


# ---------------------------------------------------------------- the feed's open work orders


def test_vclose_check_and_the_open_wo_note(shop: Shop) -> None:
    pkt = at_complete(shop)
    pid = pkt["id"]
    set_feed_legs(shop.settings, pid, [leg("L670001", st="C"), leg("L670002", st="R")])
    view = fresh(shop("clerk.clara"), pid)
    vc = view["vclose_check"]
    assert vc["open_wos"] == ["L670002/001"] and vc["closed_at"] is None and vc["observed_at"] is not None
    assert vc["agrees"] is True  # neither closed
    pkt = vclose(shop, view)
    assert pkt["vclose_check"]["agrees"] is False
    row = find_log(shop("clerk.clara"), pid, "Job closed in VISUAL")
    assert row["note"] == "L670001, L670002 · VISUAL still shows open: L670002/001", row


def test_vclose_check_closed_by_the_feed(shop: Shop) -> None:
    pkt = at_complete(shop)
    pid = pkt["id"]
    set_feed_legs(shop.settings, pid, [leg("L670001", st="C"), leg("L670002", st="C")], closed=True)
    vc = fresh(shop("clerk.clara"), pid)["vclose_check"]
    assert vc["open_wos"] == [] and vc["closed_at"] is not None and vc["agrees"] is False
    pkt = vclose(shop, fresh(shop("clerk.clara"), pid))
    assert pkt["vclose_check"]["agrees"] is True
    assert find_log(shop("clerk.clara"), pid, "Job closed in VISUAL")["note"] == "L670001, L670002"
    out = ok(file_packet(shop, pkt))
    assert out["seal"]["open_wos"] is None


def test_vclose_check_before_the_feed_looked(shop: Shop) -> None:
    pkt = at_complete(shop)
    set_feed_legs(shop.settings, pkt["id"], None)
    vc = fresh(shop("clerk.clara"), pkt["id"])["vclose_check"]
    assert vc is not None and vc["observed_at"] is None and vc["open_wos"] == []


def test_file_with_open_work_orders_needs_the_confirm(shop: Shop) -> None:
    pkt = at_complete(shop)
    pid = pkt["id"]
    set_feed_legs(shop.settings, pid, [leg("L670001", st="R"), leg("L670002", lot="7A", st="R")])
    pkt = vclose(shop, fresh(shop("clerk.clara"), pid))
    sigs = q1(shop.settings, "SELECT count(*) FROM signatures")[0]
    body = refused(file_packet(shop, pkt), 409, "open_work_orders")
    assert body["work_orders"] == ["L670001/001", "L670002/7A"]
    assert body["detail"] == (
        "VISUAL still shows 2 work orders open on this packet: L670001/001, L670002/7A. File anyway?"
    )
    # nothing written, no signature taken
    assert q1(shop.settings, "SELECT count(*) FROM signatures")[0] == sigs
    assert q1(shop.settings, "SELECT count(*) FROM packet_seals WHERE packet_id = %s", (pid,))[0] == 0
    assert fresh(shop("clerk.clara"), pid)["status"] == "complete"
    refused(file_packet(shop, pkt, confirm_open_wos=False), 409, "open_work_orders")
    out = ok(file_packet(shop, pkt, confirm_open_wos=True, note="Customer pickup today"))
    assert out["status"] == "filed" and out["seal"]["open_wos"] == ["L670001/001", "L670002/7A"]
    row = q1(shop.settings, "SELECT open_wos FROM packet_seals WHERE packet_id = %s", (pid,))
    assert row[0] == ["L670001/001", "L670002/7A"]
    note = find_log(shop("clerk.clara"), pid, "Filed in the data book")["note"]
    assert note == "Customer pickup today · Filed with work orders open in VISUAL: L670001/001, L670002/7A", note


def test_cancelled_legs_do_not_count_as_open(shop: Shop) -> None:
    pkt = at_complete(shop)
    set_feed_legs(shop.settings, pkt["id"], [leg("L670001", st="C"), leg("L670002", st="X")])
    assert fresh(shop("clerk.clara"), pkt["id"])["vclose_check"]["open_wos"] == []


def test_hand_packet_has_no_feed_check(shop: Shop) -> None:
    pkt = at_complete(shop)
    assert pkt["vclose_check"] is None
    pkt = vclose(shop, pkt)
    assert find_log(shop("clerk.clara"), pkt["id"], "Job closed in VISUAL")["note"] == "L670001, L670002"
    out = ok(file_packet(shop, pkt))
    assert out["status"] == "filed" and out["seal"]["open_wos"] is None


# ---------------------------------------------------------------- the filing, the seal, the data book


def test_filing_builds_and_stores_the_data_book(shop: Shop) -> None:
    pkt = vclose(shop, at_complete(shop))
    pid = pkt["id"]
    out = ok(file_packet(shop, pkt))
    assert out["status"] == "filed" and out["databook_at"] is not None
    seal = out["seal"]
    assert seal["broken_at"] is None and seal["signed"] is True and out["seals"] == 1
    db = seal["databook"]
    assert db["pages"] > 0 and db["url"] == f"/_blob/{db['asset_id']}"
    assert db["download_url"] == f"/_blob/{db['asset_id']}?download=1"
    assert db["file_name"].endswith("_Manufacturing_Data_Book.pdf") and db["size"] > 0
    rows = log_rows(shop("clerk.clara"), pid)
    actions = [r["action"] for r in rows]
    i = actions.index("Filed in the data book")
    j = actions.index("Data book filed")
    assert j < i, actions  # newest first: the data book row comes after the filing row
    assert rows[j]["note"] == f"{db['pages']} pages"
    # the seal checks out and the stored copy is the filed data book
    view = ok(shop("clerk.clara").get(f"/api/packets/{pid}/seal"))
    assert view["valid"] is True
    row = q1(shop.settings, "SELECT databook_asset_id, databook_pages FROM packet_seals WHERE packet_id = %s", (pid,))
    assert str(row[0]) == db["asset_id"] and row[1] == db["pages"]
    r = shop("clerk.clara").get(db["url"])
    assert r.status_code == 200 and r.headers["content-type"].startswith("application/pdf")
    text = flat(pdf_text(r.content))
    assert "MANUFACTURING" in text
    project = out["unit"]["project"]["number"]
    assert project in text
    assert f"seal {view['seal']['hash'][:12]}" in text
    src = q1(shop.settings, "SELECT source, source_ref, packet_id FROM assets WHERE id = %s", (db["asset_id"],))
    assert src[0] == "databook" and src[1]["packet_id"] == pid and src[2] == pid
    # the export answers with the stored copy, unchanged
    r2 = shop("pm.paula").get(f"/api/packets/{pid}/pdf/databook")
    assert r2.status_code == 200 and r2.content == r.content


def test_reopen_keeps_the_broken_seal_and_refile_adds_a_second(shop: Shop) -> None:
    pkt = filed(shop)
    pid = pkt["id"]
    first = pkt["seal"]["databook"]
    r = shop("admin.ann").post(f"/api/packets/{pid}/reopen", json={"reason": "Wrong heat number filed", "sign": SIGN})
    assert r.status_code == 200, r.text
    out = r.json()
    assert out["status"] == "complete" and out["seal"]["broken_at"] is not None
    assert out["seal"]["broken_reason"] == "Wrong heat number filed"
    assert out["seal"]["databook"]["asset_id"] == first["asset_id"] and out["seals"] == 1
    assert shop("clerk.clara").get(first["url"]).status_code == 200
    # complete again: the export builds a preview, not the broken seal's copy
    r = shop("clerk.clara").get(f"/api/packets/{pid}/pdf/databook")
    assert r.status_code == 200 and "not yet filed" in flat(pdf_text(r.content))
    # refile: a second seal, a second data book, the first kept
    again = ok(file_packet(shop, fresh(shop("clerk.clara"), pid)))
    assert again["status"] == "filed" and again["seals"] == 2
    assert again["seal"]["broken_at"] is None and again["seal"]["databook"]["asset_id"] != first["asset_id"]
    view = ok(shop("clerk.clara").get(f"/api/packets/{pid}/seal"))
    assert view["valid"] is True and len(view["seals"]) == 2
    assert view["seals"][1]["broken_reason"] == "Wrong heat number filed"
    rows = qall(
        shop.settings,
        "SELECT databook_asset_id, broken_at IS NOT NULL FROM packet_seals WHERE packet_id = %s ORDER BY id",
        (pid,),
    )
    assert [r[1] for r in rows] == [True, False] and all(r[0] is not None for r in rows)


def test_failed_data_book_files_nothing(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None:
    from packetstation.pdf import databook
    from packetstation.pdf.style import PdfError

    def boom(*a: Any, **k: Any) -> Any:
        raise PdfError("the drawing could not be read")

    monkeypatch.setattr(databook, "build", boom)
    pkt = vclose(shop, at_complete(shop))
    pid = pkt["id"]
    body = refused(file_packet(shop, pkt), 500, "databook_failed")
    assert body["detail"].startswith("The data book could not be built:") and "Nothing was filed." in body["detail"]
    view = fresh(shop("clerk.clara"), pid)
    assert view["status"] == "complete" and view["seal"] is None and view["seals"] == 0
    assert q1(shop.settings, "SELECT count(*) FROM packet_seals WHERE packet_id = %s", (pid,))[0] == 0
    assert "Filed in the data book" not in [r["action"] for r in log_rows(shop("clerk.clara"), pid)]