PacketStation/tests/server/test_c7_short_approve.py

"""Change C7 (Gitea issue 18): every received short needs the department supervisor's approval, one receipt at a time.

SHORT_APPROVE through POST /api/packets/{id}/shorts/receipts/{receipt_id}/approve: who (the supervisor of the
packet's department, Admin with an IT reason), the signature, one approval per receipt (409), the read model (who
received and who approved each receipt, shorts_unapproved, can), and the shorts_approved precondition on QC_ACCEPT
and FILE (not on QC_ACCEPT_AND_ROUTE). 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, q1, vclose
from fakedir import FakeDirectory
from pshelpers import app_conn, needs_pg
from qchelpers import accept, accept_route, finish_inspection, to_qc
from shophelpers import (
    IT,
    SIGN,
    Shop,
    draft,
    find_log,
    finish_stage,
    fresh,
    im,
    line_of,
    ok,
    open_shop,
    refused,
    release,
    start,
    submit,
    to_station,
    to_warehouse,
)

from packetstation.settings import Settings

pytestmark = needs_pg

MSG = "Every short must be received and approved by the supervisor first."


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


def approve(shop: Shop, pkt: dict[str, Any], receipt_id: int, by: str = "sup.wes", **body: Any) -> Any:
    body.setdefault("sign", SIGN)
    return shop(by).post(f"/api/packets/{pkt['id']}/shorts/receipts/{receipt_id}/approve", json=body)


def receive(shop: Shop, pkt: dict[str, Any], item: int, qty: int, heat: str, by: str = "wh.wade") -> dict[str, Any]:
    return ok(shop(by).post(f"/api/packets/{pkt['id']}/bom/{item}/short/receipts", json={"qty": qty, "heat": heat}))


def short_at_qc(shop: Shop) -> dict[str, Any]:
    """Items 1 and 2 pulled; item 3 short 2 of 4, one piece received (HT-31), one still open; through the shop
    and at Q.C. with the inspection finished."""
    pkt = to_warehouse(shop, draft(shop))
    wade = shop("wh.wade")
    pid = pkt["id"]
    for item in (1, 2):
        cur = line_of(pkt, item)
        pkt = ok(wade.patch(f"/api/packets/{pid}/bom/{item}", json={"pulled": True}, headers=im(cur["row_version"])))
        cur = line_of(pkt, item)
        pkt = ok(wade.patch(f"/api/packets/{pid}/bom/{item}", json={"heat": "HT-1"}, headers=im(cur["row_version"])))
    pkt = ok(wade.post(f"/api/packets/{pid}/bom/3/short", json={"qty": 2, "reason": "On order", "expect_open": False}))
    pkt = receive(shop, pkt, 3, 1, "HT-31")
    pkt = start(shop, to_station(shop, release(shop, pkt)))
    pkt = submit(shop, finish_stage(shop, pkt, welds=1))
    return finish_inspection(shop, to_qc(shop, pkt))


def test_approve_receipts_then_accept(shop: Shop) -> None:
    pkt = short_at_qc(shop)
    pid = pkt["id"]
    assert pkt["shorts_unapproved"] == 2  # one open short and one receipt without approval
    assert pkt["blocked"]["qc_accept"] == [MSG]
    assert "qc_accept_and_route" in pkt["can"] and "qc_accept_and_route" not in pkt["blocked"]
    refused(accept(shop, pkt), 403, "precondition", MSG)

    [short] = pkt["shorts"]
    assert (short["item"], float(short["qty"]), float(short["qty_open"]), short["closed_at"]) == (3, 2, 1, None)
    [r1] = short["receipts"]
    assert float(r1["qty"]) == 1 and r1["heat"] == "HT-31" and r1["by"]["display_name"] and r1["approval"] is None
    assert line_of(pkt, 3)["short"]["receipts"][0]["approval"] is None

    # who: the supervisor of the packet's department; a lead, Q.C., the warehouse and Clerical are refused
    wes = fresh(shop("sup.wes"), pid)
    assert "short_approve" in wes["can"]
    for sam in ("lead.lou", "weld.andy", "qc.quinn", "wh.wade", "clerk.leo"):
        assert "short_approve" not in fresh(shop(sam), pid)["can"], sam
        refused(approve(shop, pkt, r1["id"], by=sam), 403, "forbidden", "Your access doesn't allow this step.")
    refused(approve(shop, pkt, r1["id"], by="admin.ann"), 400, "it_reason_required")
    # a signature is required; an unknown receipt is 404
    refused(approve(shop, pkt, r1["id"], sign=None), 400, "signature_required")
    refused(approve(shop, pkt, r1["id"] + 100000), 404, "not_found")

    out = ok(approve(shop, pkt, r1["id"], note="Heat matches the issue ticket"))
    a = out["shorts"][0]["receipts"][0]["approval"]
    assert a["by"]["id"] == str(shop.uid("sup.wes")) and a["by"]["display_name"], a
    assert a["note"] == "Heat matches the issue ticket" and a["at"] and a["dept_code"] == "03"
    assert line_of(out, 3)["short"]["receipts"][0]["approval"]["note"] == "Heat matches the issue ticket"
    assert out["shorts_unapproved"] == 1  # the short is still open
    log = find_log(shop("sup.wes"), pid, "Short approved · item 3")
    assert log["note"] == "1 pc · heat HT-31 · Heat matches the issue ticket" and log["signed"] is True
    row = q1(shop.settings, "SELECT dept_code, signature_id FROM short_approvals WHERE receipt_id = %s", (r1["id"],))
    assert row[0] == "03" and row[1] is not None
    # one approval per receipt
    refused(approve(shop, pkt, r1["id"]), 409, "already_approved", "This receipt is already approved.")

    # the warehouse receives the last piece: the short closes, the new receipt waits for approval
    out = receive(shop, out, 3, 1, "HT-32")
    assert out["shorts"][0]["closed_at"] is not None and line_of(out, 3)["short"] is None
    assert out["shorts_unapproved"] == 1 and fresh(shop("qc.quinn"), pid)["blocked"]["qc_accept"] == [MSG]
    r2 = out["shorts"][0]["receipts"][1]
    assert r2["heat"] == "HT-32" and r2["approval"] is None
    # Admin approves through the bypass with an IT reason
    out = ok(approve(shop, out, r2["id"], by="admin.ann", it_reason=IT))
    assert out["shorts_unapproved"] == 0 and "qc_accept" not in fresh(shop("qc.quinn"), pid)["blocked"]
    done = ok(accept(shop, fresh(shop("qc.quinn"), pid)))
    assert done["status"] == "complete"
    # an approval needs no open step: still open to the supervisor at complete, but nothing is left to approve
    refused(approve(shop, done, r2["id"]), 409, "already_approved")


def test_accept_and_route_does_not_wait_for_the_shorts(shop: Shop) -> None:
    pkt = short_at_qc(shop)
    out = ok(accept_route(shop, pkt, "09"))
    assert out["status"] == "ready" and out["dept"]["code"] == "09" and out["shorts_unapproved"] == 2
    # the packet's department is now 09: the 03 supervisor no longer approves, a 09 supervisor would
    r1 = out["shorts"][0]["receipts"][0]
    refused(approve(shop, out, r1["id"]), 403, "forbidden")


def test_file_waits_for_the_approval_and_the_filed_lock(shop: Shop) -> None:
    """A packet completed before C7 with a receipt nobody approved: FILE shows and refuses it until approved."""
    pkt = at_complete(shop)
    pid = pkt["id"]
    with app_conn(shop.settings) as conn:
        line = conn.execute("SELECT id FROM bom_lines WHERE packet_id = %s AND item = 3", (pid,)).fetchone()[0]
        sid = conn.execute(
            "INSERT INTO bom_shorts (packet_id, bom_line_id, qty, qty_open, reason, closed_at)"
            " VALUES (%s, %s, 1, 0, 'On order', now()) RETURNING id",
            (pid, line),
        ).fetchone()[0]
        rid = conn.execute(
            "INSERT INTO short_receipts (short_id, packet_id, qty, heat) VALUES (%s, %s, 1, 'HT-77') RETURNING id",
            (sid, pid),
        ).fetchone()[0]
    pkt = vclose(shop, fresh(shop("clerk.clara"), pid))
    assert pkt["shorts_unapproved"] == 1 and pkt["blocked"]["file"] == [MSG]
    refused(file_packet(shop, pkt), 403, "precondition", MSG)
    [card] = ok(shop("clerk.clara").get(f"/api/packets?unit_id={pkt['unit']['id']}"))["packets"]
    assert card["shorts_unapproved"] == 1 and card["blocked"]["file"] == [MSG]
    ok(approve(shop, pkt, rid))
    out = ok(file_packet(shop, fresh(shop("clerk.clara"), pid)))
    assert out["status"] == "filed"
    # filed: nobody approves any more
    refused(approve(shop, out, rid, by="admin.ann", it_reason=IT), 409, "already_approved")
    with app_conn(shop.settings) as conn:
        rid2 = conn.execute(
            "INSERT INTO short_receipts (short_id, packet_id, qty, heat) VALUES (%s, %s, 1, 'HT-78') RETURNING id",
            (sid, pid),
        ).fetchone()[0]
    refused(approve(shop, out, rid2), 403, "filed")