xiaoxia-saas/scripts/smoke_public_boundary_flow.py

from __future__ import annotations

import io
import time
import uuid

import os
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[:500])
    response.raise_for_status()
    return response.json() if response.content else {}


def _register_login(session: requests.Session, label: str) -> dict:
    suffix = uuid.uuid4().hex[:10]
    email = f"smoke-boundary-{label}-{suffix}@example.com"
    username = f"smoke_boundary_{label}_{suffix}"
    _json_or_raise(
        f"{label}_register",
        session.post(
            f"{BASE_URL}/auth/register",
            json={"email": email, "username": username, "password": PASSWORD, "full_name": "Boundary Smoke"},
            timeout=30,
        ),
    )
    login = _json_or_raise(
        f"{label}_login",
        session.post(f"{BASE_URL}/auth/login", json={"email": email, "password": PASSWORD}, timeout=30),
    )
    return {"Authorization": f"Bearer {login['access_token']}"}


def main() -> None:
    owner = requests.Session()
    intruder = requests.Session()
    owner_headers = _register_login(owner, "owner")
    intruder_headers = _register_login(intruder, "intruder")

    workspace = _json_or_raise(
        "owner_workspace",
        owner.post(f"{BASE_URL}/workspaces", json={"name": "Boundary Workspace"}, headers=owner_headers, timeout=30),
    )
    project = _json_or_raise(
        "owner_project",
        owner.post(
            f"{BASE_URL}/projects",
            headers=owner_headers,
            timeout=30,
        ),
    )
    project_id = project["id"]
    library = _json_or_raise(
        "owner_library",
        owner.post(
            f"{BASE_URL}/asset-libraries",
            headers=owner_headers,
            timeout=30,
        ),
    )
    library_id = library["id"]
    upload = _json_or_raise(
        "owner_upload",
        owner.post(
            f"{BASE_URL}/upload",
            files={"file": ("boundary.mp4", io.BytesIO(b"boundary-owner"), "video/mp4")},
            headers=owner_headers,
            timeout=60,
        ),
    )
    ingest_id = upload.get("ingest_job_id")
    for index in range(5):
        if not ingest_id:
            break
        ingest = owner.get(f"{BASE_URL}/ingest-jobs/{ingest_id}", headers=owner_headers, timeout=30)
        print(f"owner_ingest_{index}={ingest.status_code}")
        if ingest.status_code == 200 and ingest.json().get("result_asset_id"):
            break
        time.sleep(1)

    task = _json_or_raise(
        "owner_generation_create",
        owner.post(
            f"{BASE_URL}/generation/tasks",
            headers=owner_headers,
            timeout=30,
        ),
    )
    task_id = task["id"]
    video_id = ""
    for index in range(8):
        task_response = owner.get(f"{BASE_URL}/generation/tasks/{task_id}", headers=owner_headers, timeout=30)
        print(f"owner_generation_{index}={task_response.status_code}")
        results_response = owner.get(
            f"{BASE_URL}/generation/tasks/{task_id}/results", headers=owner_headers, timeout=30
        )
        print(f"owner_generation_results_{index}={results_response.status_code}")
        if results_response.status_code == 200:
            items = results_response.json().get("items", [])
            if items:
                video_id = items[0]["id"]
                break
        time.sleep(2)

    checks = [
        ("anon_me", requests.get(f"{BASE_URL}/auth/me", timeout=30).status_code, {401, 403}),
        ("anon_workspaces", requests.get(f"{BASE_URL}/workspaces", timeout=30).status_code, {401, 403}),
        (
            "anon_project_get",
            requests.get(f"{BASE_URL}/projects/{project_id}", timeout=30).status_code,
            {401, 403, 404},
        ),
        (
            "intruder_project_get",
            intruder.get(f"{BASE_URL}/projects/{project_id}", headers=intruder_headers, timeout=30).status_code,
            {403, 404},
        ),
        (
            "intruder_library_get",
            intruder.get(f"{BASE_URL}/asset-libraries/{library_id}", headers=intruder_headers, timeout=30).status_code,
            {403, 404},
        ),
        (
            "anon_assets",
            requests.get(f"{BASE_URL}/assets", params={"library_id": library_id}, timeout=30).status_code,
            {401, 403, 404},
        ),
        (
            "intruder_assets",
            intruder.get(
                f"{BASE_URL}/assets", params={"library_id": library_id}, headers=intruder_headers, timeout=30
            ).status_code,
            {403, 404},
        ),
        (
            "anon_generation_task",
            requests.get(f"{BASE_URL}/generation/tasks/{task_id}", timeout=30).status_code,
            {401, 403, 404},
        ),
        (
            "intruder_generation_task",
            intruder.get(f"{BASE_URL}/generation/tasks/{task_id}", headers=intruder_headers, timeout=30).status_code,
            {403, 404},
        ),
        (
            "intruder_generation_results",
            intruder.get(
                f"{BASE_URL}/generation/tasks/{task_id}/results", headers=intruder_headers, timeout=30
            ).status_code,
            {403, 404},
        ),
    ]
    if video_id:
        checks.extend(
            [
                (
                    "intruder_video_get",
                    intruder.get(
                        f"{BASE_URL}/generated-videos/{video_id}", headers=intruder_headers, timeout=30
                    ).status_code,
                    {403, 404},
                ),
                (
                    "intruder_video_download",
                    intruder.get(
                        f"{BASE_URL}/generated-videos/{video_id}/download-url", headers=intruder_headers, timeout=30
                    ).status_code,
                    {403, 404},
                ),
            ]
        )
    upload_response = intruder.post(
        f"{BASE_URL}/upload",
        files={"file": ("boundary.txt", io.BytesIO(b"boundary"), "text/plain")},
        headers=intruder_headers,
        timeout=60,
    )
    checks.append(("intruder_upload", upload_response.status_code, {403, 404}))

    failed = []
    for name, status_code, expected in checks:
        print(f"{name}={status_code}")
        if status_code not in expected:
            failed.append(f"{name} expected {sorted(expected)} got {status_code}")
    if failed:
        raise RuntimeError("; ".join(failed))
    print("public_boundary_flow=ok")


if __name__ == "__main__":
    main()