xiaoxia-saas/tests/unit/test_render_adapter.py

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

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

from __future__ import annotations

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

import pytest
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,
    asset_url_map: dict[str, str] | None = None,
) -> tuple[RenderAdapter, MagicMock, MagicMock]:
    """创建测试用的 RenderAdapter 及 mock repo。

    Args:
        plan: 模拟的剪辑计划
        clips: 模拟的片段列表
        asset_url_map: asset_id → storage_key 映射，用于 mock assets 表查询

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

    # mock assets 表查询：_download_assets 用
    if asset_url_map is not None:
        mock_query = MagicMock()
        mock_assets = []
        for aid, url in asset_url_map.items():
            m = MagicMock()
            m.id = aid
            m.storage_key = url
            mock_assets.append(m)
        mock_query.all.return_value = mock_assets
        mock_query.filter.return_value = mock_query
        mock_db.query.return_value = mock_query

    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


@pytest.fixture(autouse=True)
def _patch_validation():
    """Mock video validation to always pass in tests (mock renders don't create real files)."""
    from video_processing.video_validation import VideoValidationResult

    with patch(
        "video_processing.render_adapter.validate_video_output",
        return_value=VideoValidationResult(valid=True),
    ):
        yield


# ── 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)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        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(storage_key, 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),
        ]
        asset_url_map = {
            "asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4",
            "asset_c2.mp4": "https://test-bucket.oss.com/assets/asset_c2.mp4",
        }
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        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
        # 缩略图URL（即使生成失败也应该是空串，不为None）
        assert hasattr(result, "thumbnail_url")

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    @pytest.mark.skip(reason="thumbnail_url mock 与当前代码不匹配，待更新")
    def test_thumbnail_generated_on_success(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """渲染成功后生成缩略图，thumbnail_url 正确返回。"""

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

        mock_download.side_effect = _fake_download

        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
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(id="plan_thumb")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        # Mock extract_first_frame to return a temp file path
        import tempfile as _tf

        _fake_frame = _tf.NamedTemporaryFile(suffix=".jpg", delete=False)
        _fake_frame.write(b"fake frame")
        _fake_frame.close()
        with patch(
            "video_processing.thumbnail_generator.extract_first_frame",
            return_value=_fake_frame.name,
        ):
            result = adapter.render_plan(
                "plan_thumb",
                work_dir=tmp_path / "work",
            )
        from pathlib import Path as _P

        _P(_fake_frame.name).unlink(missing_ok=True)

        assert result.success
        # cover_url from upload_to_oss (mocked globally)
        assert result.thumbnail_url == "https://oss.example.com/out.mp4"

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    @pytest.mark.skip(reason="thumbnail_url mock 与当前代码不匹配，待更新")
    def test_thumbnail_failure_does_not_block(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """缩略图生成失败不影响主流程，thumbnail_url 为空串。"""

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

        mock_download.side_effect = _fake_download

        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
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(id="plan_thumb_fail")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        with patch(
            "video_processing.thumbnail_generator.extract_first_frame",
            side_effect=RuntimeError("ffmpeg not available"),
        ):
            result = adapter.render_plan(
                "plan_thumb_fail",
                work_dir=tmp_path / "work",
            )

        assert result.success  # 不阻断
        assert result.thumbnail_url == ""

        # 验证 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"]) == 1
        assert len(call_kwargs.kwargs["asset_path_map"]) == 1

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_resolution_from_config(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """输出分辨率从 plan.config.export.resolution 读取并传给 UnifiedRenderService。"""

        def _fake_download(storage_key, 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=1080,
            height=1920,
        )
        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",
            config={"export": {"resolution": "1080x1920", "fps": 30}},
        )
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

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

        assert result.success
        mock_render_cls.assert_called_once()
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["output_width"] == 1080
        assert call_kwargs.kwargs["output_height"] == 1920

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_resolution_default_when_missing(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """plan.config 无 export 配置时使用默认分辨率 1080x1920。"""

        def _fake_download(storage_key, 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=1080,
            height=1920,
        )
        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", config={})
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

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

        assert result.success
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["output_width"] == 1080
        assert call_kwargs.kwargs["output_height"] == 1920

    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_ffmpeg_error_captures_stderr(self, mock_download, mock_render_cls, tmp_path):
        """ffmpeg CalledProcessError 时 stderr 写入 error_detail 和 error_message。"""
        import subprocess

        def _fake_download(storage_key, 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

        mock_render = MagicMock()
        mock_render.render.side_effect = subprocess.CalledProcessError(
            returncode=1,
            cmd=["ffmpeg", "-i", "input.mp4", "output.mp4"],
            stderr="Invalid data found when processing input\nLast message repeated 3 times",
        )
        mock_render_cls.return_value = mock_render

        plan = FakePlan(id="plan_001", status="editing")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

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

        assert not result.success
        assert "FFmpeg渲染失败" in result.error_message
        assert "exit=1" in result.error_message
        assert "Invalid data found" in result.error_detail
        assert "Invalid data found" in result.error_message

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

        def _fake_download(storage_key, 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)]
            asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
            adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

            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(storage_key, local_path):
            idx = hash(storage_key) % 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"),
                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002"),
                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003"),
                ]
                asset_url_map = {
                    "asset_001": "https://test-bucket.oss.com/assets/001.mp4",
                    "asset_002": "https://test-bucket.oss.com/assets/002.mp4",
                    "asset_003": "https://test-bucket.oss.com/assets/003.mp4",
                }
                adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

                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:
    @staticmethod
    def _make_mock_db(asset_url_map: dict[str, str]):
        """构造 mock db，根据 asset_id 返回对应的 AssetModel.storage_key。"""
        mock_db = MagicMock()
        mock_query = MagicMock()

        def _fake_filter(query):
            # 模拟 .filter(AssetModel.id.in_([...])).all()
            mock_assets = []
            for asset_id, storage_key in asset_url_map.items():
                mock_asset = MagicMock()
                mock_asset.id = asset_id
                mock_asset.storage_key = storage_key
                mock_assets.append(mock_asset)
            mock_query.all.return_value = mock_assets
            return mock_query

        mock_query.filter.side_effect = _fake_filter
        mock_db.query.return_value = mock_query
        return mock_db

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

        asset_url_map = {
            "asset_001": "https://bucket.oss.com/videos/key1.mp4",
            "asset_002": "https://bucket.oss.com/videos/key2.mp4",
        }
        mock_db = self._make_mock_db(asset_url_map)
        adapter = RenderAdapter(mock_db)

        clips = [
            _make_clip("c1", order=0, asset_id="asset_001"),
            _make_clip("c2", order=1, asset_id="asset_002"),
        ]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 2
        assert "asset_001" in asset_path_map
        assert "asset_002" in asset_path_map
        assert len(rendered_ids) == 2
        assert len(failed_ids) == 0
        assert mock_download.call_count == 2
        # 验证传给 download_asset 的是 storage_key 而非 asset_id
        call_keys = [call[0][0] for call in mock_download.call_args_list]
        assert "https://bucket.oss.com/videos/key1.mp4" in call_keys
        assert "https://bucket.oss.com/videos/key2.mp4" in call_keys

    @patch("video_processing.render_adapter.download_asset")
    def test_empty_asset_id_skipped(self, mock_download, tmp_path):
        """空 asset_id 的片段被跳过。"""
        mock_download.return_value = True

        asset_url_map = {"asset_002": "https://bucket.oss.com/videos/key2.mp4"}
        mock_db = self._make_mock_db(asset_url_map)
        adapter = RenderAdapter(mock_db)

        clips = [
            _make_clip("c1", order=0, asset_id=""),
            _make_clip("c2", order=1, asset_id="asset_002"),
        ]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert "asset_002" in asset_path_map
        assert "c1" in failed_ids
        assert "c2" in rendered_ids
        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

        asset_url_map = {"asset_001": "https://bucket.oss.com/videos/key1.mp4"}
        mock_db = self._make_mock_db(asset_url_map)
        adapter = RenderAdapter(mock_db)

        clips = [
            _make_clip("c1", order=0, asset_id="asset_001"),
        ]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 0
        assert len(rendered_ids) == 0
        assert "c1" in failed_ids

    @patch("video_processing.render_adapter.download_asset")
    def test_partial_download_failure(self, mock_download, tmp_path):
        """部分下载失败时正确区分成功/失败。"""
        results = {
            "https://bucket.oss.com/videos/key1.mp4": True,
            "https://bucket.oss.com/videos/key2.mp4": False,
            "https://bucket.oss.com/videos/key3.mp4": True,
        }

        def _fake_download(storage_key, local_path):
            local_path.parent.mkdir(parents=True, exist_ok=True)
            local_path.write_bytes(b"fake")
            return results.get(storage_key, False)

        mock_download.side_effect = _fake_download

        asset_url_map = {
            "asset_001": "https://bucket.oss.com/videos/key1.mp4",
            "asset_002": "https://bucket.oss.com/videos/key2.mp4",
            "asset_003": "https://bucket.oss.com/videos/key3.mp4",
        }
        mock_db = self._make_mock_db(asset_url_map)
        adapter = RenderAdapter(mock_db)

        clips = [
            _make_clip("c1", order=0, asset_id="asset_001"),
            _make_clip("c2", order=1, asset_id="asset_002"),
            _make_clip("c3", order=2, asset_id="asset_003"),
        ]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 2
        assert "c1" in rendered_ids
        assert "c3" in rendered_ids
        assert "c2" in failed_ids

    @patch("video_processing.render_adapter.download_asset")
    def test_duplicate_asset_downloaded_once(self, mock_download, tmp_path):
        """同一素材被多个 clip 引用时只下载一次。"""
        mock_download.return_value = True

        asset_url_map = {"asset_shared": "https://bucket.oss.com/videos/shared.mp4"}
        mock_db = self._make_mock_db(asset_url_map)
        adapter = RenderAdapter(mock_db)

        clips = [
            _make_clip("c1", order=0, asset_id="asset_shared"),
            _make_clip("c2", order=1, asset_id="asset_shared"),
        ]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert mock_download.call_count == 1
        assert "c1" in rendered_ids
        assert "c2" in rendered_ids

    @patch("video_processing.render_adapter.download_asset")
    def test_asset_without_storage_key_fallback_to_file_url(self, mock_download, tmp_path):
        """storage_key为空但file_url有值时，fallback到file_url下载成功。"""
        mock_download.return_value = True
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_asset = MagicMock()
        mock_asset.id = "asset_fallback"
        mock_asset.storage_key = ""
        mock_asset.file_url = "videos/test_fallback.mp4"
        mock_query.all.return_value = [mock_asset]
        mock_query.filter.return_value = mock_query
        mock_db.query.return_value = mock_query

        adapter = RenderAdapter(mock_db)
        clips = [_make_clip("c1", order=0, asset_id="asset_fallback")]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert "c1" in rendered_ids
        mock_download.assert_called_once()
        # 验证用的是file_url的值
        call_args = mock_download.call_args
        assert call_args[0][0] == "videos/test_fallback.mp4"

    @patch("video_processing.render_adapter.download_asset")
    def test_asset_without_storage_key_and_file_url_skipped(self, mock_download, tmp_path):
        """storage_key和file_url都为空时跳过下载，标记为失败。"""
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_asset = MagicMock()
        mock_asset.id = "asset_no_key"
        mock_asset.storage_key = ""
        mock_asset.file_url = ""
        mock_query.all.return_value = [mock_asset]
        mock_query.filter.return_value = mock_query
        mock_db.query.return_value = mock_query

        adapter = RenderAdapter(mock_db)
        clips = [_make_clip("c1", order=0, asset_id="asset_no_key")]

        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 0
        assert "c1" in failed_ids
        mock_download.assert_not_called()


# ── _prepare_bgm 测试 ────────────────────────────────────────────────────────


class TestPrepareBgm:
    """BGM 音频准备逻辑测试。"""

    def test_bgm_disabled_returns_none(self, tmp_path):
        """BGM 未启用时返回 None。"""
        plan = FakePlan(config={"bgm": {"enabled": False}})
        adapter = RenderAdapter(MagicMock())
        result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
        assert result is None

    def test_bgm_no_config_returns_none(self, tmp_path):
        """没有 BGM 配置时返回 None。"""
        plan = FakePlan(config={})
        adapter = RenderAdapter(MagicMock())
        result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
        assert result is None

    def test_bgm_from_url(self, tmp_path):
        """从 URL 下载 BGM。"""
        plan = FakePlan(
            config={
                "bgm": {
                    "enabled": True,
                    "audio_url": "https://example.com/bgm.mp3",
                }
            }
        )
        adapter = RenderAdapter(MagicMock())

        with patch("video_processing.url_security.safe_download_file") as mock_download:

            def _fake_download(url, path, **kwargs):
                Path(path).parent.mkdir(parents=True, exist_ok=True)
                Path(path).write_bytes(b"fake mp3 data")

            mock_download.side_effect = _fake_download

            result = adapter._prepare_bgm(plan, tmp_path, "plan_001")

        assert result is not None
        assert Path(result).exists()

    def test_bgm_from_url_failure_falls_through(self, tmp_path):
        """URL 下载失败时不抛异常，继续尝试其他来源。"""
        plan = FakePlan(
            config={
                "bgm": {
                    "enabled": True,
                    "audio_url": "https://example.com/bgm.mp3",
                    "preset_id": "preset_001",
                }
            }
        )
        adapter = RenderAdapter(MagicMock())

        with patch("video_processing.url_security.safe_download_file") as mock_download:
            mock_download.side_effect = Exception("download failed")

            # 预设库返回 None（没有这个预设）
            with patch("packages.domain.preset_bgm.get_preset_bgm") as mock_preset:
                mock_preset.return_value = None
                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")

        assert result is None  # 所有来源都失败时返回 None

    def test_bgm_from_asset_library(self, tmp_path):
        """从素材库下载 BGM。"""
        plan = FakePlan(
            config={
                "bgm": {
                    "enabled": True,
                    "asset_id": "asset_bgm_001",
                }
            }
        )

        mock_db = MagicMock()
        mock_asset = MagicMock()
        mock_asset.file_url = "oss://bgm/sample.mp3"
        mock_query = MagicMock()
        mock_query.first.return_value = mock_asset
        mock_db.query.return_value = mock_query

        adapter = RenderAdapter(mock_db)

        with patch("video_processing.render_adapter.download_asset") as mock_download:

            def _fake_download(storage_key, local_path):
                Path(local_path).parent.mkdir(parents=True, exist_ok=True)
                Path(local_path).write_bytes(b"fake bgm")
                return True

            mock_download.side_effect = _fake_download

            result = adapter._prepare_bgm(plan, tmp_path, "plan_001")

        assert result is not None
        assert Path(result).exists()

    def test_bgm_from_preset_library(self, tmp_path):
        """从预设 BGM 库下载。"""
        plan = FakePlan(
            config={
                "bgm": {
                    "enabled": True,
                    "preset_id": "preset_calm",
                }
            }
        )
        adapter = RenderAdapter(MagicMock())

        with patch("packages.domain.preset_bgm.get_preset_bgm") as mock_preset:
            mock_preset.return_value = MagicMock(audio_url="https://cdn.example.com/preset_calm.mp3")

            with patch("video_processing.url_security.safe_download_file") as mock_download:

                def _fake_download(url, path, **kwargs):
                    Path(path).parent.mkdir(parents=True, exist_ok=True)
                    Path(path).write_bytes(b"preset bgm data")

                mock_download.side_effect = _fake_download

                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")

        assert result is not None
        assert Path(result).exists()

    def test_bgm_url_takes_priority_over_asset(self, tmp_path):
        """URL 优先级高于素材库。"""
        plan = FakePlan(
            config={
                "bgm": {
                    "enabled": True,
                    "audio_url": "https://example.com/bgm.mp3",
                    "asset_id": "asset_bgm_001",
                }
            }
        )
        adapter = RenderAdapter(MagicMock())

        with patch("video_processing.url_security.safe_download_file") as mock_url_download:

            def _fake_url_download(url, path, **kwargs):
                Path(path).parent.mkdir(parents=True, exist_ok=True)
                Path(path).write_bytes(b"url bgm")

            mock_url_download.side_effect = _fake_url_download

            with patch("video_processing.render_adapter.download_asset") as mock_asset_download:
                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")

        assert result is not None
        mock_url_download.assert_called_once()
        mock_asset_download.assert_not_called()  # URL 成功后不会再走素材库


# ── _get_asr_service 测试 ────────────────────────────────────────────────────


class TestGetAsrService:
    """ASR 服务初始化测试。"""

    def test_returns_service_when_available(self):
        """ASR 服务可用时返回实例。"""
        mock_service = MagicMock()
        with patch("apps.worker.services.asr_service_factory.get_asr_service") as mock_get:
            mock_get.return_value = mock_service
            result = RenderAdapter._get_asr_service()

        assert result is mock_service

    def test_returns_none_when_import_fails(self):
        """ASR 服务导入失败时返回 None（不阻断主流程）。"""
        with patch("apps.worker.services.asr_service_factory.get_asr_service") as mock_get:
            mock_get.side_effect = ImportError("asr module not found")
            result = RenderAdapter._get_asr_service()

        assert result is None

    def test_returns_none_when_init_fails(self):
        """ASR 服务初始化失败时返回 None（不阻断主流程）。"""
        with patch("apps.worker.services.asr_service_factory.get_asr_service") as mock_get:
            mock_get.side_effect = RuntimeError("ASR init failed")
            result = RenderAdapter._get_asr_service()

        assert result is None


# ── render_plan BGM/ASR 集成测试 ─────────────────────────────────────────────


class TestRenderPlanWithBgmAsr:
    """渲染流程中 BGM 和 ASR 的集成测试。"""

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_bgm_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """BGM 路径被正确传递给 UnifiedRenderService。"""

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

        mock_download.side_effect = _fake_download

        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
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(
            id="plan_bgm",
            config={
                "bgm": {
                    "enabled": True,
                    "preset_id": "preset_001",
                }
            },
        )
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}

        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        # Mock BGM 下载
        with patch.object(adapter, "_prepare_bgm", return_value=str(tmp_path / "bgm.mp3")):
            with patch.object(adapter, "_get_asr_service", return_value=None):
                result = adapter.render_plan(
                    "plan_bgm",
                    work_dir=tmp_path / "work",
                )

        assert result.success
        # 验证 UnifiedRenderService 收到了 bgm_path
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["bgm_path"] == str(tmp_path / "bgm.mp3")

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_asr_service_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """ASR 服务被正确传递给 UnifiedRenderService。"""

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

        mock_download.side_effect = _fake_download

        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
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        mock_asr = MagicMock()
        plan = FakePlan(id="plan_asr")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}

        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

        with patch.object(adapter, "_prepare_bgm", return_value=None):
            with patch.object(adapter, "_get_asr_service", return_value=mock_asr):
                result = adapter.render_plan(
                    "plan_asr",
                    work_dir=tmp_path / "work",
                )

        assert result.success
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["asr_service"] is mock_asr

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    @patch("video_processing.render_adapter.download_asset")
    def test_rendered_and_failed_clip_ids_in_result(self, mock_download, mock_render_cls, mock_upload, tmp_path):
        """渲染结果中包含成功和失败的 clip id 列表。"""
        # 成功/失败映射，通过 storage_key（file_url）区分
        download_map = {
            "https://test-bucket.oss.com/assets/good_001.mp4": True,
            "https://test-bucket.oss.com/assets/bad_002.mp4": False,
            "https://test-bucket.oss.com/assets/good_003.mp4": True,
        }

        def _fake_download(storage_key, local_path):
            local_path.parent.mkdir(parents=True, exist_ok=True)
            ok = download_map.get(storage_key, False)
            if ok:
                local_path.write_bytes(b"fake video")
            return ok

        mock_download.side_effect = _fake_download

        mock_render = MagicMock()
        mock_render.render.return_value = MagicMock(
            output_path=tmp_path / "out.mp4",
            duration=3.0,
            file_size=512,
            width=1280,
            height=720,
        )
        mock_render_cls.return_value = mock_render
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(id="plan_mixed")
        clips = [
            _make_clip("c_good1", order=0, asset_id="good_001"),
            _make_clip("c_bad", order=1, asset_id="bad_002"),
            _make_clip("c_good2", order=2, asset_id="good_003"),
        ]
        asset_url_map = {
            "good_001": "https://test-bucket.oss.com/assets/good_001.mp4",
            "bad_002": "https://test-bucket.oss.com/assets/bad_002.mp4",
            "good_003": "https://test-bucket.oss.com/assets/good_003.mp4",
        }

        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)

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

        assert result.success
        assert "c_good1" in result.rendered_clip_ids
        assert "c_good2" in result.rendered_clip_ids
        assert "c_bad" in result.failed_clip_ids
        assert len(result.rendered_clip_ids) == 2
        assert len(result.failed_clip_ids) == 1


# ── render_from_memory 测试 ──────────────────────────────────────────────────


class TestRenderFromMemory:
    """render_from_memory 内存模式渲染测试。"""

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_successful_render(self, mock_render_cls, mock_upload, tmp_path):
        """内存模式渲染成功。"""
        mock_render = MagicMock()
        mock_render.render.return_value = MagicMock(
            output_path=tmp_path / "out.mp4",
            duration=10.0,
            file_size=1024,
            width=1280,
            height=720,
        )
        mock_render_cls.return_value = mock_render
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(id="mem_plan_001", config={"bgm": {"enabled": False}})
        clips = [
            _make_clip("c1", order=0, duration=5.0),
            _make_clip("c2", order=1, duration=5.0),
        ]
        asset_path_map = {
            "asset_c1.mp4": tmp_path / "c1.mp4",
            "asset_c2.mp4": tmp_path / "c2.mp4",
        }
        # 创建假文件
        for p in asset_path_map.values():
            p.write_bytes(b"fake")

        adapter, _, _ = _make_adapter()

        result = adapter.render_from_memory(
            plan=plan,
            clips=clips,
            asset_path_map=asset_path_map,
            plan_id="mem_plan_001",
            work_dir=tmp_path / "work",
        )

        assert result.success
        assert result.duration == 10.0
        assert result.file_size == 1024
        assert result.width == 1280
        assert result.height == 720
        assert result.clip_count == 2
        assert len(result.rendered_clip_ids) == 2
        assert len(result.failed_clip_ids) == 0

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_empty_clips_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
        """clips 为空时返回失败。"""
        plan = FakePlan(id="mem_empty")
        adapter, _, _ = _make_adapter()

        result = adapter.render_from_memory(
            plan=plan,
            clips=[],
            asset_path_map={},
            plan_id="mem_empty",
            work_dir=tmp_path / "work",
        )

        assert not result.success
        assert "没有可渲染的片段" in result.error_message
        mock_render_cls.assert_not_called()

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_empty_asset_map_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
        """asset_path_map 为空时返回失败。"""
        plan = FakePlan(id="mem_no_assets")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        adapter, _, _ = _make_adapter()

        result = adapter.render_from_memory(
            plan=plan,
            clips=clips,
            asset_path_map={},
            plan_id="mem_no_assets",
            work_dir=tmp_path / "work",
        )

        assert not result.success
        assert "素材路径映射为空" in result.error_message
        mock_render_cls.assert_not_called()

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_bgm_prepared_for_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
        """内存模式下 BGM 配置也会被正确处理。"""
        mock_render = MagicMock()
        mock_render.render.return_value = MagicMock(
            output_path=tmp_path / "out.mp4",
            duration=5.0,
            file_size=512,
            width=1280,
            height=720,
        )
        mock_render_cls.return_value = mock_render
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(
            id="mem_bgm",
            config={
                "bgm": {
                    "enabled": True,
                    "preset_id": "preset_001",
                }
            },
        )
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")

        adapter, _, _ = _make_adapter()

        # mock _prepare_bgm 返回一个假的 bgm 路径
        fake_bgm_path = tmp_path / "bgm.mp3"
        fake_bgm_path.write_bytes(b"fake bgm")
        with patch.object(adapter, "_prepare_bgm", return_value=str(fake_bgm_path)):
            result = adapter.render_from_memory(
                plan=plan,
                clips=clips,
                asset_path_map=asset_path_map,
                plan_id="mem_bgm",
                work_dir=tmp_path / "work",
            )

        assert result.success
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["bgm_path"] == str(fake_bgm_path)

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_resolution_from_config_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
        """内存模式下从 plan.config.export.resolution 读取分辨率。"""
        mock_render = MagicMock()
        mock_render.render.return_value = MagicMock(
            output_path=tmp_path / "out.mp4",
            duration=5.0,
            file_size=512,
            width=720,
            height=1280,
        )
        mock_render_cls.return_value = mock_render
        mock_upload.return_value = "https://oss.example.com/out.mp4"

        plan = FakePlan(
            id="mem_res",
            config={"export": {"resolution": "720x1280"}},
        )
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")

        adapter, _, _ = _make_adapter()

        result = adapter.render_from_memory(
            plan=plan,
            clips=clips,
            asset_path_map=asset_path_map,
            plan_id="mem_res",
            work_dir=tmp_path / "work",
        )

        assert result.success
        call_kwargs = mock_render_cls.call_args
        assert call_kwargs.kwargs["output_width"] == 720
        assert call_kwargs.kwargs["output_height"] == 1280

    @patch("video_processing.render_adapter.upload_to_oss")
    @patch("video_processing.render_adapter.UnifiedRenderService")
    def test_ffmpeg_error_captured_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
        """内存模式下 FFmpeg 错误也会被正确捕获。"""
        import subprocess

        mock_render = MagicMock()
        mock_render.render.side_effect = subprocess.CalledProcessError(
            returncode=1,
            cmd=["ffmpeg", "-i", "input.mp4", "output.mp4"],
            stderr="test error output",
        )
        mock_render_cls.return_value = mock_render

        plan = FakePlan(id="mem_err")
        clips = [_make_clip("c1", order=0, duration=5.0)]
        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")

        adapter, _, _ = _make_adapter()

        result = adapter.render_from_memory(
            plan=plan,
            clips=clips,
            asset_path_map=asset_path_map,
            plan_id="mem_err",
            work_dir=tmp_path / "work",
        )

        assert not result.success
        assert "FFmpeg渲染失败" in result.error_message
        assert "test error output" in result.error_detail