test_brain_failures.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Brain failures stay speakable and never become benchmark answer timings.
Run with Python directly. No models, audio device, assistant CLI, or live inbox.
"""
from __future__ import annotations
import contextlib
import io
import json
import subprocess
import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch
import benchmark
import brain_bridge
import live_driver
import pipeline
class BrainFailureTests(unittest.TestCase):
def test_cli_errors_survive_sanitizing(self):
cases = (
(FileNotFoundError(), "cli_unavailable"),
(OSError(), "cli_unavailable"),
(subprocess.TimeoutExpired("unused", 1), "cli_timeout"),
(SimpleNamespace(returncode=1, stdout="", stderr="**failed**"), "cli_exit"),
)
for outcome, reason in cases:
with self.subTest(reason=reason):
kwargs = (
{"side_effect": outcome}
if isinstance(outcome, Exception)
else {"return_value": outcome}
)
with (
patch.object(
pipeline, "load_config", return_value=pipeline.DEFAULTS
),
patch.object(pipeline.subprocess, "run", **kwargs),
):
reply = pipeline.brain("hello")
self.assertIsInstance(reply, brain_bridge.BrainFailure)
self.assertEqual(reply.reason, reason)
self.assertTrue(reply.startswith("Sorry,"))
self.assertNotIn("**", reply)
def test_bridge_configuration_errors_have_reasons(self):
cases = (
({}, "bridge_reply_path"),
({"brain_reply_path": "/unused", "brain_transport": "ssh"}, "bridge_host"),
(
{"brain_reply_path": "/unused", "brain_transport": "sim"},
"bridge_inbox_path",
),
(
{
"brain_reply_path": "/unused",
"brain_transport": "sim",
"brain_inbox_path": [],
},
"bridge_inbox_path",
),
(
{"brain_reply_path": "/unused", "brain_transport": "invalid"},
"bridge_transport",
),
)
for cfg, reason in cases:
with self.subTest(reason=reason):
reply = brain_bridge.build_brain(cfg)("hello")
self.assertIsInstance(reply, brain_bridge.BrainFailure)
self.assertEqual(reply.reason, reason)
def test_bridge_transport_errors_have_reasons(self):
for send, extra, reason in (
(lambda *_a, **_k: None, {}, "bridge_timeout"),
(
lambda *_a, **_k: None,
{"confirm_landing": lambda _id: False, "landing_timeout_s": 0},
"bridge_not_landed",
),
):
with self.subTest(reason=reason):
reply = brain_bridge.brain_via_bridge(
"hello",
persona="test",
send=send,
read_reply=lambda: "",
timeout_s=0,
**extra,
)
self.assertIsInstance(reply, brain_bridge.BrainFailure)
self.assertEqual(reply.reason, reason)
reply = brain_bridge.brain_via_bridge(
"hello",
persona="test",
send=Mock(side_effect=OSError()),
read_reply=lambda: "",
timeout_s=0,
)
self.assertIsInstance(reply, brain_bridge.BrainFailure)
self.assertEqual(reply.reason, "bridge_send")
def test_pipeline_speaks_error_and_marks_failure(self):
with (
patch.object(pipeline, "load_config", return_value=pipeline.DEFAULTS),
patch.object(pipeline, "CLAUDE_BIN", "/this/assistant/does/not/exist"),
patch.object(
pipeline, "detect_wake", return_value=(True, "hey_jarvis", 0.9)
),
patch.object(
pipeline,
"transcribe_detailed",
return_value=pipeline.Transcript("hello", -0.2, 0.01),
),
patch.object(pipeline, "speak") as speak,
):
result = pipeline.run_pipeline("/unused.wav", "/unused-reply.wav")
self.assertEqual(result["rejected"], "brain_failure")
self.assertEqual(result["reject_reason"], "cli_unavailable")
speak.assert_called_once_with(result["reply"], "/unused-reply.wav")
self.assertEqual(
str(result["reply"]), "Sorry, the brain is not available right now."
)
def test_answer_with_error_words_is_still_an_answer(self):
text = "Sorry, the brain is not available right now."
with (
patch.object(pipeline, "load_config", return_value=pipeline.DEFAULTS),
patch.object(
pipeline.subprocess,
"run",
return_value=SimpleNamespace(returncode=0, stdout=text, stderr=""),
),
):
reply = pipeline.brain("quote the error message")
self.assertEqual(reply, text)
self.assertNotIsInstance(reply, brain_bridge.BrainFailure)
def test_live_turn_plays_error_and_keeps_listening(self):
cfg = dict(pipeline.DEFAULTS, wake_chime=False)
with (
patch.object(pipeline, "load_config", return_value=cfg),
patch.object(pipeline, "CLAUDE_BIN", "/this/assistant/does/not/exist"),
patch.object(live_driver, "listen_for_wake", return_value=0.9),
patch.object(
pipeline, "capture_request", return_value=SimpleNamespace(size=1)
),
patch.object(
pipeline,
"transcribe_detailed",
return_value=pipeline.Transcript("hello", -0.2, 0.01),
),
patch.object(pipeline, "speak") as speak,
patch.object(live_driver, "_play_wav") as play,
):
keep_listening = live_driver.run_turn(
iter(()), None, None, 0.5, "/unused-reply.wav", None, cfg, False
)
self.assertTrue(keep_listening)
self.assertIsInstance(speak.call_args.args[0], brain_bridge.BrainFailure)
self.assertEqual(speak.call_args.args[0].reason, "cli_unavailable")
play.assert_called_once_with("/unused-reply.wav", None)
def test_empty_reply_keeps_spoken_fallback_and_marks_failure(self):
for text in ("", " ", "\x00"):
with (
self.subTest(text=text),
patch.object(pipeline, "load_config", return_value=pipeline.DEFAULTS),
patch.object(
pipeline.subprocess,
"run",
return_value=SimpleNamespace(returncode=0, stdout=text, stderr=""),
),
):
reply = pipeline.brain("hello")
self.assertEqual(reply, pipeline.EMPTY_REPLY_FALLBACK)
self.assertIsInstance(reply, brain_bridge.BrainFailure)
self.assertEqual(reply.reason, "empty_reply")
def collect(self, turns):
with (
patch.object(pipeline, "load_config", return_value=pipeline.DEFAULTS),
patch.object(pipeline, "warm_models", return_value={}),
patch.object(benchmark, "ensure_clip", return_value=False),
patch.object(pipeline, "run_pipeline", side_effect=turns),
):
return benchmark.collect(len(turns), None, None)
def test_mixed_runs_keep_only_answer_timings(self):
collected = self.collect(
[
{"wake_fired": False, "wake_score": 0.2, "timings_s": {"total": 0.01}},
{
"wake_fired": True,
"rejected": "brain_failure",
"reject_reason": "cli_unavailable",
"timings_s": {"brain": 0.01, "total": 0.02},
},
{"wake_fired": True, "timings_s": {"brain": 3.0, "total": 4.0}},
{
"wake_fired": True,
"rejected": "brain_failure",
"reject_reason": "bridge_timeout",
"timings_s": {"brain": 120.0, "total": 121.0},
},
]
)
self.assertEqual(collected["runs_measured"], 1)
self.assertEqual(collected["wake_misses"], 1)
self.assertEqual(
collected["brain_failures"], {"cli_unavailable": 1, "bridge_timeout": 1}
)
self.assertEqual(collected["per_stage"], {"brain": [3.0], "total": [4.0]})
report = "\n".join(benchmark.report_lines(collected, None))
self.assertIn("2 run(s) failed in the brain", report)
self.assertIn("cli_unavailable=1", report)
with (
patch.object(benchmark, "collect", return_value=collected),
contextlib.redirect_stdout(io.StringIO()),
):
self.assertEqual(benchmark.main(["--runs", "4", "--no-ssh"]), 1)
def test_all_brain_failures_report_no_answers_and_exit_nonzero(self):
collected = self.collect(
[
{
"wake_fired": True,
"rejected": "brain_failure",
"reject_reason": "cli_unavailable",
"timings_s": {"total": 0.02},
}
]
* 2
)
self.assertEqual(collected["runs_measured"], 0)
self.assertEqual(collected["per_stage"], {})
report = "\n".join(benchmark.report_lines(collected, None))
self.assertIn("No stage timings: no run produced a brain answer.", report)
self.assertNotIn("none of the 2 run(s) fired", report)
self.assertNotIn("| Stage", report)
for json_mode in (False, True):
out, err = io.StringIO(), io.StringIO()
with (
patch.object(benchmark, "collect", return_value=collected),
contextlib.redirect_stdout(out),
contextlib.redirect_stderr(err),
):
code = benchmark.main(
["--runs", "2", "--no-ssh"] + (["--json"] if json_mode else [])
)
self.assertEqual(code, 1)
self.assertIn("the brain failed", err.getvalue())
self.assertNotIn("never fired", err.getvalue())
if json_mode:
data = json.loads(out.getvalue())
self.assertEqual(data["brain_failures"], {"cli_unavailable": 2})
self.assertEqual(data["runs_measured"], 0)
if __name__ == "__main__":
unittest.main()