xiaoxia-saas/tests/unit/test_dedup_two_phase_commit.py

"""Tests for two-phase commit pattern in dedup_helpers (#1664 follow-up).

Verifies that the new dedup_helpers.py:
1. Creates video with all dedup fields in a single commit
2. Still creates video when fingerprint computation fails
3. Creates video with fingerprint but no rate when rate computation fails
4. Never does a partial commit (no create + separate update)
"""

from __future__ import annotations

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

# 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")

import pytest
from video_processing.dedup_helpers import create_video_record_and_dedup


@pytest.fixture
def session():
    s = MagicMock()
    return s


@pytest.fixture
def mock_fingerprint():
    fp = MagicMock()
    fp.duration = 15000  # 15 seconds in ms
    fp.to_dict.return_value = {"md5": "abc123", "keyframe_phashes": ["aabb"], "color_histograms": []}
    fp.chunks = []
    fp.keyframe_phashes = ["aabb"]
    fp.color_histograms = []
    fp.md5 = "abc123"
    return fp


class TestTwoPhaseCommit:
    """Verify that dedup data is computed before commit."""

    def test_video_created_with_all_dedup_fields(self, session, mock_fingerprint):
        """When all computations succeed, video is created with all fields in one commit."""
        mock_repo = MagicMock()
        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 = {
            "duplicate_rate": 42.5,
            "visual_similarity": 0.75,
            "match_count": 2,
        }

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

        assert result == 1
        # create() should be called exactly once with the complete video object
        mock_repo.create.assert_called_once()
        created_video = mock_repo.create.call_args[0][0]
        assert created_video.duplicate_rate == 42.5
        assert created_video.visual_similarity == 0.75
        assert created_video.match_count == 2
        assert created_video.video_fingerprint is not None
        # session.commit should be called exactly once (at the end)
        session.commit.assert_called_once()

    def test_video_created_even_when_fingerprint_fails(self, session):
        """When fingerprint computation fails, video is still created (without dedup data)."""
        mock_repo = MagicMock()
        mock_deduplicator = MagicMock()
        mock_deduplicator.compute_fingerprint.side_effect = RuntimeError("cv2 not available")

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

        assert result == 1
        mock_repo.create.assert_called_once()
        created_video = mock_repo.create.call_args[0][0]
        assert created_video.duplicate_rate is None
        assert created_video.video_fingerprint is None
        session.commit.assert_called_once()
        # No dedup methods should have been called
        mock_deduplicator.check_duplicate.assert_not_called()
        mock_deduplicator.compute_duplicate_rate.assert_not_called()

    def test_video_created_with_fingerprint_but_no_rate(self, session, mock_fingerprint):
        """When rate computation fails, video is created with fingerprint but no rate."""
        mock_repo = MagicMock()
        mock_deduplicator = MagicMock()
        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
        mock_deduplicator.check_duplicate.return_value = None
        mock_deduplicator.compute_duplicate_rate.side_effect = RuntimeError("DB error")

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

        assert result == 1
        mock_repo.create.assert_called_once()
        created_video = mock_repo.create.call_args[0][0]
        # Fingerprint should be set
        assert created_video.video_fingerprint is not None
        # But duplicate_rate should be None
        assert created_video.duplicate_rate is None
        session.commit.assert_called_once()

    def test_no_separate_update_call(self, session, mock_fingerprint):
        """Verify the new pattern uses create() only, not create() + update()."""
        mock_repo = MagicMock()
        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 = {
            "duplicate_rate": 10.0,
            "visual_similarity": 0.5,
            "match_count": 1,
        }

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

        # Only create() should be called, not update()
        mock_repo.create.assert_called_once()
        mock_repo.update.assert_not_called()

    def test_commit_not_called_on_total_failure(self, session):
        """When the entire function fails, session.rollback is called instead of commit."""
        mock_repo = MagicMock()
        mock_repo.create.side_effect = RuntimeError("DB connection lost")

        with patch(
            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
            return_value=mock_repo,
        ):
            result = create_video_record_and_dedup(
                generation_task_id="task-005",
                project_id="proj-001",
                user_id="user-001",
                batch_id="",
                file_url="https://example.com/v.mp4",
                file_size=1024,
                duration=15.0,
                video_path="/tmp/fake.mp4",
                mode="smart",
                session=session,
            )

        assert result == 0
        session.commit.assert_not_called()
        session.rollback.assert_called_once()