PacketStation/tests/server/shophelpers.py

"""Helpers for the phase 5 shop tests: one signed-in client per user, a packet brought to `ready` in 03 through
the warehouse, and the shop steps (assign, start, finish the stage, submit, review).

The packet: a hand draft in 03 on a fresh unit, two work orders (target 2 h and 2.5 h), three BOM lines, a
two-page PDF drawing, heat map answered, sent to the warehouse, every line pulled with a heat by wh.wade, released.
"""

from __future__ import annotations

import secrets
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Any

from fakedir import FAKE_PASSWORD, FakeDirectory
from fastapi.testclient import TestClient
from pkthelpers import phase2_client, seed_users, step, tiny_pdf, upload
from pshelpers import ORIGIN_HEADERS, app_conn, sign_in
from psycopg.types.json import Jsonb

from packetstation.settings import Settings

SIGN = {"factor": "password", "secret": FAKE_PASSWORD}
IT = "Ticket 5501: shop lead is out today"
STALE = "Someone else already took this step."

USERS = (
    "admin.ann",
    "clerk.clara",
    "clerk.leo",
    "wh.wade",
    "qc.quinn",
    "pm.paula",
    "weld.andy",
    "lead.lou",
    "sup.wes",
    "multi.mia",
    "both.bo",
    "dup.dave",
)

WOS = [
    {"wo_base": "L670001", "lot": "001", "split": "0", "sub": "0", "qty": 2, "target_hrs": 2},
    {"wo_base": "L670002", "lot": "001", "split": "0", "sub": "0", "qty": 1, "target_hrs": 2.5},
]
BOM3 = [
    {"item": 1, "qty": 1, "part_id": "PIPE-4", "description": "4in pipe SCH40", "material": "A106-B"},
    {"item": 2, "qty": 2, "part_id": "GSKT-4", "description": "4in gasket"},
    {"item": 3, "qty": 4, "part_id": "STUD-58", "description": "5/8 stud bolt"},
]
WELD_KEYS = ("dwg", "heat", "wps", "fit", "pre", "vt", "stamp", "clean", "qty")
PLAIN_KEYS = ("dwg", "prev", "mat", "work", "insp", "qty")

MSG_CHECKLIST = "Finish the checklist first."
MSG_WELDS = "Record at least one weld first."
MSG_FILLER = "Enter the filler metal first."
MSG_QC = "Confirm the book and the spool, and give the location."
MSG_OTHER_DEPT = "Choose a different department."


# ---------------------------------------------------------------- clients


@dataclass
class Shop:
    """One app, one signed-in TestClient per user (separate cookie jars)."""

    main: TestClient
    settings: Settings
    clients: dict[str, TestClient] = field(default_factory=dict)

    def __call__(self, sam: str) -> TestClient:
        c = self.clients.get(sam)
        if c is None:
            c = self.fresh(sam)
            self.clients[sam] = c
        return c

    def fresh(self, sam: str) -> TestClient:
        """A new client with its own session (two sessions of one user, or a thread of its own)."""
        c = TestClient(self.main.app, headers=ORIGIN_HEADERS)
        r = sign_in(c, sam)
        assert r.status_code == 200, r.text
        return c

    def uid(self, sam: str) -> str:
        return user_id(self.settings, sam)

    def stall(self, dept: str = "03", name: str = "Stall 1") -> int:
        return station_id(self.settings, dept, name)


def open_shop(settings: Settings, fake_dir: FakeDirectory) -> Iterator[Shop]:
    """The body of a `shop` fixture: the app, every test user signed in once (their users rows exist)."""
    with phase2_client(settings, fake_dir) as client:
        seed_users(client, USERS)
        yield Shop(client, settings)


# ---------------------------------------------------------------- SQL


def one(settings: Settings, sql: str, params: Any = ()) -> Any:
    with app_conn(settings) as conn:
        return conn.execute(sql, params).fetchone()


def all_rows(settings: Settings, sql: str, params: Any = ()) -> list[Any]:
    with app_conn(settings) as conn:
        return list(conn.execute(sql, params).fetchall())


def run(settings: Settings, sql: str, params: Any = ()) -> None:
    with app_conn(settings) as conn:
        conn.execute(sql, params)


def user_id(settings: Settings, sam: str) -> str:
    row = one(settings, "SELECT id FROM users WHERE sam_account = %s", (sam,))
    assert row is not None, sam
    return str(row[0])


def station_id(settings: Settings, dept: str = "03", name: str = "Stall 1") -> int:
    row = one(settings, "SELECT id FROM stations WHERE dept_code = %s AND name = %s", (dept, name))
    assert row is not None, (dept, name)
    return int(row[0])


def set_stamp(settings: Settings, sam: str, stamp: str | None) -> None:
    run(settings, "UPDATE users SET welder_stamp = %s WHERE sam_account = %s", (stamp, sam))


def set_limits(settings: Settings, sam: str, station_ids: list[int]) -> None:
    """Replace a user's station limits (an empty list removes them)."""
    with app_conn(settings) as conn:
        uid = conn.execute("SELECT id FROM users WHERE sam_account = %s", (sam,)).fetchone()[0]
        conn.execute("DELETE FROM user_station_limits WHERE user_id = %s", (uid,))
        for sid in station_ids:
            conn.execute("INSERT INTO user_station_limits (user_id, station_id) VALUES (%s, %s)", (uid, sid))


def make_unit(settings: Settings, number: str | None = None, lot: str = "001") -> tuple[int, int]:
    """A project with one unit by SQL. Returns (project_id, unit_id)."""
    number = number or "L5" + secrets.token_hex(3).upper()
    with app_conn(settings) as conn:
        pid = conn.execute(
            "INSERT INTO projects (number, name, customer, po) VALUES (%s, 'Shop project', 'Acme Refining', 'PO-55')"
            " RETURNING id",
            (number,),
        ).fetchone()[0]
        uid = conn.execute(
            "INSERT INTO units (project_id, lot, name, status) VALUES (%s, %s, 'Unit one', 'open') RETURNING id",
            (pid, lot),
        ).fetchone()[0]
    return int(pid), int(uid)


def set_running_since(settings: Settings, packet_id: int, seconds_ago: float) -> None:
    """Move the open stage's running clock back in time, as if it had run that long."""
    run(
        settings,
        "UPDATE packet_stages SET clock_running_since = now() - make_interval(secs => %s)"
        " WHERE packet_id = %s AND closed_at IS NULL AND clock_running_since IS NOT NULL",
        (seconds_ago, packet_id),
    )


def set_templates(settings: Settings, value: Any) -> None:
    run(
        settings,
        "INSERT INTO settings (key, value, version, updated_at) VALUES ('checklist_templates', %s, 1, now())"
        " ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1",
        (Jsonb(value),),
    )


# ---------------------------------------------------------------- reads


def fresh(c: TestClient, pid: int) -> dict[str, Any]:
    r = c.get(f"/api/packets/{pid}")
    assert r.status_code == 200, r.text
    out: dict[str, Any] = r.json()
    return out


def log_rows(c: TestClient, pid: int) -> list[dict[str, Any]]:
    r = c.get(f"/api/packets/{pid}/log")
    assert r.status_code == 200, r.text
    out: list[dict[str, Any]] = r.json()
    return out


def last_log(c: TestClient, pid: int) -> dict[str, Any]:
    return log_rows(c, pid)[0]


def find_log(c: TestClient, pid: int, action: str) -> dict[str, Any]:
    """The newest log row with this action text."""
    rows = [r for r in log_rows(c, pid) if r["action"] == action]
    assert rows, (action, [r["action"] for r in log_rows(c, pid)])
    return rows[0]


def ok(r: Any) -> dict[str, Any]:
    assert r.status_code == 200, r.text
    out: dict[str, Any] = r.json()
    return out


def refused(r: Any, status: int, code: str, detail: str | None = None) -> dict[str, Any]:
    assert r.status_code == status, r.text
    body: dict[str, Any] = r.json()
    assert body["code"] == code, body
    if detail is not None:
        assert body["detail"] == detail, body
    return body


def weld_of(pkt: dict[str, Any], no: int) -> dict[str, Any]:
    return next(w for w in pkt["welds"] if w["weld_no"] == no)


def tag_of(pkt: dict[str, Any], no: int) -> dict[str, Any]:
    return next(t for t in pkt["heat_tags"] if t["tag_no"] == no)


def line_of(pkt: dict[str, Any], item: int) -> dict[str, Any]:
    return next(b for b in pkt["bom"] if b["item"] == item)


def im(v: int) -> dict[str, str]:
    return {"If-Match": str(v)}


# ---------------------------------------------------------------- the packet


def draft(
    shop: Shop,
    unit_id: int | None = None,
    *,
    title: str = "Spool 7 outlet",
    drawing: str = "D-5001",
    rev: str = "B",
    bom: list[dict[str, Any]] | None = None,
    wos: list[dict[str, Any]] | None = None,
    dept: str = "03",
    pages: int = 2,
) -> dict[str, Any]:
    """A hand draft by clerk.clara with the heat map answered and a PDF drawing of `pages` pages."""
    if unit_id is None:
        _, unit_id = make_unit(shop.settings)
    c = shop("clerk.clara")
    body = {
        "unit_id": unit_id,
        "dept_code": dept,
        "op": "WELD-PART",
        "title": title,
        "customer": "Acme Refining",
        "drawing": drawing,
        "rev": rev,
        "wps": "WPS-11",
        "work_orders": WOS if wos is None else wos,
        "bom": BOM3 if bom is None else bom,
    }
    r = c.post("/api/packets", json=body)
    assert r.status_code == 201, r.text
    pid = r.json()["id"]
    assert c.patch(f"/api/packets/{pid}", json={"set": {"heat_map": True}, "base": {}}).status_code == 200
    a = upload(c, tiny_pdf(pages))
    assert a.status_code == 201, a.text
    r = c.post(f"/api/packets/{pid}/pages", json={"asset_id": a.json()["id"], "type": "drawing", "sign": SIGN})
    return ok(r)


def to_warehouse(shop: Shop, pkt: dict[str, Any]) -> dict[str, Any]:
    out = ok(step(shop("clerk.clara"), pkt, "send_to_warehouse"))
    assert out["status"] == "warehouse"
    return out


def pull_all(shop: Shop, pkt: dict[str, Any]) -> dict[str, Any]:
    """wh.wade ticks every line pulled and gives it a heat."""
    c = shop("wh.wade")
    for b in list(pkt["bom"]):
        item = b["item"]
        cur = line_of(pkt, item)
        pkt = ok(c.patch(f"/api/packets/{pkt['id']}/bom/{item}", json={"pulled": True}, headers=im(cur["row_version"])))
        cur = line_of(pkt, item)
        pkt = ok(
            c.patch(
                f"/api/packets/{pkt['id']}/bom/{item}", json={"heat": f"HT-{item}0"}, headers=im(cur["row_version"])
            )
        )
    return pkt


def release(shop: Shop, pkt: dict[str, Any]) -> dict[str, Any]:
    out = ok(step(shop("wh.wade"), pkt, "wh_release"))
    assert out["status"] == "ready" and out["dept"]["code"] == "03", out["status"]
    return out


def ready_packet(shop: Shop, unit_id: int | None = None, **kw: Any) -> dict[str, Any]:
    """Draft, warehouse, every line pulled with a heat, released: `ready` in 03 with an open stage."""
    pkt = to_warehouse(shop, draft(shop, unit_id, **kw))
    return release(shop, pull_all(shop, pkt))


def ready_skipped(shop: Shop, unit_id: int | None = None, dept: str = "03", **kw: Any) -> dict[str, Any]:
    """A draft sent straight to a department (wh_skipped true)."""
    pkt = draft(shop, unit_id, **kw)
    out = ok(step(shop("clerk.clara"), pkt, "release_to_shop", dept=dept))
    assert out["status"] == "ready" and out["wh_skipped"] is True
    return out


def to_station(
    shop: Shop,
    pkt: dict[str, Any],
    *,
    by: str = "sup.wes",
    welder: str = "weld.andy",
    dept: str = "03",
    station: str = "Stall 1",
    **kw: Any,
) -> dict[str, Any]:
    """ASSIGN the packet to a station and a welder."""
    out = ok(step(shop(by), pkt, "assign", station_id=shop.stall(dept, station), welder_id=shop.uid(welder), **kw))
    assert out["status"] == "assigned"
    return out


def start(shop: Shop, pkt: dict[str, Any], by: str = "weld.andy") -> dict[str, Any]:
    out = ok(step(shop(by), pkt, "start"))
    assert out["status"] == "in_progress"
    return out


def add_weld(shop: Shop, pkt: dict[str, Any], by: str = "weld.andy", **over: Any) -> dict[str, Any]:
    body: dict[str, Any] = {"page": 1, "fx": 0.5, "fy": 0.5, "items": [1, 2], "type": "Fillet", "process": "GMAW"}
    body.update(over)
    return ok(shop(by).post(f"/api/packets/{pkt['id']}/welds", json=body))


def tick_all(
    shop: Shop, pkt: dict[str, Any], by: str = "weld.andy", keys: tuple[str, ...] = WELD_KEYS
) -> dict[str, Any]:
    for k in keys:
        pkt = ok(shop(by).put(f"/api/packets/{pkt['id']}/stage/checklist/{k}", json={"done": True}))
    return pkt


def set_filler(shop: Shop, pkt: dict[str, Any], by: str = "weld.andy", filler: str = "ER70S-6") -> dict[str, Any]:
    return ok(shop(by).patch(f"/api/packets/{pkt['id']}/stage", json={"set": {"filler": filler}, "base": {}}))


def finish_stage(shop: Shop, pkt: dict[str, Any], by: str = "weld.andy", welds: int = 1) -> dict[str, Any]:
    """Every checklist key ticked, `welds` welds marked, the filler set: SUBMIT can pass."""
    keys = WELD_KEYS if pkt["dept"]["weld"] else PLAIN_KEYS
    pkt = tick_all(shop, pkt, by, keys)
    for _ in range(welds):
        pkt = add_weld(shop, pkt, by)
    return set_filler(shop, pkt, by)


def submit(shop: Shop, pkt: dict[str, Any], by: str = "weld.andy") -> dict[str, Any]:
    out = ok(step(shop(by), pkt, "submit", sign=SIGN))
    assert out["status"] == "review"
    return out


def in_progress(shop: Shop, unit_id: int | None = None, **kw: Any) -> dict[str, Any]:
    return start(shop, to_station(shop, ready_packet(shop, unit_id, **kw)))


def in_review(shop: Shop, unit_id: int | None = None, **kw: Any) -> dict[str, Any]:
    pkt = in_progress(shop, unit_id, **kw)
    return submit(shop, finish_stage(shop, pkt))


def issue_co(shop: Shop, pkt: dict[str, Any], required: list[str], **over: Any) -> dict[str, Any]:
    """A manual change order by clerk.clara (approved at once, holds until every required department verifies)."""
    body = {"reason": "Nozzle moved 2 in on rev C", "required_depts": required, "sign": SIGN, **over}
    return ok(shop("clerk.clara").post(f"/api/packets/{pkt['id']}/change-orders", json=body))


def verify(shop: Shop, pkt: dict[str, Any], no: int, dept: str, by: str, **over: Any) -> Any:
    body = {"dept": dept, "note": "Reviewed with the crew", "sign": SIGN, **over}
    return shop(by).post(f"/api/packets/{pkt['id']}/change-orders/{no}/verify", json=body)