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

#!/usr/bin/env python3
"""Does the material list upload the way locations do? — D6, T8.6.

CR-013 accepted free text because the master workbook never arrived; the Aug 18
call was the CR-005 call again - build the upload path now. Same component as
the location list (wp-list-import.js - grep asserts there is exactly one),
same rules: paste or file, dry-run check, rejected rows named with their source
line, deactivate never delete. Description, unit, optional code - and the probe
greps that NO inventory, pricing or stock field crossed the line
IMPLEMENTATION.md section 7 draws.

Exit 0 all passed, 1 a failure, 2 could not run.
"""
import json
import os
import re
import sys
import tempfile
import time

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      # noqa: E402
from qa_gate_check import api                                        # noqa: E402
from stepper_check import dismiss_dialogs                            # noqa: E402

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
HTML = os.path.join(ROOT, "html")


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


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


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

    # ── 1. one component, no forbidden fields ─────────────────────────────────
    print("\n1. grep: one component, the lightweight scope")
    suite = open(os.path.join(HTML, "work-package-suite-app.js"), encoding="utf-8").read()
    comp_defs = [n for n in os.listdir(HTML) if n.endswith(".js")
                 and "window.WPListImport" in open(os.path.join(HTML, n), encoding="utf-8").read()]
    chk("the import component is defined once, in wp-list-import.js",
        comp_defs == ["wp-list-import.js"], ascii_(comp_defs))
    chk("locations and materials are both instances of it - not a copy beside it",
        "locList = WPListImport(" in suite and "matList = WPListImport(" in suite
        and "function locImport(dryRun){ locList.importText" in suite)
    # The 2026-08-21 outage, pinned: a Boolean server_default of sa.text('1')
    # passes on SQLite (which coerces 1) and crash-loops Postgres at deploy
    # (DatatypeMismatch). Every migration must say sa.true()/sa.false().
    import re as _re
    bad = []
    vdir = os.path.join(ROOT, "server", "alembic", "versions")
    for fn in sorted(os.listdir(vdir)):
        if not fn.endswith(".py"):
            continue
        for ln in open(os.path.join(vdir, fn), encoding="utf-8"):
            if "Boolean" in ln and "server_default" in ln and                not _re.search(r"server_default=sa\.(true|false)\(\)", ln):
                bad.append("%s: %s" % (fn, ln.strip()[:90]))
    chk("no migration gives a Boolean a non-portable server_default "
        "(sa.true()/sa.false() only)", not bad, ascii_(bad[:3]))
    model = open(os.path.join(ROOT, "server", "models.py"), encoding="utf-8").read()
    mat_block = model[model.find("class MaterialItem"):model.find("class WpFile")]
    cols = re.findall(r"^\s+(\w+): Mapped", mat_block, re.M)
    chk("no inventory, stock, price or warehouse column exists on the model - "
        "the lightweight scope, exactly",
        cols and not any(re.search(r"stock|inventor|price|on_hand|warehouse", c, re.I)
                         for c in cols), ascii_(cols))
    chk("the sample rows are obviously fake",
        "SAMPLE-EMT" in suite and "Sample " in suite)

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

        # ── 2. the server: import, report, edit, deactivate ──────────────────
        print("\n2. the routes")
        text = "description,unit,code\nSample 3/4in EMT,FT,S-EMT\n,FT\nSample strut,FT\nSample 3/4in EMT,FT,S-EMT\n"
        code, rep = api(base, "/api/projects/projA/materials/import", root, "POST",
                        {"text": text, "dry_run": True})
        chk("a dry run reports without writing",
            code == 200 and rep["dry_run"] and len(rep["created"]) == 2, ascii_(rep))
        chk("...rejected rows carry the SOURCE line and a reason",
            rep["rejected"] and rep["rejected"][0]["line"] == 3
            and "description" in rep["rejected"][0]["reason"], ascii_(rep["rejected"]))
        chk("...in-file duplicates are named with the line they repeat",
            rep["duplicates"] and "line 2" in rep["duplicates"][0]["reason"],
            ascii_(rep["duplicates"]))
        _, listing = api(base, "/api/projects/projA/materials", root)
        chk("...and nothing was written", listing["items"] == [])

        # The 2026-08-23 production 500, pinned: what Postgres refuses (VARCHAR
        # overflow, control bytes) must come back as a per-line rejection - on
        # EVERY dialect - never crash the request.
        code, rep = api(base, "/api/projects/projA/materials/import", root, "POST",
                        {"text": "Sample " + "x" * 300 + ",EA", "dry_run": True})
        chk("an over-long description is a line rejection, not a 500",
            code == 200 and rep["rejected"] and "300 characters" in rep["rejected"][0]["reason"]
            and not rep["created"], ascii_(rep))
        code, rep = api(base, "/api/projects/projA/materials/import", root, "POST",
                        {"text": "Sample widget\u0000,EA", "dry_run": True})
        chk("a control byte is a line rejection, not a 500",
            code == 200 and rep["rejected"]
            and "control characters" in rep["rejected"][0]["reason"], ascii_(rep))

        code, rep = api(base, "/api/projects/projA/materials/import", root, "POST",
                        {"text": text, "dry_run": False})
        _, listing = api(base, "/api/projects/projA/materials", root)
        chk("the real import lands both good rows", len(listing["items"]) == 2,
            ascii_(listing))
        item = listing["items"][0]
        code, _ = api(base, "/api/projects/projA/materials/%s" % item["id"], root,
                      "PATCH", {"active": False})
        _, act = api(base, "/api/projects/projA/materials", root)
        _, allrows = api(base, "/api/projects/projA/materials?include_inactive=true", root)
        chk("deactivating hides a line from new requests without deleting it",
            code == 200 and len(act["items"]) == 1 and len(allrows["items"]) == 2)
        code, rep = api(base, "/api/projects/projA/materials/import", root, "POST",
                        {"text": "Sample 3/4in EMT,FT,S-EMT", "dry_run": False})
        chk("re-importing a deactivated line brings back the SAME row",
            code == 200 and len(rep["reactivated"]) == 1, ascii_(rep))
        code, added = api(base, "/api/projects/projA/materials", root, "POST",
                          {"description": "Sample anchor", "unit": "ea"})
        chk("hand-adding works, unit normalised", code == 200 and added["unit"] == "EA")

        # ── 3. the wizard at 390px ────────────────────────────────────────────
        print("\n3. the wizard, on a phone")
        browser = cdp.Browser(exe)
        page = browser.page()
        page.clear_cookies()
        page.set_cookie("wp_session", root)
        page.viewport(390, 844, mobile=True)
        page.goto(base + "/work-package-suite.html?tab=sop&project=projA")
        dismiss_dialogs(page)
        settle(2.5)
        page.eval("currentStep = 11; updateStepUI(); if(typeof matList!=='undefined') matList.load(); locLoad();")
        settle(1.2)
        chk("step 11 offers the material list beside the location list",
            page.eval("!!document.getElementById('mat-tool') && !!document.getElementById('loc-tool')"))
        chk("...its current list renders the server rows",
            "Sample strut" in json.loads(page.eval(
                "JSON.stringify([...document.querySelectorAll('#mat-list .mat-desc')]"
                ".map(i=>i.value))")), ascii_(json.loads(page.eval(
                "JSON.stringify([...document.querySelectorAll('#mat-list .mat-desc')]"
                ".map(i=>i.value))"))))
        chk("...and nothing scrolls sideways at 390px",
            page.eval("document.documentElement.scrollWidth <= 392"),
            page.eval("document.documentElement.scrollWidth"))
        page.eval("document.getElementById('mat-paste').value='Sample wire nut,EA'")
        page.eval("document.getElementById('mat-import-btn').click()")
        settle(1.2)
        chk("an import through the UI lands and reports",
            "1 value added" in page.eval(
                "(document.getElementById('mat-report')||{textContent:''}).textContent"),
            ascii_(page.eval("(document.getElementById('mat-report')||{textContent:''}).textContent")))

        js_errors = [e for e in page.js_errors() if "beforeunload" not in e]
        chk("no JavaScript errors anywhere in this run", not js_errors,
            ascii_(js_errors[:2]))

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