Files
audita/tests/test_new_pipeline.py

115 lines
3.9 KiB
Python

import json
from audita.core.config import AuditaConfig
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
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):
revised = process_transcript(
_transcript(),
_glossary(),
AuditaConfig.from_sources(env={}, overrides=None),
)
assert [segment.id for segment in revised] == [1, 2]
assert revised[0].text == "Hello. Again."
assert revised[1].text == "Done."
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,
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",
)
result = process_transcript_result(_transcript(), _glossary(), config)
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"]] == [
"protected_glossary_guard",
"spoken_form_plausibility_review",
"meaning_reversal_review",
]
def test_external_report_can_be_written(tmp_path):
config = AuditaConfig.from_sources(env={})
result = process_transcript_result(_transcript(), _glossary(), config)
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()] == [
"protected_glossary_guard",
"spoken_form_plausibility_review",
"meaning_reversal_review",
]
assert [validator.name for validator in specs[1].module.validators()] == [
"protected_glossary_guard",
"spoken_form_plausibility_review",
"meaning_reversal_review",
]
assert [validator.name for validator in specs[2].module.validators()] == [
"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"]