xiaoxia-saas/tests/unit/test_edit_plan_adjustments_api.py

"""
片段调整 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