xiaoxia-saas/tests/unit/test_generation_presigned_url.py

"""
P0-2 修复测试：generation_tasks results 端点返回 OSS 预签名 URL

验证 list_generation_results 端点：
- 对每个生成视频的 file_url 调用 storage_service.get_download_url()
- 返回的 download_url 是预签名临时 URL（24h 有效期）
- 与 generated_videos.py 中的模式一致
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from app.api.routes.generation_tasks import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import OSSStorageService, get_storage_service
from app.dependencies import (
    get_generated_video_repository,
    get_generation_task_repository,
    get_project_repository,
)
from fastapi import FastAPI
from fastapi.testclient import TestClient

from packages.domain import (
    GeneratedVideo,
    GenerationTask,
    GenerationTaskStatus,
    Project,
    User,
)

# ---------------------------------------------------------------------------
# Stub repositories
# ---------------------------------------------------------------------------


class StubProjectRepository:
    def __init__(self, projects=None):
        self._projects = projects or {}

    def find_by_id(self, project_id):
        return self._projects.get(project_id)


class StubGenerationTaskRepository:
    def __init__(self, tasks=None):
        self._tasks = tasks or {}

    def get(self, task_id):
        return self._tasks.get(task_id)


class StubGeneratedVideoRepository:
    def __init__(self, videos=None):
        self._videos = videos or {}

    def list_by_generation_task(self, task_id):
        return [v for v in self._videos.values() if v.generation_task_id == task_id]


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


def _make_app(task, videos, storage_mock):
    app = FastAPI()
    app.include_router(router, prefix="/api/v1/generation")

    user = User(id="user-1", email="test@test.com", display_name="Test", username="testuser")
    auth = AuthenticatedUser(user=user, token_type="access")

    project = Project(id="project-1", name="Test", description="", owner_user_id="user-1")

    app.dependency_overrides[get_current_user] = lambda: auth
    app.dependency_overrides[get_project_repository] = lambda: StubProjectRepository({"project-1": project})
    app.dependency_overrides[get_generation_task_repository] = lambda: StubGenerationTaskRepository({"task-1": task})
    app.dependency_overrides[get_generated_video_repository] = lambda: StubGeneratedVideoRepository(videos)
    app.dependency_overrides[get_storage_service] = lambda: storage_mock

    return app


def test_results_endpoint_generates_presigned_urls():
    """验证 list_generation_results 为每个视频生成预签名 download_url"""
    task = GenerationTask(
        id="task-1",
        project_id="project-1",
        asset_library_id="lib-1",
        strategy_id="s1",
        status=GenerationTaskStatus.COMPLETED,
        progress=100,
        result_count=2,
        created_by_user_id="user-1",
    )

    videos = {
        "v1": GeneratedVideo.create(
            project_id="project-1",
            generation_task_id="task-1",
            name="video1.mp4",
            file_url="https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/v1.mp4",
            file_size=1024,
            duration=5.0,
            width=1280,
            height=720,
            fps=25.0,
        ),
        "v2": GeneratedVideo.create(
            project_id="project-1",
            generation_task_id="task-1",
            name="video2.mp4",
            file_url="https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/v2.mp4",
            file_size=2048,
            duration=10.0,
            width=1920,
            height=1080,
            fps=30.0,
        ),
    }

    storage_mock = MagicMock(spec=OSSStorageService)
    storage_mock.get_download_url.side_effect = (
        lambda url, expires_seconds=3600: f"{url}?signature=presigned&expires={expires_seconds}"
    )

    app = _make_app(task, videos, storage_mock)
    client = TestClient(app)

    response = client.get("/api/v1/generation/tasks/task-1/results")
    assert response.status_code == 200

    data = response.json()
    assert len(data["items"]) == 2

    # Verify presigned URLs were generated with 24h expiry
    assert storage_mock.get_download_url.call_count == 2
    for call in storage_mock.get_download_url.call_args_list:
        assert call.kwargs["expires_seconds"] == 86400

    # Verify download_url is present in response
    for item in data["items"]:
        assert item["download_url"] is not None
        assert "signature=presigned" in item["download_url"]
        assert "expires=86400" in item["download_url"]

    # Verify file_url is still the original (raw) URL
    assert data["items"][0]["file_url"] == videos["v1"].file_url


def test_results_endpoint_handles_empty_videos():
    """验证无视频时正常返回空列表"""
    task = GenerationTask(
        id="task-1",
        project_id="project-1",
        asset_library_id="lib-1",
        strategy_id="s1",
        status=GenerationTaskStatus.RUNNING,
        progress=50,
        created_by_user_id="user-1",
    )

    storage_mock = MagicMock(spec=OSSStorageService)
    app = _make_app(task, {}, storage_mock)
    client = TestClient(app)

    response = client.get("/api/v1/generation/tasks/task-1/results")
    assert response.status_code == 200
    assert response.json()["items"] == []
    storage_mock.get_download_url.assert_not_called()