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

#!/usr/bin/env python3
"""Is the Micron asset picker read-only, and does it degrade to manual entry? — D11.

Cody Schaefer's `origin/Micron-Assets` branch, merged Aug 20 2026 and adapted to
the R2 creator (decisions-2026-08-20.md). The properties this pins:

  * **Read-only, structurally.** assets_db.py holds one SELECT and nothing else;
    /api/assets has no writing verb. Picking an asset can never change Micron.
  * **Unconfigured is a first-class state.** No MICRON_DB_URL -> configured:false,
    the picker says so, and manual entry carries the package. The suite must run
    without Micron existing at all — every other probe implicitly relies on that.
  * **Broken is not a leak.** A configured-but-unusable URL 503s with a message
    that never echoes the connection string (whose parse errors can quote
    password fragments).
  * **The client honours the catalog.** Search ranks exact matches first, a
    picked row is locked to the DB's own casing and badged, imports canonicalise
    casing / fall back to manual / skip duplicates, and the import summary goes
    through the T7.9 dialog kit, not a native alert().

Boots its own throwaway SQLite + uvicorn + headless browser; run it alone, not
back to back with other probes. Exit 0 all passed, 1 a failure, 2 could not run.
"""
import io
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 sections_check import set_sop                                   # noqa: E402
from stepper_check import dismiss_dialogs                            # noqa: E402
from qa_gate_check import api                                        # noqa: E402

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


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


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


def strip_py(src):
    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

    # ── 1. read-only, structurally ─────────────────────────────────────────────
    print("\n1. read-only, structurally")
    src = strip_py(io.open(os.path.join(SERVER, "assets_db.py"), encoding="utf-8").read())
    verbs = re.findall(r"\b(INSERT|UPDATE|DELETE|MERGE|EXEC|TRUNCATE|DROP|ALTER)\b",
                       src, re.I)
    chk("assets_db.py contains no writing SQL verb", not verbs, verbs)
    chk("...and exactly one SELECT (the whole schema contract)",
        len(re.findall(r"\bSELECT\b", src, re.I)) == 1)
    app_src = io.open(os.path.join(SERVER, "app.py"), encoding="utf-8").read()
    chk("/api/assets is a GET and only a GET",
        len(re.findall(r'@app\.get\("/api/assets"\)', app_src)) == 1
        and not re.findall(r'@app\.(post|put|patch|delete)\("/api/assets', app_src))
    outside = [f for f in ("models.py", "auth.py", "notify.py")
               if "MICRON_DB_URL" in io.open(os.path.join(SERVER, f), encoding="utf-8").read()]
    chk("the connection string is env-only plumbing, not model or auth state",
        not outside, outside)

    tmpdir = tempfile.mkdtemp(prefix="wpsuite-assets-")
    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)

        # ── 2. the API's unconfigured state ────────────────────────────────────
        print("\n2. unconfigured is a first-class state")
        st, _ = api(base, "/api/assets", "not-a-session")
        chk("anonymous gets 401, same as every other /api/ path", st == 401, st)
        st, body = api(base, "/api/assets", tok["root"])
        chk("signed in, no MICRON_DB_URL: 200 with configured:false",
            st == 200 and body and body.get("configured") is False
            and body.get("assets") == [], ascii_(body))
        chk("...and the detail tells the user what to do instead",
            "manual" in (body.get("detail") or "").lower(), ascii_(body))

        # ── 3. the picker, catalog absent ──────────────────────────────────────
        print("\n3. the picker degrades to manual entry")
        browser = cdp.Browser(exe)
        page = browser.page()
        page.clear_cookies()
        page.set_cookie("wp_session", tok["root"])
        page.viewport(1440, 900)
        page.goto(base + "/wp-creation-index.html?project=projA")
        dismiss_dialogs(page)
        chk("the creator boots", wait_creator(page))
        time.sleep(1.2)
        chk("the search box is disabled and says the catalog is not configured",
            page.eval("(() => { const b=document.getElementById('asset-search');"
                      " return b.disabled && /not configured/i.test(b.placeholder); })()"))
        chk("the source note announces it (role=status, non-empty)",
            page.eval("(() => { const n=document.getElementById('asset-source-note');"
                      " return n.getAttribute('role')==='status' && n.textContent.length>0; })()"))
        chk("no assets yet: the empty state renders instead of a blank table",
            page.eval("/No assets yet/.test(document.getElementById('asset-body').textContent)"))
        page.eval("addManualAsset()")
        chk("+ Add asset adds an editable manual row",
            page.eval("pkgAssets.length") == 1
            and page.eval("pkgAssets[0].source") == "manual"
            and page.eval("!!document.querySelector('#asset-body input')"))
        page.eval("document.querySelector('#asset-body input').value='HAND-01';"
                  "document.querySelector('#asset-body input')"
                  ".dispatchEvent(new Event('input',{bubbles:true}))")
        chk("...and typing lands in the model", page.eval("pkgAssets[0].tag") == "HAND-01")

        # ── 4. the client honours the catalog (injected; no SQL Server here) ──
        print("\n4. search, pick, import — against an injected catalog")
        page.eval("pkgAssets=[]; buildAssets();"
                  "assetCatalog=['AHU-2P-014','AHU-2P-015','PUMP-01','XPUMP-PUMP-011','CT-100'];"
                  "assetCatalogIndex=new Map(assetCatalog.map(t=>[t.toLowerCase(),t]));"
                  "assetCatalogState='ready';"
                  "(() => { const b=document.getElementById('asset-search');"
                  " b.disabled=false; b.placeholder='Search asset IDs'; })()")
        page.eval("runAssetSearch('pump-01')")
        chk("an exact match outranks a longer contains-match",
            page.eval("JSON.stringify(assetResults)") == '["PUMP-01","XPUMP-PUMP-011"]',
            ascii_(page.eval("JSON.stringify(assetResults)")))
        chk("results render as real <button>s, none disabled yet",
            page.eval("(() => { const r=[...document.querySelectorAll('#asset-results button.asset-result')];"
                      " return r.length===2 && r.every(b=>!b.disabled); })()"))
        page.eval("addCatalogAsset(0)")
        chk("picking adds a catalog row: locked ID (no input), badge, source:'catalog'",
            page.eval("pkgAssets.length") == 1
            and page.eval("pkgAssets[0].source") == "catalog"
            and page.eval("(() => { const tr=document.querySelector('#asset-body tr');"
                          " return !!tr.querySelector('.asset-badge')"
                          " && !tr.cells[0].querySelector('input'); })()"))
        n0 = page.eval("pkgAssets.length")
        page.eval("addCatalogAsset(0)")
        chk("picking it again is refused (already on the package)",
            page.eval("pkgAssets.length") == n0)
        chk("normaliseAsset: no source means manual; an unknown source means manual",
            page.eval("normaliseAsset({tag:'X'}).source") == "manual"
            and page.eval("normaliseAsset({tag:'X',source:'evil'}).source") == "manual"
            and page.eval("normaliseAsset({tag:'X',source:'catalog'}).source") == "catalog")

        page.eval("void applyImportedAssets([['asset id'],['ahu-2p-015'],['NOT-IN-DB'],['AHU-2P-015']])")
        time.sleep(0.4)
        got = json.loads(page.eval(
            "JSON.stringify(pkgAssets.map(a=>({t:a.tag,s:a.source})))"))
        chk("import: a hit is canonicalised to the DB's own casing and badged catalog",
            {"t": "AHU-2P-015", "s": "catalog"} in got, ascii_(got))
        chk("...a miss is kept, visibly manual — not silently dropped",
            {"t": "NOT-IN-DB", "s": "manual"} in got, ascii_(got))
        chk("...the in-file duplicate is skipped (3 rows total: pick + hit + miss)",
            len(got) == 3, ascii_(got))
        chk("...and the summary is the T7.9 dialog, not a native alert()",
            page.eval("document.getElementById('wp-dialog').classList.contains('open')")
            and page.eval("document.getElementById('wp-dialog-cancel').style.display") == "none")
        page.eval("wpDialogOk()")

        page.eval("(() => { const b=document.getElementById('asset-search');"
                  " b.value='ct-1'; runAssetSearch(b.value);"
                  " b.dispatchEvent(new KeyboardEvent('keydown',{key:'Enter',bubbles:true})); })()")
        chk("Enter takes the first result not already on the package",
            page.eval("pkgAssets[pkgAssets.length-1].tag") == "CT-100")

        # removal reopens the row for re-adding
        page.eval("runAssetSearch('ct-100')")
        chk("a just-added result reads 'added' and is disabled",
            page.eval("(() => { const b=document.querySelector('#asset-results button');"
                      " return b.disabled && /added/.test(b.textContent); })()"))
        page.eval("removeAsset(pkgAssets.length-1)")
        chk("removing the asset makes it addable again",
            page.eval("(() => { const b=document.querySelector('#asset-results button');"
                      " return !b.disabled && /add/.test(b.textContent); })()"))

        # ── 5. configured-but-broken: a 503 that does not leak ────────────────
        print("\n5. broken is not a leak")
        browser.close()
        browser = None
        server.terminate()
        server.wait(timeout=10)
        os.environ["MICRON_DB_URL"] = "mssql+pymssql://user:S3CRETpw@127.0.0.1:1/MicronDB"
        try:
            port2 = cdp.free_port()
            base2 = "http://127.0.0.1:%d" % port2
            server = start_server(port2, db_path)
            st, body = api(base2, "/api/assets", tok["root"])
            detail = (body or {}).get("detail") or ""
            chk("a configured-but-unusable catalog answers 503, not 500",
                st == 503, (st, ascii_(body)))
            chk("...and the message never echoes the URL, login or password",
                "S3CRETpw" not in detail and "user" not in detail
                and "127.0.0.1:1" not in detail, ascii_(detail))
        finally:
            del os.environ["MICRON_DB_URL"]
    finally:
        if browser:
            browser.close()
        if server:
            server.terminate()

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