xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 11 Pull Requests 4 Actions Packages Projects Releases 7 Wiki Activity

test: 添加核心模块单元测试 (config/upload/chunked_upload/asset_diagnosis) #122

Merged
xiaoxia merged 1 commits from test/unit-tests-core-modules into develop 2026-06-30 13:03:48 +08:00
Conversation 1 Commits 1 Files Changed 4
+1046 -4
4 changed files with 1046 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_asset_diagnosis.py
+142 -4
View File
@@ -1,11 +1,28 @@
import os
import sys
from pathlib import Path
from uuid import uuid4
# 设置必要环境变量（必须在导入 app 模块之前）
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"))
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.routes.asset_diagnosis import _build_diagnosis
from packages.domain import Asset, AssetStatus, ClassificationStatus
from packages.domain import (
Asset,
AssetLibrary,
AssetLibraryKind,
AssetStatus,
ClassificationStatus,
Project,
)
def _asset(name: str, mime_type: str, *, status=AssetStatus.READY, duration=None, quality_score=None):
@@ -25,7 +42,6 @@ def _asset(name: str, mime_type: str, *, status=AssetStatus.READY, duration=None
def test_asset_diagnosis_reports_missing_video_gap():
diagnosis = _build_diagnosis(
"workspace-1",
"project-1",
[_asset("voice.mp3", "audio/mpeg"), _asset("image.jpg", "image/jpeg")],
)
@@ -39,7 +55,6 @@ def test_asset_diagnosis_scores_ready_video_assets():
used_asset = _asset("video-1.mp4", "video/mp4", duration=8)
used_asset.metadata = {"generation_use_count": 1, "review_status": "pending_review"}
diagnosis = _build_diagnosis(
"workspace-1",
"project-1",
[
used_asset,
@@ -64,7 +79,6 @@ def test_asset_diagnosis_scores_ready_video_assets():
def test_asset_diagnosis_flags_unready_and_low_quality_assets():
diagnosis = _build_diagnosis(
"workspace-1",
"project-1",
[
_asset("video.mp4", "video/mp4", duration=10, quality_score=40),
@@ -78,3 +92,127 @@ def test_asset_diagnosis_flags_unready_and_low_quality_assets():
smart_view_counts = {item.key: item.count for item in diagnosis.smart_views}
assert smart_view_counts["needs_attention"] == 2
assert smart_view_counts["high_risk"] == 1
# ---------------------------------------------------------------------------
# 路由层测试 — 验证 find_by_project 调用正确性
# ---------------------------------------------------------------------------
class _StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def get(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
class _StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
self.find_by_project_called_with: list[str] = []
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
self.find_by_project_called_with.append(project_id)
return [lib for lib in self._libraries.values() if lib.project_id == project_id]
def list_by_project(self, project_id: str) -> list[AssetLibrary]:
raise AssertionError("路由不应调用 list_by_project，应调用 find_by_project")
class _StubAssetRepository:
def __init__(self, assets: dict[str, Asset] | None = None):
self._assets = assets or {}
self.list_by_library_called_with: list[str] = []
def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50) -> list[Asset]:
self.list_by_library_called_with.append(library_id)
return [a for a in self._assets.values() if a.library_id == library_id]
def count_by_library(self, library_id: str) -> int:
return len([a for a in self._assets.values() if a.library_id == library_id])
def _dep(name: str):
from app import dependencies
return getattr(dependencies, name)
def _build_route_test_app(project_repo, library_repo, asset_repo):
from unittest.mock import MagicMock
from app.api.routes.asset_diagnosis import router
from app.auth import AuthenticatedUser, get_current_user
app = FastAPI()
app.include_router(router, prefix="/api/v1")
mock_user = MagicMock(spec=AuthenticatedUser)
mock_user.id = "user-1"
mock_user.email = "test@example.com"
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[_dep("get_project_repository")] = lambda: project_repo
app.dependency_overrides[_dep("get_asset_library_repository")] = lambda: library_repo
app.dependency_overrides[_dep("get_asset_repository")] = lambda: asset_repo
return app
class TestAssetDiagnosisRoute:
"""路由层测试 — 验证 find_by_project 调用正确性。"""
def test_returns_404_when_project_not_found(self):
project_repo = _StubProjectRepository()
library_repo = _StubAssetLibraryRepository()
asset_repo = _StubAssetRepository()
app = _build_route_test_app(project_repo, library_repo, asset_repo)
client = TestClient(app)
resp = client.get("/api/v1/projects/nonexistent/asset-diagnosis")
assert resp.status_code == 404
def test_find_by_project_called_with_correct_project_id(self):
project = Project(id="proj-123", name="Test", owner_user_id="user-1")
library = AssetLibrary(
id="lib-1", name="Lib", project_id="proj-123", kind=AssetLibraryKind.VIDEO
)
project_repo = _StubProjectRepository({"proj-123": project})
library_repo = _StubAssetLibraryRepository({"lib-1": library})
asset_repo = _StubAssetRepository()
app = _build_route_test_app(project_repo, library_repo, asset_repo)
client = TestClient(app)
resp = client.get("/api/v1/projects/proj-123/asset-diagnosis")
assert resp.status_code == 200
assert library_repo.find_by_project_called_with == ["proj-123"]
def test_list_by_library_called_for_each_library(self):
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
lib1 = AssetLibrary(id="lib-1", name="Lib1", project_id="proj-1", kind=AssetLibraryKind.VIDEO)
lib2 = AssetLibrary(id="lib-2", name="Lib2", project_id="proj-1", kind=AssetLibraryKind.VOICE)
project_repo = _StubProjectRepository({"proj-1": project})
library_repo = _StubAssetLibraryRepository({"lib-1": lib1, "lib-2": lib2})
asset_repo = _StubAssetRepository()
app = _build_route_test_app(project_repo, library_repo, asset_repo)
client = TestClient(app)
resp = client.get("/api/v1/projects/proj-1/asset-diagnosis")
assert resp.status_code == 200
assert set(asset_repo.list_by_library_called_with) == {"lib-1", "lib-2"}
def test_find_by_project_not_list_by_project(self):
"""路由调用 find_by_project 而非 list_by_project（否则会触发 AssertionError）。"""
project = Project(id="proj-1", name="Test", owner_user_id="user-1")
library = AssetLibrary(id="lib-1", name="Lib", project_id="proj-1", kind=AssetLibraryKind.VIDEO)
project_repo = _StubProjectRepository({"proj-1": project})
library_repo = _StubAssetLibraryRepository({"lib-1": library})
asset_repo = _StubAssetRepository()
app = _build_route_test_app(project_repo, library_repo, asset_repo)
client = TestClient(app)
resp = client.get("/api/v1/projects/proj-1/asset-diagnosis")
# 如果调用了 list_by_project，会抛 AssertionError → 500
assert resp.status_code == 200
tests/unit/test_chunked_upload_routes.py
+316
View File
@@ -0,0 +1,316 @@
"""
chunked_upload.py 路由单元测试
覆盖：
- init_chunked_upload 端点正常路径
- init_chunked_upload 项目/素材库不存在时返回 404
- find_by_project 调用正确性
- 文件大小校验
- OSS 凭证校验
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock
# 设置必要环境变量（必须在导入 app 模块之前）
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
# 确保 app 模块可导入
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from fastapi import FastAPI
from fastapi.testclient import TestClient
from packages.domain import AssetLibrary, AssetLibraryKind, Project
# ---------------------------------------------------------------------------
# Stub 实现（不继承 Port ABC）
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def get(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
class StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
if kind is not None:
items = [lib for lib in items if lib.kind == kind]
return items
def list_by_project(self, project_id: str) -> list[AssetLibrary]:
raise AssertionError("路由不应调用 list_by_project，应调用 find_by_project")
class StubChunkedUploadRepository:
def __init__(self):
self._uploads = {}
def add(self, upload) -> None:
self._uploads[upload.upload_id] = upload
def get(self, upload_id: str):
return self._uploads.get(upload_id)
def update(self, upload) -> None:
self._uploads[upload.upload_id] = upload
def find_by_project(self, project_id: str, skip: int = 0, limit: int = 50):
return [u for u in self._uploads.values() if u.project_id == project_id]
def count_by_project(self, project_id: str) -> int:
return len([u for u in self._uploads.values() if u.project_id == project_id])
class StubIngestJobRepository:
def add(self, job) -> None:
pass
def get(self, job_id: str):
return None
def update_status(self, job_id, status, **kwargs):
pass
def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50):
return []
def count_by_library(self, library_id: str) -> int:
return 0
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_project(id: str = "proj-1", owner_user_id: str = "user-1") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=AssetLibraryKind.VIDEO)
def _dep(name: str):
from app import dependencies
return getattr(dependencies, name)
def _build_app(
project_repo=None,
library_repo=None,
storage=None,
chunked_repo=None,
ingest_repo=None,
) -> FastAPI:
from app.api.routes.chunked_upload import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
app = FastAPI()
app.include_router(router, prefix="/api/v1")
project_repo = project_repo or StubProjectRepository()
library_repo = library_repo or StubAssetLibraryRepository()
storage = storage or MagicMock()
storage.is_configured = True
chunked_repo = chunked_repo or StubChunkedUploadRepository()
ingest_repo = ingest_repo or StubIngestJobRepository()
# Mock auth
mock_user = MagicMock(spec=AuthenticatedUser)
mock_user.id = "user-1"
mock_user.email = "test@example.com"
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[_dep("get_project_repository")] = lambda: project_repo
app.dependency_overrides[_dep("get_asset_library_repository")] = lambda: library_repo
app.dependency_overrides[get_storage_service] = lambda: storage
app.dependency_overrides[_dep("get_ingest_job_repository")] = lambda: ingest_repo
return app
def _client(**kwargs) -> TestClient:
return TestClient(_build_app(**kwargs))
# ---------------------------------------------------------------------------
# 测试用例
# ---------------------------------------------------------------------------
class TestInitChunkedUpload:
"""init_chunked_upload 端点测试。"""
def test_returns_upload_record_on_success(self):
"""正常初始化分片上传。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
file_size = 100 * 1024 * 1024 # 100MB
chunk_size = 5 * 1024 * 1024 # 5MB
total_chunks = (file_size + chunk_size - 1) // chunk_size
resp = client.post(
"/api/v1/init",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "large-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["filename"] == "large-video.mp4"
assert "upload_id" in data
def test_returns_404_when_project_not_found(self):
"""项目不存在时返回 404。"""
library = _make_library()
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(
project_repo=StubProjectRepository(),
library_repo=library_repo,
)
file_size = 100 * 1024 * 1024
total_chunks = (file_size + 5 * 1024 * 1024 - 1) // (5 * 1024 * 1024)
resp = client.post(
"/api/v1/init",
json={
"project_id": "nonexistent",
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
assert resp.status_code == 404
assert "Project not found" in resp.json()["detail"]
def test_returns_404_when_library_not_in_project(self):
"""素材库不属于该项目时返回 404。"""
project = _make_project()
project_repo = StubProjectRepository({project.id: project})
# 素材库属于另一个项目
other_library = _make_library(project_id="other-project")
library_repo = StubAssetLibraryRepository({other_library.id: other_library})
client = _client(project_repo=project_repo, library_repo=library_repo)
file_size = 100 * 1024 * 1024
total_chunks = (file_size + 5 * 1024 * 1024 - 1) // (5 * 1024 * 1024)
resp = client.post(
"/api/v1/init",
json={
"project_id": project.id,
"library_id": "nonexistent-lib",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
assert resp.status_code == 404
assert "Asset library not found" in resp.json()["detail"]
def test_rejects_file_exceeding_max_size(self):
"""超过 2GB 限制的文件被拒绝（schema 层 le=2GB 会返回 422）。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/init",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "huge.mp4",
"content_type": "video/mp4",
"file_size": 3 * 1024 * 1024 * 1024, # 3GB，超过 2GB 限制
"total_chunks": 600,
},
)
# schema le=2147483648 → 422; route-level check → 413
assert resp.status_code in (400, 413, 422)
def test_find_by_project_is_called_not_list_by_project(self):
"""验证路由调用 find_by_project 而非 list_by_project。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
file_size = 100 * 1024 * 1024
total_chunks = (file_size + 5 * 1024 * 1024 - 1) // (5 * 1024 * 1024)
resp = client.post(
"/api/v1/init",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
# 如果调用了 list_by_project，StubAssetLibraryRepository 会抛 AssertionError
assert resp.status_code == 200
class TestChunkedUploadConstants:
"""分片上传常量测试。"""
def test_max_file_size_is_2gb(self):
from app.api.routes.chunked_upload import MAX_FILE_SIZE
assert MAX_FILE_SIZE == 2 * 1024 * 1024 * 1024
def test_default_chunk_size_is_5mb(self):
from app.api.routes.chunked_upload import DEFAULT_CHUNK_SIZE
assert DEFAULT_CHUNK_SIZE == 5 * 1024 * 1024
def test_chunk_expiry_hours_is_24(self):
from app.api.routes.chunked_upload import CHUNK_EXPIRY_HOURS
assert CHUNK_EXPIRY_HOURS == 24
tests/unit/test_config_oss.py
+198
View File
@@ -0,0 +1,198 @@
"""
config.py OSS 配置字段单元测试
覆盖：
- OSS 相关字段默认值
- 环境变量覆盖
- 字段名与代码引用一致
- pydantic_settings 加载行为
"""
from __future__ import annotations
import importlib.util
import os
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
def _load_settings_class():
"""
直接加载 config.py 模块，绕过 apps/api/__init__.py 的副作用。
apps/api/__init__.py 会导入 main.py，而 main.py 依赖 app 模块。
"""
config_path = Path(__file__).resolve().parents[2] / "apps" / "api" / "app" / "config.py"
spec = importlib.util.spec_from_file_location("config_module", config_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.Settings
def _fresh_settings(**env_overrides: dict[str, str]):
"""
每次创建一个全新的 Settings 实例。
env_overrides 会注入到 os.environ。
"""
env = {
"JWT_SECRET_KEY": "unit-test-secret-key-12345",
**env_overrides,
}
with patch.dict(os.environ, env, clear=False):
Settings = _load_settings_class()
return Settings()
# ---------------------------------------------------------------------------
# 默认值测试
# ---------------------------------------------------------------------------
class TestOSSConfigDefaults:
"""OSS 配置字段默认值必须与代码引用一致。"""
def test_oss_endpoint_default(self):
settings = _fresh_settings()
assert settings.OSS_ENDPOINT == "oss-cn-hangzhou.aliiyuncs.com"
def test_oss_access_key_id_default_empty(self):
settings = _fresh_settings()
assert settings.OSS_ACCESS_KEY_ID == ""
def test_oss_access_key_secret_default_empty(self):
settings = _fresh_settings()
assert settings.OSS_ACCESS_KEY_SECRET == ""
def test_oss_bucket_name_default(self):
settings = _fresh_settings()
assert settings.OSS_BUCKET_NAME == "xiaoxia-autocut"
def test_oss_direct_upload_max_mb_default_is_2000(self):
"""PR #117 修复：默认值从 800 改为 2000。"""
settings = _fresh_settings()
assert settings.OSS_DIRECT_UPLOAD_MAX_MB == 2000
def test_oss_direct_upload_expire_seconds_default(self):
settings = _fresh_settings()
assert settings.OSS_DIRECT_UPLOAD_EXPIRE_SECONDS == 900
# ---------------------------------------------------------------------------
# 环境变量覆盖测试
# ---------------------------------------------------------------------------
class TestOSSConfigEnvOverride:
"""环境变量能正确覆盖 OSS 配置字段。"""
def test_oss_endpoint_override(self):
settings = _fresh_settings(OSS_ENDPOINT="oss-cn-shanghai.aliiyuncs.com")
assert settings.OSS_ENDPOINT == "oss-cn-shanghai.aliiyuncs.com"
def test_oss_access_key_id_override(self):
settings = _fresh_settings(OSS_ACCESS_KEY_ID="test-key-id")
assert settings.OSS_ACCESS_KEY_ID == "test-key-id"
def test_oss_access_key_secret_override(self):
settings = _fresh_settings(OSS_ACCESS_KEY_SECRET="test-key-secret")
assert settings.OSS_ACCESS_KEY_SECRET == "test-key-secret"
def test_oss_bucket_name_override(self):
settings = _fresh_settings(OSS_BUCKET_NAME="test-bucket")
assert settings.OSS_BUCKET_NAME == "test-bucket"
def test_oss_direct_upload_max_mb_override(self):
settings = _fresh_settings(OSS_DIRECT_UPLOAD_MAX_MB="4096")
assert settings.OSS_DIRECT_UPLOAD_MAX_MB == 4096
def test_oss_direct_upload_expire_seconds_override(self):
settings = _fresh_settings(OSS_DIRECT_UPLOAD_EXPIRE_SECONDS="1800")
assert settings.OSS_DIRECT_UPLOAD_EXPIRE_SECONDS == 1800
# ---------------------------------------------------------------------------
# 字段名一致性测试（防止再次出现字段名拼写错误导致 500）
# ---------------------------------------------------------------------------
class TestOSSConfigFieldNameConsistency:
"""
确保 Settings 类包含代码中实际引用的所有字段。
防止类似 OSS_DIRECT_UPLOAD_EXPRESS_SECRET 的拼写错误再次发生。
"""
def test_settings_has_oss_endpoint_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_ENDPOINT")
def test_settings_has_oss_access_key_id_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_ACCESS_KEY_ID")
def test_settings_has_oss_access_key_secret_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_ACCESS_KEY_SECRET")
def test_settings_has_oss_bucket_name_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_BUCKET_NAME")
def test_settings_has_oss_direct_upload_max_mb_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_DIRECT_UPLOAD_MAX_MB")
def test_settings_has_oss_direct_upload_expire_seconds_field(self):
settings = _fresh_settings()
assert hasattr(settings, "OSS_DIRECT_UPLOAD_EXPIRE_SECONDS")
def test_property_aliases_match_field_names(self):
"""确保 property 访问器与字段值一致。"""
settings = _fresh_settings(
OSS_ENDPOINT="ep",
OSS_ACCESS_KEY_ID="kid",
OSS_ACCESS_KEY_SECRET="ksec",
OSS_BUCKET_NAME="bkt",
)
assert settings.oss_endpoint == "ep"
assert settings.oss_access_key_id == "kid"
assert settings.oss_access_key_secret == "ksec"
assert settings.oss_bucket_name == "bkt"
# ---------------------------------------------------------------------------
# extra="ignore" 行为测试
# ---------------------------------------------------------------------------
class TestSettingsExtraIgnore:
"""extra="ignore" 确保未知环境变量不会导致启动失败。"""
def test_unknown_env_var_is_ignored(self):
settings = _fresh_settings(UNKNOWN_RANDOM_VAR="whatever")
assert not hasattr(settings, "UNKNOWN_RANDOM_VAR")
def test_settings_loads_without_error(self):
settings = _fresh_settings()
assert settings.APP_NAME == "xiaoxia-saas"
# ---------------------------------------------------------------------------
# 别名测试（MAX_UPLOAD_SIZE_MB 兼容旧配置）
# ---------------------------------------------------------------------------
class TestOSSConfigAliases:
"""OSS_DIRECT_UPLOAD_MAX_MB 支持 MAX_UPLOAD_SIZE_MB 别名。"""
def test_max_upload_size_mb_alias_works(self):
"""旧环境变量 MAX_UPLOAD_SIZE_MB 仍能生效。"""
env = {
"JWT_SECRET_KEY": "unit-test-secret-key-12345",
"MAX_UPLOAD_SIZE_MB": "3000",
}
with patch.dict(os.environ, env, clear=False):
os.environ.pop("OSS_DIRECT_UPLOAD_MAX_MB", None)
Settings = _load_settings_class()
settings = Settings()
assert settings.OSS_DIRECT_UPLOAD_MAX_MB == 3000
tests/unit/test_upload_routes.py
+390
View File
@@ -0,0 +1,390 @@
"""
upload.py 路由单元测试
覆盖：
- _require_project_and_library 中 find_by_project 调用正确性
- OSS 凭证校验（未配置时返回 503）
- prepare_direct_upload 正常路径
- 文件类型校验
- 异常处理路径
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock
# 设置必要环境变量（必须在导入 app 模块之前）
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
# 确保 app 模块可导入
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from fastapi import FastAPI
from fastapi.testclient import TestClient
from packages.domain import AssetLibrary, AssetLibraryKind, Project
# ---------------------------------------------------------------------------
# 测试用 Stub（不继承 Port ABC，因为 Port 定义 async 方法，路由实际使用同步 duck-type）
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def get(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
class StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
if kind is not None:
items = [lib for lib in items if lib.kind == kind]
return items
def list_by_project(self, project_id: str) -> list[AssetLibrary]:
"""故意保留旧方法名，验证路由不会调用它。"""
raise AssertionError("路由不应调用 list_by_project，应调用 find_by_project")
class StubIngestJobRepository:
def add(self, job) -> None:
pass
def get(self, job_id: str):
return None
def update_status(self, job_id, status, **kwargs):
pass
def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50):
return []
def count_by_library(self, library_id: str) -> int:
return 0
# ---------------------------------------------------------------------------
# 测试 Fixtures
# ---------------------------------------------------------------------------
def _make_project(id: str = "proj-1", owner_user_id: str = "user-1") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_library(
id: str = "lib-1",
project_id: str = "proj-1",
kind: AssetLibraryKind = AssetLibraryKind.VIDEO,
) -> AssetLibrary:
return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=kind)
def _build_app(
project_repo: StubProjectRepository | None = None,
library_repo: StubAssetLibraryRepository | None = None,
storage: MagicMock | None = None,
ingest_repo: StubIngestJobRepository | None = None,
) -> FastAPI:
"""构建一个最小化的 FastAPI app，只注册 upload 路由。"""
from app.api.routes.upload import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
get_asset_library_repository,
get_ingest_job_repository,
get_project_repository,
)
app = FastAPI()
app.include_router(router, prefix="/api/v1")
project_repo = project_repo or StubProjectRepository()
library_repo = library_repo or StubAssetLibraryRepository()
storage = storage or MagicMock()
storage.is_configured = True
storage.create_direct_upload_post.return_value = {
"url": "https://bucket.oss.example.com",
"method": "POST",
"storage_key": "uploads/abc/test.mp4",
"expires_at": "2026-01-01T00:00:00Z",
"fields": {"key": "uploads/abc/test.mp4"},
}
ingest_repo = ingest_repo or StubIngestJobRepository()
# Mock auth
mock_user = MagicMock(spec=AuthenticatedUser)
mock_user.id = "user-1"
mock_user.email = "test@example.com"
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_project_repository] = lambda: project_repo
app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
app.dependency_overrides[get_storage_service] = lambda: storage
app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
return app
def _client(**kwargs) -> TestClient:
app = _build_app(**kwargs)
return TestClient(app)
# ---------------------------------------------------------------------------
# 测试用例
# ---------------------------------------------------------------------------
class TestRequireProjectAndLibrary:
"""_require_project_and_library 辅助函数测试。"""
def test_returns_200_when_project_and_library_exist(self):
"""项目和素材库都存在时，正常返回。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 200
def test_returns_404_when_project_not_found(self):
"""项目不存在时返回 404。"""
library = _make_library()
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(
project_repo=StubProjectRepository(),
library_repo=library_repo,
)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": "nonexistent",
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 404
assert "Project not found" in resp.json()["detail"]
def test_returns_404_when_library_not_found(self):
"""素材库不属于该项目时返回 404。"""
project = _make_project()
project_repo = StubProjectRepository({project.id: project})
other_library = _make_library(project_id="other-project")
library_repo = StubAssetLibraryRepository({other_library.id: other_library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": "nonexistent-lib",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 404
assert "Asset library not found" in resp.json()["detail"]
def test_find_by_project_is_called_not_list_by_project(self):
"""
验证路由调用的是 find_by_project 而不是 list_by_project。
StubAssetLibraryRepository.list_by_project 会抛出 AssertionError。
"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
# 如果调用了 list_by_project，会抛 AssertionError 导致 500
assert resp.status_code == 200
class TestPrepareDirectUpload:
"""prepare_direct_upload 端点测试。"""
def test_returns_upload_credentials_when_configured(self):
"""OSS 已配置时，返回上传凭证。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
storage = MagicMock()
storage.is_configured = True
storage.create_direct_upload_post.return_value = {
"url": "https://bucket.oss.example.com",
"method": "POST",
"storage_key": "uploads/abc/test-video.mp4",
"expires_at": "2026-01-01T00:00:00Z",
"fields": {"key": "uploads/abc/test-video.mp4"},
}
client = _client(
project_repo=project_repo,
library_repo=library_repo,
storage=storage,
)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": 1024 * 1024,
},
)
assert resp.status_code == 200
data = resp.json()
assert "upload_url" in data
assert "storage_key" in data
def test_returns_503_when_oss_not_configured(self):
"""OSS 未配置时，返回 503。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
storage = MagicMock()
storage.create_direct_upload_post.side_effect = RuntimeError("OSS 未配置")
client = _client(
project_repo=project_repo,
library_repo=library_repo,
storage=storage,
)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 503
class TestCompleteDirectUpload:
"""complete_direct_upload 端点测试。"""
def test_returns_404_when_project_not_found(self):
"""项目不存在时返回 404。"""
library = _make_library()
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(
project_repo=StubProjectRepository(),
library_repo=library_repo,
)
resp = client.post(
"/api/v1/direct/complete",
json={
"project_id": "nonexistent",
"library_id": library.id,
"storage_key": "uploads/abc/test.mp4",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 404
class TestMimeTypeValidation:
"""文件类型校验测试。"""
def test_accepts_video_mp4(self):
"""video/mp4 是合法类型。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024,
},
)
assert resp.status_code == 200
def test_rejects_invalid_mime_type(self):
"""非法文件类型被拒绝。"""
project = _make_project()
library = _make_library()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
client = _client(project_repo=project_repo, library_repo=library_repo)
resp = client.post(
"/api/v1/direct/prepare",
json={
"project_id": project.id,
"library_id": library.id,
"filename": "malware.exe",
"content_type": "application/x-executable",
"file_size": 1024,
},
)
assert resp.status_code == 415