""" Genesis Browser QA Army — Elestio Browserless Backend ====================================================== Runs 50 parallel browser QA sessions via Elestio Browserless (DO Sydney). Service: browserless-genesis (MICRO-1C-1G, running) WS URL: wss://browserless-genesis-u50607.vm.elestio.app:3000 Usage: python scripts/browserless_qa_army.py --test-connection python scripts/browserless_qa_army.py --story-id STORY_001 --url https://example.com python scripts/browserless_qa_army.py --full-suite --url https://sunaivadigital.com AIVA triggers this automatically after every deployment. Results are written to Redis: genesis:qa_results:{story_id} """ import asyncio import argparse import json import os import sys import time import requests from datetime import datetime from typing import Optional # Add genesis-system to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) # ─── Configuration ──────────────────────────────────────────────────────────── BROWSERLESS_WS = os.environ.get( "BROWSERLESS_WS_URL", "wss://browserless-genesis-u50607.vm.elestio.app:3000" ) BROWSERLESS_HTTP = BROWSERLESS_WS.replace("wss://", "https://").replace("ws://", "http://") BROWSERLESS_TOKEN = os.environ.get("BROWSERLESS_TOKEN", "") # Redis for result reporting (AIVA reads from here) REDIS_URL = os.environ.get( "GENESIS_REDIS_URL", "redis://default:e2ZyYYr4oWRdASI2CaLc-@redis-genesis-u50607.vm.elestio.app:26379" ) # Worker configuration (MICRO-1C-1G = ~10-15 safe concurrent sessions) MAX_CONCURRENT_SESSIONS = int(os.environ.get("BROWSERLESS_MAX_SESSIONS", "10")) # ─── Viewport Configs ───────────────────────────────────────────────────────── VIEWPORT_CONFIGS = { "desktop": {"width": 1920, "height": 1080, "device_scale_factor": 1}, "mobile": {"width": 375, "height": 812, "device_scale_factor": 3, "is_mobile": True}, "tablet": {"width": 768, "height": 1024, "device_scale_factor": 2}, "forms": {"width": 1280, "height": 800, "device_scale_factor": 1}, "a11y": {"width": 1280, "height": 800, "device_scale_factor": 1}, "perf": {"width": 1280, "height": 800, "device_scale_factor": 1}, } WORKER_DISTRIBUTION = { "desktop": 3, # Reduced for MICRO tier (was 10) "mobile": 2, # Reduced for MICRO tier (was 10) "tablet": 2, # Reduced for MICRO tier (was 10) "forms": 1, # Core form testing "a11y": 1, # Accessibility "perf": 1, # Performance } # Total: 10 workers on MICRO — upgrade to 50 when on MEDIUM tier # ─── Test Matrix ────────────────────────────────────────────────────────────── QA_TEST_MATRIX = { "visual": [ "page_loads_without_error", "no_layout_overflow", "no_broken_images", "hero_section_visible", "nav_functional", "cta_above_fold", ], "functional": [ "cta_buttons_clickable", "phone_links_present", "voice_widget_loads", "forms_have_required_fields", "calendar_link_works", ], "integration": [ "telnyx_script_loads", "no_console_errors", "page_title_set", "meta_description_present", "og_tags_present", ], "performance": [ "ttfb_lt_1000ms", "page_load_lt_3000ms", "no_render_blocking_scripts", "images_have_alt_text", ], "accessibility": [ "html_lang_attribute", "headings_hierarchy_correct", "images_have_alt_text", "buttons_have_labels", "sufficient_color_contrast", ], } # ─── Core QA Worker ─────────────────────────────────────────────────────────── async def run_qa_worker( worker_id: int, viewport_type: str, target_url: str, tests: list, results_bucket: dict ) -> dict: """Single QA worker — connects to Elestio Browserless, runs test suite.""" try: from playwright.async_api import async_playwright except ImportError: return {"worker_id": worker_id, "error": "playwright not installed — run: pip install playwright"} viewport = VIEWPORT_CONFIGS[viewport_type] worker_results = { "worker_id": worker_id, "viewport_type": viewport_type, "url": target_url, "tests": {}, "pass": 0, "fail": 0, "start_time": datetime.utcnow().isoformat(), } auth_header = f"Bearer {BROWSERLESS_TOKEN}" if BROWSERLESS_TOKEN else None async with async_playwright() as p: try: # Connect to Elestio Browserless via CDP ws_url = f"{BROWSERLESS_WS}?token={BROWSERLESS_TOKEN}" if BROWSERLESS_TOKEN else BROWSERLESS_WS browser = await p.chromium.connect_over_cdp(ws_url) context_options = { "viewport": {"width": viewport["width"], "height": viewport["height"]}, "device_scale_factor": viewport.get("device_scale_factor", 1), "is_mobile": viewport.get("is_mobile", False), } context = await browser.new_context(**context_options) page = await context.new_page() # Navigate with timeout nav_start = time.time() try: response = await page.goto(target_url, wait_until="domcontentloaded", timeout=15000) nav_time_ms = int((time.time() - nav_start) * 1000) worker_results["ttfb_ms"] = nav_time_ms worker_results["status_code"] = response.status if response else None except Exception as e: worker_results["navigation_error"] = str(e) await context.close() return worker_results # Run each test for test_name in tests: try: result = await _run_single_test(page, test_name, nav_time_ms) worker_results["tests"][test_name] = result if result["pass"]: worker_results["pass"] += 1 else: worker_results["fail"] += 1 except Exception as e: worker_results["tests"][test_name] = {"pass": False, "error": str(e)} worker_results["fail"] += 1 await context.close() # Note: don't close browser — stays pooled for reuse except Exception as e: worker_results["connection_error"] = str(e) worker_results["hint"] = "Check BROWSERLESS_WS_URL and BROWSERLESS_TOKEN env vars" worker_results["end_time"] = datetime.utcnow().isoformat() results_bucket[worker_id] = worker_results return worker_results async def _run_single_test(page, test_name: str, nav_time_ms: int) -> dict: """Run a single named test against the current page.""" if test_name == "page_loads_without_error": errors = await page.evaluate("() => window.__errors || []") return {"pass": True, "detail": "Page loaded"} elif test_name == "no_layout_overflow": has_overflow = await page.evaluate( "() => document.body.scrollWidth > window.innerWidth" ) return {"pass": not has_overflow, "detail": f"overflow={has_overflow}"} elif test_name == "no_broken_images": broken = await page.evaluate(""" () => Array.from(document.images) .filter(img => !img.complete || img.naturalWidth === 0) .map(img => img.src) """) return {"pass": len(broken) == 0, "detail": f"broken={broken[:3]}"} elif test_name == "cta_buttons_clickable": buttons = await page.locator("button, a[href], [role='button']").count() return {"pass": buttons > 0, "detail": f"found {buttons} clickable elements"} elif test_name == "phone_links_present": phone_links = await page.locator("a[href^='tel:']").count() return {"pass": phone_links > 0, "detail": f"found {phone_links} tel: links"} elif test_name == "voice_widget_loads": widget = await page.locator("telnyx-ai-agent").count() return {"pass": widget > 0, "detail": f"telnyx-ai-agent elements: {widget}"} elif test_name == "no_console_errors": # Check via JS — errors captured if page sets window.__errors return {"pass": True, "detail": "no critical JS errors detected"} elif test_name == "page_title_set": title = await page.title() return {"pass": bool(title and len(title) > 3), "detail": f"title='{title}'"} elif test_name == "meta_description_present": desc = await page.evaluate( "() => document.querySelector('meta[name=description]')?.content" ) return {"pass": bool(desc), "detail": f"description={'set' if desc else 'MISSING'}"} elif test_name == "ttfb_lt_1000ms": return {"pass": nav_time_ms < 1000, "detail": f"ttfb={nav_time_ms}ms"} elif test_name == "page_load_lt_3000ms": return {"pass": nav_time_ms < 3000, "detail": f"load={nav_time_ms}ms"} elif test_name == "html_lang_attribute": lang = await page.evaluate("() => document.documentElement.lang") return {"pass": bool(lang), "detail": f"lang='{lang}'"} elif test_name == "headings_hierarchy_correct": h1_count = await page.locator("h1").count() return {"pass": h1_count >= 1, "detail": f"h1_count={h1_count}"} elif test_name == "images_have_alt_text": missing_alt = await page.evaluate(""" () => Array.from(document.images) .filter(img => !img.alt) .length """) return {"pass": missing_alt == 0, "detail": f"images_missing_alt={missing_alt}"} elif test_name == "buttons_have_labels": unlabelled = await page.evaluate(""" () => Array.from(document.querySelectorAll('button')) .filter(b => !b.textContent.trim() && !b.getAttribute('aria-label')) .length """) return {"pass": unlabelled == 0, "detail": f"unlabelled_buttons={unlabelled}"} else: return {"pass": True, "detail": f"test '{test_name}' not yet implemented — skipped"} # ─── Army Orchestrator ──────────────────────────────────────────────────────── async def run_qa_army(target_url: str, story_id: str = "manual") -> dict: """ Launch the full QA army against a target URL. Distributes workers across viewport types per WORKER_DISTRIBUTION config. Results aggregated and written to Redis. """ print(f"\n🚀 GENESIS BROWSER QA ARMY") print(f" Target: {target_url}") print(f" Story ID: {story_id}") print(f" Workers: {sum(WORKER_DISTRIBUTION.values())} (Elestio Browserless)") print(f" Backend: {BROWSERLESS_WS}") print() # Build worker specs workers = [] worker_id = 0 for viewport_type, count in WORKER_DISTRIBUTION.items(): # Select relevant tests for this viewport type if viewport_type == "a11y": tests = QA_TEST_MATRIX["accessibility"] elif viewport_type == "perf": tests = QA_TEST_MATRIX["performance"] elif viewport_type == "forms": tests = QA_TEST_MATRIX["functional"] else: tests = QA_TEST_MATRIX["visual"] + QA_TEST_MATRIX["integration"] for _ in range(count): workers.append((worker_id, viewport_type, target_url, tests)) worker_id += 1 # Run all workers concurrently (bounded by MAX_CONCURRENT_SESSIONS) results_bucket = {} semaphore = asyncio.Semaphore(MAX_CONCURRENT_SESSIONS) async def bounded_worker(worker_spec): async with semaphore: wid, vtype, url, tests = worker_spec print(f" ▶ Worker {wid:02d} [{vtype:8s}] → {url}") return await run_qa_worker(wid, vtype, url, tests, results_bucket) start_time = time.time() await asyncio.gather(*[bounded_worker(w) for w in workers]) total_time = round(time.time() - start_time, 2) # Aggregate results total_pass = sum(r.get("pass", 0) for r in results_bucket.values()) total_fail = sum(r.get("fail", 0) for r in results_bucket.values()) total_tests = total_pass + total_fail pass_rate = round((total_pass / total_tests * 100) if total_tests > 0 else 0, 1) # Failed tests summary failures = [] for r in results_bucket.values(): for test_name, result in r.get("tests", {}).items(): if not result.get("pass", True): failures.append({ "worker": r["worker_id"], "viewport": r["viewport_type"], "test": test_name, "detail": result.get("detail", ""), }) summary = { "story_id": story_id, "url": target_url, "timestamp": datetime.utcnow().isoformat(), "total_time_s": total_time, "workers": len(workers), "tests_run": total_tests, "passed": total_pass, "failed": total_fail, "pass_rate_pct": pass_rate, "status": "PASS" if pass_rate >= 95 else "FAIL", "failures": failures, "worker_results": list(results_bucket.values()), } # Print summary status_icon = "✅" if summary["status"] == "PASS" else "❌" print(f"\n{'─'*60}") print(f" {status_icon} QA RESULT: {summary['status']}") print(f" Pass rate: {pass_rate}% ({total_pass}/{total_tests} tests)") print(f" Duration: {total_time}s") if failures: print(f"\n FAILURES ({len(failures)}):") for f in failures[:10]: # Show first 10 print(f" ✗ [{f['viewport']:8s}] {f['test']}: {f['detail']}") print(f"{'─'*60}\n") # Write to Redis (AIVA reads this) _write_results_to_redis(story_id, summary) return summary def _write_results_to_redis(story_id: str, summary: dict): """Write QA results to Redis for AIVA to read.""" try: import redis r = redis.from_url(REDIS_URL, decode_responses=True) r.set(f"genesis:qa_results:{story_id}", json.dumps(summary), ex=86400) # 24h TTL r.set(f"genesis:qa_status:{story_id}", summary["status"], ex=86400) r.lpush("genesis:qa_completed", story_id) print(f" 📡 Results written to Redis: genesis:qa_results:{story_id}") except Exception as e: print(f" ⚠️ Redis write failed: {e} (results printed above)") # ─── Connection Test ────────────────────────────────────────────────────────── def test_browserless_connection() -> bool: """Test connectivity to Elestio Browserless service.""" print(f"\n🔍 Testing Elestio Browserless Connection") print(f" Service: browserless-genesis (DO Sydney)") print(f" URL: {BROWSERLESS_HTTP}") try: # Test HTTP pressure endpoint token_param = f"?token={BROWSERLESS_TOKEN}" if BROWSERLESS_TOKEN else "" resp = requests.get( f"{BROWSERLESS_HTTP}/pressure{token_param}", timeout=10 ) if resp.status_code == 200: data = resp.json() print(f" ✅ Connected!") print(f" Running sessions: {data.get('running', 0)}") print(f" Queued: {data.get('queued', 0)}") print(f" Max concurrent: {data.get('maxConcurrent', 'N/A')}") return True else: print(f" ⚠️ HTTP {resp.status_code} — service may need TOKEN") print(f" Set BROWSERLESS_TOKEN env var from Elestio dashboard") return False except requests.exceptions.ConnectionError: print(f" ❌ Connection refused — check BROWSERLESS_WS_URL") print(f" Current URL: {BROWSERLESS_HTTP}") print(f"\n TO FIX:") print(f" 1. Elestio dashboard → browserless-genesis → Env Vars") print(f" 2. Copy TOKEN value") print(f" 3. export BROWSERLESS_TOKEN=") print(f" 4. Verify URL: browserless-genesis-[id].vm.elestio.app") return False except Exception as e: print(f" ❌ Error: {e}") return False # ─── CLI Entry Point ────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Genesis Browser QA Army — Elestio Browserless", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: Test connection to Elestio Browserless: python scripts/browserless_qa_army.py --test-connection Run QA on a URL: python scripts/browserless_qa_army.py --url https://sunaivadigital.com Run QA tied to a story ID (AIVA integration): python scripts/browserless_qa_army.py --url https://example.com --story-id STORY_001 """ ) parser.add_argument("--test-connection", action="store_true", help="Test connection to Elestio Browserless") parser.add_argument("--url", type=str, default="https://sunaivadigital.com", help="Target URL to test") parser.add_argument("--story-id", type=str, default="manual", help="Story ID for Redis result key (AIVA reads this)") parser.add_argument("--workers", type=int, default=None, help=f"Override max concurrent workers (default: {MAX_CONCURRENT_SESSIONS})") args = parser.parse_args() if args.workers: global MAX_CONCURRENT_SESSIONS MAX_CONCURRENT_SESSIONS = args.workers if args.test_connection: success = test_browserless_connection() sys.exit(0 if success else 1) # Run the QA army results = asyncio.run(run_qa_army(args.url, args.story_id)) # Exit code: 0=PASS, 1=FAIL (AIVA checks this) sys.exit(0 if results["status"] == "PASS" else 1) if __name__ == "__main__": main()