eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Improvements around reporting and work directory retention options

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-23 12:08:44 -05:00
parent c4e2db75f1
commit 3277587e3a
11 changed files with 620 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
tests/test_cli.py
View File

@@ -1,6 +1,8 @@
import pytest
from audita.cli import main
from audita.reporting import ProcessResult, RunReport
from audita.schemas import parse_transcript_json
def test_cli_help_uses_audita_program_name(capsys):
@@ -17,6 +19,7 @@ def test_process_help_includes_glossary_pass_flag(capsys):
assert exc.value.code == 0
output = capsys.readouterr().out
assert "--report-json" in output
assert "--glossary-max-llm-passes" in output
assert "--grammar-max-llm-passes" in output
assert "--glossary-confidence-threshold" in output
@@ -24,8 +27,61 @@ def test_process_help_includes_glossary_pass_flag(capsys):
assert "--grammar-validation-enabled" in output
assert "--grammar-validation-confidence-threshold" in output
assert "--grammar-spoken-form-validation-confidence-threshold" in output
assert "--work-dir-retention" in output
assert "--normalize-max-segment-gap" in output
assert "--normalize-ellipsis-gap" in output
assert "--normalize-max-segment-duration" in output
assert "--normalize-max-segment-tokens" in output
assert "--confidence-threshold" not in output
def test_cli_process_writes_report_json(monkeypatch, tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Fixed."}
]
"""
)
report = RunReport(
status="success",
config={"model": "m", "base_url": "b"},
normalization={"source_segment_count": 1, "normalized_segment_count": 1, "merge_count": 0},
stages=[],
applied_changes=[],
skipped_corrections=[],
totals={"output_segment_count": 1, "applied_change_count": 0, "skipped_correction_count": 0},
work_dir_retention="auto",
work_dir_retained=False,
work_dir=None,
error=None,
)
result = ProcessResult(
transcript=transcript,
report=report,
run_dir=tmp_path / "run",
work_dir_retained=False,
)
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
monkeypatch.setattr("audita.cli.load_transcript", lambda path: [])
monkeypatch.setattr("audita.cli.load_glossary", lambda path: object())
monkeypatch.setattr("audita.cli.process_transcript_result", lambda *args, **kwargs: result)
output_path = tmp_path / "out.json"
report_path = tmp_path / "report.json"
exit_code = main(
[
"process",
"transcript.json",
"--glossary",
"glossary.yaml",
"--output",
str(output_path),
"--report-json",
str(report_path),
]
)
assert exit_code == 0
assert report_path.exists()

15
tests/test_config.py
View File

@@ -18,6 +18,7 @@ from audita.config import (
DEFAULT_NORMALIZE_MAX_SEGMENT_GAP,
DEFAULT_NORMALIZE_MAX_SEGMENT_TOKENS,
DEFAULT_WORK_DIR,
DEFAULT_WORK_DIR_RETENTION,
)
from audita.errors import AuditaConfigError
@@ -50,6 +51,8 @@ def test_config_uses_defaults_with_api_key():
assert config.normalize_max_segment_tokens == DEFAULT_NORMALIZE_MAX_SEGMENT_TOKENS
assert config.normalize_max_segment_tokens == 2048
assert config.work_dir == Path(DEFAULT_WORK_DIR)
assert config.work_dir_retention == DEFAULT_WORK_DIR_RETENTION
assert config.work_dir_retention == "auto"
def test_config_env_overrides_defaults():
@@ -70,6 +73,7 @@ def test_config_env_overrides_defaults():
"AUDITA_NORMALIZE_MAX_SEGMENT_DURATION": "45.0",
"AUDITA_NORMALIZE_MAX_SEGMENT_TOKENS": "512",
"AUDITA_WORK_DIR": "/tmp/custom-audita",
"AUDITA_WORK_DIR_RETENTION": "always",
}
)
@@ -87,6 +91,7 @@ def test_config_env_overrides_defaults():
assert config.normalize_max_segment_duration == 45.0
assert config.normalize_max_segment_tokens == 512
assert config.work_dir == Path("/tmp/custom-audita")
assert config.work_dir_retention == "always"
def test_config_cli_overrides_env():
@@ -105,6 +110,7 @@ def test_config_cli_overrides_env():
"AUDITA_NORMALIZE_MAX_SEGMENT_DURATION": "45.0",
"AUDITA_NORMALIZE_MAX_SEGMENT_TOKENS": "512",
"AUDITA_WORK_DIR": "/tmp/env-audita",
"AUDITA_WORK_DIR_RETENTION": "always",
},
overrides=ConfigOverrides(
max_section_tokens=100,
@@ -121,6 +127,7 @@ def test_config_cli_overrides_env():
normalize_max_segment_duration=30.0,
normalize_max_segment_tokens=256,
work_dir=Path("/tmp/cli-audita"),
work_dir_retention="never",
),
)
@@ -138,6 +145,7 @@ def test_config_cli_overrides_env():
assert config.normalize_max_segment_duration == 30.0
assert config.normalize_max_segment_tokens == 256
assert config.work_dir == Path("/tmp/cli-audita")
assert config.work_dir_retention == "never"
def test_config_requires_api_key():
@@ -195,6 +203,13 @@ def test_config_rejects_invalid_grammar_validation_enabled():
)
def test_config_rejects_invalid_work_dir_retention():
with pytest.raises(AuditaConfigError):
AuditaConfig.from_sources(
env={"OPENROUTER_API_KEY": "key", "AUDITA_WORK_DIR_RETENTION": "sometimes"}
)
def test_legacy_confidence_threshold_env_is_ignored():
config = AuditaConfig.from_sources(
env={"OPENROUTER_API_KEY": "key", "AUDITA_CONFIDENCE_THRESHOLD": "0.9"}

4
tests/test_corrections.py
View File

@@ -33,6 +33,9 @@ def test_apply_corrections_uses_threshold_and_preserves_id_order():
assert [segment.speaker for segment in result.transcript] == ["Eric", "Mike"]
assert result.transcript[0].text == "I ask Chauntea for help."
assert result.skipped == []
assert len(result.applied_corrections) == 1
assert result.applied_corrections[0].segment_text_before == "I ask Chontia for help."
assert result.applied_corrections[0].segment_text_after == "I ask Chauntea for help."
def test_apply_corrections_ignores_below_threshold():
@@ -75,6 +78,7 @@ def test_apply_corrections_allows_multiple_distinct_spans_in_one_segment():
assert result.transcript[0].text == "I ask Chauntea for guidance."
assert result.skipped == []
assert len(result.applied_corrections) == 2
def test_apply_corrections_skips_missing_substring():

170
tests/test_pipeline.py
View File

@@ -4,7 +4,8 @@ import pytest
from audita.config import AuditaConfig
from audita.errors import AuditaError
from audita.pipeline import process_transcript
from audita.io import write_report
from audita.pipeline import process_transcript, process_transcript_result
from audita.schemas import (
CorrectionCandidate,
GrammarSpokenFormValidationDecision,
@@ -56,6 +57,7 @@ def _config(
grammar_validation_enabled=False,
grammar_validation_confidence_threshold=0.8,
grammar_spoken_form_validation_confidence_threshold=0.8,
work_dir_retention="auto",
):
return AuditaConfig(
api_key="key",
@@ -69,6 +71,7 @@ def _config(
grammar_validation_confidence_threshold=grammar_validation_confidence_threshold,
grammar_spoken_form_validation_confidence_threshold=grammar_spoken_form_validation_confidence_threshold,
work_dir=tmp_path / "work",
work_dir_retention=work_dir_retention,
)
@@ -713,6 +716,8 @@ def test_pipeline_writes_stage_metadata_for_unresolved_retries(tmp_path):
assert metadata["grammar_validation_enabled"] is False
assert metadata["grammar_validation_confidence_threshold"] == 0.8
assert metadata["grammar_spoken_form_validation_confidence_threshold"] == 0.8
assert metadata["work_dir_retention"] == "auto"
assert metadata["work_dir_retained"] is True
assert [item["stage"] for item in metadata["stages"]] == ["glossary", "grammar"]
assert [item["pass_number"] for item in metadata["stages"][0]["passes"]] == [1, 2]
assert metadata["stages"][0]["passes"][0]["retry_segment_count"] == 1
@@ -1387,3 +1392,166 @@ def test_unresolved_grammar_skip_preserves_diagnostics(tmp_path):
diagnostics = json.loads((run_dirs[0] / "skipped-corrections.json").read_text(encoding="utf-8"))
assert diagnostics["skipped_corrections"][0]["stage"] == "grammar"
assert "more than once" in diagnostics["skipped_corrections"][0]["reason"]
def test_process_transcript_result_returns_report_with_applied_changes(tmp_path):
correction = CorrectionCandidate(
id=1,
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[correction]),
CorrectionSet(corrections=[]),
]
)
result = process_transcript_result(
_transcript(),
_glossary(),
_config(tmp_path),
llm_client=fake_client,
)
assert result.transcript[0].text == "I ask Chauntea."
assert result.work_dir_retained is False
assert result.report.status == "success"
assert result.report.work_dir_retained is False
assert result.report.work_dir is None
assert result.report.totals["applied_change_count"] == 1
assert result.report.skipped_corrections == []
assert result.report.applied_changes[0].stage == "glossary"
assert result.report.applied_changes[0].pass_number == 1
assert result.report.applied_changes[0].segment_text_before == "I ask Chontia."
assert result.report.applied_changes[0].segment_text_after == "I ask Chauntea."
report_path = tmp_path / "result-report.json"
write_report(report_path, result.report)
written = json.loads(report_path.read_text(encoding="utf-8"))
assert written["totals"]["applied_change_count"] == 1
assert written["applied_changes"][0]["stage"] == "glossary"
def test_auto_retains_work_dir_on_success_with_skipped_corrections(tmp_path):
correction = CorrectionCandidate(
id=1,
original_text="Different text.",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[correction]),
CorrectionSet(corrections=[]),
]
)
result = process_transcript_result(
_transcript(),
_glossary(),
_config(tmp_path, glossary_max_llm_passes=1, work_dir_retention="auto"),
llm_client=fake_client,
)
assert result.work_dir_retained is True
assert result.run_dir.exists()
assert (result.run_dir / "report.json").exists()
report_json = json.loads((result.run_dir / "report.json").read_text(encoding="utf-8"))
assert report_json["work_dir_retained"] is True
assert report_json["skipped_corrections"][0]["stage"] == "glossary"
def test_never_retention_removes_work_dir_after_success_even_with_skipped_corrections(tmp_path):
correction = CorrectionCandidate(
id=1,
original_text="Different text.",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[correction]),
CorrectionSet(corrections=[]),
]
)
result = process_transcript_result(
_transcript(),
_glossary(),
_config(tmp_path, glossary_max_llm_passes=1, work_dir_retention="never"),
llm_client=fake_client,
)
assert result.work_dir_retained is False
assert not result.run_dir.exists()
assert result.report.work_dir_retained is False
assert result.report.skipped_corrections[0].stage == "glossary"
def test_always_preserves_work_dir_after_clean_success(tmp_path):
correction = CorrectionCandidate(
id=1,
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[correction]),
CorrectionSet(corrections=[]),
]
)
result = process_transcript_result(
_transcript(),
_glossary(),
_config(tmp_path, work_dir_retention="always"),
llm_client=fake_client,
)
assert result.work_dir_retained is True
assert result.run_dir.exists()
assert (result.run_dir / "report.json").exists()
report_json = json.loads((result.run_dir / "report.json").read_text(encoding="utf-8"))
assert report_json["work_dir_retention"] == "always"
assert report_json["skipped_corrections"] == []
def test_failure_preserves_work_dir_and_writes_failure_report(tmp_path):
transcript = parse_source_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "He became visible."}
]
"""
)
grammar_correction = CorrectionCandidate(
id=1,
original_text="visible",
corrected_text="invisible",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[]),
CorrectionSet(corrections=[grammar_correction]),
],
validation_responses=[GrammarValidationSet(validations=[])],
)
with pytest.raises(AuditaError):
process_transcript_result(
transcript,
_glossary(),
_config(tmp_path, grammar_validation_enabled=True),
llm_client=fake_client,
)
run_dirs = list((tmp_path / "work").iterdir())
assert len(run_dirs) == 1
report_json = json.loads((run_dirs[0] / "report.json").read_text(encoding="utf-8"))
assert report_json["status"] == "failed"
assert report_json["work_dir_retained"] is True
assert report_json["error"] is not None