xiaoxia-saas/scripts/smoke_public_upload_flow.py

"""Public upload smoke for the SaaS Web-domain API path.

This intentionally uses https://saas.xiaoxiajianji.com/api/v1 so the Web nginx
proxy is tested together with the API upload path.
"""

from __future__ import annotations

import io
import os
import time
import uuid

import requests

BASE_URL = "https://saas.xiaoxiajianji.com/api/v1"
PASSWORD = os.environ.get("SMOKE_TEST_PASSWORD", "changeme")


def _json_or_raise(name: str, response: requests.Response) -> dict:
    print(f"{name}={response.status_code}")
    if response.status_code >= 400:
        print(response.text[:1000])
    response.raise_for_status()
    if not response.content:
        return {}
    return response.json()


def main() -> None:
    session = requests.Session()
    suffix = uuid.uuid4().hex[:10]
    email = f"smoke-upload-{suffix}@example.com"
    username = f"smoke_upload_{suffix}"

    _json_or_raise(
        "register",
        session.post(
            f"{BASE_URL}/auth/register",
            json={"email": email, "username": username, "password": PASSWORD, "full_name": "Upload Smoke"},
            timeout=30,
        ),
    )
    login = _json_or_raise(
        "login",
        session.post(f"{BASE_URL}/auth/login", json={"email": email, "password": PASSWORD}, timeout=30),
    )
    headers = {"Authorization": f"Bearer {login['access_token']}"}

    workspace = _json_or_raise(
        "workspace",
        session.post(f"{BASE_URL}/workspaces", json={"name": "Upload Smoke Workspace"}, headers=headers, timeout=30),
    )

    project = _json_or_raise(
        "project",
        session.post(
            f"{BASE_URL}/projects",
            headers=headers,
            timeout=30,
        ),
    )
    project_id = project["id"]

    library = _json_or_raise(
        "library",
        session.post(
            f"{BASE_URL}/asset-libraries",
            json={
                "project_id": project_id,
                "name": "Smoke Video Library",
                "kind": "video",
            },
            headers=headers,
            timeout=30,
        ),
    )
    library_id = library["id"]

    upload = _json_or_raise(
        "upload",
        session.post(
            f"{BASE_URL}/upload",
            files={"file": ("smoke.txt", io.BytesIO(b"xiaoxia upload smoke"), "text/plain")},
            headers=headers,
            timeout=60,
        ),
    )

    job_id = upload["ingest_job_id"]
    latest_job = {}
    for attempt in range(10):
        latest_job = _json_or_raise(
            f"ingest_{attempt}",
            session.get(f"{BASE_URL}/ingest-jobs/{job_id}", headers=headers, timeout=30),
        )
        if latest_job.get("status") in {"completed", "failed"}:
            break
        time.sleep(1)

    if latest_job.get("status") == "failed":
        raise RuntimeError(f"ingest job failed: {latest_job.get('error_message')}")

    print("public_upload_flow=ok")


if __name__ == "__main__":
    main()