xiaoxia-saas/tests/unit/test_duplicate_rate_scope.py

"""Tests for Issue #1660 — 查重率百分比计算 + 跨项目查重."""

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

import pytest

sys.modules.setdefault("cv2", MagicMock())

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps" / "api"))
sys.path.insert(0, str(ROOT / "packages"))
sys.path.insert(0, str(ROOT / "apps" / "worker"))


def _make_fingerprint(md5="abc123", phashes=None, duration_ms=10000):
    from video_processing.dedup import VideoFingerprint

    return VideoFingerprint(
        md5=md5,
        keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
        color_histograms=[],
        duration=duration_ms,
        resolution=(1920, 1080),
    )


def _make_video(vid, fingerprint_dict, project_id="proj1", duration=10.0):
    from packages.domain import GeneratedVideo

    return GeneratedVideo(
        id=vid,
        project_id=project_id,
        generation_task_id="task1",
        name=f"video-{vid}",
        file_url=f"https://example.com/{vid}.mp4",
        file_size=1000,
        duration=duration,
        width=1920,
        height=1080,
        fps=25.0,
        video_fingerprint=fingerprint_dict,
    )


class TestCheckDuplicateScopeProject:
    """test_check_duplicate_scope_project：项目内查重（默认行为）."""

    def test_default_scope_queries_by_project(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint(md5="unique_md5")
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = []
            result = deduplicator.check_duplicate(fingerprint, "proj1", session)

        mock_repo.list_by_project.assert_called_once_with("proj1")
        assert result is None

    def test_project_scope_finds_duplicate(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint(md5="same_md5")
        session = MagicMock()

        existing = _make_video("vid2", {"md5": "same_md5", "keyframe_phashes": ["aa"]})

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = [existing]
            result = deduplicator.check_duplicate(fingerprint, "proj1", session)

        assert result is not None
        assert result["duplicate"] is True
        assert result["duplicate_of"] == "vid2"


class TestCheckDuplicateScopeUser:
    """test_check_duplicate_scope_user：跨项目查重."""

    def test_user_scope_queries_by_user(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint(md5="unique_md5")
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_user.return_value = []
            result = deduplicator.check_duplicate(
                fingerprint,
                "proj1",
                session,
                scope="user",
                user_id="user_123",
            )

        mock_repo.list_by_user.assert_called_once()
        assert result is None

    def test_user_scope_finds_cross_project_duplicate(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint(md5="cross_proj_md5")
        session = MagicMock()

        # Existing video from a different project
        existing = _make_video("vid_other", {"md5": "cross_proj_md5"}, project_id="proj_other")

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_user.return_value = [existing]
            result = deduplicator.check_duplicate(
                fingerprint,
                "proj1",
                session,
                scope="user",
                user_id="user_123",
            )

        assert result is not None
        assert result["duplicate"] is True
        assert result["duplicate_of"] == "vid_other"


class TestDurationPrefilter:
    """test_duration_prefilter：时长 ±15% 过滤."""

    def test_duration_prefilter_passes_correct_range(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint(duration_ms=30000)  # 30s video
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_user.return_value = []
            deduplicator.check_duplicate(
                fingerprint,
                "proj1",
                session,
                scope="user",
                user_id="user1",
                duration_sec=30.0,
            )

        # Should pass duration_min=25.5, duration_max=34.5 (30 ± 15%)
        call_args = mock_repo.list_by_user.call_args
        assert call_args[1]["duration_min"] == pytest.approx(25.5, abs=0.1)
        assert call_args[1]["duration_max"] == pytest.approx(34.5, abs=0.1)

    def test_no_duration_prefilter_when_zero(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint()
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_user.return_value = []
            deduplicator.check_duplicate(
                fingerprint,
                "proj1",
                session,
                scope="user",
                user_id="user1",
                duration_sec=0,
            )

        call_args = mock_repo.list_by_user.call_args
        assert call_args[1]["duration_min"] == 0
        assert call_args[1]["duration_max"] == 0


class TestComputeDuplicateRateFormula:
    """test_compute_duplicate_rate_formula：验证 0.4 * frame_match_rate + 0.6 * temporal_coverage_rate."""

    def test_formula_with_matching_frames(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        # 10 frames with varied phashes (2 unique) → not bad fingerprint
        # All close in hamming distance to existing → frame_match_rate = 1.0
        phashes = ["aa00aa00aa00aa00", "ab00ab00ab00ab00"] * 5
        fingerprint = _make_fingerprint(md5="new", phashes=phashes, duration_ms=20000)
        session = MagicMock()

        # 5 unique phashes to pass _is_bad_fingerprint check (PR #1688)
        existing = _make_video(
            "vid2",
            {
                "md5": "other",
                "keyframe_phashes": [
                    "aa00aa00aa00aa00",
                    "ab00ab00ab00ab00",
                    "ac00ac00ac00ac00",
                    "aa10aa10aa10aa10",
                    "ba00ba00ba00ba00",
                ],
            },
        )

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = [existing]
            deduplicator._get_existing_chunks = MagicMock(return_value=[])
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        # frame_match_rate=1.0, temporal_coverage depends on segments
        # duplicate_rate = (1.0 * 0.4 + temporal_coverage * 0.6) * 100
        assert rate["duplicate_rate"] >= 40.0  # At minimum, frame_match contributes 40%

    def test_no_match_returns_zero(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        # Completely different phashes
        fingerprint = _make_fingerprint(md5="new", phashes=["ff00ff00ff00ff00"])
        session = MagicMock()

        existing = _make_video(
            "vid2",
            {"md5": "other", "keyframe_phashes": ["00ff00ff00ff00ff"]},
        )

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = [existing]
            deduplicator._get_existing_chunks = MagicMock(return_value=[])
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        # Very different phashes, match_ratio < 0.3 → skipped
        assert rate["duplicate_rate"] == 0.0


class TestComputeDuplicateRateReturnDict:
    """test_compute_duplicate_rate_return_dict：验证返回 dict 含三个字段."""

    def test_return_structure(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint()
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = []
            result = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        assert isinstance(result, dict)
        assert set(result.keys()) == {"duplicate_rate", "visual_similarity", "match_count"}
        assert isinstance(result["duplicate_rate"], float)
        assert isinstance(result["visual_similarity"], float)
        assert isinstance(result["match_count"], int)
        assert 0 <= result["duplicate_rate"] <= 100
        assert 0 <= result["visual_similarity"] <= 1


class TestBackwardCompat:
    """test_backward_compat：不传 scope 时行为不变."""

    def test_default_scope_is_project(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint()
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = []

            # Call without scope parameter
            result = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        # Should use list_by_project (not list_by_user)
        mock_repo.list_by_project.assert_called_once_with("proj1")
        mock_repo.list_by_user.assert_not_called()
        assert result["duplicate_rate"] == 0.0

    def test_check_duplicate_default_scope_backward_compat(self):
        from video_processing.dedup import VideoDeduplicator

        deduplicator = VideoDeduplicator()
        fingerprint = _make_fingerprint()
        session = MagicMock()

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_project.return_value = []
            result = deduplicator.check_duplicate(fingerprint, "proj1", session)

        mock_repo.list_by_project.assert_called_once_with("proj1")
        assert result is None


class TestListByUserRepository:
    """直接测试 generated_video_repository.list_by_user() 的真实实现，覆盖 diff 代码行。"""

    def _make_repo(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
        from packages.adapters.sqlalchemy_impl.models import Base, GeneratedVideoModel

        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        session = Session()
        repo = SQLAlchemyGeneratedVideoRepository(session)
        return repo, session

    def _insert_video(self, session, video_id, user_id, project_id, duration, **kw):
        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

        row = GeneratedVideoModel(
            id=video_id,
            user_id=user_id,
            project_id=project_id,
            generation_task_id=f"task-{video_id[:8]}",
            name=f"video-{video_id[:8]}.mp4",
            file_url=f"https://example.com/{video_id}.mp4",
            file_size=1024,
            duration=duration,
            width=1280,
            height=720,
            fps=25.0,
            status="completed",
        )
        session.add(row)
        session.flush()
        return row

    def test_list_by_user_returns_cross_project_videos(self):
        """list_by_user 返回该用户所有项目的视频。"""
        repo, session = self._make_repo()
        self._insert_video(session, "v1", "user-a", "proj-1", 30.0)
        self._insert_video(session, "v2", "user-a", "proj-2", 45.0)
        self._insert_video(session, "v3", "user-b", "proj-1", 20.0)

        results = repo.list_by_user("user-a")
        assert len(results) == 2
        ids = {r.id for r in results}
        assert ids == {"v1", "v2"}
        session.close()

    def test_list_by_user_with_duration_filter(self):
        """list_by_user 支持 duration_min/duration_max 过滤。"""
        repo, session = self._make_repo()
        self._insert_video(session, "v1", "user-a", "proj-1", 10.0)
        self._insert_video(session, "v2", "user-a", "proj-1", 30.0)
        self._insert_video(session, "v3", "user-a", "proj-1", 60.0)

        results = repo.list_by_user("user-a", duration_min=20.0, duration_max=50.0)
        assert len(results) == 1
        assert results[0].id == "v2"
        session.close()

    def test_list_by_user_empty_result(self):
        """list_by_user 无匹配时返回空列表。"""
        repo, session = self._make_repo()
        self._insert_video(session, "v1", "user-a", "proj-1", 30.0)

        results = repo.list_by_user("user-nonexistent")
        assert results == []
        session.close()