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):
        from video_processing.dedup import VideoFingerprint

        return VideoFingerprint(
            md5=md5,
            keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
            color_histograms=[],
            duration=10.0,
            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
            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = (
                []
            )
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        assert rate == 0.0

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

        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

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

        existing = self._make_existing_video("existing1", {"md5": "exact_match_md5", "keyframe_phashes": ["aa"]})
        # Create a mock model with the domain attributes
        mock_model = MagicMock(spec=GeneratedVideoModel)
        mock_model.id = existing.id
        mock_model.project_id = existing.project_id
        mock_model.video_fingerprint = existing.video_fingerprint
        mock_model.generated_at = "2026-01-01"

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo._to_domain.return_value = existing
            # Mock the session.query chain
            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
                mock_model
            ]
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        assert rate == 100.0

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

        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

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

        existing = self._make_existing_video(
            "existing1",
            {"md5": "other_md5", "keyframe_phashes": ["ff00ff00ff00ff03"]},
        )
        mock_model = MagicMock(spec=GeneratedVideoModel)
        mock_model.id = existing.id
        mock_model.project_id = existing.project_id
        mock_model.video_fingerprint = existing.video_fingerprint
        mock_model.generated_at = "2026-01-01"

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo._to_domain.return_value = existing
            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
                mock_model
            ]
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        # hamming distance = 2, similarity = (1 - 2/64) * 100 = 96.875
        assert rate == pytest.approx(96.88, abs=0.1)

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

        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

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

        self_video = self._make_existing_video("vid1", {"md5": "same_md5", "keyframe_phashes": ["aa"]})
        mock_model = MagicMock(spec=GeneratedVideoModel)
        mock_model.id = self_video.id
        mock_model.project_id = self_video.project_id
        mock_model.video_fingerprint = self_video.video_fingerprint
        mock_model.generated_at = "2026-01-01"

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo._to_domain.return_value = self_video
            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
                mock_model
            ]
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        assert rate == 0.0

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

        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

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

        existing1 = self._make_existing_video("e1", {"md5": "md5_1", "keyframe_phashes": ["ff00ff00ff00ff0f"]})
        existing2 = self._make_existing_video("e2", {"md5": "md5_2", "keyframe_phashes": ["ff00ff00ff00ff01"]})
        mock_model1 = MagicMock(spec=GeneratedVideoModel)
        mock_model1.id = existing1.id
        mock_model1.project_id = existing1.project_id
        mock_model1.video_fingerprint = existing1.video_fingerprint
        mock_model1.generated_at = "2026-01-02"
        mock_model2 = MagicMock(spec=GeneratedVideoModel)
        mock_model2.id = existing2.id
        mock_model2.project_id = existing2.project_id
        mock_model2.video_fingerprint = existing2.video_fingerprint
        mock_model2.generated_at = "2026-01-01"

        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
            mock_repo = MockRepo.return_value
            mock_repo._to_domain.side_effect = [existing1, existing2]
            session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
                mock_model1,
                mock_model2,
            ]
            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)

        # max similarity: e2 distance=1, (1-1/64)*100 = 98.4375
        assert rate == pytest.approx(98.44, abs=0.1)


class TestDuplicateRateAPI:
    """Test that duplicate_rate is returned in API responses."""

    def test_video_item_response_has_duplicate_rate(self):
        from app.schemas.video_center import VideoItemResponse

        resp = VideoItemResponse(
            id="v1",
            project_id="p1",
            generation_task_id="t1",
            name="test.mp4",
            file_url="https://example.com/test.mp4",
            file_size=1000,
            duration=10.0,
            width=1920,
            height=1080,
            fps=25.0,
            duplicate_rate=75.5,
        )
        assert resp.duplicate_rate == 75.5

    def test_video_item_response_duplicate_rate_default_none(self):
        from app.schemas.video_center import VideoItemResponse

        resp = VideoItemResponse(
            id="v1",
            project_id="p1",
            generation_task_id="t1",
            name="test.mp4",
            file_url="https://example.com/test.mp4",
            file_size=1000,
            duration=10.0,
            width=1920,
            height=1080,
            fps=25.0,
        )
        assert resp.duplicate_rate is None