Files
CosScene/server/scripts/admin_endpoint_smoke.py
2026-05-09 16:40:29 +08:00

83 lines
3.3 KiB
Python

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())