core-hub/scripts/check_console_visual.py

#!/usr/bin/env python3
from __future__ import annotations

import os
import socket
import struct
import sys
import tempfile
import threading
import time
from pathlib import Path
from typing import Any

import httpx
import uvicorn

try:
    from playwright.sync_api import sync_playwright
except ImportError as exc:  # pragma: no cover - exercised only when dev deps are missing
    raise SystemExit(
        "Playwright is required for visual checks. Run `uv sync --extra dev` and "
        "`uv run --extra dev playwright install chromium`."
    ) from exc

OPERATOR_TOKEN = "operator-token"
OPERATOR_HEADERS = {"Authorization": f"Bearer {OPERATOR_TOKEN}"}
DEFAULT_OUTPUT_DIR = Path(".local/visual-checks/console")
VIEWPORTS = {
    "desktop": {"width": 1440, "height": 1100},
    "mobile": {"width": 390, "height": 1100},
}

LAYOUT_CHECK = """
({ fullKey, keyPrefix }) => {
  const failures = [];
  const viewportWidth = window.innerWidth;
  const doc = document.documentElement;

  if (doc.scrollWidth > viewportWidth + 1) {
    failures.push(`document horizontal overflow: ${doc.scrollWidth}px > ${viewportWidth}px`);
  }

  const rectFor = (label, selector) => {
    const element = document.querySelector(selector);
    if (!element) {
      failures.push(`missing element: ${label} (${selector})`);
      return null;
    }
    const rect = element.getBoundingClientRect();
    if (rect.width <= 0 || rect.height <= 0) {
      failures.push(`empty box: ${label}`);
      return null;
    }
    return {
      label,
      selector,
      left: rect.left,
      right: rect.right,
      top: rect.top,
      bottom: rect.bottom,
      width: rect.width,
      height: rect.height,
    };
  };

  const overlaps = (a, b) => (
    a.left < b.right - 1 && a.right > b.left + 1 &&
    a.top < b.bottom - 1 && a.bottom > b.top + 1
  );

  const compareNonOverlap = (items) => {
    for (let i = 0; i < items.length; i += 1) {
      for (let j = i + 1; j < items.length; j += 1) {
        if (items[i] && items[j] && overlaps(items[i], items[j])) {
          failures.push(`overlap: ${items[i].label} intersects ${items[j].label}`);
        }
      }
    }
  };

  compareNonOverlap([
    rectFor("top navigation", ".wn-topnav"),
    rectFor("sidebar", ".wn-sidebar"),
    rectFor("main", ".console-main"),
  ]);

  compareNonOverlap([
    rectFor("overview", "#overview"),
    rectFor("counters", ".kpi-grid"),
    rectFor("readiness gates", "section[aria-labelledby='gates-heading']"),
    rectFor("registry", "#registry"),
    rectFor("migration", "#migration"),
    rectFor("access", "#access"),
    rectFor("events", "#events"),
  ]);

  compareNonOverlap([
    rectFor("top navigation brand", ".wn-topnav__brand"),
    rectFor("top navigation state", ".wn-topnav__right"),
  ]);

  const viewportBoundedSelectors = [
    ".wn-topnav",
    ".wn-sidebar",
    ".console-main",
    ".wn-page-header",
    ".kpi-grid",
    ".wn-card",
    ".gate",
    ".gate-list",
    ".table-wrap",
    ".section-head",
    ".wn-btn",
  ];
  for (const selector of viewportBoundedSelectors) {
    for (const element of document.querySelectorAll(selector)) {
      const rect = element.getBoundingClientRect();
      if (rect.left < -1 || rect.right > viewportWidth + 1) {
        const bounds = `${Math.round(rect.left)}..${Math.round(rect.right)}`;
        failures.push(`${selector} escapes viewport: ${bounds} of ${viewportWidth}`);
      }
    }
  }

  const intrinsicSelectors = [
    ".wn-btn",
    ".wn-tag",
    ".wn-sidebar__item",
    ".stat-card span",
    ".stat-card strong",
    ".wn-topnav__brand",
    ".wn-topnav__right",
  ];
  for (const selector of intrinsicSelectors) {
    for (const element of document.querySelectorAll(selector)) {
      if (element.scrollWidth > Math.ceil(element.clientWidth) + 1) {
        failures.push(`${selector} text overflows its own box`);
      }
    }
  }

  const openApi = document.querySelector(".wn-btn[href='/api/v2/openapi.json']");
  if (!openApi) {
    failures.push("missing OpenAPI control");
  } else {
    const rect = openApi.getBoundingClientRect();
    if (rect.width < 44 || rect.height < 32) {
      failures.push("OpenAPI control is too small to operate comfortably");
    }
  }

  const text = document.body.innerText;
  if (!text.includes("Operator console") || !text.includes("ops-hub")) {
    failures.push("expected console content is not visible");
  }
  if (!text.includes(keyPrefix)) {
    failures.push("non-secret key prefix is not visible in access table");
  }
  if (text.includes(fullKey) || document.documentElement.outerHTML.includes(fullKey)) {
    failures.push("full API key leaked into rendered console");
  }
  if (document.documentElement.outerHTML.includes("fullKey")) {
    failures.push("fullKey field name leaked into rendered console");
  }

  return failures;
}
"""


def configure_environment(database_path: Path) -> None:
    os.environ["CORE_HUB_DATABASE_URL"] = f"sqlite+aiosqlite:///{database_path}"
    os.environ["CORE_HUB_API_TOKEN"] = OPERATOR_TOKEN
    os.environ["CORE_HUB_AUTO_CREATE_TABLES"] = "1"

    from core_hub.config import get_settings
    from core_hub.db import get_engine, get_sessionmaker

    get_settings.cache_clear()
    get_engine.cache_clear()
    get_sessionmaker.cache_clear()


def reset_environment_caches() -> None:
    from core_hub.config import get_settings
    from core_hub.db import get_engine, get_sessionmaker

    get_settings.cache_clear()
    get_engine.cache_clear()
    get_sessionmaker.cache_clear()


def start_server() -> tuple[uvicorn.Server, threading.Thread, str]:
    from core_hub.app import create_app

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    host, port = sock.getsockname()
    config = uvicorn.Config(create_app(), log_level="warning", lifespan="on")
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
    thread.start()

    base_url = f"http://{host}:{port}"
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        try:
            response = httpx.get(f"{base_url}/healthz", timeout=0.5)
        except httpx.HTTPError:
            time.sleep(0.1)
            continue
        if response.status_code == 200:
            return server, thread, base_url
        time.sleep(0.1)

    server.should_exit = True
    thread.join(timeout=5)
    raise RuntimeError("Timed out waiting for the visual check server to start")


def stop_server(server: uvicorn.Server, thread: threading.Thread) -> None:
    server.should_exit = True
    thread.join(timeout=5)


def seed_console_data(base_url: str) -> dict[str, Any]:
    with httpx.Client(base_url=base_url, headers=OPERATOR_HEADERS, timeout=5) as client:
        hub = client.post(
            "/api/v2/hubs",
            json={
                "slug": "ops-hub",
                "name": "Ops Hub",
                "domain": "ops.coulomb.social",
                "hubKind": "domain",
                "hubFamily": "vsm",
                "vsmFunction": "OPS",
                "vsmSystem": "1",
            },
        ).raise_for_status().json()
        manifest = client.post(
            "/api/v2/hub-capability-manifests",
            json={"hubId": hub["id"], "manifestVersion": "1.0"},
        ).raise_for_status().json()
        client.post(f"/api/v2/hub-capability-manifests/{manifest['id']}/activate").raise_for_status()
        consumer = client.post(
            "/api/v2/api-consumers",
            json={
                "name": "ops-hub",
                "description": "API consumer for the VSM Operations hub",
                "hubCapabilityManifestId": manifest["id"],
                "rateLimitPerMinute": 120,
            },
        ).raise_for_status().json()
        key_payload = client.post(
            f"/api/v2/api-consumers/{consumer['id']}/api-keys",
            json={"scopes": "framework:read hub:ops-hub:write"},
        ).raise_for_status().json()

    runtime_headers = {"Authorization": f"Bearer {key_payload['fullKey']}"}
    with httpx.Client(base_url=base_url, headers=runtime_headers, timeout=5) as client:
        widget = client.post(
            "/api/v2/widgets",
            json={
                "hubId": hub["id"],
                "name": "Readiness",
                "widgetType": "ops-readiness-gate",
                "capabilityRef": "ops:readiness",
                "viewContext": "ops-hub/readiness",
                "policyScope": "ops-registry",
            },
        ).raise_for_status().json()
        client.post(
            "/api/v2/interaction-events",
            json={
                "widgetId": widget["id"],
                "eventType": "ops-endpoint-verified",
                "metadata": {"expectedStatus": 401},
            },
        ).raise_for_status()

    return key_payload


def png_width(path: Path) -> int:
    with path.open("rb") as handle:
        header = handle.read(24)
    if len(header) < 24 or header[:8] != b"\x89PNG\r\n\x1a\n":
        raise RuntimeError(f"{path} is not a valid PNG screenshot")
    return struct.unpack(">II", header[16:24])[0]


def check_protected_route(base_url: str) -> None:
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch()
        try:
            context = browser.new_context(viewport={"width": 390, "height": 800})
            page = context.new_page()
            response = page.goto(f"{base_url}/console", wait_until="domcontentloaded")
            status = response.status if response else None
            if status != 401:
                raise RuntimeError(f"Unauthenticated /console returned {status}, expected 401")
            context.close()
        finally:
            browser.close()


def check_viewports(base_url: str, key_payload: dict[str, Any], output_dir: Path) -> list[Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    screenshots: list[Path] = []
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch()
        try:
            for name, viewport in VIEWPORTS.items():
                context = browser.new_context(
                    viewport=viewport,
                    extra_http_headers=OPERATOR_HEADERS,
                    device_scale_factor=1,
                )
                page = context.new_page()
                response = page.goto(f"{base_url}/console", wait_until="networkidle")
                status = response.status if response else None
                if status != 200:
                    raise RuntimeError(f"Authenticated /console returned {status}, expected 200")

                failures = page.evaluate(
                    LAYOUT_CHECK,
                    {
                        "fullKey": key_payload["fullKey"],
                        "keyPrefix": key_payload["apiKey"]["keyPrefix"],
                    },
                )
                if failures:
                    joined = "\n - ".join(failures)
                    raise RuntimeError(f"{name} layout failures:\n - {joined}")

                screenshot = output_dir / f"console-{name}.png"
                page.screenshot(path=str(screenshot), full_page=True)
                if screenshot.stat().st_size < 1024:
                    raise RuntimeError(f"{screenshot} looks empty")
                width = png_width(screenshot)
                if width != viewport["width"]:
                    raise RuntimeError(
                        f"{screenshot} width is {width}px, expected {viewport['width']}px"
                    )
                screenshots.append(screenshot)
                context.close()
        finally:
            browser.close()
    return screenshots


def main() -> int:
    output_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_OUTPUT_DIR
    unauthorized_screenshot = output_dir / "console-unauthorized.png"
    if unauthorized_screenshot.exists():
        unauthorized_screenshot.unlink()

    with tempfile.TemporaryDirectory(prefix="core-hub-console-visual-") as tmp:
        configure_environment(Path(tmp) / "core-hub.db")
        server, thread, base_url = start_server()
        try:
            key_payload = seed_console_data(base_url)
            check_protected_route(base_url)
            screenshots = check_viewports(base_url, key_payload, output_dir)
        finally:
            stop_server(server, thread)
            reset_environment_caches()

    if unauthorized_screenshot.exists():
        raise RuntimeError("unauthorized console screenshot was created unexpectedly")

    print("Console visual checks passed.")
    for screenshot in screenshots:
        print(f"- {screenshot}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())