tests/test_api.py - master - asigatchov/vb-ai-api

tests/test_api.py
327 строк13 KB
Alexander Sigatchov
api rally actions
10 мар 2026, 09:38
10 мар 2026, 09:3895f0449
import sys
import json
from pathlib import Path
from uuid import uuid4

from fastapi.testclient import TestClient
from sqlalchemy import select

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import main
from src.database.models import Project, Rally


def auth_headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def create_user(client: TestClient) -> str:
    response = client.post(
        "/api/auth/register",
        json={"name": "Test User", "email": "test@example.com", "password": "secret123"},
    )
    assert response.status_code == 200
    return response.json()["token"]


def test_full_api_flow(tmp_path: Path):
    db_path = tmp_path / "test.db"
    uploads_dir = tmp_path / "uploads"
    uploads_dir.mkdir()

    main.UPLOADS_DIR = uploads_dir
    app = main.create_app(db_path=db_path)

    with TestClient(app) as client:
        token = create_user(client)
        headers = auth_headers(token)

        me = client.get("/api/auth/me", headers=headers)
        assert me.status_code == 200
        assert me.json()["email"] == "test@example.com"

        folder = client.post("/api/folders", json={"name": "Training"}, headers=headers)
        assert folder.status_code == 200
        folder_id = folder.json()["id"]

        files = {"file": ("match.mp4", b"fake-video-content", "video/mp4")}
        data = {"title": "Quarterfinal", "folder": folder_id}
        project = client.post("/api/projects", headers=headers, files=files, data=data)
        assert project.status_code == 200
        project_id = project.json()["id"]
        assert project.json()["folder"] == "Training"
        assert project.json()["thumbnailUrl"].startswith(f"/uploads/{me.json()['id']}/")

        chunk_a = b"chunk-a-"
        chunk_b = b"chunk-b"
        chunk_upload_id = None
        for index, payload in enumerate((chunk_a, chunk_b)):
            chunk_files = {"file": (f"part-{index}.bin", payload, "application/octet-stream")}
            chunk_data = {
                "title": "Chunked Match",
                "folder": folder_id,
                "chunkIndex": str(index),
                "totalChunks": "2",
                "originalFileName": "chunked.mp4",
            }
            if chunk_upload_id:
                chunk_data["uploadId"] = chunk_upload_id
            chunk_response = client.post("/api/projects/chunks", headers=headers, files=chunk_files, data=chunk_data)
            assert chunk_response.status_code == 200
            body = chunk_response.json()
            chunk_upload_id = body["uploadId"]
            assert body["chunkIndex"] == index
            assert body["totalChunks"] == 2
            assert body["uploadedChunks"] == index + 1
            if index == 0:
                assert body["completed"] is False
                assert body["project"] is None
            else:
                assert body["completed"] is True
                assert body["project"]["title"] == "Chunked Match"
                assert body["project"]["folder"] == "Training"

        annotations_payload = {
            "keypoints": [
                {"id": 1, "name": "left_shoulder", "x": 120.5, "y": 80.0, "visible": True},
                {"id": 2, "name": "right_shoulder", "x": 200.5, "y": 81.5, "visible": False},
            ],
            "bbox": [100, 60, 220, 300],
            "imageWidth": 1280,
            "imageHeight": 720,
        }
        save_annotations = client.post(
            f"/api/projects/{project_id}/annotations",
            json=annotations_payload,
            headers=headers,
        )
        assert save_annotations.status_code == 200
        assert save_annotations.json()["success"] is True
        uploads_root = Path(app.state.uploads_dir)
        court_json_path = uploads_root / me.json()["id"] / project_id / "court.json"
        assert court_json_path.exists()
        court_payload = json.loads(court_json_path.read_text(encoding="utf-8"))
        assert court_payload["images"][0]["width"] == 1280

        export = client.get(f"/api/projects/{project_id}/export", headers=headers)
        assert export.status_code == 200
        export_body = export.json()
        assert export_body["images"][0]["width"] == 1280
        assert export_body["annotations"][0]["num_keypoints"] == 1

        rally = client.post(
            f"/api/projects/{project_id}/rallies",
            json={
                "startFrame": 100,
                "endFrame": 180,
                "actions": [
                    {"actionType": "serve", "startFrame": 100, "endFrame": 115, "description": "Serve"},
                    {"actionType": "reception", "startFrame": 116, "endFrame": 130, "description": "Reception"},
                ],
            },
            headers=headers,
        )
        assert rally.status_code == 200
        rally_id = rally.json()["id"]
        assert rally.json()["type"] == "unknown"
        assert len(rally.json()["actions"]) == 2

        update_rally = client.put(
            f"/api/rallies/{rally_id}",
            json={"description": "Updated note", "type": "attack"},
            headers=headers,
        )
        assert update_rally.status_code == 200
        assert update_rally.json()["description"] == "Updated note"
        assert update_rally.json()["type"] == "attack"

        create_action = client.post(
            f"/api/rallies/{rally_id}/actions",
            json={"actionType": "block", "startFrame": 150, "endFrame": 155, "description": "Block"},
            headers=headers,
        )
        assert create_action.status_code == 200
        action_id = create_action.json()["id"]

        update_action = client.put(
            f"/api/rallies/actions/{action_id}",
            json={"actionType": "defense", "description": "Dig"},
            headers=headers,
        )
        assert update_action.status_code == 200
        assert update_action.json()["actionType"] == "defense"
        assert update_action.json()["description"] == "Dig"

        list_actions = client.get(f"/api/rallies/{rally_id}/actions", headers=headers)
        assert list_actions.status_code == 200
        assert len(list_actions.json()) == 3

        delete_action = client.delete(f"/api/rallies/actions/{action_id}", headers=headers)
        assert delete_action.status_code == 200
        assert delete_action.json()["success"] is True

        tracks_dir = uploads_root / me.json()["id"] / project_id / "tracks"
        tracks_dir.mkdir(parents=True)
        (tracks_dir / "track_0008.json").write_text(
            '{"startFrame": 200, "endFrame": 260, "positions": [[[11.5, 22.0], 210], [[13.0, 23.0], 211]]}'
        )
        (tracks_dir / "track_0013.json").write_text('{"frames": {"start": 300, "end": 390}}')
        (tracks_dir / "track_0015.json").write_text('{"start_frame": 542, "last_frame": 800, "fps": 30.0}')
        (tracks_dir / "broken.json").write_text("{}")

        import_rallies = client.post(
            f"/api/projects/{project_id}/rallies/import-tracks",
            json={"replaceExisting": False},
            headers=headers,
        )
        assert import_rallies.status_code == 200
        assert import_rallies.json()["imported"] == 3
        assert import_rallies.json()["skipped"] == 0
        rallies_after_import = client.get(f"/api/projects/{project_id}/rallies", headers=headers)
        assert rallies_after_import.status_code == 200
        imported_rally = next(row for row in rallies_after_import.json() if row["description"] == "0008")
        assert imported_rally["ballPositions"] == [
            {"frame": 210, "x": 11.5, "y": 22.0},
            {"frame": 211, "x": 13.0, "y": 23.0},
        ]

        db = app.state.session_factory()
        try:
            stored_rally = db.scalar(
                select(Rally).where(
                    Rally.project_id == project_id,
                    Rally.description == "0008",
                )
            )
            assert stored_rally is not None
            assert stored_rally.track_json is not None
            assert json.loads(stored_rally.track_json)["positions"]

            db.add(
                Rally(
                    id=str(uuid4()),
                    project_id=project_id,
                    start_frame=200,
                    end_frame=260,
                    type="unknown",
                    description="0008",
                    track_json=None,
                )
            )
            db.commit()
        finally:
            db.close()

        rallies_with_fallback = client.get(f"/api/projects/{project_id}/rallies", headers=headers)
        assert rallies_with_fallback.status_code == 200
        fallback_rally = next(row for row in rallies_with_fallback.json() if row["description"] == "0008")
        assert fallback_rally["ballPositions"] == [
            {"frame": 210, "x": 11.5, "y": 22.0},
            {"frame": 211, "x": 13.0, "y": 23.0},
        ]

        delete_rally = client.delete(f"/api/rallies/{rally_id}", headers=headers)
        assert delete_rally.status_code == 200
        assert delete_rally.json()["success"] is True

        logout = client.post("/api/auth/logout", headers=headers)
        assert logout.status_code == 200
        after_logout = client.get("/api/auth/me", headers=headers)
        assert after_logout.status_code == 401


def test_chunk_upload_stores_video_metadata_in_project(tmp_path: Path):
    db_path = tmp_path / "test.db"
    uploads_dir = tmp_path / "uploads"
    uploads_dir.mkdir()

    source_video = Path(__file__).resolve().parent / "video" / "f2d66b33-19dd-468c-ac3b-354ff03dfd01.mp4"
    assert source_video.exists()

    main.UPLOADS_DIR = uploads_dir
    app = main.create_app(db_path=db_path)

    with TestClient(app) as client:
        token = create_user(client)
        headers = auth_headers(token)
        me = client.get("/api/auth/me", headers=headers)
        assert me.status_code == 200

        upload_id = None
        chunk_size = 512 * 1024
        video_bytes = source_video.read_bytes()
        chunks = [video_bytes[i : i + chunk_size] for i in range(0, len(video_bytes), chunk_size)]
        assert len(chunks) > 1

        final_body = None
        for index, payload in enumerate(chunks):
            chunk_files = {"file": (f"part-{index}.bin", payload, "application/octet-stream")}
            chunk_data = {
                "title": "Chunked Real Video",
                "chunkIndex": str(index),
                "totalChunks": str(len(chunks)),
                "originalFileName": source_video.name,
            }
            if upload_id:
                chunk_data["uploadId"] = upload_id

            response = client.post("/api/projects/chunks", headers=headers, files=chunk_files, data=chunk_data)
            assert response.status_code == 200
            body = response.json()
            upload_id = body["uploadId"]
            assert body["chunkIndex"] == index
            assert body["totalChunks"] == len(chunks)
            assert body["uploadedChunks"] == index + 1

            if index < len(chunks) - 1:
                assert body["completed"] is False
                assert body["project"] is None
            else:
                assert body["completed"] is True
                assert body["project"] is not None
                final_body = body

        assert final_body is not None
        project_payload = final_body["project"]
        assert isinstance(project_payload["fps"], (int, float))
        assert project_payload["fps"] > 0
        assert isinstance(project_payload["frameWidth"], int)
        assert project_payload["frameWidth"] > 0
        assert isinstance(project_payload["frameHeight"], int)
        assert project_payload["frameHeight"] > 0

        project_id = project_payload["id"]
        project_response = client.get(f"/api/projects/{project_id}", headers=headers)
        assert project_response.status_code == 200
        reloaded = project_response.json()
        assert reloaded["fps"] == project_payload["fps"]
        assert reloaded["frameWidth"] == project_payload["frameWidth"]
        assert reloaded["frameHeight"] == project_payload["frameHeight"]

        db = app.state.session_factory()
        try:
            stored_project = db.scalar(select(Project).where(Project.id == project_id))
            assert stored_project is not None
            assert stored_project.fps == project_payload["fps"]
            assert stored_project.frame_width == project_payload["frameWidth"]
            assert stored_project.frame_height == project_payload["frameHeight"]
        finally:
            db.close()


def test_public_reels_endpoint_is_accessible_without_auth(tmp_path: Path):
    db_path = tmp_path / "test.db"
    uploads_dir = tmp_path / "uploads"
    uploads_dir.mkdir()

    main.UPLOADS_DIR = uploads_dir
    app = main.create_app(db_path=db_path)

    with TestClient(app) as client:
        response = client.get("/api/projects/reels/public?page=1&pageSize=3")
        assert response.status_code == 200
        body = response.json()
        assert body["page"] == 1
        assert body["pageSize"] == 20
        assert body["total"] == 0
        assert body["items"] == []