"""Hermetic browser tests for the header status pill + panel. /api/cdn is mocked with page.route — the throwaway server's probe thread runs against LIVE upstream, so its real data must never leak into these assertions. Everything else (login, home) hits the throwaway server, which never touches the real data. Skips if playwright is absent. """ import asyncio import json import pytest async_playwright = pytest.importorskip("playwright.async_api").async_playwright BASE = "http://127.0.0.1:8399" LAUNCH_ARGS = ["--autoplay-policy=no-user-gesture-required"] POLL = 100 def _cdn(state, kbps, origin="OK"): return { "state": state, "kbps": kbps, "ts": 1750000000, "hist": [{"t": 1750000000 - 180 * (10 - i), "kbps": 20 + i * 10} for i in range(11)], "origin": {"state": origin, "msg": "http 200", "ts": 1750000000}, } async def _open(pw, body): """logged-in home page with /api/cdn mocked. alice has no history, so home renders without touching the live upstream.""" browser = await pw.chromium.launch(args=LAUNCH_ARGS) page = await (await browser.new_context()).new_page() await page.route("**/api/cdn", lambda r: r.fulfill(status=200, body=json.dumps(body), headers={"Content-Type": "application/json"})) await page.goto(BASE + "/") await page.wait_for_selector("input#pw") await page.fill("#user", "alice") await page.fill("#pw", "alicepw12345") await page.keyboard.press("Enter") await page.wait_for_selector("#q", timeout=30000) return browser, page def _js_click(page, sel): # JS click: this playwright/chromium combo's actionability check # ("element stable") hangs non-deterministically on header elements return page.evaluate(f"document.querySelector({json.dumps(sel)}).click()") def test_pill_ok(): asyncio.run(_pill_ok()) async def _pill_ok(): async with async_playwright() as pw: browser, page = await _open(pw, _cdn("OK", 120)) try: await page.wait_for_function( "document.querySelector('#pill .c').className.includes('ok')", timeout=5000, polling=POLL) assert await page.text_content("#pill span") == "up" # tap opens the panel: detail rows + sparkline with one point # per hist sample await _js_click(page, "#pill") await page.wait_for_selector("#status-panel .sp-svg polyline") pts = await page.get_attribute("#status-panel .sp-svg polyline", "points") assert len(pts.split()) == 11 assert "CDN: OK — 120 KB/s" in await page.text_content("#status-panel") # tap again closes; so does a click away await _js_click(page, "#pill") assert not await page.locator("#status-panel").count() await _js_click(page, "#pill") await page.wait_for_selector("#status-panel") await _js_click(page, "body") assert not await page.locator("#status-panel").count() finally: await browser.close() def test_pill_degraded(): asyncio.run(_pill_degraded()) async def _pill_degraded(): async with async_playwright() as pw: browser, page = await _open(pw, _cdn("DEGRADED", 23)) try: await page.wait_for_function( "document.querySelector('#pill .c').className.includes('degraded')", timeout=5000, polling=POLL) assert await page.text_content("#pill span") == "cdn 23 KB/s" await _js_click(page, "#pill") await page.wait_for_selector("#status-panel") assert "CDN: DEGRADED — 23 KB/s" in await page.text_content("#status-panel") finally: await browser.close() def test_pill_origin_down(): asyncio.run(_pill_origin_down()) async def _pill_origin_down(): async with async_playwright() as pw: browser, page = await _open(pw, _cdn("OK", 120, origin="DOWN")) try: await page.wait_for_function( "document.querySelector('#pill .o').className.includes('down')", timeout=5000, polling=POLL) assert await page.text_content("#pill span") == "upstream down" cls = await page.get_attribute("#pill", "class") assert "odown" in cls finally: await browser.close()