audita/tests/test_corrections.py

import pytest

from audita.corrections import apply_corrections
from audita.errors import AuditaValidationError
from audita.schemas import CorrectionCandidate, parse_transcript_json


def _transcript():
    return parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 10.0, "end": 11.0, "text": "I ask Chontia for help."},
          {"id": 2, "speaker": "Mike", "start": 0.0, "end": 1.0, "text": "Then Lyra."}
        ]
        """
    )


def test_apply_corrections_uses_threshold_and_preserves_id_order():
    transcript = _transcript()
    corrections = [
        CorrectionCandidate(
            id=1,
            original_text="Chontia",
            corrected_text="Chauntea",
            confidence=0.8,
        )
    ]

    result = apply_corrections(transcript, corrections, confidence_threshold=0.8)

    assert [segment.speaker for segment in result.transcript] == ["Eric", "Mike"]
    assert result.transcript[0].text == "I ask Chauntea for help."
    assert result.skipped == []


def test_apply_corrections_ignores_below_threshold():
    transcript = _transcript()
    corrections = [
        CorrectionCandidate(
            id=1,
            original_text="Chontia",
            corrected_text="Chauntea",
            confidence=0.79,
        )
    ]

    result = apply_corrections(transcript, corrections, confidence_threshold=0.8)

    assert result.transcript[0].text == "I ask Chontia for help."
    assert result.skipped == []


def test_apply_corrections_allows_multiple_distinct_spans_in_one_segment():
    transcript = _transcript()
    first = CorrectionCandidate(
        id=1,
        original_text="Chontia",
        corrected_text="Chauntea",
        confidence=0.8,
    )
    second = CorrectionCandidate(
        id=1,
        original_text="help",
        corrected_text="guidance",
        confidence=0.9,
    )

    result = apply_corrections(transcript, [first, second], confidence_threshold=0.8)

    assert result.transcript[0].text == "I ask Chauntea for guidance."
    assert result.skipped == []


def test_apply_corrections_skips_missing_substring():
    transcript = _transcript()
    correction = CorrectionCandidate(
        id=1,
        original_text="Different text.",
        corrected_text="Chauntea",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert result.transcript[0].text == "I ask Chontia for help."
    assert len(result.skipped) == 1
    assert result.skipped[0].id == 1
    assert result.skipped[0].actual_text == "I ask Chontia for help."
    assert "does not match any substring" in result.skipped[0].reason


def test_apply_corrections_skips_missing_id():
    transcript = _transcript()
    correction = CorrectionCandidate(
        id=99,
        original_text="Missing.",
        corrected_text="Still missing.",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert [segment.text for segment in result.transcript] == ["I ask Chontia for help.", "Then Lyra."]
    assert len(result.skipped) == 1
    assert result.skipped[0].id == 99
    assert "does not exist" in result.skipped[0].reason


def test_apply_corrections_skips_no_op():
    transcript = _transcript()
    correction = CorrectionCandidate(
        id=1,
        original_text="Chontia",
        corrected_text="Chontia",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert result.transcript[0].text == "I ask Chontia for help."
    assert len(result.skipped) == 1
    assert "identical" in result.skipped[0].reason


def test_apply_corrections_replaces_all_repeated_substrings():
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Bane met Bane."}
        ]
        """
    )
    correction = CorrectionCandidate(
        id=1,
        original_text="Bane",
        corrected_text="Bain",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert result.transcript[0].text == "Bain met Bain."
    assert result.skipped == []


def test_apply_corrections_requires_unique_match_when_configured():
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "there and there"}
        ]
        """
    )
    correction = CorrectionCandidate(
        id=1,
        original_text="there",
        corrected_text="their",
        confidence=0.8,
    )

    result = apply_corrections(
        transcript,
        [correction],
        confidence_threshold=0.8,
        replacement_mode="require_unique",
    )

    assert result.transcript[0].text == "there and there"
    assert len(result.skipped) == 1
    assert "more than once" in result.skipped[0].reason


def test_apply_corrections_skips_when_guard_rejects_replacement():
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hrank moves."}
        ]
        """
    )
    correction = CorrectionCandidate(
        id=1,
        original_text="Hrank",
        corrected_text="Frank",
        confidence=0.8,
    )

    result = apply_corrections(
        transcript,
        [correction],
        confidence_threshold=0.8,
        replacement_mode="require_unique",
        correction_guard=lambda before, after: "protected term changed" if before != after else None,
    )

    assert result.transcript[0].text == "Hrank moves."
    assert len(result.skipped) == 1
    assert result.skipped[0].reason == "protected term changed"
    assert result.skipped[0].actual_text == "Hrank moves."


def test_apply_corrections_without_guard_remains_permissive():
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Frank moves."}
        ]
        """
    )
    correction = CorrectionCandidate(
        id=1,
        original_text="Frank",
        corrected_text="Hrank",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert result.transcript[0].text == "Hrank moves."
    assert result.skipped == []


def test_apply_corrections_skips_empty_original_text():
    transcript = _transcript()
    correction = CorrectionCandidate(
        id=1,
        original_text="",
        corrected_text="Chauntea",
        confidence=0.8,
    )

    result = apply_corrections(transcript, [correction], confidence_threshold=0.8)

    assert result.transcript[0].text == "I ask Chontia for help."
    assert len(result.skipped) == 1
    assert "empty" in result.skipped[0].reason


def test_apply_corrections_rejects_invalid_threshold():
    with pytest.raises(AuditaValidationError):
        apply_corrections(_transcript(), [], confidence_threshold=1.1)


def test_apply_corrections_rejects_invalid_replacement_mode():
    with pytest.raises(AuditaValidationError):
        apply_corrections(_transcript(), [], confidence_threshold=0.8, replacement_mode="unknown")