canned-prompts/service/tools/smoke.py

#!/usr/bin/env python3
"""Service-level smoke checks.

The checks a `rapp` smoke contract can assert about a *running service*.
Deployment-level checks — NetworkPolicies present, external secrets ready,
private-Service-only — are properties of the cluster, not of this process, and
belong to `rapp-canned-prompts` rather than here.

Exits non-zero if any required check fails, so a deployment gate can call it
directly. stdlib only: this runs inside the runtime image, which carries no
test dependencies.
"""

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass


@dataclass
class Check:
    name: str
    ok: bool
    detail: str


def get(base: str, path: str, timeout: float) -> tuple[int, dict]:
    request = urllib.request.Request(base.rstrip("/") + path, method="GET")
    request.add_header("Accept", "application/json")
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        try:
            return exc.code, json.loads(exc.read().decode("utf-8"))
        except Exception:  # noqa: BLE001
            return exc.code, {}
    except Exception as exc:  # noqa: BLE001
        return 0, {"error": str(exc)}


def run(base: str, expect_migration: str | None, timeout: float) -> list[Check]:
    checks: list[Check] = []

    status, body = get(base, "/healthz", timeout)
    checks.append(
        Check("liveness-ok", status == 200 and body.get("status") == "ok",
              f"{status} {body}")
    )

    status, body = get(base, "/readyz", timeout)
    ready = status == 200 and body.get("ready") is True
    checks.append(Check("readiness-ok", ready, f"{status} {body.get('detail')}"))

    status, body = get(base, "/state/health", timeout)
    checks.append(
        Check("state-health-ok", status == 200 and body.get("status") == "ok",
              f"{status} {body.get('db')}")
    )

    migration = body.get("migration")
    if expect_migration:
        checks.append(
            Check("migration-at-head", migration == expect_migration,
                  f"running {migration}, expected {expect_migration}")
        )
    else:
        # Without an expected revision this can only confirm the schema is
        # stamped at all — say so rather than implying it verified the head.
        checks.append(
            Check("migration-stamped", bool(migration),
                  f"running {migration} (no --expect-migration given)")
        )

    status, body = get(base, "/index", timeout)
    checks.append(Check("index-queryable", status == 200 and "entries" in body,
                        f"{status}"))

    status, body = get(base, "/packages", timeout)
    checks.append(Check("registry-queryable", status == 200 and "packages" in body,
                        f"{status}"))
    return checks


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base", default="http://127.0.0.1:8000")
    parser.add_argument("--expect-migration", default=None,
                        help="revision the deployed schema should be at")
    parser.add_argument("--timeout", type=float, default=5.0)
    args = parser.parse_args(argv)

    checks = run(args.base, args.expect_migration, args.timeout)
    width = max(len(c.name) for c in checks)
    for check in checks:
        print(f"{'PASS' if check.ok else 'FAIL'}  {check.name:<{width}}  {check.detail}")

    failed = [c.name for c in checks if not c.ok]
    if failed:
        print(f"\n{len(failed)} check(s) failed: {', '.join(failed)}", file=sys.stderr)
        return 1
    print(f"\nall {len(checks)} checks passed")
    return 0


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