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, all identical to existing → frame_match_rate = 1.0
        phashes = ["aa00aa00aa00aa00"] * 10
        fingerprint = _make_fingerprint(md5="new", phashes=phashes, duration_ms=20000)
        session = MagicMock()

        existing = _make_video(
            "vid2",
            {"md5": "other", "keyframe_phashes": ["aa00aa00aa00aa00"] * 5},
        )

        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