xiaoxia-saas/tests/unit/test_dedup_helpers_user_id.py

"""测试 create_video_record_and_dedup 传递 user_id 到查重逻辑.

验证 P0 修复：查重范围从项目级扩大到用户级。
dedup_helpers 必须把 user_id 传给 compute_duplicate_rate。
"""

from __future__ import annotations

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

import pytest

# Mock cv2/numpy before imports
sys.modules.setdefault("cv2", MagicMock())
sys.modules.setdefault("numpy", 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"))

import os

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")


class TestDedupHelpersUserIdPassthrough:
    """验证 dedup_helpers 把 user_id 传递给 compute_duplicate_rate."""

    def test_user_id_passed_to_compute_duplicate_rate(self):
        """create_video_record_and_dedup 必须传 user_id 给 compute_duplicate_rate."""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        session = MagicMock()
        mock_video_repo = MagicMock()

        mock_fingerprint = MagicMock()
        mock_fingerprint.to_dict.return_value = {"md5": "test", "keyframe_phashes": ["aa"]}

        mock_deduplicator = MagicMock()
        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
        mock_deduplicator.check_duplicate.return_value = None
        mock_deduplicator.compute_duplicate_rate.return_value = 42.5

        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=mock_video_repo,
            ),
            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
        ):
            result = create_video_record_and_dedup(
                generation_task_id="task-001",
                project_id="proj-001",
                user_id="user-abc",
                batch_id="",
                file_url="https://example.com/video.mp4",
                file_size=1024,
                duration=15.0,
                video_path="/tmp/fake_video.mp4",
                mode="smart",
                session=session,
            )

        # 验证 compute_duplicate_rate 被调用且 user_id 正确传递
        mock_deduplicator.compute_duplicate_rate.assert_called_once()
        call_kwargs = mock_deduplicator.compute_duplicate_rate.call_args
        assert (
            call_kwargs.kwargs.get("user_id") == "user-abc"
        ), f"user_id 应传递给 compute_duplicate_rate，实际: {call_kwargs}"

    def test_empty_user_id_still_works(self):
        """user_id 为空时仍然正常执行（回退到 project 级比较）."""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        session = MagicMock()
        mock_video_repo = MagicMock()

        mock_fingerprint = MagicMock()
        mock_fingerprint.to_dict.return_value = {"md5": "test"}

        mock_deduplicator = MagicMock()
        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
        mock_deduplicator.check_duplicate.return_value = None
        mock_deduplicator.compute_duplicate_rate.return_value = 0.0

        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=mock_video_repo,
            ),
            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
        ):
            result = create_video_record_and_dedup(
                generation_task_id="task-002",
                project_id="proj-002",
                user_id="",
                batch_id="",
                file_url="https://example.com/video.mp4",
                file_size=1024,
                duration=10.0,
                video_path="/tmp/fake.mp4",
                mode="smart",
                session=session,
            )

        mock_deduplicator.compute_duplicate_rate.assert_called_once()
        call_kwargs = mock_deduplicator.compute_duplicate_rate.call_args
        assert call_kwargs.kwargs.get("user_id") == ""

    def test_duplicate_rate_saved_to_video_record(self):
        """compute_duplicate_rate 的返回值应写入 generated_video.duplicate_rate."""
        from video_processing.dedup_helpers import create_video_record_and_dedup

        session = MagicMock()
        mock_video_repo = MagicMock()

        mock_fingerprint = MagicMock()
        mock_fingerprint.to_dict.return_value = {"md5": "test"}

        mock_deduplicator = MagicMock()
        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
        mock_deduplicator.check_duplicate.return_value = None
        mock_deduplicator.compute_duplicate_rate.return_value = 78.5

        with (
            patch(
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
                return_value=mock_video_repo,
            ),
            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
        ):
            result = create_video_record_and_dedup(
                generation_task_id="task-003",
                project_id="proj-003",
                user_id="user-xyz",
                batch_id="",
                file_url="https://example.com/v.mp4",
                file_size=2048,
                duration=20.0,
                video_path="/tmp/fake2.mp4",
                mode="smart",
                session=session,
            )

        # 验证 update 被调用（包含 duplicate_rate 的记录）
        mock_video_repo.update.assert_called_once()
        updated_video = mock_video_repo.update.call_args[0][0]
        assert updated_video.duplicate_rate == 78.5