generated from coulomb/repo-seed
Add console visual checks
This commit is contained in:
379
scripts/check_console_visual.py
Executable file
379
scripts/check_console_visual.py
Executable file
@@ -0,0 +1,379 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user