灵应
6057ec18ce
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 27h6m12s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 27h6m12s
- test_edit_plans_api: 状态校验断言改为包含'无效'关键词 - test_edit_plan_service: can_generate断言适配新错误消息 - test_config_schemas_and_ai_endpoints: AI推荐端点断言适配 - test_video_upload_fix_and_generated_api: 直接检查generated_videos router路由， 绕过FastAPI 0.139.0 _IncludedRouter惰性包装
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""
|
||
测试视频上传失败排查修复：
|
||
1. chunked_upload ALLOWED_MIME_TYPES 与 upload.py 保持一致
|
||
2. generated_video_repository list 方法无 N+1 查询
|
||
3. 成片库 API 端点可正常返回数据
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime, timezone
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
# ─── 1. chunked_upload ALLOWED_MIME_TYPES 一致性 ───
|
||
|
||
|
||
class TestChunkedUploadMIMEConsistency:
|
||
"""chunked_upload.py 的 ALLOWED_MIME_TYPES 必须包含 upload.py 的所有类型。"""
|
||
|
||
def _get_upload_mime_types(self) -> set:
|
||
from apps.api.app.api.routes.upload import ALLOWED_MIME_TYPES
|
||
|
||
return set(ALLOWED_MIME_TYPES)
|
||
|
||
def _get_chunked_mime_types(self) -> set:
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
return set(ALLOWED_MIME_TYPES)
|
||
|
||
def test_chunked_upload_has_all_upload_mime_types(self):
|
||
"""chunked_upload 白名单必须覆盖 upload.py 的全部类型。"""
|
||
upload_types = self._get_upload_mime_types()
|
||
chunked_types = self._get_chunked_mime_types()
|
||
missing = upload_types - chunked_types
|
||
assert not missing, f"chunked_upload 缺少以下 MIME 类型: {missing}"
|
||
|
||
def test_chunked_upload_supports_video_mpeg(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "video/mpeg" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_video_matroska(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "video/x-matroska" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_video_3gpp(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "video/3gpp" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_audio_flac(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "audio/flac" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_audio_aac(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "audio/aac" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_audio_m4a(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "audio/x-m4a" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_audio_webm(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "audio/webm" in ALLOWED_MIME_TYPES
|
||
|
||
def test_chunked_upload_supports_extra_image_types(self):
|
||
from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
|
||
|
||
assert "image/bmp" in ALLOWED_MIME_TYPES
|
||
assert "image/tiff" in ALLOWED_MIME_TYPES
|
||
assert "image/svg+xml" in ALLOWED_MIME_TYPES
|
||
|
||
def test_both_have_same_core_video_types(self):
|
||
"""两条路径的核心视频类型必须一致。"""
|
||
upload_types = self._get_upload_mime_types()
|
||
chunked_types = self._get_chunked_mime_types()
|
||
core_video = {"video/mp4", "video/quicktime", "video/webm"}
|
||
for vt in core_video:
|
||
assert vt in upload_types, f"upload.py 缺少 {vt}"
|
||
assert vt in chunked_types, f"chunked_upload.py 缺少 {vt}"
|
||
|
||
|
||
# ─── 2. GeneratedVideo Repository N+1 修复验证 ───
|
||
|
||
|
||
class TestGeneratedVideoRepositoryNoNPlus1:
|
||
"""list_by_project 和 list_by_generation_task 应使用 _to_domain 而非 self.get。"""
|
||
|
||
def _make_model(self, video_id: str, project_id: str = "proj-1", task_id: str = "task-1"):
|
||
model = MagicMock()
|
||
model.id = video_id
|
||
model.project_id = project_id
|
||
model.generation_task_id = task_id
|
||
model.name = f"video-{video_id}.mp4"
|
||
model.file_url = f"https://oss.example.com/{video_id}.mp4"
|
||
model.file_size = 1024
|
||
model.duration = 5.0
|
||
model.thumbnail_url = None
|
||
model.width = 1280
|
||
model.height = 720
|
||
model.fps = 25.0
|
||
model.status = "completed"
|
||
model.review_status = "pending_review"
|
||
model.generation_params = json.dumps({"mode": "one_take"})
|
||
model.video_fingerprint = None
|
||
model.is_duplicate = False
|
||
model.duplicate_of = None
|
||
model.generated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||
model.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||
return model
|
||
|
||
def test_list_by_project_uses_to_domain(self):
|
||
"""list_by_project 不应调用 self.get（N+1），应使用 _to_domain。"""
|
||
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
||
SQLAlchemyGeneratedVideoRepository,
|
||
)
|
||
|
||
session = MagicMock()
|
||
models = [self._make_model(f"v{i}") for i in range(5)]
|
||
|
||
query_mock = MagicMock()
|
||
query_mock.filter.return_value.all.return_value = models
|
||
session.query.return_value = query_mock
|
||
|
||
repo = SQLAlchemyGeneratedVideoRepository(session)
|
||
with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
|
||
result = repo.list_by_project("proj-1")
|
||
mock_get.assert_not_called()
|
||
|
||
assert len(result) == 5
|
||
assert all(v.id.startswith("v") for v in result)
|
||
|
||
def test_list_by_generation_task_uses_to_domain(self):
|
||
"""list_by_generation_task 不应调用 self.get（N+1），应使用 _to_domain。"""
|
||
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
||
SQLAlchemyGeneratedVideoRepository,
|
||
)
|
||
|
||
session = MagicMock()
|
||
models = [self._make_model(f"v{i}", task_id="task-42") for i in range(3)]
|
||
|
||
query_mock = MagicMock()
|
||
query_mock.filter.return_value.all.return_value = models
|
||
session.query.return_value = query_mock
|
||
|
||
repo = SQLAlchemyGeneratedVideoRepository(session)
|
||
with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
|
||
result = repo.list_by_generation_task("task-42")
|
||
mock_get.assert_not_called()
|
||
|
||
assert len(result) == 3
|
||
|
||
def test_list_by_project_returns_empty_when_no_videos(self):
|
||
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
|
||
SQLAlchemyGeneratedVideoRepository,
|
||
)
|
||
|
||
session = MagicMock()
|
||
query_mock = MagicMock()
|
||
query_mock.filter.return_value.all.return_value = []
|
||
session.query.return_value = query_mock
|
||
|
||
repo = SQLAlchemyGeneratedVideoRepository(session)
|
||
result = repo.list_by_project("empty-project")
|
||
assert result == []
|
||
|
||
|
||
# ─── 3. 成片库 API 端点可用性确认 ───
|
||
|
||
|
||
class TestGeneratedVideosAPIAvailability:
|
||
"""确认成片库 API 路由注册正确，端点可正常返回数据。"""
|
||
|
||
def test_generated_videos_routes_registered(self):
|
||
"""成片库路由已注册到 router。"""
|
||
from app.api.routes.generated_videos import router as gv_router
|
||
|
||
# 直接检查 generated_videos router 自身注册的路由
|
||
paths = [r.path for r in gv_router.routes if hasattr(r, "path")]
|
||
assert len(paths) > 0, "generated_videos router 没有注册任何路由"
|
||
# 验证关键端点存在
|
||
assert "" in paths, "列表端点不存在"
|
||
assert "/{video_id}" in paths, "详情端点不存在"
|
||
|
||
def test_generated_videos_list_endpoint_exists(self):
|
||
"""GET /generated-videos 端点存在。"""
|
||
from apps.api.app.api.routes.generated_videos import router
|
||
|
||
paths = [r.path for r in router.routes if hasattr(r, "path")]
|
||
assert "" in paths, "GET /generated-videos 列表端点不存在"
|
||
|
||
def test_generated_videos_detail_endpoint_exists(self):
|
||
"""GET /generated-videos/{video_id} 端点存在。"""
|
||
from apps.api.app.api.routes.generated_videos import router
|
||
|
||
paths = [r.path for r in router.routes if hasattr(r, "path")]
|
||
assert "/{video_id}" in paths, "GET /generated-videos/{{video_id}} 详情端点不存在"
|
||
|
||
def test_generated_videos_review_endpoint_exists(self):
|
||
"""PATCH /generated-videos/{video_id}/review 端点存在。"""
|
||
from apps.api.app.api.routes.generated_videos import router
|
||
|
||
paths = [r.path for r in router.routes if hasattr(r, "path")]
|
||
assert "/{video_id}/review" in paths, "PATCH review 端点不存在"
|
||
|
||
def test_generated_videos_download_url_endpoint_exists(self):
|
||
"""GET /generated-videos/{video_id}/download-url 端点存在。"""
|
||
from apps.api.app.api.routes.generated_videos import router
|
||
|
||
paths = [r.path for r in router.routes if hasattr(r, "path")]
|
||
assert "/{video_id}/download-url" in paths, "download-url 端点不存在"
|
||
|
||
def test_generated_video_response_schema_complete(self):
|
||
"""GeneratedVideoResponse 包含所有必要字段。"""
|
||
from apps.api.app.schemas.generated_video import GeneratedVideoResponse
|
||
|
||
fields = GeneratedVideoResponse.model_fields
|
||
required_fields = [
|
||
"id",
|
||
"project_id",
|
||
"name",
|
||
"file_url",
|
||
"status",
|
||
"review_status",
|
||
"download_url",
|
||
]
|
||
for field in required_fields:
|
||
assert field in fields, f"GeneratedVideoResponse 缺少字段: {field}"
|
||
|
||
def test_list_generated_videos_response_schema(self):
|
||
"""ListGeneratedVideosResponse 包含 items 列表。"""
|
||
from apps.api.app.schemas.generated_video import ListGeneratedVideosResponse
|
||
|
||
fields = ListGeneratedVideosResponse.model_fields
|
||
assert "items" in fields, "ListGeneratedVideosResponse 缺少 items 字段"
|
||
|
||
def test_generation_task_results_endpoint_exists(self):
|
||
"""GET /generation/tasks/{task_id}/results 端点存在。"""
|
||
from apps.api.app.api.routes.generation_tasks import router
|
||
|
||
paths = [r.path for r in router.routes if hasattr(r, "path")]
|
||
assert "/tasks/{task_id}/results" in paths, "generation results 端点不存在"
|
||
|
||
|
||
# ─── 4. GeneratedVideo Use Cases 可用性 ───
|
||
|
||
|
||
class TestGeneratedVideoUseCases:
|
||
"""确认成片库 Use Case 层可正常工作。"""
|
||
|
||
def _make_video(self, video_id: str, project_id: str = "proj-1"):
|
||
from packages.domain import GeneratedVideo
|
||
|
||
return GeneratedVideo(
|
||
id=video_id,
|
||
project_id=project_id,
|
||
generation_task_id="task-1",
|
||
name=f"video-{video_id}.mp4",
|
||
file_url=f"https://oss.example.com/{video_id}.mp4",
|
||
file_size=1024,
|
||
duration=5.0,
|
||
width=1280,
|
||
height=720,
|
||
fps=25.0,
|
||
status="completed",
|
||
)
|
||
|
||
def test_list_generated_videos_use_case(self):
|
||
"""ListGeneratedVideosUseCase 可正常列出视频。"""
|
||
from packages.application import ListGeneratedVideosUseCase
|
||
|
||
repo = MagicMock()
|
||
videos = [self._make_video(f"v{i}") for i in range(3)]
|
||
repo.list_by_project.return_value = videos
|
||
|
||
use_case = ListGeneratedVideosUseCase(repo)
|
||
result = use_case.execute("proj-1")
|
||
|
||
repo.list_by_project.assert_called_once_with("proj-1")
|
||
assert len(result) == 3
|
||
|
||
def test_get_generated_video_use_case(self):
|
||
"""GetGeneratedVideoUseCase 可正常获取单个视频。"""
|
||
from packages.application import GetGeneratedVideoUseCase
|
||
|
||
repo = MagicMock()
|
||
video = self._make_video("v1")
|
||
repo.get.return_value = video
|
||
|
||
use_case = GetGeneratedVideoUseCase(repo)
|
||
result = use_case.execute("v1")
|
||
|
||
repo.get.assert_called_once_with("v1")
|
||
assert result.id == "v1"
|
||
|
||
def test_list_by_task_use_case(self):
|
||
"""ListGeneratedVideosByTaskUseCase 可按任务列出视频。"""
|
||
from packages.application import ListGeneratedVideosByTaskUseCase
|
||
|
||
repo = MagicMock()
|
||
videos = [self._make_video(f"v{i}") for i in range(2)]
|
||
repo.list_by_generation_task.return_value = videos
|
||
|
||
use_case = ListGeneratedVideosByTaskUseCase(repo)
|
||
result = use_case.execute("task-1")
|
||
|
||
repo.list_by_generation_task.assert_called_once_with("task-1")
|
||
assert len(result) == 2
|