import asyncio from dataclasses import dataclass from httpx import ASGITransport, AsyncClient from app.main import app @dataclass class CheckResult: method: str path: str status: int async def run_checks() -> list[CheckResult]: results: list[CheckResult] = [] transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://testserver") as client: login = await client.post( "/api/v1/admin/auth/login", json={"account": "13900000001", "password": "demo123456"}, ) if login.status_code != 200: raise RuntimeError(f"admin login failed: {login.status_code} {login.text}") token = login.json()["access_token"] headers = {"Authorization": f"Bearer {token}"} checks: list[tuple[str, str]] = [ ("GET", "/api/v1/admin/module-design"), ("GET", "/api/v1/admin/spots?page=1&page_size=5"), ("GET", "/api/v1/admin/events?page=1&page_size=5"), ("GET", "/api/v1/admin/shooting?page=1&page_size=5"), ("GET", "/api/v1/admin/promotions?page=1&page_size=5"), ("GET", "/api/v1/admin/app-nav-configs?page=1&page_size=5"), ("GET", "/api/v1/admin/membership/plans?page=1&page_size=5"), ("GET", "/api/v1/admin/membership/user-memberships?page=1&page_size=5"), ("GET", "/api/v1/admin/points/ledger?page=1&page_size=5"), ("GET", "/api/v1/admin/notifications?page=1&page_size=5"), ("GET", "/api/v1/admin/reports?page=1&page_size=5"), ("GET", "/api/v1/admin/audit-logs?page=1&page_size=5"), ("GET", "/api/v1/admin/system-configs?page=1&page_size=5"), ("GET", "/api/v1/admin/user-options"), ("GET", "/api/v1/admin/spot-tag-options"), ("GET", "/api/v1/admin/promotion-link-options?link_type=spot"), ] for method, path in checks: resp = await client.request(method, path, headers=headers) results.append(CheckResult(method=method, path=path, status=resp.status_code)) for list_path, detail_template in [ ("/api/v1/admin/spots?page=1&page_size=1", "/api/v1/admin/spots/{id}"), ("/api/v1/admin/events?page=1&page_size=1", "/api/v1/admin/events/{id}"), ("/api/v1/admin/shooting?page=1&page_size=1", "/api/v1/admin/shooting/{id}"), ]: list_resp = await client.get(list_path, headers=headers) results.append(CheckResult(method="GET", path=list_path, status=list_resp.status_code)) if list_resp.status_code != 200: continue items = list_resp.json().get("items", []) if not items: continue detail_path = detail_template.format(id=items[0]["id"]) detail_resp = await client.get(detail_path, headers=headers) results.append(CheckResult(method="GET", path=detail_path, status=detail_resp.status_code)) return results async def main(): results = await run_checks() has_error = False for item in results: line = f"{item.method} {item.path} -> {item.status}" print(line) if item.status >= 400: has_error = True if has_error: raise SystemExit(1) if __name__ == "__main__": asyncio.run(main())