From 2e53c59cd7beb9054a2db5d530a8e4c499ae9797 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 17 Aug 2026 15:35:46 +0800 Subject: [PATCH] fix: remove unused typing.Any import + add clips API response structure tests - Remove unused 'from typing import Any' (Ruff F401) - Add 28 unit/integration tests covering: - _clip_to_response: enum conversion, datetime formatting, asset_url param - _build_asset_url_map: empty list, storage failure, asset not found, no storage_key, success path, exception handling, mixed scenarios - Route integration: list/detail/split/merge with asset URL resolution - Incremental coverage: 100% (95/95 added lines covered) --- .../app/api/routes/templates_editor/clips.py | 1 - .../unit/test_clips_api_response_structure.py | 482 ++++++++++++++++++ 2 files changed, 482 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_clips_api_response_structure.py diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py index 8dd67334a..7c6b5e77a 100755 --- a/apps/api/app/api/routes/templates_editor/clips.py +++ b/apps/api/app/api/routes/templates_editor/clips.py @@ -16,7 +16,6 @@ from __future__ import annotations import logging -from typing import Any from app.auth import AuthenticatedUser, get_current_user from app.core.storage import get_storage_service diff --git a/tests/unit/test_clips_api_response_structure.py b/tests/unit/test_clips_api_response_structure.py new file mode 100644 index 000000000..cbf876124 --- /dev/null +++ b/tests/unit/test_clips_api_response_structure.py @@ -0,0 +1,482 @@ +"""片段管理路由 clips.py 增量覆盖率测试. + +覆盖 PR fix/clips-api-response-structure 新增代码: +- _clip_to_response: 枚举转换、日期格式化、asset_url 参数 +- _build_asset_url_map: 批量素材 URL 解析（空列表/异常/正常路径） +- 路由层 asset_repo 注入与 URL 拼接逻辑 +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from enum import Enum +from unittest.mock import MagicMock, patch + +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 + +# --------------------------------------------------------------------------- +# 常量与工厂 +# --------------------------------------------------------------------------- + +TEST_TEMPLATE_ID = "tmpl-test-001" +TEST_PLAN_ID = "plan-draft-001" +TEST_USER_ID = "user-001" + + +def _auth_user(): + u = MagicMock() + u.user.id = TEST_USER_ID + u.user_id = TEST_USER_ID + return u + + +def _clip(**overrides): + """构造 mock clip，支持 Enum 类型字段""" + c = MagicMock() + c.id = overrides.get("id", "clip-001") + c.plan_id = overrides.get("plan_id", TEST_PLAN_ID) + c.clip_type = overrides.get("clip_type", "video") + c.order = overrides.get("order", 0) + c.duration = overrides.get("duration", 10.0) + c.start_time = overrides.get("start_time", 0.0) + c.text_content = overrides.get("text_content", "") + c.transition_effect = overrides.get("transition_effect", "cut") + c.transition_duration = overrides.get("transition_duration", 0.0) + c.playback_speed = overrides.get("playback_speed", 1.0) + c.asset_id = overrides.get("asset_id", "") + c.status = overrides.get("status", "ready") + c.template_clip_config_id = overrides.get("template_clip_config_id", "") + c.config = overrides.get("config", {}) + c.created_at = overrides.get("created_at", None) + c.updated_at = overrides.get("updated_at", None) + return c + + +def _services(plan_svc_overrides=None): + tpl = MagicMock() + plan = MagicMock() + if plan_svc_overrides: + for k, v in plan_svc_overrides.items(): + setattr(plan, k, v) + return tpl, plan + + +# --------------------------------------------------------------------------- +# 单元测试: _clip_to_response +# --------------------------------------------------------------------------- + + +class TestClipToResponse: + """_clip_to_response 纯函数测试 — 覆盖行 53-80""" + + def test_basic_fields(self): + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(id="c1", order=3, duration=5.5, text_content="hello") + resp = _clip_to_response(c) + assert resp.id == "c1" + assert resp.order == 3 + assert resp.duration == 5.5 + assert resp.text_content == "hello" + assert resp.asset_url is None + + def test_enum_clip_type(self): + """Enum 值应被 .value 解包""" + from app.api.routes.templates_editor.clips import _clip_to_response + + class ClipType(str, Enum): + VIDEO = "video" + AUDIO = "audio" + + c = _clip(clip_type=ClipType.VIDEO) + resp = _clip_to_response(c) + assert resp.clip_type == "video" + + def test_plain_string_clip_type(self): + """非 Enum 字符串直接用 str()""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(clip_type="main") + resp = _clip_to_response(c) + assert resp.clip_type == "main" + + def test_enum_transition_effect(self): + from app.api.routes.templates_editor.clips import _clip_to_response + + class Transition(str, Enum): + FADE = "fade" + + c = _clip(transition_effect=Transition.FADE) + resp = _clip_to_response(c) + assert resp.transition_effect == "fade" + + def test_default_transition_when_none(self): + """transition_effect 缺失时默认 cut""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip() + del c.transition_effect # 触发 getattr default + resp = _clip_to_response(c) + assert resp.transition_effect == "cut" + + def test_asset_url_passed(self): + """asset_url 参数应透传到响应""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(asset_id="a1") + resp = _clip_to_response(c, asset_url="https://signed-url.example.com/video.mp4") + assert resp.asset_url == "https://signed-url.example.com/video.mp4" + + def test_asset_url_none_by_default(self): + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip() + resp = _clip_to_response(c) + assert resp.asset_url is None + + def test_datetime_isoformat(self): + """datetime 对象应被 isoformat()""" + from datetime import datetime + from app.api.routes.templates_editor.clips import _clip_to_response + + dt = datetime(2026, 8, 17, 12, 0, 0) + c = _clip(created_at=dt, updated_at=dt) + resp = _clip_to_response(c) + assert "2026-08-17" in resp.created_at + assert "2026-08-17" in resp.updated_at + + def test_none_datetime_empty_string(self): + """None 日期应格式化为空字符串""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(created_at=None, updated_at=None) + resp = _clip_to_response(c) + assert resp.created_at == "" + assert resp.updated_at == "" + + def test_string_datetime_passthrough(self): + """已经是字符串的日期直接 str()""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(created_at="2026-08-17T00:00:00") + resp = _clip_to_response(c) + assert resp.created_at == "2026-08-17T00:00:00" + + def test_none_defaults_for_optional_fields(self): + """None/缺失字段的默认值""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(asset_id=None, status=None, template_clip_config_id=None) + resp = _clip_to_response(c) + assert resp.asset_id == "" + assert resp.status == "pending" + assert resp.template_clip_config_id == "" + + def test_zero_duration_fallback(self): + """duration=0 → playback_speed 默认 1.0""" + from app.api.routes.templates_editor.clips import _clip_to_response + + c = _clip(playback_speed=None) + resp = _clip_to_response(c) + assert resp.playback_speed == 1.0 + + +# --------------------------------------------------------------------------- +# 单元测试: _build_asset_url_map +# --------------------------------------------------------------------------- + + +class TestBuildAssetUrlMap: + """_build_asset_url_map 测试 — 覆盖行 93-118""" + + def test_empty_list(self): + """空 asset_ids 直接返回空 dict""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + repo = MagicMock() + result = _build_asset_url_map([], repo) + assert result == {} + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_storage_service_failure(self, mock_get_storage): + """存储服务获取失败时返回全 None""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + mock_get_storage.side_effect = RuntimeError("storage unavailable") + repo = MagicMock() + result = _build_asset_url_map(["a1", "a2"], repo) + assert result == {"a1": None, "a2": None} + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_asset_not_found(self, mock_get_storage): + """asset_id 找不到对应素材 → None""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + mock_get_storage.return_value = storage + repo = MagicMock() + repo.find_by_id.return_value = None + + result = _build_asset_url_map(["missing-id"], repo) + assert result == {"missing-id": None} + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_no_storage_key(self, mock_get_storage): + """素材没有 storage_key → None""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + mock_get_storage.return_value = storage + repo = MagicMock() + asset = MagicMock() + asset.storage_key = "" + repo.find_by_id.return_value = asset + + result = _build_asset_url_map(["a1"], repo) + assert result == {"a1": None} + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_successful_url_generation(self, mock_get_storage): + """正常路径：返回签名 URL""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + storage.get_download_url.return_value = "https://cdn.example.com/signed.mp4" + mock_get_storage.return_value = storage + repo = MagicMock() + asset = MagicMock() + asset.storage_key = "videos/test.mp4" + repo.find_by_id.return_value = asset + + result = _build_asset_url_map(["a1"], repo) + assert result == {"a1": "https://cdn.example.com/signed.mp4"} + storage.get_download_url.assert_called_once_with("videos/test.mp4", expires_seconds=3600) + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_exception_during_url_generation(self, mock_get_storage): + """单个 asset 生成 URL 异常 → None，不影响其他""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + storage.get_download_url.side_effect = [Exception("boom"), "https://ok.com/v2"] + mock_get_storage.return_value = storage + + repo = MagicMock() + asset1 = MagicMock() + asset1.storage_key = "v1.mp4" + asset2 = MagicMock() + asset2.storage_key = "v2.mp4" + repo.find_by_id.side_effect = [asset1, asset2] + + result = _build_asset_url_map(["a1", "a2"], repo) + assert result["a1"] is None + assert result["a2"] == "https://ok.com/v2" + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_skip_empty_asset_id(self, mock_get_storage): + """空字符串 asset_id 被跳过""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + mock_get_storage.return_value = storage + repo = MagicMock() + + result = _build_asset_url_map(["", "a1"], repo) + # "" not in result because it's skipped by `if not aid: continue` + assert "" not in result + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_multiple_assets_mixed(self, mock_get_storage): + """混合场景：正常+异常+缺失""" + from app.api.routes.templates_editor.clips import _build_asset_url_map + + storage = MagicMock() + storage.get_download_url.return_value = "https://cdn.com/ok.mp4" + mock_get_storage.return_value = storage + + repo = MagicMock() + good_asset = MagicMock() + good_asset.storage_key = "good.mp4" + # a1=good, a2=not found, a3=good + repo.find_by_id.side_effect = [good_asset, None, good_asset] + + result = _build_asset_url_map(["a1", "a2", "a3"], repo) + assert result["a1"] == "https://cdn.com/ok.mp4" + assert result["a2"] is None + assert result["a3"] == "https://cdn.com/ok.mp4" + + +# --------------------------------------------------------------------------- +# 集成测试: 路由层 asset_repo 注入 +# --------------------------------------------------------------------------- + + +class TestClipRoutesAssetIntegration: + """路由层测试 — 覆盖 asset_url 在 list/detail/split/merge 中的拼接逻辑""" + + def _create_app(self, plan_svc_config=None): + from app.api.routes import templates_editor as editor_module + from app.dependencies import get_asset_repository + + mock_clip_1 = _clip(id="c1", asset_id="asset-001") + mock_clip_2 = _clip(id="c2", asset_id="") + + mock_tpl_svc = MagicMock() + mock_plan_svc = MagicMock() + mock_plan_svc.list_clips.return_value = [mock_clip_1, mock_clip_2] + mock_plan_svc.count_clips.return_value = 2 + mock_plan_svc.get_clip.return_value = mock_clip_1 + mock_plan_svc.create_clip.return_value = _clip(id="c-new", asset_id="") + mock_plan_svc.update_clip.return_value = _clip(id="c1", duration=15.0) + mock_plan_svc.delete_clip.return_value = True + mock_plan_svc.split_clip.return_value = { + "left_clip": _clip(id="c-left", asset_id="asset-L"), + "right_clip": _clip(id="c-right", asset_id="asset-R"), + } + mock_plan_svc.merge_clips.return_value = _clip(id="c-merged", asset_id="asset-M") + + if plan_svc_config: + for k, v in plan_svc_config.items(): + setattr(mock_plan_svc, k, v) + + def _deps(): + return mock_tpl_svc, mock_plan_svc + + mock_asset_repo = MagicMock() + + app = FastAPI() + app.include_router( + editor_module.router, + prefix="/api/v1/templates/{template_id}/editor", + ) + app.dependency_overrides[editor_module.get_current_user] = _auth_user + app.dependency_overrides[editor_module.get_draft_plan_id] = lambda: TEST_PLAN_ID + app.dependency_overrides[editor_module.get_editor_services] = _deps + app.dependency_overrides[get_asset_repository] = lambda: mock_asset_repo + + return TestClient(app), mock_plan_svc, mock_asset_repo + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_list_clips_includes_asset_urls(self, mock_get_storage): + """GET /clips 应为有 asset_id 的片段返回签名 URL""" + storage = MagicMock() + storage.get_download_url.return_value = "https://cdn.com/c1.mp4" + mock_get_storage.return_value = storage + + client, _, asset_repo = self._create_app() + asset = MagicMock() + asset.storage_key = "videos/c1.mp4" + asset_repo.find_by_id.return_value = asset + + resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips") + assert resp.status_code == 200 + data = resp.json() + items = data["items"] + assert len(items) == 2 + # c1 has asset_id → should have url + assert items[0]["asset_url"] == "https://cdn.com/c1.mp4" + # c2 has empty asset_id → None + assert items[1]["asset_url"] is None + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_get_clip_detail_with_asset_url(self, mock_get_storage): + """GET /clips/{clip_id} 应返回素材签名 URL""" + storage = MagicMock() + storage.get_download_url.return_value = "https://cdn.com/detail.mp4" + mock_get_storage.return_value = storage + + client, _, asset_repo = self._create_app() + asset = MagicMock() + asset.storage_key = "videos/detail.mp4" + asset_repo.find_by_id.return_value = asset + + resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001") + assert resp.status_code == 200 + assert resp.json()["asset_url"] == "https://cdn.com/detail.mp4" + + def test_get_clip_detail_no_asset(self): + """片段没有 asset_id 时不应调用 URL 解析""" + client, plan_svc, asset_repo = self._create_app() + # 返回没有 asset_id 的片段 + plan_svc.get_clip.return_value = _clip(id="c-no-asset", asset_id="") + + resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/c-no-asset") + assert resp.status_code == 200 + assert resp.json()["asset_url"] is None + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_split_clip_returns_asset_urls(self, mock_get_storage): + """POST /clips/{clip_id}/split 返回的左右片段应带签名 URL""" + storage = MagicMock() + storage.get_download_url.side_effect = ["https://cdn.com/L.mp4", "https://cdn.com/R.mp4"] + mock_get_storage.return_value = storage + + client, _, asset_repo = self._create_app() + asset_l = MagicMock() + asset_l.storage_key = "videos/L.mp4" + asset_r = MagicMock() + asset_r.storage_key = "videos/R.mp4" + asset_repo.find_by_id.side_effect = [asset_l, asset_r] + + resp = client.post( + f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001/split", + json={"split_time": 5.0}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["left_clip"]["asset_url"] == "https://cdn.com/L.mp4" + assert data["right_clip"]["asset_url"] == "https://cdn.com/R.mp4" + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_merge_clips_returns_asset_url(self, mock_get_storage): + """POST /clips/merge 返回的合并片段应带签名 URL""" + storage = MagicMock() + storage.get_download_url.return_value = "https://cdn.com/M.mp4" + mock_get_storage.return_value = storage + + client, _, asset_repo = self._create_app() + asset = MagicMock() + asset.storage_key = "videos/M.mp4" + asset_repo.find_by_id.return_value = asset + + resp = client.post( + f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge", + json={"clip_ids": ["c1", "c2"]}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["merged_clip"]["asset_url"] == "https://cdn.com/M.mp4" + assert data["deleted_clip_ids"] == ["c1", "c2"] + + def test_merge_clips_not_found(self): + """merge 时某片段不存在应返回 404""" + client, plan_svc, _ = self._create_app() + plan_svc.get_clip.return_value = None + + resp = client.post( + f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge", + json={"clip_ids": ["nonexistent-1", "nonexistent-2"]}, + ) + assert resp.status_code == 404 + + def test_delete_clip_success(self): + """DELETE /clips/{clip_id} 成功返回 204""" + client, _, _ = self._create_app() + resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001") + assert resp.status_code == 204 + + def test_delete_clip_not_found(self): + """DELETE 片段不存在返回 404""" + client, plan_svc, _ = self._create_app() + plan_svc.delete_clip.return_value = False + resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/bad-id") + assert resp.status_code == 404