xiangqinxiaochengxu/backend/app/scripts/smoke_test.py

from __future__ import annotations

import asyncio
import os
from pathlib import Path
from typing import Any

import httpx

BASE_URL = os.getenv("BASE_URL", "http://127.0.0.1:8000/api/v1")
USER_TOKEN = os.getenv("USER_TOKEN", "")
ACTIVITY_ID = os.getenv("ACTIVITY_ID", "")
SHARE_TOKEN = os.getenv("SHARE_TOKEN", "")


async def fetch(
    client: httpx.AsyncClient,
    path: str,
    *,
    method: str = "GET",
    json: dict | None = None,
    files: dict | None = None,
    auth: bool = False,
) -> tuple[int, Any]:
    headers = {}
    if auth and USER_TOKEN:
        headers["Authorization"] = f"Bearer {USER_TOKEN}"

    res = await client.request(method, path, json=json, files=files, headers=headers)
    try:
        body = res.json()
    except Exception:
        body = res.text
    return res.status_code, body


async def main() -> int:
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=20.0) as client:
        checks: list[tuple[str, str, dict[str, Any]]] = [
            ("health", "/health", {}),
            ("public-config", "/auth/public-config", {}),
            ("announcements", "/announcements?page_size=1", {}),
            ("activities", "/activities?page_size=1", {"auth": True}),
        ]

        if SHARE_TOKEN:
            checks.append(("activity-share", f"/activities/share/{SHARE_TOKEN}", {"auth": bool(USER_TOKEN)}))
        if ACTIVITY_ID:
            checks.extend(
                [
                    ("activity-detail", f"/activities/{ACTIVITY_ID}", {"auth": True}),
                    ("activity-match-state", f"/activities/{ACTIVITY_ID}/match-state", {"auth": True}),
                    ("activity-candidates", f"/activities/{ACTIVITY_ID}/candidates", {"auth": True}),
                    ("activity-my-choices", f"/activities/{ACTIVITY_ID}/my-choices", {"auth": True}),
                    ("activity-liked-me", f"/activities/{ACTIVITY_ID}/liked-me", {"auth": True}),
                    ("activity-mutual-matches", f"/activities/{ACTIVITY_ID}/mutual-matches", {"auth": True}),
                ]
            )

        failed = False
        for name, path, kwargs in checks:
            status_code, body = await fetch(client, path, **kwargs)
            ok = status_code == 200
            print(f"{name}: {'OK' if ok else 'FAIL'} {status_code} {body}")
            failed = failed or not ok

        if USER_TOKEN:
            status_code, body = await fetch(client, "/users/me", auth=True)
            ok = status_code == 200
            print(f"users-me: {'OK' if ok else 'FAIL'} {status_code} {body}")
            failed = failed or not ok

        avatar_path = Path(__file__).resolve().parent / "fixtures" / "avatar_demo.jpg"
        if USER_TOKEN and avatar_path.exists():
            files = {"file": (avatar_path.name, avatar_path.read_bytes(), "image/jpeg")}
            status_code, body = await fetch(client, "/upload/avatar", method="POST", files=files, auth=True)
            ok = status_code == 200
            print(f"upload-avatar: {'OK' if ok else 'FAIL'} {status_code} {body}")
            failed = failed or not ok

            status_code, body = await fetch(client, "/upload/profile-image", method="POST", files=files, auth=True)
            ok = status_code == 200
            print(f"upload-profile-image: {'OK' if ok else 'FAIL'} {status_code} {body}")
            failed = failed or not ok

        return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))