Move python implementation under python/ in preparation for the upcoming Go rewrite
This commit is contained in:
659
python/tests/test_new_cli.py
Normal file
659
python/tests/test_new_cli.py
Normal file
@@ -0,0 +1,659 @@
|
||||
import json
|
||||
import io
|
||||
import pytest
|
||||
|
||||
from audita.cli import main
|
||||
from audita.core.errors import AuditaConfigError
|
||||
from audita.core.reporting import ProcessResult, RunReport
|
||||
from audita.core.schemas import parse_transcript_json
|
||||
|
||||
|
||||
def test_cli_help_uses_audita_program_name(capsys):
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["--help"])
|
||||
|
||||
assert exc.value.code == 0
|
||||
assert capsys.readouterr().out.startswith("usage: audita ")
|
||||
|
||||
|
||||
def test_process_help_exposes_framework_flags(capsys):
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["process", "--help"])
|
||||
|
||||
assert exc.value.code == 0
|
||||
output = capsys.readouterr().out
|
||||
assert "--report-json" in output
|
||||
assert "--llm-api-key" in output
|
||||
assert "--llm-concurrency" in output
|
||||
assert "--llm-timeout-seconds" in output
|
||||
assert "--validation-llm-api-key" in output
|
||||
assert "--validation-llm-concurrency" in output
|
||||
assert "--validation-llm-timeout-seconds" in output
|
||||
assert "--validation-model" in output
|
||||
assert "--validation-base-url" in output
|
||||
assert "--validation-max-retries" in output
|
||||
assert "--validation-max-prompt-tokens" in output
|
||||
assert "--target-sections" in output
|
||||
assert "--modules" in output
|
||||
assert "--model" in output
|
||||
assert "--base-url" in output
|
||||
assert "--max-retries" in output
|
||||
assert "--max-section-tokens" in output
|
||||
assert "--min-section-tokens" in output
|
||||
assert "--glossary-confidence-threshold" in output
|
||||
assert "--grammar-confidence-threshold" in output
|
||||
assert "--homophones-confidence-threshold" in output
|
||||
assert "--spoken-word-confidence-threshold" in output
|
||||
assert "--work-dir-retention" in output
|
||||
assert "--normalize-max-segment-gap" in output
|
||||
assert "--grammar-validation-enabled" 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},
|
||||
pipeline=["glossary_1", "homophones", "glossary_2", "spoken_word", "grammar"],
|
||||
modules=[],
|
||||
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()
|
||||
|
||||
|
||||
def test_cli_process_passes_modules_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["module_keys"] = overrides.module_keys
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--modules",
|
||||
"grammar",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["module_keys"] == "grammar"
|
||||
|
||||
|
||||
def test_cli_process_passes_llm_api_key_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["llm_api_key"] = overrides.llm_api_key
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--llm-api-key",
|
||||
"cli-key",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["llm_api_key"] == "cli-key"
|
||||
|
||||
|
||||
def test_cli_process_passes_llm_concurrency_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["llm_concurrency"] = overrides.llm_concurrency
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--llm-concurrency",
|
||||
"3",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["llm_concurrency"] == 3
|
||||
|
||||
|
||||
def test_cli_process_passes_llm_timeout_seconds_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["llm_timeout_seconds"] = overrides.llm_timeout_seconds
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--llm-timeout-seconds",
|
||||
"900",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["llm_timeout_seconds"] == 900.0
|
||||
|
||||
|
||||
def test_cli_process_passes_validation_llm_overrides_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["validation_llm_api_key"] = overrides.validation_llm_api_key
|
||||
captured["validation_llm_concurrency"] = overrides.validation_llm_concurrency
|
||||
captured["validation_llm_timeout_seconds"] = overrides.validation_llm_timeout_seconds
|
||||
captured["validation_model"] = overrides.validation_model
|
||||
captured["validation_base_url"] = overrides.validation_base_url
|
||||
captured["validation_max_retries"] = overrides.validation_max_retries
|
||||
captured["validation_max_prompt_tokens"] = overrides.validation_max_prompt_tokens
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--validation-llm-api-key",
|
||||
"validator-key",
|
||||
"--validation-llm-concurrency",
|
||||
"4",
|
||||
"--validation-llm-timeout-seconds",
|
||||
"180",
|
||||
"--validation-model",
|
||||
"validator-model",
|
||||
"--validation-base-url",
|
||||
"http://localhost:9000/v1",
|
||||
"--validation-max-retries",
|
||||
"2",
|
||||
"--validation-max-prompt-tokens",
|
||||
"1024",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured == {
|
||||
"validation_llm_api_key": "validator-key",
|
||||
"validation_llm_concurrency": 4,
|
||||
"validation_llm_timeout_seconds": 180.0,
|
||||
"validation_model": "validator-model",
|
||||
"validation_base_url": "http://localhost:9000/v1",
|
||||
"validation_max_retries": 2,
|
||||
"validation_max_prompt_tokens": 1024,
|
||||
}
|
||||
|
||||
|
||||
def test_cli_process_passes_target_sections_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["target_sections"] = overrides.target_sections
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--target-sections",
|
||||
"4",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["target_sections"] == 4
|
||||
|
||||
|
||||
def test_cli_process_passes_min_section_tokens_override_to_config(monkeypatch, tmp_path):
|
||||
captured = {}
|
||||
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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_from_sources(*, overrides=None):
|
||||
captured["min_section_tokens"] = overrides.min_section_tokens
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", _fake_from_sources)
|
||||
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)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--min-section-tokens",
|
||||
"5000",
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert captured["min_section_tokens"] == 5000
|
||||
|
||||
|
||||
def test_cli_process_writes_failure_diagnostics_and_keeps_stdout_empty(monkeypatch, tmp_path, capsys):
|
||||
monkeypatch.setattr(
|
||||
"audita.cli.AuditaConfig.from_sources",
|
||||
lambda overrides=None: (_ for _ in ()).throw(AuditaConfigError("bad config")),
|
||||
)
|
||||
|
||||
report_path = tmp_path / "external-report.json"
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--work-dir",
|
||||
str(tmp_path / "work"),
|
||||
"--report-json",
|
||||
str(report_path),
|
||||
]
|
||||
)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert exit_code == 1
|
||||
assert captured.out == ""
|
||||
assert "audita: error: bad config" in captured.err
|
||||
assert "audita: exit code: 1" in captured.err
|
||||
run_dir = next((tmp_path / "work").iterdir())
|
||||
assert f"audita: run directory: {run_dir}" in captured.err
|
||||
assert f"audita: error log: {run_dir / 'error.log'}" in captured.err
|
||||
assert f"audita: report: {run_dir / 'report.json'}" in captured.err
|
||||
assert (run_dir / "error.log").exists()
|
||||
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
|
||||
assert report["status"] == "failed"
|
||||
assert report["error"] == "bad config"
|
||||
assert report["error_details"]["phase"] == "config"
|
||||
assert report["error_details"]["type"] == "AuditaConfigError"
|
||||
external = json.loads(report_path.read_text(encoding="utf-8"))
|
||||
assert external["error"] == "bad config"
|
||||
|
||||
|
||||
def test_cli_process_writes_failure_diagnostics_for_unexpected_exceptions(monkeypatch, tmp_path, capsys):
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
|
||||
monkeypatch.setattr("audita.cli.load_transcript", lambda path: (_ for _ in ()).throw(RuntimeError("boom")))
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--work-dir",
|
||||
str(tmp_path / "work"),
|
||||
]
|
||||
)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert exit_code == 1
|
||||
assert captured.out == ""
|
||||
assert "audita: error: boom" in captured.err
|
||||
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["error_details"]["phase"] == "transcript_load"
|
||||
assert report["error_details"]["type"] == "RuntimeError"
|
||||
assert "RuntimeError: boom" in (run_dir / "error.log").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
class _BrokenWriter:
|
||||
def write(self, _message):
|
||||
raise OSError(9, "Bad file descriptor")
|
||||
|
||||
def flush(self):
|
||||
raise OSError(9, "Bad file descriptor")
|
||||
|
||||
|
||||
def test_cli_process_progress_falls_back_to_stdout_when_stderr_is_invalid(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},
|
||||
pipeline=["grammar"],
|
||||
modules=[],
|
||||
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,
|
||||
)
|
||||
|
||||
def _fake_pipeline(*args, **kwargs):
|
||||
kwargs["progress"]("progress-line")
|
||||
return result
|
||||
|
||||
stdout_capture = io.StringIO()
|
||||
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", _fake_pipeline)
|
||||
monkeypatch.setattr("audita.cli.sys.stderr", _BrokenWriter())
|
||||
monkeypatch.setattr("audita.cli.sys.stdout", stdout_capture)
|
||||
|
||||
output_path = tmp_path / "out.json"
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--output",
|
||||
str(output_path),
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 0
|
||||
assert "progress-line" in stdout_capture.getvalue()
|
||||
|
||||
|
||||
def test_cli_process_failure_summary_falls_back_to_stdout_when_stderr_is_invalid(monkeypatch, tmp_path):
|
||||
stdout_capture = io.StringIO()
|
||||
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
|
||||
monkeypatch.setattr("audita.cli.load_transcript", lambda path: (_ for _ in ()).throw(RuntimeError("boom")))
|
||||
monkeypatch.setattr("audita.cli.sys.stderr", _BrokenWriter())
|
||||
monkeypatch.setattr("audita.cli.sys.stdout", stdout_capture)
|
||||
|
||||
exit_code = main(
|
||||
[
|
||||
"process",
|
||||
"transcript.json",
|
||||
"--glossary",
|
||||
"glossary.yaml",
|
||||
"--work-dir",
|
||||
str(tmp_path / "work"),
|
||||
]
|
||||
)
|
||||
|
||||
assert exit_code == 1
|
||||
assert "audita: error: boom" in stdout_capture.getvalue()
|
||||
Reference in New Issue
Block a user