audita/python/tests/test_protected_validator.py

from audita.core.schemas import parse_glossary_yaml, parse_transcript_json
from audita.framework.models import CorrectionProposal, ModuleRunSpec
from audita.validators import (
    GlossaryStageProtectedGlossaryTermsValidator,
    ProtectedGlossaryTermsValidator,
    ProtectedVocabulary,
)
from audita.validators.base import ValidationContext


def _glossary():
    return parse_glossary_yaml(
        """
        glossary:
          - name: "Hrank"
            aliases:
              - "Greenfield"
            category: pc
            summary: "Hrank Greenfield is a player character."
          - name: "Popov"
            category: npc
            summary: "Popov is an allied NPC."
          - name: "Jesters"
            aliases:
              - "Jester"
            category: faction
            summary: "The Jesters are a faction."
          - name: "Svend"
            category: pc
            summary: "Svend is a player character."
          - name: "Godfrey"
            category: npc
            summary: "Godfrey is an NPC."
          - name: "Lyra"
            category: npc
            summary: "Lyra is an NPC."
          - name: "Loviator"
            category: deity
            summary: "Loviator is a deity."
        """
    )


def test_protected_vocabulary_blocks_replacing_protected_term():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert (
        vocabulary.violation_reason("Hrank moves.", "Frank moves.")
        == "correction changes protected glossary term usage"
    )


def test_protected_vocabulary_glossary_stage_allows_glossary_to_glossary_changes():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert vocabulary.violation_reason("Hrank moves.", "Popov moves.") == "correction changes protected glossary term usage"
    assert vocabulary.glossary_stage_violation_reason("Hrank moves.", "Popov moves.") is None


def test_protected_vocabulary_glossary_stage_still_blocks_glossary_to_nonglossary_changes():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert (
        vocabulary.glossary_stage_violation_reason("Hrank moves.", "Frank moves.")
        == "correction changes protected glossary term usage"
    )


def test_protected_vocabulary_blocks_noncanonical_capitalization():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert (
        vocabulary.violation_reason("Popov moves.", "POPOV moves.")
        == "correction changes protected glossary term capitalization"
    )
    assert (
        vocabulary.violation_reason("Jesters", "jesters")
        == "correction changes protected glossary term capitalization"
    )
    assert (
        vocabulary.glossary_stage_violation_reason("Popov moves.", "POPOV moves.")
        == "correction changes protected glossary term capitalization"
    )


def test_protected_vocabulary_allows_corrections_toward_protected_terms():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert vocabulary.violation_reason("Pawpaw moves.", "Popov moves.") is None
    assert vocabulary.violation_reason("gestures", "Jesters") is None
    assert vocabulary.violation_reason("gestures", "jesters") is None
    assert vocabulary.violation_reason("rank", "Hrank") is None
    assert vocabulary.violation_reason("rank", "hrank") is None
    assert vocabulary.violation_reason("spend", "Svend") is None


def test_protected_vocabulary_allows_unchanged_noncanonical_terms_and_quote_wrapping():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert vocabulary.violation_reason("jesters advance.", "jesters advance.") is None
    before = (
        "When you say that, Popov will say, when I was in that room with the jesters, "
        "I just knew that Godfrey and Lyra came directly from Loviator herself."
    )
    after = (
        'When you say that, Popov will say, "When I was in that room with the jesters, '
        'I just knew that Godfrey and Lyra came directly from Loviator herself."'
    )
    assert vocabulary.violation_reason(before, after) is None


def test_protected_vocabulary_allows_inferred_and_explicit_plurals():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())
    explicit = ProtectedVocabulary.from_glossary(
        parse_glossary_yaml(
            """
            glossary:
              - name: "Mox"
                plural: "Moxen"
                category: faction
                summary: "The Mox are a faction."
            """
        )
    )

    assert vocabulary.violation_reason("Godfrey's", "Godfreys") is None
    assert vocabulary.violation_reason("gesture", "Jesters") is None
    assert explicit.violation_reason("Mox's", "Moxen") is None


def test_protected_vocabulary_does_not_match_embedded_substrings():
    vocabulary = ProtectedVocabulary.from_glossary(_glossary())

    assert vocabulary.violation_reason("The shrank spell worked.", "The shrank spell works.") is None


def test_protected_glossary_terms_validator_returns_proposal_indexed_decisions():
    validator = ProtectedGlossaryTermsValidator("protected_glossary_guard")
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Frank moves."},
          {"id": 2, "speaker": "Eric", "start": 1.0, "end": 2.0, "text": "Pawpaw waits."}
        ]
        """
    )
    proposals = [
        CorrectionProposal(
            proposal_index=0,
            module_instance="glossary_primary",
            module_key="glossary",
            id=1,
            original_text="Hrank",
            corrected_text="Frank",
            confidence=0.9,
        ),
        CorrectionProposal(
            proposal_index=1,
            module_instance="glossary_primary",
            module_key="glossary",
            id=2,
            original_text="Pawpaw",
            corrected_text="Popov",
            confidence=0.9,
        ),
    ]
    result = validator.validate(
        ValidationContext(
            proposals=proposals,
            transcript=transcript,
            glossary=_glossary(),
            config=None,  # type: ignore[arg-type]
            run_spec=ModuleRunSpec(instance_name="glossary_primary", module_key="glossary", module=None),  # type: ignore[arg-type]
            run_dir=transcript[0].__class__.__module__ and __import__("pathlib").Path("."),
        )
    )

    assert [decision.proposal_index for decision in result.decisions] == [0, 1]
    assert result.decisions[0].approved is False
    assert result.decisions[0].reason == "correction changes protected glossary term usage"
    assert result.decisions[1].approved is True


def test_protected_glossary_terms_validator_allows_nonglossary_to_lowercase_glossary_replacement():
    validator = ProtectedGlossaryTermsValidator("protected_glossary_guard")
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "gestures advance."},
          {"id": 2, "speaker": "Eric", "start": 1.0, "end": 2.0, "text": "rank moves."}
        ]
        """
    )
    proposals = [
        CorrectionProposal(
            proposal_index=0,
            module_instance="homophones",
            module_key="homophones",
            id=1,
            original_text="gestures",
            corrected_text="jesters",
            confidence=0.9,
        ),
        CorrectionProposal(
            proposal_index=1,
            module_instance="homophones",
            module_key="homophones",
            id=2,
            original_text="rank",
            corrected_text="hrank",
            confidence=0.9,
        ),
    ]

    result = validator.validate(
        ValidationContext(
            proposals=proposals,
            transcript=transcript,
            glossary=_glossary(),
            config=None,  # type: ignore[arg-type]
            run_spec=ModuleRunSpec(instance_name="homophones", module_key="homophones", module=None),  # type: ignore[arg-type]
            run_dir=__import__("pathlib").Path("."),
        )
    )

    assert [decision.approved for decision in result.decisions] == [True, True]


def test_glossary_stage_protected_glossary_terms_validator_allows_glossary_to_glossary_replacement():
    validator = GlossaryStageProtectedGlossaryTermsValidator("glossary_stage_protected_glossary_guard")
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hrank moves."},
          {"id": 2, "speaker": "Eric", "start": 1.0, "end": 2.0, "text": "Hrank moves."}
        ]
        """
    )
    proposals = [
        CorrectionProposal(
            proposal_index=0,
            module_instance="glossary_1",
            module_key="glossary",
            id=1,
            original_text="Hrank",
            corrected_text="Popov",
            confidence=0.9,
        ),
        CorrectionProposal(
            proposal_index=1,
            module_instance="glossary_1",
            module_key="glossary",
            id=2,
            original_text="Hrank",
            corrected_text="POPOV",
            confidence=0.9,
        ),
    ]

    result = validator.validate(
        ValidationContext(
            proposals=proposals,
            transcript=transcript,
            glossary=_glossary(),
            config=None,  # type: ignore[arg-type]
            run_spec=ModuleRunSpec(instance_name="glossary_1", module_key="glossary", module=None),  # type: ignore[arg-type]
            run_dir=__import__("pathlib").Path("."),
        )
    )

    assert [decision.proposal_index for decision in result.decisions] == [0, 1]
    assert result.decisions[0].approved is True
    assert result.decisions[1].approved is True
    assert result.decisions[1].reason is None


def test_protected_glossary_terms_validator_uses_proposal_span_only():
    validator = ProtectedGlossaryTermsValidator("protected_glossary_guard")
    glossary = parse_glossary_yaml(
        """
        glossary:
          - name: "Svend"
            category: pc
            summary: "Svend is a player character."
          - name: "Jesters"
            category: faction
            summary: "The Jesters are a faction."
        """
    )
    transcript = parse_transcript_json(
        """
        [
          {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "You have to keep it bind. Svend sees the jesters."}
        ]
        """
    )
    proposals = [
        CorrectionProposal(
            proposal_index=0,
            module_instance="spoken_word",
            module_key="spoken_word",
            id=1,
            original_text="keep it bind",
            corrected_text="keep in mind",
            confidence=0.9,
        )
    ]

    result = validator.validate(
        ValidationContext(
            proposals=proposals,
            transcript=transcript,
            glossary=glossary,
            config=None,  # type: ignore[arg-type]
            run_spec=ModuleRunSpec(instance_name="spoken_word", module_key="spoken_word", module=None),  # type: ignore[arg-type]
            run_dir=__import__("pathlib").Path("."),
        )
    )

    assert result.decisions[0].approved is True