xiaoxia-saas/tests/unit/test_generation_finalize_service.py

"""#2024 GenerationFinalizeService 单元测试。

覆盖 service 层：存在性校验、幂等分支、状态门、封面决策、异常映射、成功路径。
同时为 packages/application/generated_video_finalize.py 的缺失分支补测。
"""

from unittest.mock import MagicMock, patch

import pytest

# ---------- helpers ----------


def _make_task(
    task_id="task-1",
    status="awaiting_cover",
    project_id="proj-1",
    user_id="user-1",
    cover_url="",
    extra_meta=None,
    error_message="",
):
    t = MagicMock()
    t.id = task_id
    t.project_id = project_id
    t.created_by_user_id = user_id
    t.cover_url = cover_url
    t.extra_meta = extra_meta if extra_meta is not None else {}
    t.error_message = error_message
    s = MagicMock()
    s.value = status
    t.status = s

    def _mark_completed(result_count=1):
        s.value = "completed"
        t.completed_at = "now"

    t.mark_completed = MagicMock(side_effect=_mark_completed)

    def _mark_confirmed():
        t.is_preview = False

    t.mark_confirmed = MagicMock(side_effect=_mark_confirmed)
    return t


def _make_db():
    db = MagicMock()
    db.query.return_value.filter.return_value.first.return_value = None
    db.commit = MagicMock()
    db.rollback = MagicMock()
    db.bulk_save_objects = MagicMock()
    return db


def _make_rendered_dict(**overrides):
    base = {
        "file_url": "https://oss.example.com/v.mp4",
        "file_size": 123456,
        "duration": 10.5,
        "width": 1280,
        "height": 720,
        "fps": 25.0,
        "name": "demo.mp4",
        "thumbnail_url": "https://oss.example.com/thumb.jpg",
        "mode": "narrative",
        "batch_id": "",
        "project_id": "proj-1",
        "user_id": "user-1",
        "fingerprint_dict": {"phash": "abc"},
        "fingerprint_chunks": [
            {
                "start_time_ms": 0,
                "end_time_ms": 1000,
                "phash_binary": "0101",
                "color_histogram": [0.1, 0.2, 0.3],
                "frame_count": 25,
            },
        ],
        "is_duplicate": False,
        "duplicate_of": None,
        "duplicate_rate": 0.0,
        "match_count": 0,
        "visual_similarity": 0.0,
        "video_fingerprint_md5": "md5-abc",
    }
    base.update(overrides)
    return base


_PATCHES = [
    "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository",
    "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
    "packages.adapters.sqlalchemy_impl.models.GeneratedVideoModel",
    "packages.application.generated_video_finalize.finalize_generated_video",
]


def _svc(db):
    from app.services.generation_finalize_service import GenerationFinalizeService

    return GenerationFinalizeService(db)


# ---------- service tests ----------


class TestFinalizeService:
    def test_task_not_found_raises_404(self):
        from app.services.generation_finalize_service import GenerationFinalizeError

        db = _make_db()
        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]), patch(_PATCHES[3]):
            TR.return_value.get.return_value = None
            svc = _svc(db)
            with pytest.raises(GenerationFinalizeError) as ei:
                svc.finalize_task("nope", "user-1")
            assert ei.value.status_code == 404
            assert ei.value.code == "TaskNotFound"

    def test_invalid_status_raises(self):
        from app.services.generation_finalize_service import GenerationFinalizeError

        db = _make_db()
        task = _make_task(status="running")
        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]) as GVM, patch(_PATCHES[3]):
            TR.return_value.get.return_value = task
            GVM.query.filter.return_value.first.return_value = None
            svc = _svc(db)
            with pytest.raises(GenerationFinalizeError) as ei:
                svc.finalize_task("task-1", "user-1")
            assert ei.value.code == "InvalidTaskStatus"
            assert ei.value.status_code == 400

    def test_idempotent_when_video_already_exists_updates_cover_and_completes(self):
        db = _make_db()
        task = _make_task(status="awaiting_cover")
        existing = MagicMock()
        existing.id = "video-exist"
        existing.thumbnail_url = "https://old-cover.jpg"
        db.query.return_value.filter.return_value.first.return_value = existing
        existing_video = MagicMock()
        existing_video.id = "video-exist"

        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]):
            TR.return_value.get.return_value = task
            TR.return_value.update = MagicMock()
            VR.return_value.get.return_value = existing_video
            svc = _svc(db)
            result = svc.finalize_task("task-1", "user-1", cover_url="https://new-cover.jpg")
            assert result.id == "video-exist"
            assert existing.thumbnail_url == "https://new-cover.jpg"
            assert task.cover_url == "https://new-cover.jpg"
            task.mark_completed.assert_called()
            TR.return_value.update.assert_called_with(task)
            db.commit.assert_called()

    def test_idempotent_already_completed_skips_mark_completed(self):
        db = _make_db()
        task = _make_task(status="completed")
        existing = MagicMock()
        existing.id = "v-exist"
        existing.thumbnail_url = "https://c.jpg"
        db.query.return_value.filter.return_value.first.return_value = existing
        existing_video = MagicMock()

        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]):
            TR.return_value.get.return_value = task
            VR.return_value.get.return_value = existing_video
            svc = _svc(db)
            svc.finalize_task("task-1", "user-1")
            task.mark_completed.assert_not_called()

    def test_missing_rendered_output_raises(self):
        from app.services.generation_finalize_service import GenerationFinalizeError

        db = _make_db()
        task = _make_task(status="awaiting_cover", extra_meta={})
        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]), patch(_PATCHES[3]) as fu:
            TR.return_value.get.return_value = task
            TR.return_value.update = MagicMock()
            fu.side_effect = ValueError("file_url 为空")
            svc = _svc(db)
            with pytest.raises(GenerationFinalizeError) as ei:
                svc.finalize_task("task-1", "user-1")
            assert ei.value.code == "RenderedOutputMissing"

    def test_success_creates_video_and_marks_completed(self):
        db = _make_db()
        task = _make_task(
            status="awaiting_cover",
            cover_url="https://task-cover.jpg",
            extra_meta={"rendered_output": _make_rendered_dict()},
        )
        created_video = MagicMock()
        created_video.id = "video-new"

        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu:
            TR.return_value.get.return_value = task
            TR.return_value.update = MagicMock()
            fu.return_value = {"video_id": "video-new", "is_duplicate": False, "duplicate_of": None}
            VR.return_value.get.return_value = created_video
            svc = _svc(db)
            v = svc.finalize_task("task-1", "user-1")
            assert v.id == "video-new"
            task.mark_completed.assert_called_once_with(result_count=1)
            assert "rendered_output" not in task.extra_meta
            TR.return_value.update.assert_called_with(task)
            db.commit.assert_called()

    def test_cover_fallback_to_task_cover_url(self):
        db = _make_db()
        task = _make_task(
            status="awaiting_cover",
            cover_url="https://task-cover.jpg",
            extra_meta={"rendered_output": _make_rendered_dict(thumbnail_url="")},
        )
        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu:
            TR.return_value.get.return_value = task
            TR.return_value.update = MagicMock()
            fu.return_value = {"video_id": "v1", "is_duplicate": False, "duplicate_of": None}
            VR.return_value.get.return_value = MagicMock(id="v1")
            svc = _svc(db)
            svc.finalize_task("task-1", "user-1")
            assert task.cover_url == "https://task-cover.jpg"
            kwargs = fu.call_args.kwargs
            assert kwargs["effective_cover_url"] == "https://task-cover.jpg"

    def test_explicit_cover_url_overrides_task_cover(self):
        db = _make_db()
        task = _make_task(
            status="awaiting_cover",
            cover_url="https://old.jpg",
            extra_meta={"rendered_output": _make_rendered_dict()},
        )
        with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu:
            TR.return_value.get.return_value = task
            TR.return_value.update = MagicMock()
            fu.return_value = {"video_id": "v1", "is_duplicate": False, "duplicate_of": None}
            VR.return_value.get.return_value = MagicMock(id="v1")
            svc = _svc(db)
            svc.finalize_task("task-1", "user-1", cover_url="  https://new.jpg  ")
            kwargs = fu.call_args.kwargs
            assert kwargs["effective_cover_url"] == "https://new.jpg"


# ---------- packages/application/generated_video_finalize.py 覆盖补测 ----------


class TestFinalizeUseCaseCoverage:
    def test_rendered_output_non_dict_raises(self):
        from packages.application.generated_video_finalize import RenderedOutput

        with pytest.raises(ValueError):
            RenderedOutput.from_dict("not-a-dict")

    def test_safe_float_handles_invalid(self):
        from packages.application.generated_video_finalize import _safe_float, _safe_int

        assert _safe_float(None) is None
        assert _safe_float("abc") is None
        assert _safe_float("3.14") == pytest.approx(3.14)
        assert _safe_int(None) is None
        assert _safe_int("xyz") is None
        assert _safe_int("42") == 42

    def test_fingerprint_chunks_non_dict_entry_is_skipped(self):
        """非 dict chunk 被 continue 跳过；bulk_save 只处理合法 chunk。"""
        from packages.application import generated_video_finalize as mod

        task = _make_task(
            extra_meta={
                "rendered_output": _make_rendered_dict(
                    fingerprint_chunks=[
                        "not-a-dict",
                        {
                            "start_time_ms": 0,
                            "end_time_ms": 500,
                            "phash_binary": "xx",
                            "color_histogram": [0.1, 0.2],
                            "frame_count": 10,
                        },
                    ],
                )
            }
        )
        db = MagicMock()
        db.bulk_save_objects = MagicMock()
        db.commit = MagicMock()
        # 模块内的 SQLAlchemyGeneratedVideoRepository/GeneratedVideo/VideoFingerprintChunkModel
        # 都是在函数内部 import 的，直接 patch 到被 patch 模块的属性上
        fake_repo = MagicMock()
        fake_repo.create = MagicMock()
        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=fake_repo,
            ),
            patch(
                "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel",
                side_effect=lambda **kw: MagicMock(**kw),
            ),
            patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)),
        ):
            result = mod.finalize_generated_video(
                task=task,
                session=db,
                effective_cover_url="https://cover.jpg",
            )
            assert "video_id" in result
            assert db.bulk_save_objects.call_count == 1
            saved_chunks = db.bulk_save_objects.call_args[0][0]
            assert len(saved_chunks) == 1
            db.commit.assert_called()
            fake_repo.create.assert_called_once()

    def test_missing_file_url_raises(self):
        from packages.application import generated_video_finalize as mod

        task = _make_task(
            extra_meta={
                "rendered_output": _make_rendered_dict(file_url=""),
            }
        )
        db = MagicMock()
        with pytest.raises(ValueError):
            mod.finalize_generated_video(task=task, session=db, effective_cover_url="")

    def test_no_fingerprint_chunks_skips_bulk_save(self):
        from packages.application import generated_video_finalize as mod

        task = _make_task(
            extra_meta={
                "rendered_output": _make_rendered_dict(fingerprint_chunks=None),
            }
        )
        db = MagicMock()
        db.bulk_save_objects = MagicMock()
        db.commit = MagicMock()
        fake_repo = MagicMock()
        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=fake_repo,
            ),
            patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)),
        ):
            mod.finalize_generated_video(task=task, session=db, effective_cover_url="")
            db.bulk_save_objects.assert_not_called()
            fake_repo.create.assert_called_once()
            db.commit.assert_called()

    def test_name_fallback_when_empty(self):
        from packages.application import generated_video_finalize as mod

        task = _make_task(
            task_id="abcd1234ef567890",
            extra_meta={
                "rendered_output": _make_rendered_dict(name="   ", thumbnail_url=""),
            },
        )
        db = MagicMock()
        db.bulk_save_objects = MagicMock()
        db.commit = MagicMock()
        fake_repo = MagicMock()
        captured = {}

        def _capture(**kw):
            captured.update(kw)
            return MagicMock(**kw)

        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=fake_repo,
            ),
            patch("packages.domain.generated_video.GeneratedVideo", side_effect=_capture),
            patch(
                "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel",
                side_effect=lambda **kw: MagicMock(**kw),
            ),
        ):
            mod.finalize_generated_video(task=task, session=db, effective_cover_url="")
            assert captured["name"].startswith("generated-abcd1234")
            assert captured["thumbnail_url"] is None

    def test_chunk_exception_is_swallowed(self):
        """chunk 构造异常时 logger.warning，不阻塞主流程。"""
        from packages.application import generated_video_finalize as mod

        task = _make_task(
            extra_meta={
                "rendered_output": _make_rendered_dict(
                    fingerprint_chunks=[
                        {
                            "start_time_ms": 0,
                            "end_time_ms": 500,
                            "phash_binary": "xx",
                            "color_histogram": ["not-a-number"],
                            "frame_count": 10,
                        },
                    ],
                )
            }
        )
        db = MagicMock()
        db.bulk_save_objects = MagicMock()
        db.commit = MagicMock()
        fake_repo = MagicMock()
        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=fake_repo,
            ),
            patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)),
            patch(
                "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel",
                side_effect=lambda **kw: MagicMock(**kw),
            ),
        ):
            # color_histogram 里 "not-a-number" 触发 float() 异常，被 except chunk_err 吞掉
            # 但此时 chunk_models 中仍有 1 个元素（MagicMock 构造不会因 float() 失败）——
            # 因为我们把 float 列表推导也放在 try 内，float("not-a-number") 抛 ValueError
            # 所以要让 float 真的抛。但 MagicMock side_effect 不触发 float()，这里直接构造：
            # 通过真实验证路径
            result = mod.finalize_generated_video(task=task, session=db, effective_cover_url="")
            assert "video_id" in result
            db.commit.assert_called()
            fake_repo.create.assert_called_once()