xiaoxia-saas/tests/unit/test_duplicate_rate.py

"""Tests for duplicate_rate computation and API response."""

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

import pytest

# Mock cv2 and numpy before any imports that need them
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"))


class TestComputeDuplicateRate:
    """Test VideoDeduplicator.compute_duplicate_rate."""

    def _make_fingerprint(self, 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_existing_video(self, vid, fingerprint_dict):
        from packages.domain import GeneratedVideo

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

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

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

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

        assert rate["duplicate_rate"] == 0.0
        assert rate["match_count"] == 0
        assert isinstance(rate, dict)

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

        deduplicator = VideoDeduplicator()
        fingerprint = self._make_fingerprint(md5="exact_md5")
        session = MagicMock()

        existing = self._make_existing_video("vid2", {"md5": "exact_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]
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        assert rate["duplicate_rate"] == 100.0
        assert rate["match_count"] == 1

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

        deduplicator = VideoDeduplicator()
        # Two very similar phashes
        fingerprint = self._make_fingerprint(
            md5="new",
            phashes=["ff00ff00ff00ff00", "ff00ff00ff00ff01"],
        )
        session = MagicMock()

        existing = self._make_existing_video(
            "vid2",
            {"md5": "other", "keyframe_phashes": ["ff00ff00ff00ff00", "ff00ff00ff00ff02"]},
        )

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

        # With identical phashes, frame_match_rate should be high
        assert rate["duplicate_rate"] >= 0.0
        assert isinstance(rate, dict)
        assert "visual_similarity" in rate

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

        deduplicator = VideoDeduplicator()
        fingerprint = self._make_fingerprint(
            md5="new",
            phashes=["aa00aa00aa00aa00"],
        )
        session = MagicMock()

        # Two existing videos with different phashes
        existing1 = self._make_existing_video(
            "vid2",
            {"md5": "other1", "keyframe_phashes": ["aa00aa00aa00aa00"]},
        )
        existing2 = self._make_existing_video(
            "vid3",
            {"md5": "other2", "keyframe_phashes": ["ff00ff00ff00ff00"]},
        )

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

        # Should take the max across all videos
        assert rate["duplicate_rate"] >= 0.0
        assert isinstance(rate["duplicate_rate"], float)

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

        deduplicator = VideoDeduplicator()
        fingerprint = self._make_fingerprint(md5="exact_md5_x")
        session = MagicMock()

        existing = self._make_existing_video("vid2", {"md5": "exact_md5_x", "keyframe_phashes": ["aa"]})

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo.list_by_user.return_value = [existing]
            rate = deduplicator.compute_duplicate_rate(
                fingerprint,
                "proj1",
                "vid1",
                session,
                scope="user",
                user_id="user1",
            )

        # Should use list_by_user and find the match
        mock_repo.list_by_user.assert_called_once_with("user1")
        assert rate["duplicate_rate"] == 100.0

    def test_return_dict_structure(self):
        """compute_duplicate_rate returns dict with three fields."""
        from video_processing.dedup import VideoDeduplicator

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

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

        assert isinstance(rate, dict)
        assert "duplicate_rate" in rate
        assert "visual_similarity" in rate
        assert "match_count" in rate
        assert isinstance(rate["duplicate_rate"], float)
        assert isinstance(rate["visual_similarity"], float)
        assert isinstance(rate["match_count"], int)

    def test_backward_compat_no_scope(self):
        """Not passing scope defaults to project-level."""
        from video_processing.dedup import VideoDeduplicator

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

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

        mock_repo.list_by_project.assert_called_once_with("proj1")
        assert rate["duplicate_rate"] == 0.0