The AI QA Engineer: Test Every User Flow After Every Deploy

[
  {
    "name": "Signup flow",
    "base_url": "https://staging.yourapp.com",
    "steps": [
      "Navigate to /signup",
      "Fill in email with 'test-{timestamp}@example.com'",
      "Fill in password with 'TestPassword123!'",
      "Click the 'Create Account' button",
      "Verify: the page shows a welcome message or redirects to /dashboard",
      "Screenshot: the dashboard after signup"
    ],
    "success_criteria": "User lands on dashboard with their email visible"
  },
  {
    "name": "Login + dashboard",
    "base_url": "https://staging.yourapp.com",
    "steps": [
      "Navigate to /login",
      "Enter email 'qa-user@yourcompany.com' and password 'QaPass456!'",
      "Click 'Sign In'",
      "Verify: dashboard loads with at least one widget or data panel visible",
      "Click 'Settings' in the navigation",
      "Verify: settings page loads with user profile section",
      "Screenshot: the settings page"
    ],
    "success_criteria": "Login succeeds, dashboard and settings both load correctly"
  },
  {
    "name": "Billing + checkout",
    "base_url": "https://staging.yourapp.com",
    "steps": [
      "Login as 'qa-user@yourcompany.com'",
      "Navigate to /billing",
      "Verify: current plan is shown",
      "Click 'Upgrade' or 'Change Plan'",
      "Verify: pricing table or plan selection appears",
      "Select the 'Pro' plan if available",
      "Verify: checkout form or Stripe widget loads",
      "Screenshot: the checkout page (do NOT submit payment)"
    ],
    "success_criteria": "Billing page loads, upgrade flow reaches checkout without errors"
  },
  {
    "name": "Mobile viewport",
    "base_url": "https://staging.yourapp.com",
    "viewport": {"width": 375, "height": 812},
    "steps": [
      "Navigate to /",
      "Verify: mobile navigation (hamburger menu) is visible",
      "Click the hamburger menu",
      "Verify: navigation drawer opens with links",
      "Screenshot: the mobile nav open state",
      "Navigate to /login",
      "Verify: login form is usable on mobile (no overflow, no hidden fields)"
    ],
    "success_criteria": "All pages render correctly at mobile viewport"
  }
]

import asyncio, json, time, os, base64, httpx, anthropic
from playwright.async_api import async_playwright

SANDBOX_URL = os.environ["SANDBOX_URL"]
API_KEY = os.environ["SANDBOX_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
claude = anthropic.Anthropic()


async def create_browser(client, viewport=None):
    resp = await client.post(f"{SANDBOX_URL}/api/browsers", headers=HEADERS, json={
        "browser_type": "chrome", "enable_debug_port": True,
        "idle_timeout_minutes": 10,
    })
    return resp.json()


async def wait_assigned(client, cid):
    for _ in range(40):
        s = (await client.get(f"{SANDBOX_URL}/api/containers/{cid}/status",
            headers=HEADERS)).json()
        if s["state"] == "assigned": return s
        if s["state"] == "failed": raise RuntimeError("Container failed")
        await asyncio.sleep(3)
    raise TimeoutError("Browser not ready")


async def run_flow(client, flow: dict) -> dict:
    """Run a single user flow in its own browser container."""
    flow_name = flow["name"]
    print(f"[{flow_name}] Starting...")
    screenshots = []

    browser_data = await create_browser(client)
    cid = browser_data["container_id"]

    try:
        status = await wait_assigned(client, cid)
        cdp_url = status["cdp_url"]

        async with async_playwright() as p:
            browser = await p.chromium.connect_over_cdp(cdp_url)
            ctx = browser.contexts[0]
            page = await ctx.new_page()

            viewport = flow.get("viewport", {"width": 1280, "height": 720})
            await page.set_viewport_size(viewport)

            # Execute each step using vision
            for i, step in enumerate(flow["steps"]):
                print(f"  [{flow_name}] Step {i+1}: {step[:60]}...")
                result = await execute_step(page, step, flow["base_url"])

                if result["action"] == "screenshot" or step.lower().startswith("screenshot"):
                    img = await page.screenshot()
                    screenshots.append({
                        "step": i + 1,
                        "description": step,
                        "image": base64.b64encode(img).decode(),
                    })

                if result.get("failed"):
                    # Screenshot on failure
                    img = await page.screenshot()
                    screenshots.append({
                        "step": i + 1, "description": f"FAILURE: {step}",
                        "image": base64.b64encode(img).decode(),
                    })
                    return {
                        "flow": flow_name, "status": "FAIL",
                        "failed_step": i + 1, "reason": result["reason"],
                        "screenshots": screenshots,
                    }

        return {
            "flow": flow_name, "status": "PASS",
            "screenshots": screenshots,
        }

    except Exception as e:
        return {"flow": flow_name, "status": "ERROR", "reason": str(e), "screenshots": screenshots}
    finally:
        await client.post(f"{SANDBOX_URL}/api/containers/{cid}/release", headers=HEADERS)


async def execute_step(page, step: str, base_url: str) -> dict:
    """Use Claude vision to execute one test step."""
    # Handle direct navigation steps without vision
    if step.lower().startswith("navigate to"):
        path = step.split("navigate to", 1)[1].strip().strip("'").strip('"')
        url = f"{base_url}{path}" if path.startswith("/") else path
        await page.goto(url, wait_until="networkidle", timeout=15000)
        await page.wait_for_timeout(1500)
        return {"action": "navigate"}

    # Screenshot-only steps
    if step.lower().startswith("screenshot"):
        return {"action": "screenshot"}

    # Vision-guided steps (fill, click, verify)
    img = await page.screenshot()
    b64 = base64.b64encode(img).decode()

    response = claude.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {
                    "type": "base64", "media_type": "image/png", "data": b64}},
                {"type": "text", "text": f"""You are a QA test agent. Execute this test step:

{step}

Look at the screenshot. Respond with ONE action:
  CLICK x y        — click at pixel coordinates
  TYPE text         — type into the focused input
  KEY combo         — press keys (e.g., KEY Tab, KEY Return)
  VERIFY_PASS note  — the verification passes, explain why
  VERIFY_FAIL note  — the verification fails, explain why
  WAIT seconds      — wait for content to load

Reply with ONLY the action line."""},
            ],
        }],
    )

    action = response.content[0].text.strip()

    if action.startswith("CLICK"):
        _, x, y = action.split()
        await page.mouse.click(int(x), int(y))
        await page.wait_for_timeout(2000)
        return {"action": "click"}
    elif action.startswith("TYPE"):
        await page.keyboard.type(action[5:], delay=30)
        return {"action": "type"}
    elif action.startswith("KEY"):
        await page.keyboard.press(action[4:])
        return {"action": "key"}
    elif action.startswith("VERIFY_PASS"):
        return {"action": "verify", "note": action[12:]}
    elif action.startswith("VERIFY_FAIL"):
        return {"action": "verify", "failed": True, "reason": action[12:]}
    elif action.startswith("WAIT"):
        await page.wait_for_timeout(int(action.split()[1]) * 1000)
        return {"action": "wait"}

    return {"action": "unknown"}

async def run_qa_suite(flows_file: str = "flows.json"):
    with open(flows_file) as f:
        flows = json.load(f)

    print(f"Running {len(flows)} flows in parallel...")
    start = time.time()

    async with httpx.AsyncClient(timeout=300) as client:
        tasks = [run_flow(client, flow) for flow in flows]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    elapsed = time.time() - start
    passed = [r for r in results if isinstance(r, dict) and r["status"] == "PASS"]
    failed = [r for r in results if isinstance(r, dict) and r["status"] == "FAIL"]
    errors = [r for r in results if isinstance(r, dict) and r["status"] == "ERROR"]

    # Build Slack report
    status_icon = "✅" if not failed and not errors else "❌"
    report = f"""{status_icon} *QA Run Complete* ({elapsed:.0f}s)

*Passed:* {len(passed)}/{len(flows)}
*Failed:* {len(failed)}
*Errors:* {len(errors)}
"""

    for f in failed:
        report += f"\n❌ *{f['flow']}* — Step {f['failed_step']}: {f['reason']}"

    for e in errors:
        report += f"\n⚠️ *{e['flow']}* — {e['reason']}"

    print(report)

    # Save results + screenshots
    with open("qa_results.json", "w") as f:
        json.dump([r for r in results if isinstance(r, dict)], f, indent=2)

    return results

# In your CI pipeline, after deploy to staging:
SANDBOX=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name":"qa-run","instance_type":"ab0t.micro","auto_stop_minutes":10}')
SID=$(echo "$SANDBOX" | jq -r '.sandbox_id')

# Wait for boot, then run QA
sleep 45
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/execute" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"command\": \"cd /workspace && git clone https://github.com/your-org/qa-flows.git . && pip install httpx playwright anthropic && python qa_orchestrator.py\"}"

# Cleanup
curl -s -X DELETE "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $SANDBOX_API_KEY"

{
  "name": "Desktop app — main workflow",
  "container_type": "desktop",
  "desktop_type": "ubuntu-xfce",
  "setup": "sudo dpkg -i /workspace/myapp.deb && myapp &",
  "steps": [
    "Wait for the application window to appear",
    "Click 'File' menu, then 'New Project'",
    "Type 'QA Test Project' in the project name field",
    "Click 'Create'",
    "Verify: project workspace opens with empty canvas",
    "Screenshot: the new project workspace"
  ]
}