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 (a block on both routes since C5; the feed closes a VISUAL packet), 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:
    """A hand packet on the Shipping route: no feed state, no Clerical close."""
    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 (C5)

MSG_FEED_CLOSES = "The VISUAL feed marks this job closed when it sees every work order closed."
MSG_FEED_WAIT = "Filing waits until the VISUAL feed sees the job closed."


def open_msg(wos: str) -> str:
    return f"VISUAL still shows work orders open on this packet: {wos}. Filing waits until VISUAL closes them."


def test_visual_packet_open_in_visual_cannot_be_closed_or_filed(shop: Shop) -> None:
    """C5: Clerical cannot mark a VISUAL packet closed; the open work orders block the file, and say why."""
    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
    assert "set_vclosed" not in view["can"] and "file" not in view["can"]
    assert view["blocked"]["file"] == [open_msg("L670002/001")]
    refused(shop("clerk.clara").post(f"/api/packets/{pid}/vclosed", json={"closed": True}), 403, "wrong_status")
    refused(file_packet(shop, view), 403, "precondition", open_msg("L670002/001"))


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)
    view = fresh(shop("clerk.clara"), pid)
    vc = view["vclose_check"]
    assert vc["open_wos"] == [] and vc["closed_at"] is not None and vc["agrees"] is True
    assert view["vclosed"]["at"] is not None and view["vclosed"]["by"] is None  # the feed, not a person
    assert "file" in view["can"] and "file" not in view["blocked"]
    out = ok(file_packet(shop, view))
    assert out["status"] == "filed" and 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)
    view = fresh(shop("clerk.clara"), pkt["id"])
    vc = view["vclose_check"]
    assert vc is not None and vc["observed_at"] is None and vc["open_wos"] == []
    assert view["blocked"]["file"] == [MSG_FEED_WAIT]


@pytest.mark.parametrize("route", ["databook", "shipping"])
def test_file_blocked_while_visual_shows_work_orders_open(shop: Shop, route: str) -> None:
    """C5: a block on both routes, not a warning. The old confirm_open_wos is ignored. A close set before C5
    (vclosed by hand on a VISUAL packet) does not pass while the feed shows work orders open."""
    pkt = at_complete(shop, route=route)
    pid = pkt["id"]
    set_feed_legs(shop.settings, pid, [leg("L670001", st="R"), leg("L670002", lot="7A", st="R")])
    with_hand_close = route == "databook"
    if with_hand_close:
        q1(shop.settings, "UPDATE packets SET vclosed_at = now() WHERE id = %s RETURNING id", (pid,))
    pkt = fresh(shop("clerk.clara"), pid)
    sigs = q1(shop.settings, "SELECT count(*) FROM signatures")[0]
    msg = open_msg("L670001/001, L670002/7A")
    refused(file_packet(shop, pkt), 403, "precondition", msg)
    refused(file_packet(shop, pkt, confirm_open_wos=True), 403, "precondition", msg)
    # 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"
    # the feed sees VISUAL closed: the packet files, on either route
    set_feed_legs(shop.settings, pid, [leg("L670001", st="C"), leg("L670002", lot="7A", st="X")], closed=True)
    out = ok(file_packet(shop, fresh(shop("clerk.clara"), pid), confirm_open_wos=True, note="Customer pickup today"))
    assert out["status"] == "filed" and out["seal"]["open_wos"] is None
    assert find_log(shop("clerk.clara"), pid, "Filed in the data book")["note"] == "Customer pickup today"


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)]