""" nomos-browser-qa — pixel-world + dashboard visual QA HF Space. Runs Playwright directly for low-level DOM assertions (cheap + deterministic), plus browser-use agent for any higher-level nav tasks if needed. """ from __future__ import annotations import asyncio import base64 import json import os import traceback from datetime import datetime, timezone from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field # --------------------------------------------------------------------------- # Paths / env # --------------------------------------------------------------------------- DATA_DIR = Path(os.environ.get("DATA_DIR", "/data")) try: DATA_DIR.mkdir(parents=True, exist_ok=True) except PermissionError: DATA_DIR = Path("/tmp/nomos-browser-qa") DATA_DIR.mkdir(parents=True, exist_ok=True) QA_DIR = DATA_DIR / "qa-runs" QA_DIR.mkdir(parents=True, exist_ok=True) ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY") GOOGLE_KEY = os.environ.get("GOOGLE_API_KEY") PIXEL_WORLD_URL = os.environ.get( "PIXEL_WORLD_URL", "https://nomos42-pixel-world.static.hf.space" ) DASHBOARD_BASE = os.environ.get("DASHBOARD_URL", "https://nomosdashboard.vercel.app") DASHBOARD_ROUTES = ["/nba", "/political", "/world"] DASHBOARD_PRICING = "/pricing" # --------------------------------------------------------------------------- # Lazy imports # --------------------------------------------------------------------------- PLAYWRIGHT_AVAILABLE = False PLAYWRIGHT_IMPORT_ERROR: str | None = None try: from playwright.async_api import async_playwright # type: ignore PLAYWRIGHT_AVAILABLE = True except Exception as e: # pragma: no cover PLAYWRIGHT_IMPORT_ERROR = f"{type(e).__name__}: {e}" BROWSER_USE_AVAILABLE = False BROWSER_USE_IMPORT_ERROR: str | None = None try: from browser_use import Agent as BrowserAgent # type: ignore # noqa: F401 BROWSER_USE_AVAILABLE = True except Exception as e: # pragma: no cover BROWSER_USE_IMPORT_ERROR = f"{type(e).__name__}: {e}" # --------------------------------------------------------------------------- # Schemas # --------------------------------------------------------------------------- class QAPixelRequest(BaseModel): url: str | None = None min_sprites: int = Field(default=40, ge=1) capture_screenshot: bool = True timeout_s: int = Field(default=45, ge=5, le=180) class QADashboardRequest(BaseModel): base_url: str | None = None routes: list[str] | None = None check_pricing: bool = True timeout_s: int = Field(default=60, ge=5, le=240) class QARun(BaseModel): kind: str passed: bool ts: str details: dict[str, Any] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _write_run(run: QARun) -> Path: safe_ts = run.ts.replace(":", "").replace("-", "")[:15] path = QA_DIR / f"{run.kind}-{safe_ts}.json" path.write_text(run.model_dump_json(indent=2)) return path async def _new_browser(): if not PLAYWRIGHT_AVAILABLE: raise RuntimeError( f"playwright unavailable: {PLAYWRIGHT_IMPORT_ERROR}" ) pw = await async_playwright().start() browser = await pw.chromium.launch( headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"] ) ctx = await browser.new_context(viewport={"width": 1440, "height": 900}) return pw, browser, ctx async def _shutdown(pw, browser): with _suppress(): await browser.close() with _suppress(): await pw.stop() class _suppress: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return True # --------------------------------------------------------------------------- # Pixel QA # --------------------------------------------------------------------------- async def _qa_pixel_impl(req: QAPixelRequest) -> QARun: url = req.url or PIXEL_WORLD_URL ts = datetime.now(timezone.utc).isoformat() details: dict[str, Any] = {"url": url} passed = False pw = browser = None try: pw, browser, ctx = await _new_browser() page = await ctx.new_page() console_errors: list[str] = [] page.on( "console", lambda msg: console_errors.append(f"{msg.type}: {msg.text}") if msg.type in {"error", "warning"} else None, ) page.on("pageerror", lambda err: console_errors.append(f"pageerror: {err}")) await page.goto(url, wait_until="load", timeout=req.timeout_s * 1000) # Give the pixel world time to animate in. await asyncio.sleep(3) # Count sprites: common selectors used in pixel-world builds. sprite_count = 0 for sel in [ "[data-sprite]", ".agent-sprite", ".sprite", "canvas", ".pixel-agent", ]: try: count = await page.locator(sel).count() sprite_count = max(sprite_count, count) except Exception: pass # HP-bar heuristic hp_count = 0 for sel in [".hp-bar", "[data-hp]", ".health"]: try: hp_count = max(hp_count, await page.locator(sel).count()) except Exception: pass # DOM size as a liveness fallback when selectors unknown. try: dom_size = await page.evaluate("document.querySelectorAll('*').length") except Exception: dom_size = 0 screenshot_b64 = None if req.capture_screenshot: try: shot = await page.screenshot(full_page=True, type="png") screenshot_b64 = base64.b64encode(shot).decode("ascii")[:40000] except Exception as e: details["screenshot_error"] = str(e) real_errors = [ e for e in console_errors if "error" in e.lower() and "warning" not in e.lower() ] details.update( sprite_count=sprite_count, hp_bar_count=hp_count, dom_size=dom_size, console_errors=real_errors[:20], console_warnings=[e for e in console_errors if "warning" in e.lower()][:10], screenshot_b64_preview=screenshot_b64, ) passed = ( sprite_count >= req.min_sprites or dom_size >= 200 ) and len(real_errors) == 0 except Exception as e: details["error"] = f"{type(e).__name__}: {e}" details["traceback"] = traceback.format_exc()[-2000:] finally: if pw and browser: await _shutdown(pw, browser) run = QARun(kind="pixel", passed=passed, ts=ts, details=details) _write_run(run) return run # --------------------------------------------------------------------------- # Dashboard QA # --------------------------------------------------------------------------- async def _qa_dashboard_impl(req: QADashboardRequest) -> QARun: base = (req.base_url or DASHBOARD_BASE).rstrip("/") routes = req.routes or DASHBOARD_ROUTES ts = datetime.now(timezone.utc).isoformat() details: dict[str, Any] = {"base_url": base, "routes": routes} per_route: dict[str, Any] = {} passed = True pw = browser = None try: pw, browser, ctx = await _new_browser() for route in routes: full = f"{base}{route}" page = await ctx.new_page() console_errors: list[str] = [] page.on( "pageerror", lambda err, _c=console_errors: _c.append(f"pageerror: {err}") ) page.on( "console", lambda msg, _c=console_errors: _c.append(f"{msg.type}: {msg.text}") if msg.type == "error" else None, ) try: resp = await page.goto(full, wait_until="load", timeout=req.timeout_s * 1000) status = resp.status if resp else None title = await page.title() dom_size = await page.evaluate("document.querySelectorAll('*').length") except Exception as e: per_route[route] = {"error": f"{type(e).__name__}: {e}"} passed = False await page.close() continue route_passed = ( status is not None and status < 400 and len(console_errors) == 0 ) if not route_passed: passed = False per_route[route] = { "status": status, "title": title, "dom_size": dom_size, "console_errors": console_errors[:15], "passed": route_passed, } await page.close() # Pricing page (optional) if req.check_pricing: pricing_url = f"{base}{DASHBOARD_PRICING}" page = await ctx.new_page() pricing_info: dict[str, Any] = {"url": pricing_url} try: resp = await page.goto( pricing_url, wait_until="load", timeout=req.timeout_s * 1000 ) status = resp.status if resp else None body = await page.content() has_stripe = ( "stripe.com" in body.lower() or "buy.stripe" in body.lower() or "checkout.stripe" in body.lower() ) pricing_info.update( status=status, has_stripe_link=has_stripe, body_len=len(body) ) except Exception as e: pricing_info["error"] = f"{type(e).__name__}: {e}" finally: await page.close() details["pricing"] = pricing_info except Exception as e: details["error"] = f"{type(e).__name__}: {e}" details["traceback"] = traceback.format_exc()[-2000:] passed = False finally: if pw and browser: await _shutdown(pw, browser) details["per_route"] = per_route run = QARun(kind="dashboard", passed=passed, ts=ts, details=details) _write_run(run) return run # --------------------------------------------------------------------------- # FastAPI app # --------------------------------------------------------------------------- app = FastAPI(title="nomos-browser-qa", version="0.1.0") BOOT_TS = datetime.now(timezone.utc).isoformat() @app.get("/") def root(): return {"service": "nomos-browser-qa", "status": "ok", "boot": BOOT_TS} @app.get("/api/status") def status(): return { "service": "nomos-browser-qa", "boot_ts": BOOT_TS, "now_ts": datetime.now(timezone.utc).isoformat(), "playwright_available": PLAYWRIGHT_AVAILABLE, "playwright_import_error": PLAYWRIGHT_IMPORT_ERROR, "browser_use_available": BROWSER_USE_AVAILABLE, "browser_use_import_error": BROWSER_USE_IMPORT_ERROR, "providers": {"anthropic": bool(ANTHROPIC_KEY), "google": bool(GOOGLE_KEY)}, "pixel_world_url": PIXEL_WORLD_URL, "dashboard_base": DASHBOARD_BASE, "qa_runs_cached": len(list(QA_DIR.glob("*.json"))), } @app.get("/api/latest-qa") def latest_qa(): runs: list[dict] = [] paths = sorted(QA_DIR.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)[:10] for p in paths: try: data = json.loads(p.read_text()) # Scrub heavy screenshots from summary if isinstance(data.get("details"), dict): data["details"].pop("screenshot_b64_preview", None) runs.append(data) except Exception as e: # pragma: no cover runs.append({"file": p.name, "error": str(e)}) return {"runs": runs, "count": len(runs)} @app.post("/api/qa-pixel", response_model=QARun) async def qa_pixel(req: QAPixelRequest): if not PLAYWRIGHT_AVAILABLE: raise HTTPException(503, f"playwright unavailable: {PLAYWRIGHT_IMPORT_ERROR}") return await _qa_pixel_impl(req) @app.post("/api/qa-dashboard", response_model=QARun) async def qa_dashboard(req: QADashboardRequest): if not PLAYWRIGHT_AVAILABLE: raise HTTPException(503, f"playwright unavailable: {PLAYWRIGHT_IMPORT_ERROR}") return await _qa_dashboard_impl(req)