Project-SDE-WP-Suite/tests/hold_check.py

#!/usr/bin/env python3
"""Does clearing the last constraint clear the hold? — CR-015 / A1 / D4, T7.3.

`CR-015` is the highest-severity item in the plan and it was reproduced live in
front of the Micron team: a package on hold with every constraint cleared, stuck.
The root cause was hold state STORED rather than derived — `submitHold()` wrote
`prevStatus='Issue'`, destroying the status the hold interrupted, and the clear
path fell into the "Mark it as Issued now?" offer because `'Issue'` is not in
`STATUS_ORDER` (`indexOf` gives -1, which reads as "before Issued"). Declining
that offer left the package on hold with nothing open, forever.

  1. the clear-last-constraint path, specifically (the regression this file exists for)
  2. D4: an Urgent package gets the audited override as the primary action; Normal and High are unchanged
  3. Issue (Hold) branches from every released state
  4. the server refuses what the browser refuses, and writes the history

The dialog stubs record every native dialog: the release path must fire NONE.
The prompt stub is how the audited override is driven — its return value is the
reason, and None is the user backing out.

Self-contained: throwaway SQLite, its own uvicorn, headless Edge or Chrome.
Exit 0 all passed, 1 a failure, 2 could not run.
"""
import ast
import json
import os
import re
import sys
import tempfile
import time
import urllib.error
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import cdp                                                           # noqa: E402
from browser_check import seed, start_server, chk, _PASS, _FAIL, _c  # noqa: E402
from sections_check import set_sop                                   # noqa: E402
from stepper_check import dismiss_dialogs                            # noqa: E402

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def ascii_(v, n=300):
    return re.sub(r"\s+", " ", str(v)).encode("ascii", "replace").decode()[:n]


def settle(seconds=0.8):
    time.sleep(seconds)


def wait_creator(page, tries=40):
    for _ in range(tries):
        if page.eval("!!window.wpCreatorReady"):
            return True
        time.sleep(0.3)
    return False


def api(base, path, token, method="GET", body=None):
    """Returns (http_status, parsed_body). 4xx is a result here, not an error —
    half of what this file checks is that the server says no."""
    req = urllib.request.Request(base + path, method=method)
    req.add_header("Cookie", "wp_session=" + token)
    req.add_header("Accept", "application/json")
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, data, timeout=15) as r:
            return r.status, json.loads(r.read().decode() or "null")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode() or "null")
        except Exception:
            return e.code, None


def audit(base, tok, wp_id, action):
    _, rows = api(base, "/api/audit?entity_type=wp&entity_id=%s&action=%s" % (wp_id, action), tok)
    return rows or []


# In-page dialog stubs. Installed once per load; __dlg records what fired so a
# path that must be dialog-free can prove it, and __pReturn is the prompt answer.
STUBS_JS = """(() => {
  window.__dlg = [];
  window.__pReturn = 'Boom lift arrives Friday - releasing to hold schedule';
  window.prompt = m => { window.__dlg.push(['prompt', String(m)]); return window.__pReturn; };
  window.alert  = m => { window.__dlg.push(['alert',  String(m)]); };
  window.confirm= m => { window.__dlg.push(['confirm',String(m)]); return false; };
  return true;
})()"""


def dlg(page):
    return json.loads(page.eval("JSON.stringify(window.__dlg||[])"))


def reset_dlg(page):
    page.eval("window.__dlg=[]")


def constraint_btn(page, index, which):
    """Click Open/Cleared/N-A on constraint row `index` — the same buttons a
    user clicks, re-queried each time because buildConstraints() rebuilds them."""
    labels = {"open": "Open", "cleared": "Cleared", "na": "N/A"}
    page.eval("""(() => {
      const tr = document.querySelectorAll('#constraint-body tr')[%d];
      [...tr.querySelectorAll('.cstatus button')]
        .find(b => b.textContent.trim() === %s).click();
    })()""" % (index, json.dumps(labels[which])))
    settle(0.4)


def click_status(page, val):
    page.eval("document.querySelector('#status-group .radio-pill[data-val=%s]').click()"
              % json.dumps(val))
    settle(0.4)


def status_of(page):
    return page.eval("getRadio('status')")


def strip_js_comments(src):
    """Good enough for grepping: block comments, then line comments that are not
    inside a string (approximated by requiring the // not be preceded by : which
    covers the https:// case that bit BL-017)."""
    src = re.sub(r"/\*.*?\*/", "", src, flags=re.S)
    return "\n".join(re.sub(r"(?<![:'\"])//.*$", "", ln) for ln in src.split("\n"))


def main():
    exe = cdp.find_browser()
    if not exe:
        print("no headless-capable browser found; set WP_BROWSER.")
        return 2

    tmpdir = tempfile.mkdtemp(prefix="wpsuite-hold-")
    db_path = os.path.join(tmpdir, "check.db")
    server = None
    browser = None
    try:
        tok = seed(db_path)
        set_sop(db_path, {})
        port = cdp.free_port()
        base = "http://127.0.0.1:%d" % port
        server = start_server(port, db_path)
        root = tok["root"]

        # ── 1. the clear-last-constraint path, driven in the browser ─────────
        print("\n1. CR-015: the stale hold")
        browser = cdp.Browser(exe)
        page = browser.page()
        page.clear_cookies()
        page.set_cookie("wp_session", root)
        page.viewport(1440, 900)
        page.goto(base + "/wp-creation-index.html?project=projA")
        dismiss_dialogs(page)
        chk("the creator boots", wait_creator(page))
        settle(1.5)
        page.eval(STUBS_JS)
        page.eval("window.__sentinel = 1")

        n = page.eval("pkgConstraints.length")
        for i in range(n):
            constraint_btn(page, i, "cleared")
        offers = [d for d in dlg(page) if d[0] == "confirm" and "release-ready" in d[1]]
        chk("the release-ready offer on an unreleased package is unchanged",
            len(offers) == 1, ascii_(dlg(page)))
        reset_dlg(page)
        click_status(page, "In Progress")
        chk("with everything cleared the package moves to In Progress",
            status_of(page) == "In Progress", status_of(page))

        constraint_btn(page, 0, "open")
        chk("flagging a constraint open on an in-progress package opens the hold modal",
            page.eval("document.getElementById('hold-modal').classList.contains('open')"))
        page.eval("document.getElementById('hold-details').value='Boom lift recalled for inspection'")
        page.eval("submitHold()")
        settle(0.4)
        chk("submitting the hold puts the package on hold", status_of(page) == "Issue",
            status_of(page))
        h = json.loads(page.eval("JSON.stringify(pkgHolds[pkgHolds.length-1]||{})"))
        chk("the hold entry records timestamp, user, and the status it interrupted",
            bool(h.get("ts")) and bool(h.get("by")) and h.get("from") == "In Progress",
            ascii_(h))

        reset_dlg(page)
        constraint_btn(page, 0, "cleared")
        chk("clearing the last open constraint clears the hold",
            status_of(page) != "Issue", status_of(page))
        chk("...and returns the package to its PRIOR status, not to Issued and not to a guess",
            status_of(page) == "In Progress", status_of(page))
        chk("...with no dialog of any kind", dlg(page) == [], ascii_(dlg(page)))
        chk("...and no refresh — the same document is still running",
            page.eval("window.__sentinel") == 1)
        rel = json.loads(page.eval("JSON.stringify(pkgHolds[pkgHolds.length-1]||{})"))
        chk("the release is in the history with timestamp, user, and what cleared it",
            rel.get("released") is True and bool(rel.get("ts")) and bool(rel.get("by"))
            and rel.get("to") == "In Progress", ascii_(rel))

        # Done-when items two and three: the same round trip, again.
        constraint_btn(page, 1, "open")
        page.eval("document.getElementById('hold-details').value='Second issue'")
        page.eval("submitHold()")
        settle(0.4)
        chk("logging a NEW constraint places it back on hold", status_of(page) == "Issue")
        constraint_btn(page, 1, "cleared")
        chk("clearing that constraint releases it again", status_of(page) == "In Progress",
            status_of(page))
        chk("the history now holds two hold/release pairs",
            page.eval("pkgHolds.filter(h=>h.released).length") == 2
            and page.eval("pkgHolds.filter(h=>!h.released).length") == 2,
            ascii_(page.eval("JSON.stringify(pkgHolds.map(h=>!!h.released))")))

        # The return status must survive persistence, not live in a JS variable:
        # collect the package, put it back mid-hold, and clear.
        constraint_btn(page, 2, "open")
        page.eval("document.getElementById('hold-details').value='Third issue'")
        page.eval("submitHold()")
        settle(0.4)
        page.eval("""(() => {
          const p = collectPackage();
          loadPackageIntoForm(JSON.parse(JSON.stringify(p)));
        })()""")
        settle(0.8)
        chk("a saved-and-reloaded package is still on hold", status_of(page) == "Issue",
            status_of(page))
        constraint_btn(page, 2, "cleared")
        chk("...and clearing its constraint returns it to the status recorded on the hold",
            status_of(page) == "In Progress", status_of(page))

        # ── 2. D4: the Urgent override ────────────────────────────────────────
        print("\n2. D4: Urgent surfaces the audited path; Normal and High do not")
        page.eval("newPackage()")
        settle(0.8)
        page.eval(STUBS_JS)
        chk("a new package starts with no override recorded",
            page.eval("pkgGateOverride === null || pkgGateOverride === undefined") is True
            or page.eval("!pkgGateOverride") is True)

        chk("a Normal package with open constraints shows NO override button",
            page.eval("!document.querySelector('.rb-act')"))
        reset_dlg(page)
        click_status(page, "Issued")
        alerts = [d for d in dlg(page) if d[0] == "alert"]
        chk("...and the status control still hard-blocks it with the same refusal",
            len(alerts) == 1 and "still open" in alerts[0][1], ascii_(dlg(page)))
        chk("...and the status snapped back", status_of(page) == "Draft", status_of(page))

        page.eval("document.getElementById('wp_priority').value='High'")
        constraint_btn(page, 0, "open")   # any change redraws the banner
        chk("High is treated exactly like Normal — no override button",
            page.eval("!document.querySelector('.rb-act')"))

        page.eval("document.getElementById('wp_priority').value='Urgent'")
        constraint_btn(page, 0, "open")
        chk("an Urgent package with open constraints offers the override as the "
            "banner's primary action, a real button",
            page.eval("(document.querySelector('.rb-act')||{}).tagName") == "BUTTON")

        reset_dlg(page)
        page.eval("window.__pReturn = null")   # back out first
        page.eval("document.querySelector('.rb-act').click()")
        settle(0.4)
        chk("backing out of the prompt releases nothing", status_of(page) == "Draft",
            status_of(page))
        chk("...and records no override", page.eval("!pkgGateOverride"))

        reset_dlg(page)
        page.eval("window.__pReturn = 'Client directive 42 - install proceeds at risk'")
        page.eval("document.querySelector('.rb-act').click()")
        settle(0.4)
        prompts = [d for d in dlg(page) if d[0] == "prompt"]
        open_names = json.loads(page.eval(
            "JSON.stringify(pkgConstraints.filter(c=>c.status==='open').map(c=>c.name))"))
        chk("the prompt names every open constraint it is about to cross",
            len(prompts) == 1 and all(nm in prompts[0][1] for nm in open_names),
            ascii_(prompts))
        chk("taking the override releases the package", status_of(page) == "Issued",
            status_of(page))
        ov = json.loads(page.eval("JSON.stringify(pkgGateOverride||{})"))
        chk("the override records actor, timestamp, reason and the constraints it crossed",
            bool(ov.get("by")) and bool(ov.get("at"))
            and ov.get("reason", "").startswith("Client directive")
            and sorted(ov.get("constraints") or []) == sorted(open_names), ascii_(ov))

        src = strip_js_comments(
            open(os.path.join(ROOT, "html", "wp-creation-app.js"), encoding="utf-8").read())
        chk("the client writes a gate override in exactly one place — confirmEarlyRelease()",
            src.count("pkgGateOverride={") == 1, src.count("pkgGateOverride={"))
        chk("the hold history has exactly two writers — the hold and the release",
            src.count("pkgHolds.push(") == 2, src.count("pkgHolds.push("))

        # ── 3. the hold branches from every released state ────────────────────
        print("\n3. Issue (Hold) is a branch, not a step")
        page.eval("newPackage()")
        settle(0.8)
        page.eval(STUBS_JS)
        n = page.eval("pkgConstraints.length")
        for i in range(n):
            constraint_btn(page, i, "na")
        reset_dlg(page)
        for st in ("Issued", "In Progress", "QC"):
            click_status(page, st)
            click_status(page, "Issue")
            opened = page.eval("document.getElementById('hold-modal').classList.contains('open')")
            page.eval("cancelHold()")
            settle(0.3)
            chk("from %s: Issue (hold) opens the log modal, and cancel restores %s" % (st, st),
                opened and status_of(page) == st, status_of(page))

        page.close()
        browser.close()
        browser = None

        # ── 4. the server refuses what the browser refuses, and writes history ─
        print("\n4. the server: gates and history")
        CON = lambda st: [{"name": "Boom lift", "status": st, "comment": ""},
                          {"name": "Permits", "status": "cleared", "comment": ""}]

        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpN1", "project_id": "projA", "number": "N-1", "subject": "normal",
            "status": "Issued", "data": {"priority": "Normal", "constraints": CON("open")}})
        chk("Normal + open constraint: refused", code == 409, code)

        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpU1", "project_id": "projA", "number": "U-1", "subject": "urgent bare",
            "status": "Issued", "data": {"priority": "Urgent", "constraints": CON("open")}})
        chk("Urgent with NO override: refused — urgency alone is not a reason", code == 409, code)

        ov = {"reason": "Client directive 42", "at": "2026-08-19T00:00:00Z", "by": "Root"}
        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpU2", "project_id": "projA", "number": "U-2", "subject": "urgent unnamed",
            "status": "Issued",
            "data": {"priority": "Urgent", "constraints": CON("open"), "gateOverride": ov}})
        chk("Urgent + override that names nothing: refused — it must say what it covers",
            code == 409, code)

        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpU3", "project_id": "projA", "number": "U-3", "subject": "urgent wrong name",
            "status": "Issued",
            "data": {"priority": "Urgent", "constraints": CON("open"),
                     "gateOverride": dict(ov, constraints=["Permits"])}})
        chk("Urgent + override naming a DIFFERENT constraint: refused — no riding through "
            "on an old reason", code == 409, code)

        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpN2", "project_id": "projA", "number": "N-2", "subject": "normal named",
            "status": "Issued",
            "data": {"priority": "Normal", "constraints": CON("open"),
                     "gateOverride": dict(ov, constraints=["Boom lift"])}})
        chk("Normal + a fully-named override: still refused — the path is Urgent-only",
            code == 409, code)

        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpU4", "project_id": "projA", "number": "U-4", "subject": "urgent covered",
            "status": "Issued",
            "data": {"priority": "Urgent", "constraints": CON("open"),
                     "gateOverride": dict(ov, constraints=["Boom lift"])}})
        chk("Urgent + an override naming the open constraint: released", code == 200, code)
        rows = audit(base, root, "wpU4", "gate_overridden")
        chk("...and the audit log names the constraints it crossed",
            rows and "Boom lift" in ((rows[0].get("detail") or {}).get("constraints") or []),
            ascii_(rows[:1]))

        # A1: the predecessor gate is untouched — refusable only with a reason.
        api(base, "/api/wps", root, "POST", {
            "id": "wpPred", "project_id": "projA", "number": "P-0", "subject": "upstream",
            "status": "In Progress", "data": {"constraints": CON("cleared")}})
        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpA1c", "project_id": "projA", "number": "A1-1", "subject": "downstream",
            "status": "Issued",
            "data": {"constraints": CON("cleared"), "predecessors": ["wpPred"]}})
        chk("A1: an unclosed predecessor still refuses a release", code == 409, code)
        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpA1c", "project_id": "projA", "number": "A1-1", "subject": "downstream",
            "status": "Issued",
            "data": {"constraints": CON("cleared"), "predecessors": ["wpPred"],
                     "gateOverride": {"reason": "Staged ahead of turnover", "by": "Root",
                                      "at": "2026-08-19T00:00:00Z"}}})
        chk("A1: the reasoned override still releases", code == 200, code)
        chk("A1: and still logs", bool(audit(base, root, "wpA1c", "gate_overridden")))

        # The hold history, through the upsert (how the browser and outbox save).
        api(base, "/api/wps", root, "POST", {
            "id": "wpH1", "project_id": "projA", "number": "H-1", "subject": "hold history",
            "status": "In Progress", "data": {"constraints": CON("cleared")}})
        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpH1", "project_id": "projA", "number": "H-1", "subject": "hold history",
            "status": "Issue",
            "data": {"constraints": CON("open"),
                     "holds": [{"ts": "2026-08-19T01:00:00Z", "constraint": "Boom lift",
                                "details": "Lift recalled for inspection", "by": "Root",
                                "from": "In Progress"}]}})
        rows = audit(base, root, "wpH1", "hold_logged")
        chk("going on hold writes hold_logged with the reason and the interrupted status",
            code == 200 and rows
            and (rows[0].get("detail") or {}).get("reason") == "Lift recalled for inspection"
            and (rows[0].get("detail") or {}).get("from") == "In Progress",
            ascii_((code, rows[:1])))
        code, _ = api(base, "/api/wps", root, "POST", {
            "id": "wpH1", "project_id": "projA", "number": "H-1", "subject": "hold history",
            "status": "In Progress",
            "data": {"constraints": CON("cleared"),
                     "holds": [{"ts": "2026-08-19T01:00:00Z", "constraint": "Boom lift",
                                "details": "Lift recalled for inspection", "by": "Root",
                                "from": "In Progress"},
                               {"ts": "2026-08-19T02:00:00Z", "released": True,
                                "constraint": "Boom lift", "by": "Root",
                                "details": "Hold released - last open constraint cleared",
                                "to": "In Progress"}]}})
        rows = audit(base, root, "wpH1", "hold_released")
        chk("coming off hold writes hold_released with where it went back to",
            code == 200 and rows
            and (rows[0].get("detail") or {}).get("to") == "In Progress",
            ascii_((code, rows[:1])))

        # The /status endpoint is not a side door: same gates, same history.
        api(base, "/api/wps", root, "POST", {
            "id": "wpS1", "project_id": "projA", "number": "S-1", "subject": "endpoint",
            "status": "In Progress", "data": {"constraints": CON("cleared")}})
        code, _ = api(base, "/api/wps/wpS1/status", root, "POST", {"status": "Issue"})
        chk("/status can place a package on hold", code == 200, code)
        chk("...and writes hold_logged", bool(audit(base, root, "wpS1", "hold_logged")))
        api(base, "/api/wps", root, "POST", {
            "id": "wpS1", "project_id": "projA", "number": "S-1", "subject": "endpoint",
            "status": "Issue", "data": {"constraints": CON("open")}})
        code, _ = api(base, "/api/wps/wpS1/status", root, "POST", {"status": "Issued"})
        chk("/status cannot walk a held package past its open constraint", code == 409, code)
        api(base, "/api/wps", root, "POST", {
            "id": "wpS1", "project_id": "projA", "number": "S-1", "subject": "endpoint",
            "status": "Issue", "data": {"constraints": CON("cleared")}})
        code, _ = api(base, "/api/wps/wpS1/status", root, "POST", {"status": "In Progress"})
        chk("/status releases once the record is clear", code == 200, code)
        chk("...and writes hold_released", bool(audit(base, root, "wpS1", "hold_released")))

        # Grep half of D4's "no unwritten release": every status assignment in
        # app.py sits in a function that runs the gates. (Constructor-built rows
        # in seed scripts are out of scope — S13 owns seeding.)
        tree = ast.parse(open(os.path.join(ROOT, "server", "app.py"), encoding="utf-8").read())
        bad = []
        for fn in [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)]:
            assigns = [n for n in ast.walk(fn) if isinstance(n, ast.Assign)
                       and any(isinstance(t, ast.Attribute) and t.attr == "status"
                               for t in n.targets)]
            if not assigns:
                continue
            calls = {c.func.id for c in ast.walk(fn)
                     if isinstance(c, ast.Call) and isinstance(c.func, ast.Name)}
            if "enforce_release_gates" not in calls:
                bad.append(fn.name)
        chk("every status assignment in server/app.py is inside a function that runs "
            "the release gates", not bad, ascii_(bad))

    finally:
        if browser is not None:
            try:
                browser.close()
            except Exception:
                pass
        if server is not None:
            try:
                server.terminate()
            except Exception:
                pass

    print("\n" + "-" * 54)
    print("%d/%d checks passed." % (len(_PASS), len(_PASS) + len(_FAIL)))
    for f in _FAIL:
        print("  - " + f)
    return 1 if _FAIL else 0


if __name__ == "__main__":
    sys.exit(main())