cubelinux/harness_cube_panel.py

#!/usr/bin/env python3
"""
harness_cube_panel.py - REAL verification harness for the CUBE (Squared Plus)
control-panel circuit. Asserts on every check; exits non-zero on any failure.

Circuit exercised (end-to-end, against the LIVE system):
  A. FRONTEND  - both consoles served; renamed to CUBE; HTML well-formed
                 (validated by stdlib HTMLParser; node is root-only here).
  B. BACKEND   - full worker-profile loop via /api/cube-vm/*:
                 create -> list(shows) -> read(matches) -> status(running)
                 -> delete -> list(gone).
  C. PERSIST   - schematic DUAL-CUBE-WORKING-PLAN title == CUBE (Squared Plus);
                 project-continuity registry seeded.
  D. SYSTEMD   - both refresh timers enabled+active; last service fire exit 0.
"""
import os, sys, json, base64, subprocess, urllib.request, datetime, socket
from html.parser import HTMLParser

API = "http://127.0.0.1"
SOCK = (
    os.environ.get("CUBE_SOCKET") or "/run/user/1000/cubelinux/cubed-names.sock"
)
NS_SCHEMA = "cube-schematic"
NS_PROJ = "project-continuity"
TEST_ID = "_harness_%d" % os.getpid()

results = []
def check(name, ok, detail=""):
    results.append((name, ok, detail))
    line = ("  PASS " if ok else "  FAIL ") + name
    if (not ok) and detail:
        line += "  -- " + detail
    print(line)
    return ok

def cubed_rpc(req, timeout=5):
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM); s.settimeout(timeout)
    s.connect(SOCK); s.sendall((json.dumps(req) + "\n").encode())
    buf = b""
    while b"\n" not in buf:
        c = s.recv(4096)
        if not c: break
        buf += c
    s.close(); return json.loads(buf.split(b"\n", 1)[0])

def get_json(path):
    return json.loads(urllib.request.urlopen(API + path, timeout=5).read().decode())

def post_json(path, body):
    req = urllib.request.Request(API + path, data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"}, method="POST")
    return json.loads(urllib.request.urlopen(req, timeout=10).read().decode())

def html_well_formed(url):
    """Real HTML well-formedness via stdlib HTMLParser (tag balance, no stray closes)."""
    class V(HTMLParser):
        def __init__(s):
            super().__init__(convert_charrefs=True)
            s.void = {'meta','link','br','img','input','hr','area','base',
                      'col','embed','source','track','wbr','!doctype'}
            s.stack = []; s.errs = []
        def handle_starttag(s, t, a):
            if t not in s.void: s.stack.append(t)
        def handle_endtag(s, t):
            if t in s.void: return
            if s.stack and s.stack[-1] == t: s.stack.pop()
            elif t in s.stack:
                while s.stack and s.stack.pop() != t: pass
            else: s.errs.append(t)
    try:
        data = urllib.request.urlopen(url, timeout=10).read().decode("utf-8", "replace")
    except Exception:
        return False
    v = V(); v.feed(data)
    return (not v.stack) and (not v.errs)

def systemd_state(unit):
    en = subprocess.run(["systemctl", "--user", "is-enabled", unit],
                        capture_output=True, text=True).stdout.strip()
    ac = subprocess.run(["systemctl", "--user", "is-active", unit],
                        capture_output=True, text=True).stdout.strip()
    res = subprocess.run(["systemctl", "--user", "show", "-p", "Result",
                         unit.replace(".timer", ".service")],
                        capture_output=True, text=True).stdout.strip()
    return en, ac, res

def circuit():
    ok_all = True
    print("\n[A] FRONTEND")
    idx = urllib.request.urlopen(API + "/", timeout=5).read().decode()
    vm = urllib.request.urlopen(API + "/cube-vm.html", timeout=5).read().decode()
    ok_all &= check("console served + renamed CUBE⊞",
                    "CUBE⊞ · Squared Plus · AI Command Center" in idx,
                    "title mark missing")
    ok_all &= check("panel served + renamed CUBE⊞",
                    "<h1>CUBE⊞</h1>" in vm, "h1 mark missing")
    ok_all &= check("console HTML well-formed", html_well_formed(API + "/"))
    ok_all &= check("panel HTML well-formed", html_well_formed(API + "/cube-vm.html"))

    print("\n[B] BACKEND worker-profile loop")
    prof = {"name": "Harness Worker", "tools": ["read_file", "shell_ro"],
            "model": "local-default",
            "expert": {"domain": "test", "prompt": "x", "knowledge": "y"},
            "recall": {"coords": [], "namespaces": ["hermes"]},
            "runtime": {"vcpus": 2, "mem_mb": 2048}}
    r = post_json("/api/cube-vm/config", {"id": TEST_ID, "profile": prof})
    ok_all &= check("create profile", r.get("ok") is True, str(r))
    cfgs = get_json("/api/cube-vm/configs").get("profiles", [])
    ok_all &= check("profile appears in roster", any(p["id"] == TEST_ID for p in cfgs))
    rd = get_json("/api/cube-vm/config/" + TEST_ID)
    ok_all &= check("individual read matches",
                    rd.get("profile", {}).get("name") == "Harness Worker", str(rd))
    st = get_json("/api/cube-vm/status")
    ok_all &= check("VM status reports running", st.get("running") is True,
                    str(st.get("boot_state")))
    d = post_json("/api/cube-vm/delete", {"id": TEST_ID})
    ok_all &= check("delete profile", d.get("ok") is True)
    cfgs2 = get_json("/api/cube-vm/configs").get("profiles", [])
    ok_all &= check("profile gone after delete",
                    not any(p["id"] == TEST_ID for p in cfgs2))

    print("\n[C] PERSISTENCE (CUBE store)")
    sch = cubed_rpc({"cmd": "read", "namespace": NS_SCHEMA, "name": "DUAL-CUBE-WORKING-PLAN"})
    title = json.loads(base64.b64decode(sch["value"]).decode()).get("title", "")
    ok_all &= check("schematic title == CUBE⊞ (Squared Plus) in cube",
                    "CUBE⊞ (Squared Plus)" in title, title[:40])
    sop = cubed_rpc({"cmd": "read", "namespace": NS_PROJ, "name": "SOP"})
    ok_all &= check("project-continuity SOP seeded", sop.get("ok") is True)
    plist = cubed_rpc({"cmd": "list", "namespace": NS_PROJ})
    ok_all &= check("project-continuity registry populated",
                    len(plist.get("entries", [])) >= 5,
                    str(len(plist.get("entries", []))))

    print("\n[D] SYSTEMD timers")
    for unit in ("cube-schematic-refresh.timer", "project-continuity-refresh.timer"):
        en, ac, res = systemd_state(unit)
        ok_all &= check("%s enabled+active (Result=%s)" % (unit, res),
                        en == "enabled" and ac == "active", "%s/%s" % (en, ac))

    return ok_all

if __name__ == "__main__":
    print("=== CUBE harness run @ %s ===" % datetime.datetime.now().isoformat(timespec="seconds"))
    print("TEST_ID=%s" % TEST_ID)
    ok = circuit()
    npass = sum(1 for _, o, _ in results if o)
    nfail = len(results) - npass
    print("\n=== SUMMARY: %d/%d checks passed, %d failed ===" % (npass, len(results), nfail))
    sys.exit(0 if ok else 1)