xiaoxia-saas/tests/unit/test_render_adapter.py

"""RenderAdapter 单元测试 — Phase 2.

测试适配层的计划加载、素材下载、引擎调用、结果上传等逻辑。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any
from unittest.mock import MagicMock, patch

from video_processing.render_adapter import RenderAdapter

# ── Fixtures ──────────────────────────────────────────────────────────────────


@dataclass
class FakeClip:
    """模拟 EditPlanClip。"""

    id: str
    plan_id: str = "plan_001"
    clip_type: str = "main"
    order: int = 0
    asset_id: str = ""
    text_content: str = ""
    start_time: float = 0.0
    duration: float = 0.0
    transition_effect: str = "cut"
    status: str = "ready"
    config: dict[str, Any] = field(default_factory=dict)


@dataclass
class FakePlan:
    """模拟 EditPlan。"""

    id: str = "plan_001"
    name: str = "测试计划"
    status: str = "editing"
    config: dict[str, Any] = field(default_factory=dict)


def _make_clip(
    clip_id: str,
    clip_type: str = "main",
    order: int = 0,
    asset_id: str | None = None,
    duration: float = 5.0,
    status: str = "ready",
    transition_effect: str = "cut",
    config: dict[str, Any] | None = None,
) -> FakeClip:
    # asset_id 为 None 时生成默认值，为空字符串时保留空串
    if asset_id is None:
        asset_id = f"asset_{clip_id}.mp4"
    return FakeClip(
        id=clip_id,
        clip_type=clip_type,
        order=order,
        asset_id=asset_id,
        duration=duration,
        status=status,
        transition_effect=transition_effect,
        config=config or {},
    )


def _make_adapter(
    plan: FakePlan | None = None,
    clips: list[FakeClip] | None = None,
) -> tuple[RenderAdapter, MagicMock, MagicMock]:
    """创建测试用的 RenderAdapter 及 mock repo。

    Returns:
        (adapter, mock_plan_repo, mock_clip_repo)
    """
    mock_db = MagicMock()
    adapter = RenderAdapter(mock_db)

    # 替换内部 repo
    mock_plan_repo = MagicMock()
    mock_clip_repo = MagicMock()
    adapter._plan_repo = mock_plan_repo
    adapter._clip_repo = mock_clip_repo

    # 设置默认返回
    if plan is not None:
        mock_plan_repo.get.return_value = plan
    if clips is not None:
        mock_clip_repo.list_by_plan.return_value = clips

    return adapter, mock_plan_repo, mock_clip_repo


# ── validate_plan 测试 ───────────────────────────────────────────────────────


class TestValidatePlan:
    def test_plan_not_found(self):
        """计划不存在时校验失败。"""
        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
        mock_plan_repo.get.return_value = None

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert not valid
        assert len(errors) == 1
        assert "不存在" in errors[0]
        assert ready_count == 0
        assert total_count == 0

    def test_no_clips(self):
        """没有任何片段时校验失败。"""
        plan = FakePlan(id="plan_001", status="editing")
        adapter, _, mock_clip_repo = _make_adapter(plan=plan, clips=[])

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert not valid
        assert any("没有任何片段" in e for e in errors)

    def test_no_ready_clips(self):
        """没有 ready 片段时校验失败。"""
        plan = FakePlan(id="plan_001", status="editing")
        clips = [
            _make_clip("c1", status="pending"),
            _make_clip("c2", status="pending"),
        ]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert not valid
        assert any("没有就绪" in e for e in errors)
        assert ready_count == 0
        assert total_count == 2

    def test_ready_clip_no_asset(self):
        """ready 片段没有 asset_id 时报错。"""
        plan = FakePlan(id="plan_001", status="editing")
        clips = [
            _make_clip("c1", asset_id=""),
        ]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert not valid
        assert any("没有分配素材" in e for e in errors)

    def test_valid_plan(self):
        """正常计划校验通过。"""
        plan = FakePlan(id="plan_001", status="editing")
        clips = [
            _make_clip("c1", order=0, duration=3.0),
            _make_clip("c2", order=1, duration=4.0),
        ]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert valid
        assert len(errors) == 0
        assert ready_count == 2
        assert total_count == 2

    def test_wrong_status(self):
        """计划状态不正确时报错。"""
        plan = FakePlan(id="plan_001", status="draft")
        clips = [_make_clip("c1")]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        valid, errors, _, _, _ = adapter.validate_plan("plan_001")

        assert not valid
        assert any("状态不正确" in e for e in errors)

    def test_mixed_status_with_warnings(self):
        """混合状态时有 pending/failed 警告。"""
        plan = FakePlan(id="plan_001", status="editing")
        clips = [
            _make_clip("c1", order=0, status="ready"),
            _make_clip("c2", order=1, status="pending"),
            _make_clip("c3", order=2, status="failed"),
        ]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")

        assert valid
        assert any("pending" in w for w in warnings)
        assert any("failed" in w for w in warnings)
        assert ready_count == 1
        assert total_count == 3


# ── render_plan 测试 ─────────────────────────────────────────────────────────


class TestRenderPlan:
    def test_plan_not_found(self):
        """计划不存在时返回失败。"""
        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
        mock_plan_repo.get.return_value = None

        result = adapter.render_plan("plan_001")

        assert not result.success
        assert "不存在" in result.error_message

    def test_no_ready_clips(self):
        """没有 ready 片段时返回失败。"""
        plan = FakePlan(id="plan_001", status="editing")
        clips = [_make_clip("c1", status="pending")]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        result = adapter.render_plan("plan_001")

        assert not result.success
        assert "没有可渲染" in result.error_message
        assert result.clip_count == 0

    @patch("video_processing.render_adapter.download_asset")
    def test_all_assets_download_fail(self, mock_download):
        """所有素材下载失败时返回失败。"""
        mock_download.return_value = False

        plan = FakePlan(id="plan_001", status="editing")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        result = adapter.render_plan("plan_001")

        assert not result.success
        assert "素材下载失败" in result.error_message

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_successful_render(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """完整渲染流程成功。"""

        # 素材下载成功
        def _fake_download(asset_id, local_path):
            local_path.parent.mkdir(parents=True, exist_ok=True)
            local_path.write_bytes(b"fake video data")
            return True

        mock_download.side_effect = _fake_download

        # 渲染成功
        mock_render = MagicMock()
        mock_render.render.return_value = MagicMock(
            output_path=tmp_path / "output.mp4",
            duration=10.0,
            file_size=102400,
            width=1280,
            height=720,
        )
        mock_render_cls.return_value = mock_render

        # 上传成功
        mock_upload.return_value = "https://oss.example.com/rendered/plan_001/job_001.mp4"

        plan = FakePlan(id="plan_001", status="editing")
        clips = [
            _make_clip("c1", order=0, duration=5.0),
            _make_clip("c2", order=1, duration=5.0),
        ]
        adapter, _, _ = _make_adapter(plan=plan, clips=clips)

        result = adapter.render_plan(
            "plan_001",
            job_id="job_001",
            work_dir=tmp_path / "work",
        )

        assert result.success
        assert result.output_url.startswith("https://")
        assert result.duration == 10.0
        assert result.width == 1280
        assert result.height == 720
        assert result.clip_count == 2

        # 验证 UnifiedRenderService 被正确调用
        mock_render_cls.assert_called_once()
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["plan"] is plan
        assert len(call_kwargs.kwargs["clips"]) == 2
        assert len(call_kwargs.kwargs["asset_path_map"]) == 2

    @patch("video_processing.render_adapter.download_asset")
    def test_progress_callback(self, mock_download, tmp_path):
        """进度回调被正确触发。"""

        def _fake_download(asset_id, local_path):
            local_path.parent.mkdir(parents=True, exist_ok=True)
            local_path.write_bytes(b"fake data")
            return True

        mock_download.side_effect = _fake_download

        # 模拟渲染异常，避免走到最后
        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
            mock_render = MagicMock()
            mock_render.render.side_effect = RuntimeError("render error")
            mock_render_cls.return_value = mock_render

            plan = FakePlan(id="plan_001", status="editing")
            clips = [_make_clip("c1", order=0, duration=5.0)]
            adapter, _, _ = _make_adapter(plan=plan, clips=clips)

            progress_values = []

            def progress_cb(progress: float, stage: str) -> None:
                progress_values.append((progress, stage))

            adapter.render_plan(
                "plan_001",
                work_dir=tmp_path / "work",
                progress_cb=progress_cb,
            )

            # 即使渲染失败，前期进度也应该上报了
            assert len(progress_values) > 0
            # 第一个进度应该是加载计划
            assert progress_values[0][1] == "加载剪辑计划"

    @patch("video_processing.render_adapter.download_asset")
    def test_partial_asset_download(self, mock_download, tmp_path):
        """部分素材下载失败时，只使用成功的素材。"""
        download_results = [True, False, True]  # 3个素材中2个成功

        def _fake_download(asset_id, local_path):
            idx = hash(asset_id) % 3
            if download_results[idx]:
                local_path.parent.mkdir(parents=True, exist_ok=True)
                local_path.write_bytes(b"fake data")
                return True
            return False

        mock_download.side_effect = _fake_download

        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
            mock_render = MagicMock()
            mock_render.render.return_value = MagicMock(
                output_path=tmp_path / "out.mp4",
                duration=5.0,
                file_size=1024,
                width=1280,
                height=720,
            )
            mock_render_cls.return_value = mock_render

            with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"):
                plan = FakePlan(id="plan_001", status="editing")
                clips = [
                    _make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"),
                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"),
                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"),
                ]
                adapter, _, _ = _make_adapter(plan=plan, clips=clips)

                result = adapter.render_plan(
                    "plan_001",
                    work_dir=tmp_path / "work",
                )

                # 至少有部分素材成功，渲染应该进行
                # （具体成功数量取决于 hash 结果，但至少1个成功就能渲染）
                assert result.success or "素材下载失败" in result.error_message


# ── _download_assets 测试 ────────────────────────────────────────────────────


class TestDownloadAssets:
    @patch("video_processing.render_adapter.download_asset")
    def test_all_download_success(self, mock_download, tmp_path):
        """全部素材下载成功。"""
        mock_download.return_value = True

        clips = [
            _make_clip("c1", order=0, asset_id="key1.mp4"),
            _make_clip("c2", order=1, asset_id="key2.mp4"),
        ]

        result = RenderAdapter._download_assets(clips, tmp_path)

        assert len(result) == 2
        assert "key1.mp4" in result
        assert "key2.mp4" in result
        assert mock_download.call_count == 2

    @patch("video_processing.render_adapter.download_asset")
    def test_empty_asset_id_skipped(self, mock_download, tmp_path):
        """空 asset_id 的片段被跳过。"""
        clips = [
            _make_clip("c1", order=0, asset_id=""),
            _make_clip("c2", order=1, asset_id="key2.mp4"),
        ]
        mock_download.return_value = True

        result = RenderAdapter._download_assets(clips, tmp_path)

        assert len(result) == 1
        assert "key2.mp4" in result
        assert mock_download.call_count == 1  # 只调用了一次下载

    @patch("video_processing.render_adapter.download_asset")
    def test_all_download_fail(self, mock_download, tmp_path):
        """全部下载失败返回空字典。"""
        mock_download.return_value = False

        clips = [
            _make_clip("c1", order=0, asset_id="key1.mp4"),
        ]

        result = RenderAdapter._download_assets(clips, tmp_path)

        assert len(result) == 0