130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
#!/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
|
|
|
|
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('/')}")
|
|
|
|
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())
|