""" 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 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) def count_pending_by_user(self, user_id): return 0 def count_pending_total(self): return 0 def list_by_user_filtered(self, user_id, *, status=None, limit=None, offset=0): items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id] if status: items = [t for t in items if getattr(t, "status", None) == status] if offset: items = items[offset:] if limit is not None: items = items[:limit] return items def count_by_user_filtered(self, user_id, *, status=None): items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id] if status: items = [t for t in items if getattr(t, "status", None) == status] return len(items) def list_by_project_filtered(self, project_id, *, status=None, limit=None, offset=0): items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id] if status: items = [t for t in items if getattr(t, "status", None) == status] if offset: items = items[offset:] if limit is not None: items = items[:limit] return items def count_by_project_filtered(self, project_id, *, status=None): items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id] if status: items = [t for t in items if getattr(t, "status", None) == status] return len(items) 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()