video-accessibility/backend/tests/unit/test_whisper_snap.py

"""Tests for the improved snap_pause_point algorithm (A1/A2/A3)."""

import sys
from unittest.mock import MagicMock

# faster_whisper ships only in the Docker image; stub it so pytest can run locally.
if 'faster_whisper' not in sys.modules:
    sys.modules['faster_whisper'] = MagicMock()

import pytest
from app.services.whisper_service import (
    WhisperService,
    WordTimestamp,
    SpeechGap,
    SentenceBoundary,
)


# ── fixtures ────────────────────────────────────────────────────────────────

@pytest.fixture
def svc():
    """WhisperService instance with default settings."""
    return WhisperService()


def _word(start: float, end: float, text: str = "word") -> WordTimestamp:
    return WordTimestamp(word=text, start=start, end=end)


def _gap(start: float, end: float, gap_type: str = "sentence") -> SpeechGap:
    return SpeechGap(start=start, end=end, duration=end - start, gap_type=gap_type)


def _boundary(
    time: float,
    btype: str = "sentence_end",
    has_prev: bool = True,
    has_next: bool = True,
    gap: SpeechGap | None = None,
) -> SentenceBoundary:
    return SentenceBoundary(
        time=time,
        boundary_type=btype,
        word_index=0,
        has_previous_sentence=has_prev,
        has_next_sentence=has_next,
        gap=gap,
    )


# ── A2: forward-preferred snap ───────────────────────────────────────────────

class TestForwardPreferredSnap:
    def test_picks_forward_over_equidistant_backward(self, svc):
        """Gemini=10.5s; forward boundary@11.2s and backward@9.8s — must pick forward."""
        gap = _gap(11.2, 11.8)
        boundaries = [
            _boundary(9.8, gap=_gap(9.8, 10.0)),
            _boundary(11.2, gap=gap),
        ]
        words = [_word(9.0, 9.5), _word(10.0, 10.5), _word(11.0, 11.2)]
        gaps = [_gap(9.8, 10.0), gap]

        pause, _, warning, _ = svc.snap_pause_point(10.5, words, gaps, boundaries)

        assert pause == pytest.approx(11.5, abs=0.01)  # midpoint of 11.2–11.8
        assert warning is None

    def test_forward_boundary_within_window_is_preferred(self, svc):
        """Even a slightly farther forward boundary beats a closer backward one."""
        gap_fwd = _gap(12.0, 12.6)
        gap_bwd = _gap(10.1, 10.4)
        boundaries = [
            _boundary(10.1, gap=gap_bwd),
            _boundary(12.0, gap=gap_fwd),
        ]
        words = [_word(9.0, 10.1), _word(10.5, 12.0)]
        gaps = [gap_bwd, gap_fwd]

        pause, _, _, _ = svc.snap_pause_point(10.5, words, gaps, boundaries)

        assert pause == pytest.approx(12.3, abs=0.01)  # midpoint of 12.0–12.6

    def test_falls_back_to_backward_when_no_forward_within_window(self, svc):
        """No forward boundary within snap_forward_window → use backward (within 1.5s)."""
        # Boundary at 9.2s: distance = 10.5 - 9.2 = 1.3s ≤ snap_backward_window (1.5s) ✓
        gap = _gap(9.0, 9.4)
        boundaries = [_boundary(9.0, gap=gap)]
        words = [_word(7.0, 9.0), _word(9.4, 10.5)]
        gaps = [gap]

        pause, _, warning, _ = svc.snap_pause_point(10.5, words, gaps, boundaries)

        assert pause == pytest.approx(9.2, abs=0.01)  # midpoint of 9.0–9.4

    def test_no_boundary_in_any_window_returns_gemini_with_warning(self, svc):
        """Boundary exists but outside both windows → exact Gemini point + warning."""
        # Put boundaries 10s away in both directions (beyond any window)
        boundaries = [
            _boundary(0.1, gap=_gap(0.0, 0.5)),
            _boundary(50.0, gap=_gap(49.0, 50.0)),
        ]
        words = [_word(9.0, 12.0)]
        gaps = []

        pause, resume, warning, _ = svc.snap_pause_point(10.5, words, gaps, boundaries)

        assert pause == pytest.approx(10.5)
        assert warning is not None
        assert "snap windows" in warning.lower()

    def test_not_during_speaking_uses_exact_point(self, svc):
        """Pause point far from all words → no snap, exact point returned."""
        boundaries = [_boundary(5.0, gap=_gap(4.8, 5.3))]
        words = [_word(0.0, 3.0)]  # speech ends at 3s; pause at 7s
        # Gap covers 3.0–10.0; pause at 7.0 is inside it
        gaps = [_gap(3.0, 10.0)]

        pause, _, warning, natural_gap_ms = svc.snap_pause_point(7.0, words, gaps, boundaries)

        assert pause == pytest.approx(7.0)
        assert warning is None
        # natural_gap covers the pause (7.0 is inside gap 3.0–10.0)
        assert natural_gap_ms > 0


# ── A1: natural_gap_ms returned correctly ───────────────────────────────────

class TestNaturalGapMs:
    def test_case_c_returns_gap_duration(self, svc):
        """Case C (gap midpoint) must return gap.duration * 1000 as natural_gap_ms."""
        gap = _gap(10.0, 11.2)
        boundaries = [_boundary(10.0, gap=gap)]
        words = [_word(9.0, 10.0), _word(11.2, 12.0)]
        gaps = [gap]

        _, _, _, natural_gap_ms = svc.snap_pause_point(10.5, words, gaps, boundaries)

        assert natural_gap_ms == pytest.approx(1200.0, abs=1.0)  # 1.2s gap

    def test_no_gap_returns_zero(self, svc):
        """Fallback case with no gap → natural_gap_ms == 0."""
        b = _boundary(10.0, gap=None)  # no gap attached
        words = [_word(9.0, 10.1), _word(10.1, 11.0)]
        gaps = []

        _, _, _, natural_gap_ms = svc.snap_pause_point(10.5, words, gaps, [b])

        assert natural_gap_ms == 0.0

    def test_not_during_speaking_reads_gap_from_gaps_list(self, svc):
        """Not-during-speaking path should read natural gap from the gaps list."""
        gap = _gap(5.0, 6.0)  # covers pause at 5.5s
        words = [_word(0.0, 3.0)]  # all speech before 3s
        gaps = [gap]

        _, _, _, natural_gap_ms = svc.snap_pause_point(5.5, words, gaps, [])

        assert natural_gap_ms == pytest.approx(1000.0, abs=1.0)


# ── A3: minimum gap validation ───────────────────────────────────────────────

class TestMinGapValidation:
    def test_short_gap_triggers_forward_search(self, svc):
        """Case C gap < min_acceptable_gap → searches forward for a better gap."""
        short_gap = _gap(10.0, 10.1)  # 0.1s < 0.2s threshold
        good_gap = _gap(11.5, 12.0)   # 0.5s — acceptable
        boundaries = [_boundary(10.0, gap=short_gap)]
        words = [_word(9.0, 10.0), _word(10.2, 11.5)]
        gaps = [short_gap, good_gap]

        pause, _, _, natural_gap_ms = svc.snap_pause_point(10.5, words, gaps, boundaries)

        # Should snap forward to midpoint of good_gap (11.5+12.0)/2 = 11.75
        assert pause == pytest.approx(11.75, abs=0.01)
        assert natural_gap_ms == pytest.approx(500.0, abs=1.0)

    def test_short_gap_no_forward_alternative_keeps_original(self, svc):
        """Short gap, no acceptable gap ahead → stays at original point with warning."""
        short_gap = _gap(10.0, 10.1)
        boundaries = [_boundary(10.0, gap=short_gap)]
        words = [_word(9.0, 10.0), _word(10.2, 14.0)]
        gaps = [short_gap]  # no other gap

        pause, _, warning, _ = svc.snap_pause_point(10.5, words, gaps, boundaries)

        # Falls back to midpoint of short_gap since no alternative
        assert pause == pytest.approx(10.05, abs=0.01)
        assert warning is None  # no warning for "stayed at original"

    def test_fallback_no_gap_triggers_forward_search(self, svc):
        """Fallback case (no gap on boundary) with no nearby gap → searches forward."""
        b = _boundary(10.0, gap=None)
        good_gap = _gap(11.0, 11.8)
        words = [_word(9.0, 10.0), _word(10.1, 11.0)]
        gaps = [good_gap]

        pause, _, warning, natural_gap_ms = svc.snap_pause_point(10.5, words, gaps, [b])

        assert pause == pytest.approx(11.4, abs=0.01)
        assert natural_gap_ms == pytest.approx(800.0, abs=1.0)
        assert warning is not None  # warns that it snapped forward


# ── refine_all_pause_points integration ─────────────────────────────────────

class TestRefineAllPausePointsIntegration:
    def test_stores_natural_gap_ms_on_placement(self, svc):
        """refine_all_pause_points must persist natural_gap_ms onto each placement."""
        gap = _gap(10.0, 11.0)
        words = [_word(8.0, 10.0), _word(11.0, 12.0)]
        gaps = [gap]
        placements = [{"ad_cue_index": 0, "pause_point": 10.5, "ad_duration": 3.0}]

        refined, _ = svc.refine_all_pause_points(placements, words, gaps)

        assert "natural_gap_ms" in refined[0]
        assert refined[0]["natural_gap_ms"] == pytest.approx(1000.0, abs=1.0)

    def test_no_whisper_data_returns_original_with_zero_gap(self, svc):
        """No words → _is_during_speaking=False → exact point, no warning, natural_gap_ms=0."""
        placements = [{"ad_cue_index": 0, "pause_point": 5.0, "ad_duration": 2.0}]

        refined, warnings = svc.refine_all_pause_points(placements, [], [])

        assert refined[0]["pause_point"] == pytest.approx(5.0)
        assert refined[0].get("natural_gap_ms", 0) == 0.0
        # No words → not-during-speaking path → no snap → no warning
        assert len(warnings) == 0