ewoooc/scripts/check_observability_pages.py

#!/usr/bin/env python3
"""Production smoke check for AI observability pages.

The goal is to catch broken observability pages quickly after UI, route,
schema, or deployment changes. It verifies the ten war-room pages return HTTP
200 and do not expose raw framework/database errors to the user.
"""

from __future__ import annotations

import argparse
import sys
import urllib.error
import urllib.request

from observability_contract import CSS_ASSET_CHECKS, OBSERVABILITY_PAGES

MIN_HTML_BYTES = 2500

HEALTH_CHECK = (
    "/health",
    "App Health",
    ("healthy",),
)

GLOBAL_REQUIRED_MARKERS = [
    "momo-observability-mode",
    "observability-system.css",
    "AI 觀測台",
    "momo-obs-link",
]

ERROR_NEEDLES = [
    "Traceback",
    "UndefinedError",
    "ProgrammingError",
    "Internal Server Error",
    'relation "',
    "relation "",
    "查詢失敗:",
]


def fetch_page(base_url: str, path: str, timeout: int) -> tuple[int, str]:
    request = urllib.request.Request(
        base_url.rstrip("/") + path,
        headers={"User-Agent": "momo-observability-smoke/1.0"},
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        html = response.read().decode("utf-8", "ignore")
        return response.status, html


def main() -> int:
    parser = argparse.ArgumentParser(description="Smoke check AI observability pages")
    parser.add_argument("--base-url", default="https://mo.wooo.work")
    parser.add_argument("--timeout", type=int, default=12)
    args = parser.parse_args()

    failed = False
    print(f"Observability page smoke: {args.base_url.rstrip('/')}")

    health_path, health_label, health_markers = HEALTH_CHECK
    try:
        status, body = fetch_page(args.base_url, health_path, args.timeout)
        missing_markers = [marker for marker in health_markers if marker not in body]
        if status != 200 or missing_markers:
            issues = []
            if status != 200:
                issues.append("bad_status")
            if missing_markers:
                issues.append(f"missing_health_markers={missing_markers}")
            print(f"- {health_label}: HTTP {status}, issues={issues}, FAIL")
            failed = True
        else:
            print(f"- {health_label}: HTTP {status}, markers=ok")
    except urllib.error.HTTPError as exc:
        print(f"- {health_label}: HTTP {exc.code}, FAIL")
        failed = True
    except Exception as exc:
        print(f"- {health_label}: {type(exc).__name__}: {exc}, FAIL")
        failed = True

    for page in OBSERVABILITY_PAGES:
        try:
            status, html = fetch_page(args.base_url, page.url, args.timeout)
        except urllib.error.HTTPError as exc:
            print(f"- {page.short_label}: HTTP {exc.code}, FAIL")
            failed = True
            continue
        except Exception as exc:
            print(f"- {page.short_label}: {type(exc).__name__}: {exc}, FAIL")
            failed = True
            continue

        found = [needle for needle in ERROR_NEEDLES if needle in html]
        missing_markers = [
            marker for marker in page.markers
            if marker not in html
        ]
        missing_global_markers = [
            marker for marker in GLOBAL_REQUIRED_MARKERS
            if marker not in html
        ]
        too_small = len(html.encode("utf-8")) < MIN_HTML_BYTES
        if status != 200 or found or missing_markers or missing_global_markers or too_small:
            issues = []
            if status != 200:
                issues.append("bad_status")
            if found:
                issues.extend(found)
            if missing_markers:
                issues.append(f"missing_markers={missing_markers}")
            if missing_global_markers:
                issues.append(f"missing_global_markers={missing_global_markers}")
            if too_small:
                issues.append(f"html_too_small={len(html.encode('utf-8'))}B")
            print(f"- {page.short_label}: HTTP {status}, issues={issues}, FAIL")
            failed = True
        else:
            print(f"- {page.short_label}: HTTP {status}, issues=none")

    for asset in CSS_ASSET_CHECKS:
        try:
            status, body = fetch_page(args.base_url, asset.url, args.timeout)
        except urllib.error.HTTPError as exc:
            print(f"- {asset.label}: HTTP {exc.code}, FAIL")
            failed = True
            continue
        except Exception as exc:
            print(f"- {asset.label}: {type(exc).__name__}: {exc}, FAIL")
            failed = True
            continue

        missing_markers = [marker for marker in asset.markers if marker not in body]
        if status != 200 or missing_markers:
            issues = []
            if status != 200:
                issues.append("bad_status")
            if missing_markers:
                issues.append(f"missing_asset_markers={missing_markers}")
            print(f"- {asset.label}: HTTP {status}, issues={issues}, FAIL")
            failed = True
        else:
            print(f"- {asset.label}: HTTP {status}, markers=ok")

    if failed:
        print("Observability page smoke: FAIL")
        return 1

    print("Observability page smoke: PASS")
    return 0


if __name__ == "__main__":
    sys.exit(main())