scripts/run_session_ibrowse_qa.py

#!/usr/bin/env python3
"""
run_session_ibrowse_qa.py
Executes required iBrowse tests following backend/frontend UI changes:
- 1-2 focused iBrowse tests
- 1 exploratory debugging E2E test
Target: http://127.0.0.1:8096 (PocketBase Studio UI)
Service: http://homelab.local:3000 (Homelab iBrowse)
"""

import os
import sys
import json
import time
import urllib.request
import urllib.error
from typing import Dict, Any, List

def load_credentials():
    env_file = os.path.expanduser("~/.ibrowse.env")
    token = os.environ.get("IBROWSE_API_KEY", "")
    host = os.environ.get("IBROWSE_BASE_URL") or os.environ.get("IBROWSE_HOST", "")
    if os.path.exists(env_file):
        with open(env_file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    k, v = line.split("=", 1)
                    k, v = k.strip(), v.strip()
                    if k == "IBROWSE_API_KEY" and not token:
                        token = v
                    elif (k == "IBROWSE_BASE_URL" or k == "IBROWSE_HOST") and not host:
                        host = v
    return host or "http://homelab.local:3000", token

HOST, TOKEN = load_credentials()
TARGET_URL = "http://127.0.0.1:8096"

TESTS = [
    {
        "id": "focus_nav_vault_settings",
        "name": "Focus Test 1: Navigation, Vault, Jobs & Settings HUD",
        "goal": (
            "Focus QA audit on Studio PB Dashboard, Vault, Jobs, and Settings. "
            "1. On 'Dashboard', inspect cards (stats) and verify the 'Ultimele joburi' table renders. "
            "2. Click the 'Vault' navigation tab, verify the songs list table renders with track titles and genre tags. "
            "3. Click the 'Jobs' navigation tab, verify the jobs table shows job type, status pills, and created timestamps. "
            "4. Click the 'Settings' navigation tab, verify system settings and zerogpu_daily_budget entries are visible. "
            "5. Monitor DevTools Console and Network log for any uncaught JS errors or 5xx responses. "
            "Provide a concise pass/fail verdict."
        ),
        "priority": "high",
    },
    {
        "id": "focus_baker_ling_dict",
        "name": "Focus Test 2: Lyrics Baker, Lingvistică & Dicționar Interactive Flow",
        "goal": (
            "Focus QA audit on Studio PB Lyrics Baker, Lingvistică, and Dicționar tabs at http://127.0.0.1:8096/. "
            "1. Click the 'Lyrics Baker' tab. Verify topic input (#bakerTopicInput), topic chips (e.g. #chipNoapteChisinau), "
            "manuscript canvas (#bakerLyricsCanvas), and action buttons: #bakerComposeBtn ('Generează Versuri'), "
            "#bakerRhymesBtn ('Găsește Rime'), and #bakerSaveBtn ('Salvează piesa în Vault' - disabled when empty canvas). "
            "Enter 'dor' into #bakerRhymesInput and click #bakerRhymesBtn, confirming rhymes appear in #bakerRhymesList (e.g. fraților). "
            "Do NOT click #bakerComposeBtn because full LLM lyric synthesis takes >80s and is tested in backend unit suites. "
            "2. Click the 'Lingvistică' tab. Verify dialect converter input (#lingTextInput), level buttons (#lingLevelBtn1..4), "
            "and thesaurus input (#thesaurusInput). Click #lingLevelBtn2, type 'harbuz' into #thesaurusInput, click #thesaurusBtn, "
            "and verify thesaurus results render in #thesaurusResults (e.g. harbuz -> pepene verde). "
            "3. Click the 'Dicționar' tab. Enter 'harbuz' into #dictSearchInput and verify filtered dictionary rows render in #dictTableBody. "
            "4. Monitor DevTools console for uncaught exceptions and network tab for /api/ endpoint responses. "
            "Provide a concise pass/fail verdict."
        ),
        "priority": "high",
    },
    {
        "id": "exploratory_e2e_debug",
        "name": "Exploratory Test: Adversarial E2E Debugging & Responsive Sweep",
        "goal": (
            "Adversarial E2E QA sweep as a skeptical senior QA engineer on Studio PB (http://127.0.0.1:8096). "
            "1. Fast-cycle through all 9 navigation tabs: Dashboard, Vault, Lyrics Baker, Lingvistică, Dicționar, "
            "Personas, Albume, Jobs, Settings. Verify no component crashes, blank screens, or visual overlaps. "
            "2. In Personas tab, verify persona cards render with names and system prompts. "
            "3. In Albume tab, verify album manifests and track lists render. "
            "4. Check desktop viewport layout: ensure no horizontal scrollbars, clipped text, or broken buttons. "
            "5. Scrutinize DevTools Console for JavaScript errors, unhandled rejections, slow requests (>3s), and failed requests. "
            "Deliver an overall health verdict with score and anomaly breakdown."
        ),
        "priority": "high",
    }
]

def submit_job(test_spec: Dict[str, Any]) -> Dict[str, Any]:
    url = f"{HOST}/v1/runs"
    payload = {
        "goal": test_spec["goal"],
        "url": TARGET_URL,
        "priority": test_spec.get("priority", "high")
    }
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        headers={
            "Authorization": f"Bearer {TOKEN}",
            "Content-Type": "application/json",
            "User-Agent": "iBrowse-Auditor/1.0"
        }
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        err_body = e.read().decode("utf-8")
        return {"status": "error", "code": e.code, "error": err_body}
    except Exception as e:
        return {"status": "error", "error": str(e)}

def poll_job(job_id: str, timeout_sec: int = 600, interval_sec: int = 5) -> Dict[str, Any]:
    url = f"{HOST}/v1/runs/{job_id}"
    t0 = time.time()
    last_status = None
    while time.time() - t0 < timeout_sec:
        req = urllib.request.Request(
            url,
            headers={"Authorization": f"Bearer {TOKEN}"}
        )
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read().decode("utf-8"))
                status = data.get("status")
                if status != last_status:
                    elapsed = int(time.time() - t0)
                    print(f"    [{elapsed}s] {job_id}: status={status}", flush=True)
                    last_status = status
                if status in ["succeeded", "completed", "failed", "cancelled", "error"]:
                    return data
        except Exception as e:
            elapsed = int(time.time() - t0)
            print(f"    [{elapsed}s] {job_id}: poll error: {e}", flush=True)
        time.sleep(interval_sec)
    return {"status": "timeout", "job_id": job_id, "elapsed": round(time.time() - t0, 1)}

def main():
    selected_tests = TESTS
    if len(sys.argv) > 1 and not sys.argv[1].startswith("-"):
        filter_str = sys.argv[1].lower()
        selected_tests = [t for t in TESTS if filter_str in t["id"].lower() or filter_str in t["name"].lower()]
        if not selected_tests:
            print(f"No tests match filter '{sys.argv[1]}'. Available: {[t['id'] for t in TESTS]}")
            sys.exit(1)

    print(f"=== iBrowse QA Session Runner ===")
    print(f"iBrowse Host: {HOST}")
    print(f"Target URL:   {TARGET_URL}")
    print(f"Tests to run: {len(selected_tests)}")
    print("=" * 40)

    # 1. Health check
    try:
        req = urllib.request.Request(f"{HOST}/v1/health")
        with urllib.request.urlopen(req, timeout=5) as r:
            h = json.loads(r.read().decode("utf-8"))
            print(f"Health check OK: version={h.get('version')}, active={h.get('active_jobs')}, queue={h.get('queue_depth')}")
    except Exception as e:
        print(f"Health check failed: {e}")
        sys.exit(1)

    # 2. Enqueue jobs
    submitted = []
    for t in selected_tests:
        print(f"\nSubmitting {t['name']}...")
        res = submit_job(t)
        job_id = res.get("job_id") or res.get("jobId")
        if not job_id:
            print(f"  FAILED to submit: {res}")
            sys.exit(1)
        print(f"  Queued: {job_id} (execution_id={res.get('execution_id')})")
        submitted.append({**t, "job_id": job_id, "submission": res})

    # 3. Poll all jobs until completion
    print("\n" + "=" * 40)
    print("Polling jobs concurrently...")
    results = {}
    pending = {item["job_id"]: item for item in submitted}

    t0 = time.time()
    max_wait = 900  # 15 min max

    while pending and (time.time() - t0 < max_wait):
        for jid in list(pending.keys()):
            item = pending[jid]
            url = f"{HOST}/v1/runs/{jid}"
            req = urllib.request.Request(url, headers={"Authorization": f"Bearer {TOKEN}"})
            try:
                with urllib.request.urlopen(req, timeout=10) as resp:
                    data = json.loads(resp.read().decode("utf-8"))
                    status = data.get("status")
                    if status in ["succeeded", "completed", "failed", "cancelled", "error", "blocked"]:
                        elapsed = int(time.time() - t0)
                        print(f"  [{elapsed}s] DONE: {item['name']} -> status={status}")
                        results[jid] = {**item, "data": data}
                        del pending[jid]
            except Exception as e:
                pass
        if pending:
            time.sleep(5)

    if pending:
        for jid, item in pending.items():
            print(f"  TIMEOUT: {item['name']} ({jid}) did not complete in {max_wait}s")
            results[jid] = {**item, "data": {"status": "timeout"}}

    # 4. Process and format results
    print("\n" + "=" * 50)
    print("=== FINAL iBrowse QA SESSION SUMMARY ===")
    out_evidence = []

    for item in submitted:
        jid = item["job_id"]
        res = results.get(jid, {})
        data = res.get("data", {})
        status = data.get("status")
        inner_res = data.get("result", {})
        message = inner_res.get("message", "") if isinstance(inner_res, dict) else str(inner_res)
        anomalies = data.get("anomalies", {})
        console_errs = anomalies.get("console_errors", []) if isinstance(anomalies, dict) else []
        slow_reqs = anomalies.get("slow_requests", []) if isinstance(anomalies, dict) else []
        aborted_reqs = anomalies.get("aborted_requests", []) if isinstance(anomalies, dict) else []
        steps = inner_res.get("steps", 0) if isinstance(inner_res, dict) else 0
        duration_s = round((inner_res.get("durationMs", 0) / 1000), 1) if isinstance(inner_res, dict) else 0

        print(f"\n[{status.upper() if status else 'UNKNOWN'}] {item['name']}")
        print(f"  Job ID:           {jid}")
        print(f"  Steps:            {steps}")
        print(f"  Duration:         {duration_s}s")
        print(f"  Console Errors:   {len(console_errs)}")
        print(f"  Slow Requests:    {len(slow_reqs)}")
        print(f"  Aborted Requests: {len(aborted_reqs)}")
        if message:
            # Print first 5 lines of message
            preview = "\n".join(message.strip().split("\n")[:6])
            print(f"  Summary Preview:\n{preview}")

        out_evidence.append({
            "test_id": item["id"],
            "name": item["name"],
            "job_id": jid,
            "status": status,
            "steps": steps,
            "duration_s": duration_s,
            "anomalies": anomalies,
            "message": message
        })

    # 5. Save results to data/qa_evidence
    evidence_path = "moldovan-qwen/data/qa_evidence/session_ibrowse_qa_results.json"
    os.makedirs(os.path.dirname(evidence_path), exist_ok=True)
    with open(evidence_path, "w", encoding="utf-8") as f:
        json.dump(out_evidence, f, indent=2, ensure_ascii=False)
    print(f"\nSaved full QA evidence to: {evidence_path}")

    # Check if all tests succeeded
    all_ok = all(e["status"] in ["succeeded", "completed"] for e in out_evidence)
    if all_ok:
        print("\n ALL iBrowse TESTS PASSED.")
        sys.exit(0)
    else:
        print("\n⚠️ SOME iBrowse TESTS FAILED OR TIMED OUT.")
        sys.exit(1)

if __name__ == "__main__":
    main()