diff --git a/packages/adapters/in_memory/asset_library_repository.py b/packages/adapters/in_memory/asset_library_repository.py index 361938881..97b7012af 100644 --- a/packages/adapters/in_memory/asset_library_repository.py +++ b/packages/adapters/in_memory/asset_library_repository.py @@ -16,6 +16,9 @@ class InMemoryAssetLibraryRepository: def get(self, library_id: str) -> AssetLibrary | None: return self._libraries.get(library_id) + def find_by_id(self, library_id: str) -> AssetLibrary | None: + return self.get(library_id) + def find_by_project(self, project_id: str, kind: AssetLibraryKind | None = None) -> list[AssetLibrary]: items = [library for library in self._libraries.values() if library.project_id == project_id] if kind is not None: diff --git a/tests/unit/test_asset_library_delete.py b/tests/unit/test_asset_library_delete.py new file mode 100644 index 000000000..648dcafa4 --- /dev/null +++ b/tests/unit/test_asset_library_delete.py @@ -0,0 +1,254 @@ +""" +DELETE /asset-libraries/{library_id} 单元测试 + +覆盖: +- 正常删除空素材库(204) +- 删除含素材的库(同时删除库内素材) +- 素材库不存在(404) +- 无权限访问(403) +""" + +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 fastapi import FastAPI +from fastapi.testclient import TestClient + +from packages.adapters.in_memory.asset_library_repository import InMemoryAssetLibraryRepository +from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository +from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus + +# --------------------------------------------------------------------------- +# Stub Project Repository +# --------------------------------------------------------------------------- + + +class StubProject: + """最小化 Project stub,支持 can_access""" + + def __init__(self, project_id: str, owner_id: str): + self.id = project_id + self._owner_id = owner_id + + def can_access(self, user_id: str) -> bool: + return user_id == self._owner_id + + +class StubProjectRepository: + def __init__(self, projects: dict[str, StubProject] | None = None): + self._projects = projects or {} + + def find_by_id(self, project_id: str): + return self._projects.get(project_id) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_auth_user(user_id: str = "user-001"): + from app.auth import AuthenticatedUser + + from packages.domain.entities import User + + user = User(id=user_id, email="test@example.com", display_name="测试用户") + return AuthenticatedUser(user=user) + + +def _create_test_app( + library_repo: InMemoryAssetLibraryRepository, + asset_repo: InMemoryAssetRepository, + project_repo: StubProjectRepository, + user_id: str = "user-001", +): + from app.api.routes import asset_libraries as module + from app.api.routes.asset_libraries import router + + app = FastAPI() + app.include_router(router, prefix="/api/v1/asset-libraries") + + app.dependency_overrides[module.get_current_user] = lambda: _make_auth_user(user_id) + app.dependency_overrides[module.get_asset_library_repository] = lambda: library_repo + app.dependency_overrides[module.get_asset_repository] = lambda: asset_repo + app.dependency_overrides[module.get_project_repository] = lambda: project_repo + + return app + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestDeleteAssetLibrary: + """DELETE /asset-libraries/{library_id} 测试""" + + def test_delete_empty_library(self): + """删除空素材库 → 204""" + from datetime import datetime, timezone + + lib_repo = InMemoryAssetLibraryRepository() + asset_repo = InMemoryAssetRepository() + proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")}) + + now = datetime.now(timezone.utc) + library = AssetLibrary( + id="lib-1", + project_id="proj-1", + name="视频素材库", + kind=AssetLibraryKind.VIDEO, + asset_count=0, + total_size=0, + created_at=now, + updated_at=now, + ) + lib_repo.create(library) + + app = _create_test_app(lib_repo, asset_repo, proj_repo) + client = TestClient(app) + + response = client.delete("/api/v1/asset-libraries/lib-1") + assert response.status_code == 204 + + # 验证库已删除 + assert lib_repo.find_by_id("lib-1") is None + + def test_delete_library_with_assets(self): + """删除含素材的库 → 库和素材都被删除""" + from datetime import datetime, timezone + + lib_repo = InMemoryAssetLibraryRepository() + asset_repo = InMemoryAssetRepository() + proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")}) + + now = datetime.now(timezone.utc) + library = AssetLibrary( + id="lib-1", + project_id="proj-1", + name="视频素材库", + kind=AssetLibraryKind.VIDEO, + asset_count=3, + total_size=1000, + created_at=now, + updated_at=now, + ) + lib_repo.create(library) + + # 创建 3 个素材 + for i in range(3): + asset = Asset.create( + project_id="proj-1", + library_id="lib-1", + name=f"video_{i}.mp4", + storage_key=f"uploads/video_{i}.mp4", + mime_type="video/mp4", + status=AssetStatus.READY, + ) + asset_repo.create(asset) + + # 创建一个不属于该库的素材(不应被删除) + other_asset = Asset.create( + project_id="proj-1", + library_id="lib-other", + name="other.mp4", + storage_key="uploads/other.mp4", + mime_type="video/mp4", + status=AssetStatus.READY, + ) + asset_repo.create(other_asset) + + app = _create_test_app(lib_repo, asset_repo, proj_repo) + client = TestClient(app) + + response = client.delete("/api/v1/asset-libraries/lib-1") + assert response.status_code == 204 + + # 库已删除 + assert lib_repo.find_by_id("lib-1") is None + # 库内素材已删除 + assert asset_repo.find_by_library("lib-1") == [] + # 其他素材未受影响 + assert asset_repo.get(other_asset.id) is not None + + def test_delete_nonexistent_library(self): + """删除不存在的素材库 → 404""" + lib_repo = InMemoryAssetLibraryRepository() + asset_repo = InMemoryAssetRepository() + proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")}) + + app = _create_test_app(lib_repo, asset_repo, proj_repo) + client = TestClient(app) + + response = client.delete("/api/v1/asset-libraries/nonexistent-id") + assert response.status_code == 404 + assert "素材库不存在" in response.json()["detail"] + + def test_delete_library_access_denied(self): + """无权限用户删除素材库 → 403""" + from datetime import datetime, timezone + + lib_repo = InMemoryAssetLibraryRepository() + asset_repo = InMemoryAssetRepository() + # 项目属于 user-002,当前用户是 user-001 + proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-002")}) + + now = datetime.now(timezone.utc) + library = AssetLibrary( + id="lib-1", + project_id="proj-1", + name="视频素材库", + kind=AssetLibraryKind.VIDEO, + asset_count=0, + total_size=0, + created_at=now, + updated_at=now, + ) + lib_repo.create(library) + + app = _create_test_app(lib_repo, asset_repo, proj_repo, user_id="user-001") + client = TestClient(app) + + response = client.delete("/api/v1/asset-libraries/lib-1") + assert response.status_code == 403 + assert "Access denied" in response.json()["detail"] + + # 库未被删除 + assert lib_repo.find_by_id("lib-1") is not None + + def test_delete_library_project_not_found(self): + """素材库所属项目不存在 → 404""" + from datetime import datetime, timezone + + lib_repo = InMemoryAssetLibraryRepository() + asset_repo = InMemoryAssetRepository() + # 空的项目仓库,找不到项目 + proj_repo = StubProjectRepository({}) + + now = datetime.now(timezone.utc) + library = AssetLibrary( + id="lib-1", + project_id="proj-missing", + name="视频素材库", + kind=AssetLibraryKind.VIDEO, + asset_count=0, + total_size=0, + created_at=now, + updated_at=now, + ) + lib_repo.create(library) + + app = _create_test_app(lib_repo, asset_repo, proj_repo) + client = TestClient(app) + + response = client.delete("/api/v1/asset-libraries/lib-1") + assert response.status_code == 404