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()