Update the CLI to use a safe, best-effort console emitter so logging issues cannot fail the run
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import json
|
||||
import io
|
||||
import pytest
|
||||
|
||||
from audita.cli import main
|
||||
@@ -569,3 +570,90 @@ def test_cli_process_writes_failure_diagnostics_for_unexpected_exceptions(monkey
|
||||
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()
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
@@ -11,6 +14,15 @@ ROOT = Path(__file__).resolve().parents[1]
|
||||
LAUNCHER = ROOT / "audita"
|
||||
|
||||
|
||||
def _load_launcher_module():
|
||||
loader = importlib.machinery.SourceFileLoader("audita_launcher", str(LAUNCHER))
|
||||
spec = importlib.util.spec_from_file_location("audita_launcher", LAUNCHER, loader=loader)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_root_launcher_exists_and_is_executable():
|
||||
assert LAUNCHER.is_file()
|
||||
assert os.access(LAUNCHER, os.X_OK)
|
||||
@@ -73,3 +85,33 @@ def test_root_launcher_preserves_child_exit_code_and_writes_fallback_error_log(t
|
||||
assert "audita: subprocess exited with status 120" in result.stderr
|
||||
run_dir = next(work_dir.iterdir())
|
||||
assert (run_dir / "error.log").exists()
|
||||
|
||||
|
||||
class _BrokenWriter:
|
||||
def write(self, _message):
|
||||
raise OSError(9, "Bad file descriptor")
|
||||
|
||||
def flush(self):
|
||||
raise OSError(9, "Bad file descriptor")
|
||||
|
||||
|
||||
def test_root_launcher_falls_back_to_stdout_when_stderr_is_invalid(monkeypatch, tmp_path):
|
||||
launcher = _load_launcher_module()
|
||||
work_dir = tmp_path / "work"
|
||||
stdout_capture = io.StringIO()
|
||||
|
||||
monkeypatch.setattr(launcher.shutil, "which", lambda _name: None)
|
||||
monkeypatch.setattr(
|
||||
launcher.sys,
|
||||
"argv",
|
||||
["audita", "process", "transcript.json", "--glossary", "glossary.yaml", "--work-dir", str(work_dir)],
|
||||
)
|
||||
monkeypatch.setattr(launcher.sys, "stderr", _BrokenWriter())
|
||||
monkeypatch.setattr(launcher.sys, "stdout", stdout_capture)
|
||||
|
||||
exit_code = launcher.main()
|
||||
|
||||
assert exit_code == 1
|
||||
assert "audita: error: uv is required to run this launcher" in stdout_capture.getvalue()
|
||||
run_dir = next(work_dir.iterdir())
|
||||
assert (run_dir / "error.log").exists()
|
||||
|
||||
Reference in New Issue
Block a user