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