scripts/smoke_admin_api.py

#!/usr/bin/env python3
import argparse
import os
import sys
from dataclasses import dataclass

import httpx


@dataclass
class Result:
    name: str
    method: str
    path: str
    status: str
    ok: bool
    detail: str | None = None
def _print_results(results: list[Result]) -> None:
    for item in results:
        status = "OK" if item.ok else "FAIL"
        print(f"{status:4} {item.method:6} {item.path:50} {item.status}")
        if item.detail:
            print(f"      {item.detail}")


def _request(
    client: httpx.Client,
    results: list[Result],
    name: str,
    method: str,
    path: str,
    headers: dict[str, str],
    params: dict[str, str] | None = None,
) -> httpx.Response | None:
    try:
        response = client.request(method, path, headers=headers, params=params)
    except Exception as exc:  # pragma: no cover - runtime/network error
        results.append(Result(name=name, method=method, path=path, status="ERR", ok=False, detail=str(exc)))
        return None

    ok = response.status_code == 200
    detail = None
    if not ok:
        snippet = response.text[:200].replace("\n", " ")
        detail = f"{response.status_code} {snippet}"
    results.append(
        Result(
            name=name,
            method=method,
            path=path,
            status=str(response.status_code),
            ok=ok,
            detail=detail,
        )
    )
    return response


def main() -> int:
    parser = argparse.ArgumentParser(description="Smoke test admin API endpoints.")
    parser.add_argument("--base-url", default=os.getenv("BASE_URL", "http://127.0.0.1:7860"))
    parser.add_argument("--email", default=os.getenv("ADMIN_EMAIL"))
    parser.add_argument("--password", default=os.getenv("ADMIN_PASSWORD"))
    parser.add_argument("--group-id", default=os.getenv("ADMIN_GROUP_ID"))
    parser.add_argument("--timeout", type=float, default=10.0)
    args = parser.parse_args()

    if not args.email or not args.password:
        print("Missing credentials. Provide --email/--password or set ADMIN_EMAIL/ADMIN_PASSWORD.")
        return 1

    base_url = args.base_url
    results: list[Result] = []

    with httpx.Client(base_url=base_url, timeout=args.timeout) as client:
        login_res = client.post(
            "/auth/login",
            json={"email": args.email, "password": args.password},
        )
        if login_res.status_code != 200:
            snippet = login_res.text[:200].replace("\n", " ")
            print(f"Login failed: {login_res.status_code} {snippet}")
            return 1

        token = login_res.json().get("token")
        if not token:
            print("Login response missing token.")
            return 1

        headers = {"Authorization": f"Bearer {token}"}

        groups_res = _request(client, results, "groups.list", "GET", "/groups", headers)
        group_id = args.group_id
        if not group_id:
            if not groups_res:
                _print_results(results)
                return 1
            try:
                items = groups_res.json().get("items", [])
            except Exception:
                items = []
            if items:
                group_id = items[0].get("id")

        if not group_id:
            _print_results(results)
            print("No group_id available. Create a group in /admin/groups or pass --group-id.")
            return 2

        gid = group_id
        endpoints = [
            ("group.get", "GET", f"/groups/{gid}", None),
            ("config.get", "GET", f"/groups/{gid}/config", None),
            ("ad_filter.get", "GET", f"/groups/{gid}/ad-filter/config", None),
            ("ad_filter.reports", "GET", f"/groups/{gid}/ad-filter/reports", {"range": "day"}),
            ("spam.get", "GET", f"/groups/{gid}/spam/config", None),
            ("verification.get", "GET", f"/groups/{gid}/verification/config", None),
            ("auto_replies.list", "GET", f"/groups/{gid}/auto-replies", None),
            ("ban_rules.list", "GET", f"/groups/{gid}/ban-rules", None),
            ("delete_rules.list", "GET", f"/groups/{gid}/delete-rules", None),
            ("posts.list", "GET", f"/groups/{gid}/posts", None),
            ("posts.search", "GET", f"/groups/{gid}/posts/search", None),
            ("schedules.list", "GET", f"/groups/{gid}/schedules", None),
            ("broadcasts.list", "GET", f"/groups/{gid}/broadcasts", None),
            ("logs.actions", "GET", f"/groups/{gid}/logs/actions", None),
            ("logs.deletions", "GET", f"/groups/{gid}/logs/deletions", None),
            ("logs.admin", "GET", f"/groups/{gid}/logs/admin", None),
            ("stats.messages", "GET", f"/groups/{gid}/stats/messages", {"range": "day"}),
            ("stats.wordcloud", "GET", f"/groups/{gid}/stats/wordcloud", {"range": "day"}),
            ("invites.list", "GET", f"/groups/{gid}/invites", None),
            ("invites.rankings", "GET", f"/groups/{gid}/invites/rankings", {"range": "day"}),
            ("verified_users.list", "GET", f"/groups/{gid}/verified-users", None),
            ("verified_users.forms", "GET", f"/groups/{gid}/verified-users/forms", None),
            ("verified_users.reviews", "GET", f"/groups/{gid}/verified-users/reviews", None),
            ("verified_users.nearby", "GET", f"/groups/{gid}/verified-users/nearby", None),
            ("member_events.list", "GET", f"/groups/{gid}/member-events", None),
        ]

        for name, method, path, params in endpoints:
            _request(client, results, name, method, path, headers, params=params)

    _print_results(results)
    failed = [item for item in results if not item.ok]
    if failed:
        print(f"Failures: {len(failed)}")
        return 1
    return 0


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