xiaoxia-saas/tests/unit/test_batch_dedup_helpers_1743.py

"""#1743 dedup_helpers 批次查重 + worker 重选 plan 重试函数测试。

覆盖：
- 批次任务且无历史重复时走 check_batch_duplicate，batch_similarity 透传到返回 dict
- 批次任务历史已重复 → 不再做批次查重，is_duplicate=True / duplicate_of 透传
- 非批次任务 batch_similarity 恒为 None
- _reselect_plan_for_batch_retry：成功返回新 plan_id；异常返回 None（保留首版）
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
for sub in ("apps/worker", "apps/api", "packages", ""):
    p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT)
    if p not in sys.path:
        sys.path.insert(0, p)

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
from packages.adapters.sqlalchemy_impl.models import Base  # noqa: E402


@pytest.fixture
def session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    s = sessionmaker(bind=engine)()
    try:
        yield s
    finally:
        s.close()


def _dedup_kwargs(batch_id):
    return dict(
        generation_task_id="task-batch-1",
        project_id="proj-1",
        user_id="user-1",
        batch_id=batch_id,
        file_url="https://oss.example.com/v.mp4",
        file_size=1024,
        duration=10.0,
        video_path="/tmp/fake.mp4",
        mode="edit_plan",
        width=1280,
        height=720,
        fps=25.0,
    )


class TestBatchDedupInHelpers:
    """dedup_helpers 内部 lazy import video_processing.dedup（依赖 cv2），
    复用 test_generated_video_creation_logic 的 mock 模块注册模式。"""

    @classmethod
    def setup_class(cls):
        import sys

        if "cv2" not in sys.modules:
            sys.modules["cv2"] = MagicMock()

        # 保存 setup 前的真实模块引用，teardown 原样还原（沙箱无 cv2 时 reimport 会失败，
        # 若直接 pop 掉 MagicMock 会让后续测试（如 test_dedup_1702）拿到残缺模块）
        import video_processing

        cls._orig_dedup = sys.modules.get("video_processing.dedup")
        cls._orig_dedup_attr = getattr(video_processing, "dedup", None)
        cls._orig_thumb = sys.modules.get("video_processing.thumbnail_generator")
        cls._orig_thumb_attr = getattr(video_processing, "thumbnail_generator", None)

        mock_dedup = MagicMock()
        mock_dedup.VideoDeduplicator = MagicMock()
        sys.modules["video_processing.dedup"] = mock_dedup

        mock_thumb = MagicMock()
        mock_thumb.extract_first_frame = MagicMock()
        sys.modules["video_processing.thumbnail_generator"] = mock_thumb

        video_processing.dedup = mock_dedup
        video_processing.thumbnail_generator = mock_thumb

    @classmethod
    def teardown_class(cls):
        import sys

        import video_processing

        # 还原 setup 前状态：原本有真模块→放回；原本没有→移除 mock
        if cls._orig_dedup is not None:
            sys.modules["video_processing.dedup"] = cls._orig_dedup
        else:
            sys.modules.pop("video_processing.dedup", None)
        if cls._orig_dedup_attr is not None:
            video_processing.dedup = cls._orig_dedup_attr
        elif hasattr(video_processing, "dedup"):
            delattr(video_processing, "dedup")

        if cls._orig_thumb is not None:
            sys.modules["video_processing.thumbnail_generator"] = cls._orig_thumb
        else:
            sys.modules.pop("video_processing.thumbnail_generator", None)
        if cls._orig_thumb_attr is not None:
            video_processing.thumbnail_generator = cls._orig_thumb_attr
        elif hasattr(video_processing, "thumbnail_generator"):
            delattr(video_processing, "thumbnail_generator")

    def test_batch_similarity_returned_when_batch_duplicate_found(self, session):
        """批次任务 + 无历史重复 + 批次查重命中 → 返回 batch_similarity 与 is_duplicate。"""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        with patch("video_processing.dedup.VideoDeduplicator") as mock_cls:
            dedup = mock_cls.return_value
            dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
            dedup.check_duplicate.return_value = None  # 历史无重复
            dedup.check_batch_duplicate.return_value = {
                "duplicate_of": "video-existing",
                "reason": "batch_similar",
                "similarity": 0.601,
            }
            dedup.compute_duplicate_rate.return_value = {
                "duplicate_rate": 0.0,
                "visual_similarity": 0.0,
                "match_count": 0,
            }

            result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-abc"))

        assert result["video_count"] == 1
        assert result["batch_similarity"] == pytest.approx(0.601)
        assert result["is_duplicate"] is True
        assert result["duplicate_of"] == "video-existing"
        # 批次查重确实被调用（历史查重为 None 才走批次）
        dedup.check_batch_duplicate.assert_called_once()

    def test_batch_check_skipped_when_historical_duplicate(self, session):
        """历史查重已命中 → 不再批次查重，batch_similarity 为 None。"""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        with patch("video_processing.dedup.VideoDeduplicator") as mock_cls:
            dedup = mock_cls.return_value
            dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
            dedup.check_duplicate.return_value = {
                "duplicate_of": "video-old",
                "reason": "global",
                "similarity": 0.85,
            }
            dedup.compute_duplicate_rate.return_value = {
                "duplicate_rate": 85.0,
                "visual_similarity": 0.85,
                "match_count": 3,
            }

            result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-abc"))

        assert result["is_duplicate"] is True
        assert result["duplicate_of"] == "video-old"
        assert result["batch_similarity"] is None
        dedup.check_batch_duplicate.assert_not_called()

    def test_non_batch_never_runs_batch_check(self, session):
        """非批次任务（batch_id 为空）→ 不调用批次查重，batch_similarity None。"""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        with patch("video_processing.dedup.VideoDeduplicator") as mock_cls:
            dedup = mock_cls.return_value
            dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
            dedup.check_duplicate.return_value = None
            dedup.check_batch_duplicate.return_value = {"similarity": 0.99}
            dedup.compute_duplicate_rate.return_value = {
                "duplicate_rate": 0.0,
                "visual_similarity": 0.0,
                "match_count": 0,
            }

            result = create_video_record_and_dedup(session=session, **_dedup_kwargs(""))

        assert result["video_count"] == 1
        assert result["batch_similarity"] is None
        assert result["is_duplicate"] is False
        dedup.check_batch_duplicate.assert_not_called()

    def test_batch_no_duplicate_returns_none_similarity(self, session):
        """批次任务但批次查重也未命中 → batch_similarity None（驱动不重渲）。"""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        with patch("video_processing.dedup.VideoDeduplicator") as mock_cls:
            dedup = mock_cls.return_value
            dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
            dedup.check_duplicate.return_value = None
            dedup.check_batch_duplicate.return_value = None
            dedup.compute_duplicate_rate.return_value = {
                "duplicate_rate": 0.0,
                "visual_similarity": 0.0,
                "match_count": 0,
            }

            result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-xyz"))

        assert result["batch_similarity"] is None
        assert result["is_duplicate"] is False


class TestReselectPlanForBatchRetry:
    def test_success_returns_new_plan_id(self):
        """重选成功 → 返回新 plan_id。"""
        from worker_app.tasks.generation import _reselect_plan_for_batch_retry

        fake_svc = MagicMock()
        fake_svc.reselect_plan_for_variant.return_value = MagicMock(id="new-plan-999")
        with patch("app.services.edit_plan_service.EditPlanService", return_value=fake_svc):
            with patch("worker_app.tasks.generation.SessionLocal") as mock_session_local:
                mock_session_local.return_value = MagicMock()
                new_id = _reselect_plan_for_batch_retry(
                    "task-1",
                    "old-plan",
                    {"task_asset_ids": ["a1", "a2"], "user_id": "u-1"},
                )
        assert new_id == "new-plan-999"
        fake_svc.reselect_plan_for_variant.assert_called_once()
        args, kwargs = fake_svc.reselect_plan_for_variant.call_args
        assert args[0] == "old-plan"
        assert args[1] == ["a1", "a2"]
        assert kwargs["created_by_user_id"] == "u-1"

    def test_failure_returns_none_and_keeps_first_version(self):
        """重选抛异常 → 返回 None（调用方放弃重渲、保留首版），不抛出。"""
        from worker_app.tasks.generation import _reselect_plan_for_batch_retry

        fake_svc = MagicMock()
        fake_svc.reselect_plan_for_variant.side_effect = RuntimeError("db down")
        with patch("app.services.edit_plan_service.EditPlanService", return_value=fake_svc):
            with patch("worker_app.tasks.generation.SessionLocal") as mock_session_local:
                mock_session_local.return_value = MagicMock()
                result = _reselect_plan_for_batch_retry("task-1", "old-plan", {"task_asset_ids": [], "user_id": "u-1"})
        assert result is None