364 lines
13 KiB
Python
364 lines
13 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from audita.core.config import AuditaConfig
|
|
from audita.core.errors import AuditaLLMError
|
|
from audita.core.io import write_report
|
|
from audita.core.schemas import parse_glossary_yaml, parse_source_transcript_json
|
|
from audita.modules import default_module_specs
|
|
from audita.pipeline import process_transcript, process_transcript_result
|
|
|
|
|
|
class FakeStructuredLLMClient:
|
|
def __init__(self, responses):
|
|
self._responses = list(responses)
|
|
self.calls = []
|
|
|
|
def run_structured(self, *, stage_name, messages, response_model, config):
|
|
self.calls.append(
|
|
{
|
|
"stage_name": stage_name,
|
|
"messages": list(messages),
|
|
"response_model": response_model,
|
|
}
|
|
)
|
|
if not self._responses:
|
|
raise AuditaLLMError("FakeStructuredLLMClient received more calls than expected.")
|
|
response = self._responses.pop(0)
|
|
if isinstance(response, Exception):
|
|
raise response
|
|
return response_model.model_validate(response)
|
|
|
|
|
|
def _glossary():
|
|
return parse_glossary_yaml(
|
|
"""
|
|
glossary:
|
|
- name: "Jesters"
|
|
category: faction
|
|
summary: "A faction."
|
|
"""
|
|
)
|
|
|
|
|
|
def _transcript():
|
|
return parse_source_transcript_json(
|
|
"""
|
|
[
|
|
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello."},
|
|
{"speaker": "Eric", "start": 1.5, "end": 2.0, "text": "Again."},
|
|
{"speaker": "Mike", "start": 10.0, "end": 11.0, "text": "Done."}
|
|
]
|
|
"""
|
|
)
|
|
|
|
|
|
def test_process_transcript_runs_noop_framework(tmp_path):
|
|
llm_client = FakeStructuredLLMClient(
|
|
[
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
]
|
|
)
|
|
revised = process_transcript(
|
|
_transcript(),
|
|
_glossary(),
|
|
AuditaConfig.from_sources(env={}, overrides=None),
|
|
llm_client=llm_client,
|
|
)
|
|
|
|
assert [segment.id for segment in revised] == [1, 2]
|
|
assert revised[0].text == "Hello. Again."
|
|
assert revised[1].text == "Done."
|
|
assert [call["stage_name"] for call in llm_client.calls] == [
|
|
"glossary_primary:proposal",
|
|
"homophones:proposal",
|
|
"glossary_secondary:proposal",
|
|
]
|
|
|
|
|
|
def test_process_transcript_result_writes_report_and_preserves_skips_per_policy(tmp_path):
|
|
config = AuditaConfig.from_sources(
|
|
env={},
|
|
overrides=None,
|
|
)
|
|
config = AuditaConfig(
|
|
api_key=config.api_key,
|
|
model=config.model,
|
|
base_url=config.base_url,
|
|
max_retries=config.max_retries,
|
|
max_section_tokens=config.max_section_tokens,
|
|
glossary_confidence_threshold=config.glossary_confidence_threshold,
|
|
homophones_confidence_threshold=config.homophones_confidence_threshold,
|
|
normalize_max_segment_gap=config.normalize_max_segment_gap,
|
|
normalize_ellipsis_gap=config.normalize_ellipsis_gap,
|
|
normalize_max_segment_duration=config.normalize_max_segment_duration,
|
|
normalize_max_segment_tokens=config.normalize_max_segment_tokens,
|
|
work_dir=tmp_path / "work",
|
|
work_dir_retention="always",
|
|
)
|
|
|
|
llm_client = FakeStructuredLLMClient(
|
|
[
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
]
|
|
)
|
|
result = process_transcript_result(_transcript(), _glossary(), config, llm_client=llm_client)
|
|
|
|
assert result.work_dir_retained is True
|
|
assert result.report.pipeline == [
|
|
"glossary_primary",
|
|
"homophones",
|
|
"glossary_secondary",
|
|
"spoken_word",
|
|
"grammar",
|
|
]
|
|
assert result.report.totals["applied_change_count"] == 0
|
|
assert (result.run_dir / "report.json").exists()
|
|
assert (result.run_dir / "normalization" / "summary.json").exists()
|
|
assert [validator["name"] for validator in result.report.modules[0].to_dict()["validators"]] == [
|
|
"proposal_confidence_guard",
|
|
"protected_glossary_guard",
|
|
"spoken_form_plausibility_review",
|
|
"meaning_reversal_review",
|
|
]
|
|
|
|
|
|
def test_external_report_can_be_written(tmp_path):
|
|
config = AuditaConfig.from_sources(env={})
|
|
llm_client = FakeStructuredLLMClient(
|
|
[
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
{"corrections": []},
|
|
]
|
|
)
|
|
result = process_transcript_result(_transcript(), _glossary(), config, llm_client=llm_client)
|
|
report_path = tmp_path / "report.json"
|
|
write_report(report_path, result.report)
|
|
|
|
payload = json.loads(report_path.read_text(encoding="utf-8"))
|
|
assert payload["pipeline"][0] == "glossary_primary"
|
|
assert payload["totals"]["applied_change_count"] == 0
|
|
|
|
|
|
def test_default_module_specs_expose_final_validator_order():
|
|
specs = default_module_specs()
|
|
|
|
assert [validator.name for validator in specs[0].module.validators()] == [
|
|
"proposal_confidence_guard",
|
|
"protected_glossary_guard",
|
|
"spoken_form_plausibility_review",
|
|
"meaning_reversal_review",
|
|
]
|
|
assert [validator.name for validator in specs[1].module.validators()] == [
|
|
"proposal_confidence_guard",
|
|
"protected_glossary_guard",
|
|
"spoken_form_plausibility_review",
|
|
"meaning_reversal_review",
|
|
]
|
|
assert [validator.name for validator in specs[2].module.validators()] == [
|
|
"proposal_confidence_guard",
|
|
"protected_glossary_guard",
|
|
"spoken_form_plausibility_review",
|
|
"meaning_reversal_review",
|
|
]
|
|
assert [validator.name for validator in specs[3].module.validators()] == ["protected_glossary_guard"]
|
|
assert [validator.name for validator in specs[4].module.validators()] == ["protected_glossary_guard"]
|
|
|
|
|
|
def test_process_transcript_result_missing_api_key_writes_failed_report(tmp_path):
|
|
config = AuditaConfig.from_sources(
|
|
env={},
|
|
overrides=None,
|
|
)
|
|
config = AuditaConfig(
|
|
api_key=None,
|
|
model=config.model,
|
|
base_url=config.base_url,
|
|
max_retries=config.max_retries,
|
|
max_section_tokens=config.max_section_tokens,
|
|
glossary_confidence_threshold=config.glossary_confidence_threshold,
|
|
homophones_confidence_threshold=config.homophones_confidence_threshold,
|
|
normalize_max_segment_gap=config.normalize_max_segment_gap,
|
|
normalize_ellipsis_gap=config.normalize_ellipsis_gap,
|
|
normalize_max_segment_duration=config.normalize_max_segment_duration,
|
|
normalize_max_segment_tokens=config.normalize_max_segment_tokens,
|
|
work_dir=tmp_path / "work",
|
|
work_dir_retention="never",
|
|
)
|
|
|
|
with pytest.raises(AuditaLLMError, match="OPENROUTER_API_KEY"):
|
|
process_transcript_result(_transcript(), _glossary(), config)
|
|
|
|
run_dir = next((tmp_path / "work").iterdir())
|
|
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
|
|
|
|
assert report["status"] == "failed"
|
|
assert report["normalization"]["normalized_segment_count"] == 2
|
|
assert report["pipeline"] == [
|
|
"glossary_primary",
|
|
"homophones",
|
|
"glossary_secondary",
|
|
"spoken_word",
|
|
"grammar",
|
|
]
|
|
assert report["modules"] == []
|
|
assert report["applied_changes"] == []
|
|
assert report["skipped_corrections"] == []
|
|
assert report["work_dir_retained"] is True
|
|
assert report["work_dir"] == str(run_dir)
|
|
assert "OPENROUTER_API_KEY" in report["error"]
|
|
|
|
|
|
def test_process_transcript_result_preserves_partial_progress_when_later_module_fails(tmp_path):
|
|
transcript = parse_source_transcript_json(
|
|
"""
|
|
[
|
|
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "There were gestures at the dam."}
|
|
]
|
|
"""
|
|
)
|
|
config = AuditaConfig.from_sources(
|
|
env={},
|
|
overrides=None,
|
|
)
|
|
config = AuditaConfig(
|
|
api_key=config.api_key,
|
|
model=config.model,
|
|
base_url=config.base_url,
|
|
max_retries=config.max_retries,
|
|
max_section_tokens=config.max_section_tokens,
|
|
glossary_confidence_threshold=config.glossary_confidence_threshold,
|
|
homophones_confidence_threshold=config.homophones_confidence_threshold,
|
|
normalize_max_segment_gap=config.normalize_max_segment_gap,
|
|
normalize_ellipsis_gap=config.normalize_ellipsis_gap,
|
|
normalize_max_segment_duration=config.normalize_max_segment_duration,
|
|
normalize_max_segment_tokens=config.normalize_max_segment_tokens,
|
|
work_dir=tmp_path / "work",
|
|
work_dir_retention="never",
|
|
)
|
|
llm_client = FakeStructuredLLMClient(
|
|
[
|
|
{
|
|
"corrections": [
|
|
{
|
|
"id": 1,
|
|
"original_text": "gestures",
|
|
"corrected_text": "Jesters",
|
|
"confidence": 0.95,
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"validations": [
|
|
{
|
|
"correction_index": 0,
|
|
"approved": True,
|
|
"confidence": 0.97,
|
|
"reason": "Likely spoken-form correction in context.",
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"validations": [
|
|
{
|
|
"correction_index": 0,
|
|
"approved": True,
|
|
"confidence": 0.99,
|
|
"reason": "Does not reverse the segment meaning.",
|
|
}
|
|
]
|
|
},
|
|
AuditaLLMError("Simulated homophones proposal failure."),
|
|
]
|
|
)
|
|
|
|
with pytest.raises(AuditaLLMError, match="Simulated homophones proposal failure"):
|
|
process_transcript_result(transcript, _glossary(), config, llm_client=llm_client)
|
|
|
|
run_dir = next((tmp_path / "work").iterdir())
|
|
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
|
|
|
|
assert report["status"] == "failed"
|
|
assert report["normalization"]["normalized_segment_count"] == 1
|
|
assert [module["instance_name"] for module in report["modules"]] == ["glossary_primary"]
|
|
assert report["applied_changes"][0]["corrected_text"] == "Jesters"
|
|
assert report["applied_changes"][0]["segment_text_after"] == "There were Jesters at the dam."
|
|
assert report["totals"]["applied_change_count"] == 1
|
|
assert report["skipped_corrections"] == []
|
|
assert report["pipeline"][1] == "homophones"
|
|
assert "Simulated homophones proposal failure." in report["error"]
|
|
|
|
|
|
def test_process_transcript_result_preserves_partial_skips_and_validator_diagnostics_on_failure(tmp_path):
|
|
config = AuditaConfig.from_sources(
|
|
env={},
|
|
overrides=None,
|
|
)
|
|
config = AuditaConfig(
|
|
api_key=config.api_key,
|
|
model=config.model,
|
|
base_url=config.base_url,
|
|
max_retries=config.max_retries,
|
|
max_section_tokens=config.max_section_tokens,
|
|
glossary_confidence_threshold=0.8,
|
|
homophones_confidence_threshold=config.homophones_confidence_threshold,
|
|
normalize_max_segment_gap=config.normalize_max_segment_gap,
|
|
normalize_ellipsis_gap=config.normalize_ellipsis_gap,
|
|
normalize_max_segment_duration=config.normalize_max_segment_duration,
|
|
normalize_max_segment_tokens=config.normalize_max_segment_tokens,
|
|
work_dir=tmp_path / "work",
|
|
work_dir_retention="never",
|
|
)
|
|
llm_client = FakeStructuredLLMClient(
|
|
[
|
|
{
|
|
"corrections": [
|
|
{
|
|
"id": 1,
|
|
"original_text": "Hello",
|
|
"corrected_text": "Jesters",
|
|
"confidence": 0.40,
|
|
},
|
|
{
|
|
"id": 1,
|
|
"original_text": "Hello",
|
|
"corrected_text": "Jesters",
|
|
"confidence": 0.95,
|
|
},
|
|
]
|
|
},
|
|
{
|
|
"validations": [
|
|
{
|
|
"correction_index": 99,
|
|
"approved": True,
|
|
"confidence": 0.98,
|
|
"reason": "Malformed response for testing.",
|
|
}
|
|
]
|
|
},
|
|
]
|
|
)
|
|
|
|
with pytest.raises(AuditaLLMError, match="unknown correction_index"):
|
|
process_transcript_result(_transcript(), _glossary(), config, llm_client=llm_client)
|
|
|
|
run_dir = next((tmp_path / "work").iterdir())
|
|
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
|
|
validator_dir = run_dir / "glossary_primary"
|
|
|
|
assert report["status"] == "failed"
|
|
assert report["modules"] == []
|
|
assert len(report["skipped_corrections"]) == 1
|
|
assert report["skipped_corrections"][0]["reason"] == "proposal confidence below threshold"
|
|
assert report["skipped_corrections"][0]["source"] == "validator:proposal_confidence_guard"
|
|
assert "unknown correction_index" in report["error"]
|
|
assert (validator_dir / "spoken_form_plausibility_review-prompt-0000.json").exists()
|
|
assert (validator_dir / "spoken_form_plausibility_review-response-0000.json").exists()
|