Spaces:
Running on Zero
Running on Zero
AndrianBalanescu
chore(security): set visibility to private, sanitize internal Tailscale IPs, and drop qa_evidence tracking
4fb8728 | #!/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() | |