"""Tests for VTT retiming service for pause-insert accessible videos.""" import pytest from app.services.vtt_retimer import VTTRetimerService class TestVTTRetimerPauseInsert: """Test VTT retiming for pause-insert accessible videos.""" @pytest.fixture def retimer(self): """Create a VTTRetimerService instance.""" return VTTRetimerService() def test_cue_spanning_single_pause(self, retimer): """Cue 8s-12s with pause at 10s (4s freeze) should split into 8s-10s + 14s-16s.""" vtt = """WEBVTT 00:00:08.000 --> 00:00:12.000 Hello world """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 # freeze = 3.0 + 1.0 = 4.0s }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Should split into two segments assert len(cues) == 2 # Segment 1: 8s-10s (before freeze, no offset) assert cues[0]["start_time"] == 8.0 assert cues[0]["end_time"] == 10.0 assert cues[0]["text"] == "Hello world" # Segment 2: 14s-16s (after freeze, +4s offset) # Maps source 10s-12s to accessible 14s-16s assert cues[1]["start_time"] == 14.0 assert cues[1]["end_time"] == 16.0 assert cues[1]["text"] == "Hello world" def test_cue_entirely_before_pause(self, retimer): """Cue 2s-4s with pause at 10s should remain unchanged.""" vtt = """WEBVTT 00:00:02.000 --> 00:00:04.000 Early caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # No offset applied (cue is before pause) assert len(cues) == 1 assert cues[0]["start_time"] == 2.0 assert cues[0]["end_time"] == 4.0 assert cues[0]["text"] == "Early caption" def test_cue_entirely_after_pause(self, retimer): """Cue 15s-18s with pause at 10s (4s freeze) should shift to 19s-22s.""" vtt = """WEBVTT 00:00:15.000 --> 00:00:18.000 Later caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 # freeze = 4.0s }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Full offset applied (cue is after pause) assert len(cues) == 1 assert cues[0]["start_time"] == 19.0 # 15 + 4 assert cues[0]["end_time"] == 22.0 # 18 + 4 assert cues[0]["text"] == "Later caption" def test_cue_starting_at_pause_point(self, retimer): """Cue 10s-12s with pause at 10s (4s freeze) should shift to 14s-16s.""" vtt = """WEBVTT 00:00:10.000 --> 00:00:12.000 Caption at pause """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Cue starts AT pause point, so gets full offset # No split needed (pause_point is not STRICTLY between start and end) assert len(cues) == 1 assert cues[0]["start_time"] == 14.0 # 10 + 4 assert cues[0]["end_time"] == 16.0 # 12 + 4 def test_cue_ending_at_pause_point(self, retimer): """Cue 8s-10s with pause at 10s should remain at 8s-10s.""" vtt = """WEBVTT 00:00:08.000 --> 00:00:10.000 Caption before pause """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Cue ends AT pause point, so no offset (pause not strictly between) # Uses offset_at(8) = 0 (pause at 10 is not <= 8) assert len(cues) == 1 assert cues[0]["start_time"] == 8.0 assert cues[0]["end_time"] == 10.0 def test_cue_spanning_multiple_pauses(self, retimer): """Cue 5s-25s with pauses at 10s and 20s should split into 3 segments.""" vtt = """WEBVTT 00:00:05.000 --> 00:00:25.000 Long caption """ analysis = { "placements": [ {"pause_point": 10.0, "ad_duration": 3.0}, # 4s freeze {"pause_point": 20.0, "ad_duration": 5.0} # 6s freeze ] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Should split into 3 segments assert len(cues) == 3 # Segment 1: 5s-10s (no offset, before first pause) assert cues[0]["start_time"] == 5.0 assert cues[0]["end_time"] == 10.0 assert cues[0]["text"] == "Long caption" # Segment 2: 14s-24s (source 10s-20s + 4s offset from first pause) assert cues[1]["start_time"] == 14.0 assert cues[1]["end_time"] == 24.0 assert cues[1]["text"] == "Long caption" # Segment 3: 30s-35s (source 20s-25s + 10s total offset) assert cues[2]["start_time"] == 30.0 # 20 + 4 + 6 assert cues[2]["end_time"] == 35.0 # 25 + 4 + 6 assert cues[2]["text"] == "Long caption" def test_short_segment_filtered(self, retimer): """Very short segments (<100ms) should be filtered out.""" vtt = """WEBVTT 00:00:09.950 --> 00:00:10.050 Tiny caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Both segments are < 100ms, should be filtered out # Segment 1: 9.95s-10.0s = 50ms # Segment 2: 14.0s-14.05s = 50ms assert len(cues) == 0 def test_no_pauses_passthrough(self, retimer): """No placements should return original VTT unchanged.""" vtt = """WEBVTT 00:00:05.000 --> 00:00:08.000 Caption """ analysis = {"placements": []} result = retimer.retime_for_pause_insert(vtt, analysis) # Should return original unchanged assert result == vtt def test_multiple_cues_with_pause(self, retimer): """Multiple cues with a pause between them.""" vtt = """WEBVTT 00:00:02.000 --> 00:00:05.000 First caption 00:00:15.000 --> 00:00:18.000 Second caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 # 4s freeze }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) assert len(cues) == 2 # First cue: before pause, no offset assert cues[0]["start_time"] == 2.0 assert cues[0]["end_time"] == 5.0 assert cues[0]["text"] == "First caption" # Second cue: after pause, +4s offset assert cues[1]["start_time"] == 19.0 # 15 + 4 assert cues[1]["end_time"] == 22.0 # 18 + 4 assert cues[1]["text"] == "Second caption" def test_cue_with_pause_at_start_of_video(self, retimer): """Pause at 0.0 should shift all subsequent cues.""" vtt = """WEBVTT 00:00:05.000 --> 00:00:08.000 Caption after start """ analysis = { "placements": [{ "pause_point": 0.0, "ad_duration": 3.0 # 4s freeze }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) assert len(cues) == 1 assert cues[0]["start_time"] == 9.0 # 5 + 4 assert cues[0]["end_time"] == 12.0 # 8 + 4 def test_consolidated_pauses_same_pause_point(self, retimer): """Multiple placements at the same pause point should be handled correctly.""" vtt = """WEBVTT 00:00:15.000 --> 00:00:18.000 Caption after pauses """ analysis = { "placements": [ {"pause_point": 10.0, "ad_duration": 2.0}, # 3s freeze {"pause_point": 10.0, "ad_duration": 3.0} # 4s freeze (consolidated at same point) ] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) assert len(cues) == 1 # Total offset = 3s + 4s = 7s assert cues[0]["start_time"] == 22.0 # 15 + 7 assert cues[0]["end_time"] == 25.0 # 18 + 7 def test_multiline_cue_text_preserved(self, retimer): """Multi-line cue text should be preserved when retiming.""" vtt = """WEBVTT 00:00:08.000 --> 00:00:12.000 Line one Line two """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) assert len(cues) == 2 # Both segments should have the multi-line text assert cues[0]["text"] == "Line one\nLine two" assert cues[1]["text"] == "Line one\nLine two" class TestVTTRetimerHelpers: """Test helper methods of VTTRetimerService.""" @pytest.fixture def retimer(self): """Create a VTTRetimerService instance.""" return VTTRetimerService() def test_offset_at_before_pause(self, retimer): """offset_at should return 0 for timestamps before any pause.""" pauses = [(10.0, 4.0)] assert retimer._offset_at(5.0, pauses) == 0.0 def test_offset_at_at_pause(self, retimer): """offset_at should include offset for timestamps AT pause point.""" pauses = [(10.0, 4.0)] assert retimer._offset_at(10.0, pauses) == 4.0 def test_offset_at_after_pause(self, retimer): """offset_at should include offset for timestamps after pause.""" pauses = [(10.0, 4.0)] assert retimer._offset_at(15.0, pauses) == 4.0 def test_offset_at_multiple_pauses(self, retimer): """offset_at should sum offsets from all applicable pauses.""" pauses = [(10.0, 4.0), (20.0, 6.0)] assert retimer._offset_at(25.0, pauses) == 10.0 # 4 + 6 def test_offset_before_before_pause(self, retimer): """offset_before should return 0 for timestamps before any pause.""" pauses = [(10.0, 4.0)] assert retimer._offset_before(5.0, pauses) == 0.0 def test_offset_before_at_pause(self, retimer): """offset_before should NOT include offset for timestamps AT pause point.""" pauses = [(10.0, 4.0)] assert retimer._offset_before(10.0, pauses) == 0.0 def test_offset_before_after_pause(self, retimer): """offset_before should include offset for timestamps after pause.""" pauses = [(10.0, 4.0)] assert retimer._offset_before(15.0, pauses) == 4.0 def test_build_pause_list(self, retimer): """_build_pause_list should create sorted list of (pause_point, effective_offset).""" placements = [ {"pause_point": 20.0, "ad_duration": 5.0}, {"pause_point": 10.0, "ad_duration": 3.0}, ] pauses = retimer._build_pause_list(placements) assert len(pauses) == 2 # Should be sorted by pause_point assert pauses[0] == (10.0, 4.0) # 3.0 + 1.0 silence assert pauses[1] == (20.0, 6.0) # 5.0 + 1.0 silence def test_build_pause_list_filters_invalid(self, retimer): """_build_pause_list should filter out invalid placements.""" placements = [ {"pause_point": 10.0, "ad_duration": 3.0}, # valid {"pause_point": None, "ad_duration": 3.0}, # invalid: no pause_point {"pause_point": 20.0, "ad_duration": 0}, # invalid: zero duration {"pause_point": 30.0}, # invalid: no ad_duration ] pauses = retimer._build_pause_list(placements) assert len(pauses) == 1 assert pauses[0] == (10.0, 4.0) def test_filter_short_segments(self, retimer): """_filter_short_segments should remove segments shorter than MIN_SEGMENT_DURATION.""" cues = [ {"start_time": 0.0, "end_time": 1.0, "text": "long enough"}, {"start_time": 2.0, "end_time": 2.05, "text": "too short"}, # 50ms {"start_time": 3.0, "end_time": 3.1, "text": "just enough"}, # 100ms ] filtered = retimer._filter_short_segments(cues) assert len(filtered) == 2 assert filtered[0]["text"] == "long enough" assert filtered[1]["text"] == "just enough" def test_build_pause_list_uses_actual_freeze_duration(self, retimer): """_build_pause_list should use actual_freeze_duration when available.""" placements = [ {"pause_point": 10.0, "ad_duration": 3.0, "actual_freeze_duration": 4.5}, {"pause_point": 20.0, "ad_duration": 5.0}, # No actual, should calculate ] pauses = retimer._build_pause_list(placements) assert len(pauses) == 2 # First pause: should use actual_freeze_duration (4.5s) assert pauses[0] == (10.0, 4.5) # Second pause: should use calculated (5.0 + 1.0 = 6.0s) assert pauses[1] == (20.0, 6.0) class TestVTTRetimerActualFreezeDuration: """Test VTT retiming with actual_freeze_duration field.""" @pytest.fixture def retimer(self): """Create a VTTRetimerService instance.""" return VTTRetimerService() def test_uses_actual_freeze_duration_when_provided(self, retimer): """Should use actual_freeze_duration instead of calculated value.""" vtt = """WEBVTT 00:00:15.000 --> 00:00:18.000 Later caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0, "actual_freeze_duration": 4.5 # Actual is 0.5s longer than calculated }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Should use actual_freeze_duration (4.5s) not calculated (4.0s) assert len(cues) == 1 assert cues[0]["start_time"] == 19.5 # 15 + 4.5 assert cues[0]["end_time"] == 22.5 # 18 + 4.5 def test_falls_back_to_calculated_when_actual_not_provided(self, retimer): """Should use calculated value when actual_freeze_duration not provided.""" vtt = """WEBVTT 00:00:15.000 --> 00:00:18.000 Later caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0 # No actual_freeze_duration }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Should use calculated (3.0 + 1.0 = 4.0s) assert len(cues) == 1 assert cues[0]["start_time"] == 19.0 # 15 + 4 assert cues[0]["end_time"] == 22.0 # 18 + 4 def test_mixed_actual_and_calculated(self, retimer): """Should handle mix of actual and calculated freeze durations.""" vtt = """WEBVTT 00:00:25.000 --> 00:00:28.000 Late caption """ analysis = { "placements": [ {"pause_point": 10.0, "ad_duration": 3.0, "actual_freeze_duration": 4.2}, {"pause_point": 20.0, "ad_duration": 5.0} # No actual ] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Total offset = 4.2 (actual) + 6.0 (calculated: 5+1) = 10.2 assert len(cues) == 1 assert cues[0]["start_time"] == pytest.approx(35.2, rel=1e-3) # 25 + 10.2 assert cues[0]["end_time"] == pytest.approx(38.2, rel=1e-3) # 28 + 10.2 def test_cue_spanning_pause_with_actual_duration(self, retimer): """Cue spanning pause should use actual_freeze_duration for split timing.""" vtt = """WEBVTT 00:00:08.000 --> 00:00:12.000 Spanning caption """ analysis = { "placements": [{ "pause_point": 10.0, "ad_duration": 3.0, "actual_freeze_duration": 4.3 # Slightly longer than calculated }] } result = retimer.retime_for_pause_insert(vtt, analysis) cues = retimer._parse_vtt(result) # Should split into two segments assert len(cues) == 2 # Segment 1: 8s-10s (before freeze, no offset) assert cues[0]["start_time"] == 8.0 assert cues[0]["end_time"] == 10.0 # Segment 2: 14.3s-16.3s (after freeze, +4.3s offset) assert cues[1]["start_time"] == pytest.approx(14.3, rel=1e-3) assert cues[1]["end_time"] == pytest.approx(16.3, rel=1e-3)