diff --git a/tests/unit/test_asset_diagnosis.py b/tests/unit/test_asset_diagnosis.py index 084c4e0d3..4f6e6795c 100644 --- a/tests/unit/test_asset_diagnosis.py +++ b/tests/unit/test_asset_diagnosis.py @@ -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 diff --git a/tests/unit/test_chunked_upload_routes.py b/tests/unit/test_chunked_upload_routes.py new file mode 100644 index 000000000..dcbe489c9 --- /dev/null +++ b/tests/unit/test_chunked_upload_routes.py @@ -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 diff --git a/tests/unit/test_config_oss.py b/tests/unit/test_config_oss.py new file mode 100644 index 000000000..24e3b5263 --- /dev/null +++ b/tests/unit/test_config_oss.py @@ -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 diff --git a/tests/unit/test_upload_routes.py b/tests/unit/test_upload_routes.py new file mode 100644 index 000000000..785d4331a --- /dev/null +++ b/tests/unit/test_upload_routes.py @@ -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