feat: 新增 DELETE /asset-libraries/{library_id} 接口
CI/CD Pipeline / Frontend Lint (push) Failing after 25h29m22s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 25h29m22s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1684h4m26s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 1684h4m28s
CI/CD Pipeline / Build Production Runtime Images (push) Failing after 1684h4m30s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 1684h4m30s
CI/CD Pipeline / Deploy Production (push) Failing after 1684h35m56s

- 删除素材库同时清理库内素材(无 FK 级联,手动处理)
- 加权限校验:检查用户项目访问权限
- InMemoryAssetLibraryRepository 补充 find_by_id 方法
- 5 个单元测试覆盖:正常删除、含素材删除、404、403、项目不存在
This commit is contained in:
灵应
2026-07-08 11:23:22 +08:00
parent 880b6b5c1b
commit 3889cd0f1a
2 changed files with 257 additions and 0 deletions
@@ -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:
+254
View File
@@ -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