From e7bdc48c491c1364fad4d6478998d836d301a96a Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 16 Jul 2026 11:30:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=89=87=E6=AE=B5=E8=B0=83=E6=95=B4=20?= =?UTF-8?q?-=20=E8=B0=83=E9=80=9F=20+=20=E9=9F=B3=E9=87=8F=20+=20=E8=A3=81?= =?UTF-8?q?=E5=89=AA=20+=20=E6=89=B9=E9=87=8F=E8=B0=83=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PUT /clips/{id}/speed 调速（0.25~4.0倍） - PUT /clips/{id}/volume 音量调节（0~2.0倍） - PUT /clips/{id}/trim 裁剪（trim_start + trim_end） - PUT /clips/{id}/adjustments 统一调整（speed+volume+trim） - POST /{plan_id}/clips/batch-speed 批量调速 - 裁剪时长校验，不能超过片段总时长 - 音量/裁剪存储在 clip.config，零侵入数据模型 - 25个单元测试全覆盖 --- apps/api/app/api/routes/edit_plans.py | 2 + .../app/api/routes/edit_plans_adjustments.py | 311 +++++++++++ tests/unit/test_edit_plan_adjustments_api.py | 491 ++++++++++++++++++ 3 files changed, 804 insertions(+) mode change 100755 => 100644 apps/api/app/api/routes/edit_plans.py create mode 100755 apps/api/app/api/routes/edit_plans_adjustments.py create mode 100644 tests/unit/test_edit_plan_adjustments_api.py diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py old mode 100755 new mode 100644 index 0d1ed6f96..e07ecd616 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -263,10 +263,12 @@ def _to_response(p: EditPlan) -> EditPlanResponse: # ── Include sub-routers (拆分模块) ──────────────────────────────────────────── # 注意：含静态路径的子路由需放在 CRUD 路由之前，避免被 /{plan_id} 抢先匹配 +from .edit_plans_adjustments import router as adjustments_router from .edit_plans_export import router as export_router from .edit_plans_filter import router as filter_router router.include_router(export_router) +router.include_router(adjustments_router) router.include_router(filter_router) diff --git a/apps/api/app/api/routes/edit_plans_adjustments.py b/apps/api/app/api/routes/edit_plans_adjustments.py new file mode 100755 index 000000000..01adaf212 --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_adjustments.py @@ -0,0 +1,311 @@ +"""片段调整 API. + +- PUT /clips/{clip_id}/speed 调速 +- PUT /clips/{clip_id}/volume 音量调节 +- PUT /clips/{clip_id}/trim 裁剪（trim in/out） +- PUT /clips/{clip_id}/adjustments 统一调整（speed+volume+trim） +- POST /{plan_id}/clips/batch-speed 批量调速 +""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +from app.auth import AuthenticatedUser, get_current_user +from app.dependencies import get_db_session, get_project_repository +from app.services import EditPlanService +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from ._helpers import check_project_access + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +# ── Schemas ────────────────────────────────────────────────────────────────── + + +class SpeedAdjustRequest(BaseModel): + """调速请求""" + + speed: float = Field(..., ge=0.25, le=4.0, description="播放速度 0.25~4.0") + + +class VolumeAdjustRequest(BaseModel): + """音量调节请求""" + + volume: float = Field(..., ge=0.0, le=2.0, description="音量倍率 0~2.0（1.0=原音量）") + + +class TrimAdjustRequest(BaseModel): + """裁剪请求""" + + trim_start: float = Field(0.0, ge=0.0, description="开头裁剪秒数") + trim_end: float = Field(0.0, ge=0.0, description="结尾裁剪秒数") + + +class ClipAdjustmentsRequest(BaseModel): + """统一调整请求""" + + speed: Optional[float] = Field(default=None, ge=0.25, le=4.0) + volume: Optional[float] = Field(default=None, ge=0.0, le=2.0) + trim_start: Optional[float] = Field(default=None, ge=0.0) + trim_end: Optional[float] = Field(default=None, ge=0.0) + + +class BatchSpeedRequest(BaseModel): + """批量调速请求""" + + speed: float = Field(..., ge=0.25, le=4.0, description="播放速度") + + +class ClipAdjustResponse(BaseModel): + """片段调整响应""" + + clip_id: str + speed: float + volume: float + trim_start: float + trim_end: float + duration: float + + +class BatchSpeedResponse(BaseModel): + """批量调速响应""" + + updated_count: int + plan_id: str + + +# ── Helpers ────────────────────────────────────────────────────────────────── + + +def _get_clip_config(clip) -> dict: + config = getattr(clip, "config", {}) or {} + if not isinstance(config, dict): + config = {} + return config + + +def _get_volume(clip) -> float: + config = _get_clip_config(clip) + return float(config.get("volume", 1.0)) + + +def _get_trim(clip) -> tuple[float, float]: + config = _get_clip_config(clip) + trim_start = float(config.get("trim_start", 0.0)) + trim_end = float(config.get("trim_end", 0.0)) + return trim_start, trim_end + + +def _build_response(clip) -> ClipAdjustResponse: + trim_start, trim_end = _get_trim(clip) + return ClipAdjustResponse( + clip_id=clip.id, + speed=clip.playback_speed, + volume=_get_volume(clip), + trim_start=trim_start, + trim_end=trim_end, + duration=clip.duration, + ) + + +def _validate_trim(trim_start: float, trim_end: float, total_duration: float) -> None: + """验证裁剪时长不超过总时长""" + if trim_start + trim_end >= total_duration: + raise ValueError(f"裁剪总时长（{trim_start + trim_end:.2f}s）不能大于等于片段总时长（{total_duration:.2f}s）") + + +# ── Routes ─────────────────────────────────────────────────────────────────── + + +def _get_svc_and_plan_and_clip(db, clip_id, current_user, project_repository): + svc = EditPlanService(db) + clip = svc.get_clip(clip_id) + if not clip: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"片段不存在: {clip_id}", + ) + plan = svc.get_plan(clip.plan_id) + if plan and plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + return svc, plan, clip + + +@router.put("/clips/{clip_id}/speed", response_model=ClipAdjustResponse) +def adjust_speed( + clip_id: str, + body: SpeedAdjustRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> ClipAdjustResponse: + """调整片段播放速度""" + svc, _, clip = _get_svc_and_plan_and_clip(db, clip_id, current_user, project_repository) + + updated = svc.update_clip(clip_id, playback_speed=body.speed) + + logger.info( + "调整片段速度: clip_id=%s speed=%.2f by user=%s", + clip_id, + body.speed, + current_user.user.id, + ) + return _build_response(updated) + + +@router.put("/clips/{clip_id}/volume", response_model=ClipAdjustResponse) +def adjust_volume( + clip_id: str, + body: VolumeAdjustRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> ClipAdjustResponse: + """调整片段音量""" + svc, _, clip = _get_svc_and_plan_and_clip(db, clip_id, current_user, project_repository) + + # 更新 config.volume + config = dict(_get_clip_config(clip)) + config["volume"] = body.volume + updated = svc.update_clip(clip_id, config=config) + + logger.info( + "调整片段音量: clip_id=%s volume=%.2f by user=%s", + clip_id, + body.volume, + current_user.user.id, + ) + return _build_response(updated) + + +@router.put("/clips/{clip_id}/trim", response_model=ClipAdjustResponse) +def adjust_trim( + clip_id: str, + body: TrimAdjustRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> ClipAdjustResponse: + """裁剪片段（trim in/out）""" + svc, _, clip = _get_svc_and_plan_and_clip(db, clip_id, current_user, project_repository) + + # 验证裁剪时长 + try: + _validate_trim(body.trim_start, body.trim_end, clip.duration) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e + + # 更新 config + config = dict(_get_clip_config(clip)) + config["trim_start"] = body.trim_start + config["trim_end"] = body.trim_end + updated = svc.update_clip(clip_id, config=config) + + logger.info( + "裁剪片段: clip_id=%s trim_start=%.2f trim_end=%.2f by user=%s", + clip_id, + body.trim_start, + body.trim_end, + current_user.user.id, + ) + return _build_response(updated) + + +@router.put("/clips/{clip_id}/adjustments", response_model=ClipAdjustResponse) +def adjust_all( + clip_id: str, + body: ClipAdjustmentsRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> ClipAdjustResponse: + """统一调整片段的 speed / volume / trim""" + svc, _, clip = _get_svc_and_plan_and_clip(db, clip_id, current_user, project_repository) + + update_kwargs = {} + config_updates = {} + + if body.speed is not None: + update_kwargs["playback_speed"] = body.speed + + if body.volume is not None: + config_updates["volume"] = body.volume + + if body.trim_start is not None: + config_updates["trim_start"] = body.trim_start + + if body.trim_end is not None: + config_updates["trim_end"] = body.trim_end + + # 验证 trim + current_trim_start, current_trim_end = _get_trim(clip) + new_trim_start = body.trim_start if body.trim_start is not None else current_trim_start + new_trim_end = body.trim_end if body.trim_end is not None else current_trim_end + + if body.trim_start is not None or body.trim_end is not None: + try: + _validate_trim(new_trim_start, new_trim_end, clip.duration) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e + + if config_updates: + config = dict(_get_clip_config(clip)) + config.update(config_updates) + update_kwargs["config"] = config + + if not update_kwargs: + return _build_response(clip) + + updated = svc.update_clip(clip_id, **update_kwargs) + + logger.info( + "统一调整片段: clip_id=%s speed=%s volume=%s by user=%s", + clip_id, + body.speed, + body.volume, + current_user.user.id, + ) + return _build_response(updated) + + +@router.post("/{plan_id}/clips/batch-speed", response_model=BatchSpeedResponse) +def batch_adjust_speed( + plan_id: str, + body: BatchSpeedRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> BatchSpeedResponse: + """批量调整计划内所有片段的播放速度""" + svc = EditPlanService(db) + plan = svc.get_plan(plan_id) + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"剪辑计划不存在: {plan_id}", + ) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + clips = svc.list_clips(plan_id, limit=500, skip=0) + count = 0 + for clip in clips: + svc.update_clip(clip.id, playback_speed=body.speed) + count += 1 + + logger.info( + "批量调速: plan_id=%s count=%d speed=%.2f by user=%s", + plan_id, + count, + body.speed, + current_user.user.id, + ) + return BatchSpeedResponse(updated_count=count, plan_id=plan_id) diff --git a/tests/unit/test_edit_plan_adjustments_api.py b/tests/unit/test_edit_plan_adjustments_api.py new file mode 100644 index 000000000..171d589c1 --- /dev/null +++ b/tests/unit/test_edit_plan_adjustments_api.py @@ -0,0 +1,491 @@ +""" +片段调整 API 单元测试 + +覆盖： +- PUT /clips/{clip_id}/speed - 调速 +- PUT /clips/{clip_id}/volume - 音量调节 +- PUT /clips/{clip_id}/trim - 裁剪 +- PUT /clips/{clip_id}/adjustments - 统一调整 +- POST /{plan_id}/clips/batch-speed - 批量调速 +""" + +from __future__ import annotations + +import os +import sys +from datetime import datetime +from pathlib import Path +from typing import Optional +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") + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from packages.domain.config_schemas import normalize_plan_config +from packages.domain.edit_plan import EditPlan, EditPlanStatus +from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus + +# --------------------------------------------------------------------------- +# Stub Repository +# --------------------------------------------------------------------------- + + +class StubEditPlanRepository: + def __init__(self, plans: dict[str, EditPlan] | None = None): + self._plans = plans or {} + + def get(self, plan_id: str) -> Optional[EditPlan]: + return self._plans.get(plan_id) + + def create(self, plan: EditPlan) -> EditPlan: + self._plans[plan.id] = plan + return plan + + def update(self, plan: EditPlan) -> EditPlan: + self._plans[plan.id] = plan + return plan + + def list_all(self, *, status=None, skip=0, limit=50): + return list(self._plans.values())[skip : skip + limit] + + def list_by_template(self, template_id, *, status=None, skip=0, limit=50): + return [p for p in self._plans.values() if p.template_id == template_id][skip : skip + limit] + + def delete(self, plan_id: str) -> bool: + return self._plans.pop(plan_id, None) is not None + + def count(self, *, status=None, template_id=None): + return len(self._plans) + + +class StubEditPlanClipRepository: + def __init__(self, clips: dict[str, EditPlanClip] | None = None): + self._clips = clips or {} + + def list_by_plan(self, plan_id, *, status=None, skip=0, limit=100): + items = [c for c in self._clips.values() if c.plan_id == plan_id] + items.sort(key=lambda c: c.order) + return items[skip : skip + limit] + + def count(self, plan_id, *, status=None): + return len([c for c in self._clips.values() if c.plan_id == plan_id]) + + def get(self, clip_id: str) -> Optional[EditPlanClip]: + return self._clips.get(clip_id) + + def create(self, clip: EditPlanClip) -> EditPlanClip: + self._clips[clip.id] = clip + return clip + + def update(self, clip: EditPlanClip) -> EditPlanClip: + self._clips[clip.id] = clip + return clip + + def delete(self, clip_id: str) -> bool: + return self._clips.pop(clip_id, None) is not None + + def delete_by_plan(self, plan_id: str) -> int: + before = len(self._clips) + self._clips = {k: v for k, v in self._clips.items() if v.plan_id != plan_id} + return before - len(self._clips) + + +# --------------------------------------------------------------------------- +# Test Fixtures +# --------------------------------------------------------------------------- + + +def _make_sample_plan(plan_id="plan-001"): + return EditPlan( + id=plan_id, + template_id="tpl-001", + name="测试计划", + status=EditPlanStatus.EDITING, + total_duration=30.0, + config=normalize_plan_config({}), + project_id="", + created_by_user_id="user-001", + created_at=datetime(2026, 7, 16, 10, 0, 0), + updated_at=datetime(2026, 7, 16, 10, 0, 0), + ) + + +def _make_clip(clip_id, plan_id="plan-001", order=0, duration=10.0, speed=1.0): + return EditPlanClip( + id=clip_id, + plan_id=plan_id, + clip_type="video", + order=order, + asset_id="asset-001", + text_content="", + start_time=0.0, + duration=duration, + transition_effect="cut", + transition_duration=0.0, + playback_speed=speed, + status=EditPlanClipStatus.READY, + config={}, + created_at=datetime(2026, 7, 16, 10, 0, 0), + updated_at=datetime(2026, 7, 16, 10, 0, 0), + ) + + +def _create_test_app(): + import app.services.edit_plan_service as service_module + from app.api.routes.edit_plans import router + + plan = _make_sample_plan() + clips = { + "clip-001": _make_clip("clip-001", order=0, duration=10.0), + "clip-002": _make_clip("clip-002", order=1, duration=15.0), + "clip-003": _make_clip("clip-003", order=2, duration=20.0), + } + stub_plan_repo = StubEditPlanRepository({plan.id: plan}) + stub_clip_repo = StubEditPlanClipRepository(clips) + + original_plan_repo = service_module.SQLAlchemyEditPlanRepository + original_clip_repo = service_module.SQLAlchemyEditPlanClipRepository + original_gen_repo = service_module.SQLAlchemyGenerationTaskRepository + service_module.SQLAlchemyEditPlanRepository = lambda db: stub_plan_repo + service_module.SQLAlchemyEditPlanClipRepository = lambda db: stub_clip_repo + service_module.SQLAlchemyGenerationTaskRepository = lambda db: MagicMock() + + app = FastAPI() + app.include_router(router, prefix="/api/v1/edit-plans") + + def _mock_auth(): + mock = MagicMock() + mock.user.id = "user-001" + return mock + + import app.api.routes._helpers as helpers_module + + original_check = helpers_module.check_project_access + helpers_module.check_project_access = lambda *a, **kw: None + + from app.api.routes import edit_plans as main_module + + app.dependency_overrides[main_module.get_current_user] = _mock_auth + app.dependency_overrides[main_module.get_db_session] = lambda: MagicMock() + app.dependency_overrides[main_module.get_project_repository] = lambda: MagicMock() + + import app.api.routes.edit_plans_adjustments as adj_module + + app.dependency_overrides[adj_module.get_current_user] = _mock_auth + app.dependency_overrides[adj_module.get_db_session] = lambda: MagicMock() + app.dependency_overrides[adj_module.get_project_repository] = lambda: MagicMock() + + def cleanup(): + service_module.SQLAlchemyEditPlanRepository = original_plan_repo + service_module.SQLAlchemyEditPlanClipRepository = original_clip_repo + service_module.SQLAlchemyGenerationTaskRepository = original_gen_repo + helpers_module.check_project_access = original_check + + return app, stub_plan_repo, stub_clip_repo, cleanup + + +@pytest.fixture +def adj_client(): + app, plan_repo, clip_repo, cleanup = _create_test_app() + yield TestClient(app), plan_repo, clip_repo + cleanup() + + +# --------------------------------------------------------------------------- +# 调速测试 +# --------------------------------------------------------------------------- + + +class TestAdjustSpeed: + def test_speed_up(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 2.0}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["speed"] == 2.0 + assert data["clip_id"] == "clip-001" + + clip = clip_repo.get("clip-001") + assert clip.playback_speed == 2.0 + + def test_slow_down(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 0.5}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["speed"] == 0.5 + + def test_speed_clip_not_found(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-nonexist/speed", + json={"speed": 1.5}, + ) + assert resp.status_code == 404 + + def test_speed_out_of_range_low(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 0.1}, + ) + assert resp.status_code == 422 + + def test_speed_out_of_range_high(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 5.0}, + ) + assert resp.status_code == 422 + + def test_speed_default_value(self, adj_client): + c, _, clip_repo = adj_client + # 验证默认 speed + clip = clip_repo.get("clip-001") + assert clip.playback_speed == 1.0 + + +# --------------------------------------------------------------------------- +# 音量调节测试 +# --------------------------------------------------------------------------- + + +class TestAdjustVolume: + def test_set_volume(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/volume", + json={"volume": 0.5}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["volume"] == 0.5 + + clip = clip_repo.get("clip-001") + assert clip.config["volume"] == 0.5 + + def test_mute(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/volume", + json={"volume": 0.0}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["volume"] == 0.0 + + def test_boost_volume(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/volume", + json={"volume": 1.5}, + ) + assert resp.status_code == 200 + assert resp.json()["volume"] == 1.5 + + def test_volume_out_of_range(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/volume", + json={"volume": 3.0}, + ) + assert resp.status_code == 422 + + def test_volume_clip_not_found(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-nonexist/volume", + json={"volume": 1.0}, + ) + assert resp.status_code == 404 + + def test_default_volume(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 1.0}, + ) + data = resp.json() + # 默认音量应该是 1.0 + assert data["volume"] == 1.0 + + +# --------------------------------------------------------------------------- +# 裁剪测试 +# --------------------------------------------------------------------------- + + +class TestAdjustTrim: + def test_trim_start(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/trim", + json={"trim_start": 2.0, "trim_end": 0.0}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["trim_start"] == 2.0 + assert data["trim_end"] == 0.0 + + clip = clip_repo.get("clip-001") + assert clip.config["trim_start"] == 2.0 + + def test_trim_both_ends(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/trim", + json={"trim_start": 1.5, "trim_end": 2.5}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["trim_start"] == 1.5 + assert data["trim_end"] == 2.5 + + def test_trim_exceeds_duration(self, adj_client): + c, _, _ = adj_client + # 片段时长 10 秒，裁剪 8+3 = 11 > 10 + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/trim", + json={"trim_start": 8.0, "trim_end": 3.0}, + ) + assert resp.status_code == 400 + assert "不能大于等于片段总时长" in resp.json()["detail"] + + def test_trim_clip_not_found(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-nonexist/trim", + json={"trim_start": 1.0}, + ) + assert resp.status_code == 404 + + def test_default_trim_zero(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/speed", + json={"speed": 1.0}, + ) + data = resp.json() + assert data["trim_start"] == 0.0 + assert data["trim_end"] == 0.0 + + +# --------------------------------------------------------------------------- +# 统一调整测试 +# --------------------------------------------------------------------------- + + +class TestAdjustAll: + def test_adjust_speed_and_volume(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/adjustments", + json={"speed": 1.5, "volume": 0.8}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["speed"] == 1.5 + assert data["volume"] == 0.8 + + clip = clip_repo.get("clip-001") + assert clip.playback_speed == 1.5 + assert clip.config["volume"] == 0.8 + + def test_adjust_all_four(self, adj_client): + c, _, clip_repo = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/adjustments", + json={"speed": 2.0, "volume": 0.5, "trim_start": 1.0, "trim_end": 1.0}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["speed"] == 2.0 + assert data["volume"] == 0.5 + assert data["trim_start"] == 1.0 + assert data["trim_end"] == 1.0 + + clip = clip_repo.get("clip-001") + assert clip.playback_speed == 2.0 + assert clip.config["volume"] == 0.5 + assert clip.config["trim_start"] == 1.0 + assert clip.config["trim_end"] == 1.0 + + def test_adjust_empty_body(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/adjustments", + json={}, + ) + assert resp.status_code == 200 + data = resp.json() + # 保持默认值 + assert data["speed"] == 1.0 + assert data["volume"] == 1.0 + + def test_adjust_trim_exceeds(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-001/adjustments", + json={"trim_start": 9.0, "trim_end": 2.0}, + ) + assert resp.status_code == 400 + + def test_adjust_clip_not_found(self, adj_client): + c, _, _ = adj_client + resp = c.put( + "/api/v1/edit-plans/clips/clip-nonexist/adjustments", + json={"speed": 1.5}, + ) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# 批量调速测试 +# --------------------------------------------------------------------------- + + +class TestBatchSpeed: + def test_batch_speed_all(self, adj_client): + c, _, clip_repo = adj_client + resp = c.post( + "/api/v1/edit-plans/plan-001/clips/batch-speed", + json={"speed": 1.5}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["updated_count"] == 3 + assert data["plan_id"] == "plan-001" + + for cid in ["clip-001", "clip-002", "clip-003"]: + clip = clip_repo.get(cid) + assert clip.playback_speed == 1.5 + + def test_batch_speed_plan_not_found(self, adj_client): + c, _, _ = adj_client + resp = c.post( + "/api/v1/edit-plans/plan-nonexist/clips/batch-speed", + json={"speed": 2.0}, + ) + assert resp.status_code == 404 + + def test_batch_speed_invalid(self, adj_client): + c, _, _ = adj_client + resp = c.post( + "/api/v1/edit-plans/plan-001/clips/batch-speed", + json={"speed": 10.0}, + ) + assert resp.status_code == 422 -- 2.54.0