PacketStation/tests/server/test_drawing_decide.py

"""The drawing decision table (pure), then the drawing job against the fixture lookup (PostgreSQL)."""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest
from feedhelpers import (
    cos_of,
    feed_settings_over,
    gen,
    move_status,
    packet_by_key,
    packet_id,
    run_drawings,
    run_feed,
    system_audit,
)
from pshelpers import app_conn, needs_pg

from packetstation.drawings.decide import Decision, Target, decide
from packetstation.drawings.fake import FixtureLookup
from packetstation.drawings.lookup import Found
from packetstation.settings import Settings

DRAWING_FLAGS = {"drawing_needed", "rev_mismatch", "not_released", "not_found", "lookup_failed", "bad_rev"}


def target(**over: Any) -> Target:
    values: dict[str, Any] = {
        "dwg": "L900-0101",
        "visual_rev": "1",
        "attached_rev": None,
        "attached_sha": None,
        "status": "draft",
        "drawing_state": "needed",
        "has_hand_drawing": False,
        "rev_ok": None,
    }
    values.update(over)
    return Target(**values)


def found(**over: Any) -> Found:
    values: dict[str, Any] = {
        "dwg": "L900-0101",
        "ok": True,
        "rev": 1,
        "name": "L900-0101 r1",
        "obj_id": 1,
        "obj_version": 1,
        "file_id": 101,
        "file_version": 1,
        "file_size": 900,
        "base": None,
        "draft_rev": None,
        "draft_where": "",
        "in_drafting": False,
        "note": "",
    }
    values.update(over)
    return Found(**values)


def d(t: Target, f: Found, same_file: bool = False) -> Decision:
    return decide(t, f, same_file=same_file)


# ---------------------------------------------------------------- the table


def test_lookup_failed_on_a_needed_drawing() -> None:
    r = d(target(), found(ok=False, rev=None, note="M-Files lookup failed: timeout"))
    assert (r.kind, r.flag, r.drawing_state) == ("lookup_failed", "lookup_failed", "lookup_failed")
    assert "timeout" in r.note


def test_lookup_failed_keeps_what_is_attached() -> None:
    r = d(target(drawing_state="attached", attached_rev="1"), found(ok=False, rev=None, note="503"))
    assert (r.kind, r.flag, r.drawing_state) == ("lookup_failed", "lookup_failed", "attached")


def test_not_released() -> None:
    r = d(target(), found(rev=None, draft_rev=1, draft_where="in Drafting review", in_drafting=True, base="L900-05"))
    assert (r.kind, r.flag, r.drawing_state) == ("not_released", "not_released", "not_released")
    assert r.note == "Not released in M-Files yet (sheet of L900-05) - drafting status: rev 1 in Drafting review"


def test_not_released_in_drafting_without_a_rev() -> None:
    r = d(target(), found(rev=None, in_drafting=True))
    assert r.kind == "not_released" and r.note.endswith("in drafting, no released PDF yet")


def test_not_found() -> None:
    r = d(target(), found(rev=None, name=None))
    assert (r.kind, r.flag, r.drawing_state) == ("not_found", "not_found", "not_found")
    assert r.note == "Drawing not in M-Files - check the drawing number in VISUAL"


def test_bad_rev() -> None:
    r = d(target(visual_rev="A"), found(rev=1))
    assert (r.kind, r.flag) == ("bad_rev", "bad_rev")
    assert "A" in r.note and "M-Files has rev 1" in r.note and "not attached" in r.note


def test_rev_mismatch() -> None:
    r = d(target(visual_rev="2"), found(rev=3))
    assert (r.kind, r.flag, r.drawing_state) == ("rev_mismatch", "rev_mismatch", "rev_mismatch")
    assert r.note == "Rev mismatch: VISUAL calls rev 2, M-Files has rev 3"


def test_rev_mismatch_with_a_newer_draft() -> None:
    r = d(target(visual_rev="2"), found(rev=3, draft_rev=4, draft_where="in Drafting"))
    assert r.note == "Rev mismatch: VISUAL calls rev 2, M-Files has rev 3; drafting status: rev 4 in Drafting"


def test_rev_mismatch_on_an_in_shop_packet_opens_nothing() -> None:
    r = d(target(visual_rev="2", status="ready", drawing_state="attached", attached_rev="2"), found(rev=3))
    assert r.kind == "rev_mismatch" and r.drawing_state == "attached"


def test_approved_rev_attaches() -> None:
    r = d(target(visual_rev="2", rev_ok="3"), found(rev=3))
    assert (r.kind, r.flag, r.drawing_state) == ("attach", None, "attached")


def test_approved_other_rev_still_mismatches() -> None:
    assert d(target(visual_rev="2", rev_ok="4"), found(rev=3)).kind == "rev_mismatch"


def test_match_on_a_draft_attaches() -> None:
    r = d(target(), found())
    assert (r.kind, r.flag, r.drawing_state) == ("attach", None, "attached")


def test_attach_names_the_sheet() -> None:
    r = d(target(dwg="L900-0101H2"), found(dwg="L900-0101H2", base="L900-0101"))
    assert r.kind == "attach" and r.note == "sheet of drawing L900-0101"


def test_hand_drawing_at_the_same_rev_stays() -> None:
    r = d(target(has_hand_drawing=True, drawing_state="hand"), found())
    assert r.kind == "unchanged" and r.drawing_state == "hand"


def test_past_draft_same_file_unchanged() -> None:
    r = d(target(status="ready", drawing_state="attached", attached_rev="1", attached_sha=b"\x01"), found(), True)
    assert r.kind == "unchanged" and r.flag is None


def test_past_draft_new_file_same_rev_opens_a_co() -> None:
    r = d(target(status="in_progress", drawing_state="attached", attached_rev="1", attached_sha=b"\x01"), found())
    assert r.kind == "open_co"
    assert r.note == "M-Files re-released rev 1 of L900-0101 (new file)"


def test_past_draft_higher_rev_opens_a_co() -> None:
    t = target(status="ready", visual_rev="2", drawing_state="attached", attached_rev="1", attached_sha=b"\x01")
    r = d(t, found(rev=2))
    assert r.kind == "open_co"
    assert r.note == "M-Files released rev 2 of L900-0101; this packet has rev 1"


def test_past_draft_waiting_co_gets_staged() -> None:
    t = target(
        status="ready",
        visual_rev="2",
        drawing_state="attached",
        attached_rev="1",
        attached_sha=b"\x01",
        co_waiting=("L900-0101", "2"),
        co_pending=True,
    )
    assert d(t, found(rev=2)).kind == "stage_for_co"


def test_complete_packet_still_gets_a_co() -> None:
    t = target(status="complete", drawing_state="attached", attached_rev="1", attached_sha=b"\x01")
    assert d(t, found()).kind == "open_co"


# ---------------------------------------------------------------- the fixture lookup


def test_fixture_lookup_states(tmp_path: Path) -> None:
    gen.generate(tmp_path)
    one = FixtureLookup(tmp_path)
    assert one.status("L900-0101").rev == 1
    assert one.status("L900-0201").rev == 3
    nr = one.status("L900-0501")
    assert nr.rev is None and nr.in_drafting and nr.draft_rev == 1
    assert one.status("L900-0701").rev is None and one.status("L900-0701").ok
    dest = tmp_path / "out" / "x.pdf"
    assert one.download(one.status("L900-0101"), dest) > 0 and dest.read_bytes().startswith(b"%PDF-")
    (tmp_path / "state").write_text("2")
    two = FixtureLookup(tmp_path)
    f2 = two.status("L900-0101")
    assert f2.rev == 2 and f2.identity != one.status("L900-0101").identity


# ---------------------------------------------------------------- the job (PostgreSQL)


@pytest.fixture
def fx(pg_settings: Settings, tmp_path: Path) -> Settings:
    return feed_settings_over(pg_settings, tmp_path)


def _drawing_pages(conn: Any, pid: int) -> list[tuple[Any, ...]]:
    return [
        tuple(r)
        for r in conn.execute(
            "SELECT n, pn, dwg, rev, asset_id IS NOT NULL, void_at IS NOT NULL FROM packet_pages"
            " WHERE packet_id = %s AND type = 'drawing' ORDER BY n",
            (pid,),
        ).fetchall()
    ]


@needs_pg
def test_job_attaches_and_flags(fx: Settings) -> None:
    run_feed(fx, 1)
    counts = run_drawings(fx)
    assert counts["_status"] in ("ok", "degraded"), counts
    assert counts["checked"] >= 8 and counts["attached"] >= 1
    with app_conn(fx) as conn:
        wp = packet_by_key(conn, gen.KEY_WELD_PART)
        assert wp is not None
        assert wp["drawing_state"] == "attached"
        assert not set(wp["flags"]) & DRAWING_FLAGS
        assert wp["visual"].get("rev_confirmed") is True
        pages = _drawing_pages(conn, wp["id"])
        assert pages and all(p[2] == "L900-0101" and p[3] == "1" and p[4] for p in pages)
        lk = conn.execute(
            "SELECT ok, rev, asset_id IS NOT NULL, sha256 IS NOT NULL, checked_at IS NOT NULL FROM drawing_lookups"
            " WHERE dwg = 'L900-0101'"
        ).fetchone()
        assert lk == (True, 1, True, True, True)
        # the LAYOUT packet shares the drawing: attached from the same asset
        lay = packet_by_key(conn, gen.KEY_LAYOUT_A)
        assert lay is not None and lay["drawing_state"] == "attached"

        mm = packet_by_key(conn, gen.KEY_TWO_OPS)  # VISUAL rev 2, M-Files rev 3
        assert mm is not None and "rev_mismatch" in mm["flags"] and mm["drawing_state"] == "rev_mismatch"
        assert mm["visual"]["appr"]["rev"] == 3
        staged = conn.execute(
            "SELECT asset_id IS NOT NULL, rev FROM drawing_lookups WHERE dwg = 'L900-0201'"
        ).fetchone()
        assert staged == (True, 3)  # staged for Clerical Supervisor approval
        assert _drawing_pages(conn, mm["id"]) == []

        nr = packet_by_key(conn, gen.KEY_DRAFTING)
        assert nr is not None and "not_released" in nr["flags"] and nr["drawing_state"] == "not_released"
        nf = packet_by_key(conn, gen.KEY_NOT_FOUND)
        assert nf is not None and "not_found" in nf["flags"] and nf["drawing_state"] == "not_found"
        br = packet_by_key(conn, gen.KEY_REV_A)
        assert br is not None and "bad_rev" in br["flags"]
        nd = packet_by_key(conn, gen.KEY_NO_DWG)
        assert nd is not None and nd["drawing_state"] == "none"
        log = conn.execute(
            "SELECT by, by_name, role FROM packet_log WHERE packet_id = %s"
            " AND action = 'Drawing attached from M-Files'",
            (wp["id"],),
        ).fetchall()
        assert len(log) == 1 and log[0][0] is None and log[0][2] == "system"
        assert all(r[0] == "system" and r[1] is None for r in system_audit(conn, "packet.vis_update_draft"))
        assert all(r[0] == "system" for r in system_audit(conn, "drawing.fetched"))
        assert system_audit(conn, "drawing.fetched")


@needs_pg
def test_job_second_run_changes_nothing(fx: Settings) -> None:
    run_feed(fx, 1)
    run_drawings(fx)
    with app_conn(fx) as conn:
        conn.execute(
            "UPDATE drawing_lookups SET next_check_at = now() - interval '1 minute',"
            " checked_at = now() - interval '1 day'"
        )
        before = conn.execute("SELECT count(*) FROM packet_pages").fetchone()[0]
    counts = run_drawings(fx)
    assert counts.get("attached", 0) == 0 and counts.get("fetched", 0) == 0, counts
    with app_conn(fx) as conn:
        assert conn.execute("SELECT count(*) FROM packet_pages").fetchone()[0] == before


@needs_pg
def test_stage_for_co_after_state_2(fx: Settings) -> None:
    run_feed(fx, 1)
    run_drawings(fx)
    with app_conn(fx) as conn:
        pid = packet_id(conn, gen.KEY_WELD_PART)
        move_status(conn, pid, "ready")
    counts = run_feed(fx, 2)
    assert counts["cos_opened"] >= 1, counts
    with app_conn(fx) as conn:
        co = [c for c in cos_of(conn, pid) if c["review"] == "pending"]
        assert len(co) == 1 and co[0]["origin"] == "visual"
        assert co[0]["waiting"] is not None and co[0]["waiting"]["rev"] == "2"
        assert co[0]["staged"] is None
    counts = run_drawings(fx)
    assert counts.get("staged", 0) >= 1, counts
    with app_conn(fx) as conn:
        co = [c for c in cos_of(conn, pid) if c["review"] == "pending"]
        assert len(co) == 1
        assert co[0]["waiting"] is None
        staged = co[0]["staged"]
        assert staged["rev"] == "2" and staged["dwg"] == "L900-0101" and staged["asset_id"]
        # the in-shop packet keeps its rev 1 pages until the CO is approved
        assert all(p[3] == "1" for p in _drawing_pages(conn, pid))
        assert (
            conn.execute(
                "SELECT count(*) FROM packet_log WHERE packet_id = %s AND action LIKE 'Drawing for change order%%'",
                (pid,),
            ).fetchone()[0]
            == 1
        )


@needs_pg
def test_new_file_on_an_in_shop_packet_opens_a_drawing_co(fx: Settings) -> None:
    run_feed(fx, 1)
    run_drawings(fx)
    with app_conn(fx) as conn:
        pid = packet_id(conn, gen.KEY_CLOSES)  # L900-0901 rev 1, attached
        assert _drawing_pages(conn, pid)
        move_status(conn, pid, "in_progress")
        conn.execute(
            "UPDATE drawing_lookups SET checked_at = now() - interval '2 days', next_check_at = now()"
            " WHERE dwg = 'L900-0901'"
        )
    # M-Files re-releases rev 1 of L900-0901 as a new file (same rev, new version and bytes)
    d1 = fx.fixture_dir / "drawings.1.json"
    import json

    data = json.loads(d1.read_text())
    data["L900-0901"].update({"file": "L900-0901 r1b.pdf", "obj_version": 2, "file_version": 2})
    d1.write_text(json.dumps(data))
    (fx.fixture_dir / "L900-0901 r1b.pdf").write_bytes(gen.pdf_bytes("L900-0901 r1b"))
    counts = run_drawings(fx)
    assert counts.get("cos_opened", 0) == 1, counts
    with app_conn(fx) as conn:
        cos = cos_of(conn, pid)
        assert len(cos) == 1
        co = cos[0]
        assert (co["origin"], co["review"], co["ecn"], co["issued_by"]) == ("drawing", "pending", "M-FILES", None)
        assert co["reason"] == "M-Files re-released rev 1 of L900-0901 (new file)"
        assert co["staged"]["asset_id"] and co["vis"]["dwg_change"] is True
        assert all(r[0] == "system" for r in system_audit(conn, "packet.vis_open_co"))
        assert system_audit(conn, "packet.vis_open_co")