test_live_driver.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Fast, no-model tests for live_driver's turn loop, focused on the mishear guard.
live_driver.py is the real-hardware always-on path (arecord -> wake -> STT -> brain
-> speak). It must apply the same confidence guard as pipeline.run_turn, so a garbled
command never reaches the action-capable brain on the hardware path. These checks
stub the wake, capture, transcription, brain, and playback, so they run with no mic,
no model, and no audio device -- like test_brain_bridge.py.
Run: .venv/bin/python test_live_driver.py
Exit code is 0 only if every check passes.
"""
from __future__ import annotations
import os
import signal
import subprocess
import sys
import numpy as np
import live_driver
import pipeline
from pipeline import FRAME_SIZE, Transcript
PASS, FAIL = "PASS", "FAIL"
results: list[tuple[str, str]] = []
def check(name: str, ok: bool, detail: str) -> bool:
results.append((PASS if ok else FAIL, name))
print(f" [{PASS if ok else FAIL}] {name}: {detail}")
return ok
class _FakeMic:
"""A mic stand-in for turns that do not reach the post-cue flush (wake_chime off):
run_turn takes the StdinMic only to call flush(), so a counter suffices here."""
def __init__(self) -> None:
self.flushed = 0
def flush(self) -> None:
self.flushed += 1
GUARD_ON = {
"wake_word": "computah",
"stt_confidence_guard": True,
"stt_min_avg_logprob": -1.0,
"stt_max_no_speech_prob": 0.6,
"capture_vad_threshold": 0.5,
"endpoint_silence_ms": 800,
"max_request_ms": 8000,
}
GUARD_OFF = dict(GUARD_ON, stt_confidence_guard=False)
def drive(
cfg: dict,
transcript: Transcript,
output_device=None,
play_error: Exception | None = None,
empty_capture: bool = False,
) -> dict:
"""Run one live_driver turn with wake/capture/transcribe/brain/playback stubbed.
Returns what was spoken, whether the brain was called and with what, the
(path, device) args each reply playback was called with, and the run_turn
return value -- with no microphone, model, or audio device. Pass play_error to
simulate a dead output device (a raising _play_wav) and check the turn degrades
instead of crashing.
"""
state: dict = {
"spoken": None,
"brain_called": False,
"brain_arg": None,
"transcribe_arg": None,
"played": [],
"events": [],
}
real = (
live_driver.listen_for_wake,
pipeline.capture_request,
pipeline.transcribe_detailed,
pipeline.brain,
pipeline.speak,
live_driver._play_wav,
)
def fake_wake(fr, model, threshold, debug, preroll=None):
if preroll is not None:
preroll.extend(np.full(FRAME_SIZE, i, dtype=np.int16) for i in range(30))
return 0.9
def fake_capture(fr, **_):
if empty_capture:
return pipeline._empty_capture(
pipeline._EMPTY_VAD_REJECTED
if empty_capture == "vad_rejected"
else pipeline._EMPTY_NO_ONSET
)
return np.full(8 * FRAME_SIZE, 4000, dtype=np.int16)
live_driver.listen_for_wake = fake_wake
pipeline.capture_request = fake_capture
def fake_transcribe(audio):
state["transcribe_arg"] = audio
return transcript
pipeline.transcribe_detailed = fake_transcribe
def fake_brain(text, **_):
state["brain_called"] = True
state["brain_arg"] = text
return "Brain answer."
def fake_speak(text, out, **_):
state["events"].append("speak")
state["spoken"] = text
return out
def fake_play(path, dev):
# Record playback in call order relative to speak(): a stale-WAV reorder
# (play before speak) shows up as ["play", "speak"], not ["speak", "play"].
state["events"].append("play")
state["played"].append((path, dev))
if play_error is not None:
raise play_error
pipeline.brain = fake_brain
pipeline.speak = fake_speak
live_driver._play_wav = fake_play
try:
state["ran"] = live_driver.run_turn(
iter([]), _FakeMic(), object(), 0.5, "/dev/null", output_device, cfg, False
)
finally:
(
live_driver.listen_for_wake,
pipeline.capture_request,
pipeline.transcribe_detailed,
pipeline.brain,
pipeline.speak,
live_driver._play_wav,
) = real
return state
def _frame_bytes(val: int) -> bytes:
"""One 80 ms frame's worth of raw S16_LE bytes, every sample == val."""
return np.full(FRAME_SIZE, val, dtype=np.int16).tobytes()
def test_stdin_mic() -> None:
"""StdinMic.flush() is flush-to-now: it drops everything buffered up to the
instant capture starts (the cue bleed), with no wall-time guess and without
blocking on or consuming fresh post-cue frames (issue #56)."""
print("\n=== StdinMic: flush-to-now drains the pipe without a wall-time guess ===")
# frames() reconstructs full 80 ms frames from the raw pipe, in order, writable.
r, w = os.pipe()
try:
gen = live_driver.StdinMic(r).frames()
os.write(w, _frame_bytes(111) + _frame_bytes(222))
f1, f2 = next(gen), next(gen)
check(
"frames() yields written frames in order at full width",
f1.shape == (FRAME_SIZE,) and int(f1[0]) == 111 and int(f2[0]) == 222,
f"f1[0]={int(f1[0])} f2[0]={int(f2[0])} shape={f1.shape}",
)
check(
"yielded frames are writable (the wake model needs a writable array)",
f1.flags.writeable,
f"writeable={f1.flags.writeable}",
)
finally:
os.close(r)
os.close(w)
# The core fix: stale cue-bleed frames buffered before the flush never reach
# capture. The next frame after flush() is post-flush audio only.
r, w = os.pipe()
try:
mic = live_driver.StdinMic(r)
gen = mic.frames()
os.write(w, _frame_bytes(1000)) # one frame consumed during detection
next(gen)
os.write(w, _frame_bytes(1000) + _frame_bytes(1000)) # cue bleed now buffered
mic.flush() # flush-to-now drops the backlog
os.write(w, _frame_bytes(5000)) # the command, spoken after the cue
nxt = next(gen)
check(
"flush-to-now drops the buffered backlog (next frame is post-flush only)",
int(nxt[0]) == 5000,
f"want 5000 (fresh); got {int(nxt[0])} (1000 == stale bleed leaked through)",
)
finally:
os.close(r)
os.close(w)
# flush() never blocks: on an empty pipe it returns at once (the whole point vs the
# old wall-time drain). If it blocked, this turn would hang.
r, w = os.pipe()
try:
mic = live_driver.StdinMic(r)
mic.flush() # empty pipe
os.write(w, _frame_bytes(7))
nxt = next(mic.frames())
check(
"flush-to-now returns at once on an empty pipe and keeps later audio",
int(nxt[0]) == 7,
f"got {int(nxt[0])}",
)
finally:
os.close(r)
os.close(w)
# flush() also drops the sub-frame remainder and restores blocking mode, so the
# next frames() read blocks normally instead of busy-spinning on EAGAIN.
r, w = os.pipe()
try:
mic = live_driver.StdinMic(r)
mic._buf = b"\x01\x02\x03\x04" # a partial-frame remainder
mic.flush()
check(
"flush() clears the sub-frame remainder and restores blocking mode",
mic._buf == b"" and os.get_blocking(r) is True,
f"buf={mic._buf!r} blocking={os.get_blocking(r)}",
)
finally:
os.close(r)
os.close(w)
# frames() ends when arecord exits (the write end closes).
r, w = os.pipe()
try:
mic = live_driver.StdinMic(r)
os.write(w, _frame_bytes(9))
os.close(w) # arecord exits
got = list(mic.frames())
check(
"frames() ends at EOF (the writer closed)",
len(got) == 1 and int(got[0][0]) == 9,
f"n={len(got)}",
)
finally:
os.close(r)
def test_resolve_output_pcm() -> None:
"""The live driver's default output PCM comes from `-o`, then config live_output_pcm,
then the ALSA default, and never from the sounddevice output_device (issue #50)."""
print("\n=== live_driver: -o over live_output_pcm, never the sounddevice name ===")
pcm = "plughw:CARD=PowerConf,DEV=0"
cfg_with_pcm = {"live_output_pcm": pcm, "output_device": "desktop speakers"}
check(
"-o wins over config live_output_pcm",
live_driver._resolve_output_pcm("hw:CARD=Other,DEV=0", cfg_with_pcm)
== "hw:CARD=Other,DEV=0",
"explicit CLI value must override the config default",
)
check(
"config live_output_pcm is the default when -o is omitted",
live_driver._resolve_output_pcm(None, cfg_with_pcm) == pcm,
f"expected {pcm!r}",
)
check(
"the sounddevice output_device never leaks into the aplay PCM",
live_driver._resolve_output_pcm(None, {"output_device": "desktop speakers"})
is None,
"output_device is a friendly-substring and would fail as `aplay -D`",
)
check(
"an empty live_output_pcm falls back to the ALSA default (None)",
live_driver._resolve_output_pcm(None, {"live_output_pcm": ""}) is None,
"empty string is the unset sentinel, matching DEFAULTS",
)
check(
"no config and no -o means the ALSA default (None)",
live_driver._resolve_output_pcm(None, {}) is None,
"aplay with no -D plays the ALSA default device",
)
check(
"an explicit -o '' overrides config to the ALSA default",
live_driver._resolve_output_pcm("", cfg_with_pcm) is None,
"a present -o (even empty) wins; None (omitted) and '' (override) stay distinct",
)
class _FakePlaybackProcess:
def __init__(self, pid: int, return_code: int | None = None) -> None:
self.pid = pid
self.return_code = return_code
self.wait_timeouts: list[int | None] = []
def poll(self):
return self.return_code
def wait(self, timeout=None):
self.wait_timeouts.append(timeout)
return self.return_code
def test_abortable_aplay() -> None:
"""The live aplay path isolates and stops normal and root-owned playback."""
print("\n=== live_driver: abortable aplay uses a bounded process group ===")
first = _FakePlaybackProcess(101, return_code=1)
elevated = _FakePlaybackProcess(202)
processes = iter([first, elevated])
launched: list[tuple[list[str], bool]] = []
signals: list[tuple[int, signal.Signals, bool]] = []
real_popen = live_driver.subprocess.Popen
real_signal = live_driver._signal_playback_group
def fake_popen(cmd, **kwargs):
launched.append((cmd, kwargs.get("start_new_session", False)))
return next(processes)
def fake_signal(process, sig, is_elevated):
signals.append((process.pid, sig, is_elevated))
process.return_code = -int(sig)
live_driver.subprocess.Popen = fake_popen
live_driver._signal_playback_group = fake_signal
try:
finished = live_driver._play_wav(
"/tmp/reply.wav",
"plughw:CARD=PowerConf,DEV=0",
should_stop=lambda: True,
poll_ms=0,
)
finally:
live_driver.subprocess.Popen = real_popen
live_driver._signal_playback_group = real_signal
check(
"a failed user playback retries through non-interactive sudo",
launched
== [
(
[
"aplay",
"-q",
"-D",
"plughw:CARD=PowerConf,DEV=0",
"/tmp/reply.wav",
],
True,
),
(
[
"sudo",
"-n",
"aplay",
"-q",
"-D",
"plughw:CARD=PowerConf,DEV=0",
"/tmp/reply.wav",
],
True,
),
],
f"launched={launched}",
)
check(
"barge-in stops the elevated playback group and reports interruption",
finished is False
and signals == [(202, signal.SIGTERM, True)]
and elevated.wait_timeouts == [1],
f"finished={finished} signals={signals} waits={elevated.wait_timeouts}",
)
real_process = subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(30)"],
start_new_session=True,
)
live_driver._stop_playback_group(real_process, elevated=False)
check(
"normal playback termination signals the real isolated process group",
real_process.poll() is not None,
f"return_code={real_process.returncode}",
)
class _RunResult:
def __init__(self, returncode, args):
self.returncode = returncode
self.args = args
run_calls = []
real_run = live_driver.subprocess.run
elevated.return_code = None
def fake_run(cmd, **kwargs):
run_calls.append((cmd, kwargs))
return _RunResult(0, cmd)
live_driver.subprocess.run = fake_run
try:
live_driver._signal_playback_group(elevated, signal.SIGTERM, True)
finally:
live_driver.subprocess.run = real_run
check(
"root playback uses the exact non-interactive process-group signal",
len(run_calls) == 1
and run_calls[0][0] == ["sudo", "-n", "kill", "-SIGTERM", "--", "-202"]
and run_calls[0][1]["check"] is False,
f"calls={run_calls}",
)
class _StubbornProcess(_FakePlaybackProcess):
def wait(self, timeout=None):
self.wait_timeouts.append(timeout)
if len(self.wait_timeouts) == 1:
raise subprocess.TimeoutExpired("aplay", timeout)
return self.return_code
stubborn = _StubbornProcess(303)
signals = []
def signal_stubborn(process, sig, is_elevated):
signals.append((process.pid, sig, is_elevated))
process.return_code = -int(sig)
live_driver._signal_playback_group = signal_stubborn
try:
live_driver._stop_playback_group(stubborn, elevated=False)
finally:
live_driver._signal_playback_group = real_signal
check(
"a playback group that ignores TERM is killed after the bounded wait",
signals
== [
(303, signal.SIGTERM, False),
(303, signal.SIGKILL, False),
]
and stubborn.wait_timeouts == [1, 1],
f"signals={signals} waits={stubborn.wait_timeouts}",
)
detector_error_process = _FakePlaybackProcess(404)
signals = []
detector_launches = []
def signal_detector_error(process, sig, is_elevated):
signals.append((process.pid, sig, is_elevated))
process.return_code = -int(sig)
def popen_detector_error(cmd, **_kwargs):
detector_launches.append(cmd)
return detector_error_process
live_driver.subprocess.Popen = popen_detector_error
live_driver._signal_playback_group = signal_detector_error
detector_error = None
try:
live_driver._play_wav(
"reply.wav",
None,
should_stop=lambda: (_ for _ in ()).throw(
subprocess.CalledProcessError(17, ["detector"])
),
poll_ms=0,
)
except subprocess.CalledProcessError as error:
detector_error = error
finally:
live_driver.subprocess.Popen = real_popen
live_driver._signal_playback_group = real_signal
check(
"a detector error stops playback and cannot start a privileged retry",
isinstance(detector_error, subprocess.CalledProcessError)
and detector_error.cmd == ["detector"]
and detector_launches == [["aplay", "-q", "reply.wav"]]
and signals == [(404, signal.SIGTERM, False)]
and detector_error_process.wait_timeouts == [1],
f"error={detector_error!r} launches={detector_launches} signals={signals} "
f"waits={detector_error_process.wait_timeouts}",
)
def _frames(*groups):
"""Build a frame list from (value, count) groups: _frames((4000, 6), (0, 15)) is
6 loud frames then 15 of room tone."""
out = []
for val, n in groups:
out.extend(np.full(FRAME_SIZE, val, dtype=np.int16) for _ in range(n))
return out
CUE_ON = dict(GUARD_OFF, wake_chime=True)
def test_cue_pause_gate() -> None:
"""The cue is gated on a pause (#55): the hardware path peeks the post-wake audio,
skips the cue when a command is already being spoken (so the no-pause command is not
clipped by the cue window), and plays the cue only when the window is silent."""
print("\n=== live_driver.run_turn: gate the wake chime on a pause (#55) ===")
real = (
live_driver.listen_for_wake,
live_driver._play_wav,
pipeline.capture_request,
pipeline._confirm_speech,
)
live_driver.listen_for_wake = lambda fr, m, t, d, preroll=None: (
0.9
) # fire, consume nothing
pipeline._confirm_speech = lambda _pcm, _threshold: True
def run(
frames: list, store: dict, play_error=None, cfg: dict | None = None
) -> tuple[int, int]:
played = {"n": 0}
def fake_play(path, dev):
played["n"] += 1
if play_error is not None:
raise play_error
live_driver._play_wav = fake_play
def cap(fr, preroll=None, vad_threshold=None, **_):
store["frames"] = sum(1 for _ in fr) # count what capture actually receives
return np.zeros(0, dtype=np.int16) # empty -> the turn ends after the gate
pipeline.capture_request = cap
mic = _FakeMic()
live_driver.run_turn(
iter(frames),
mic,
object(),
0.5,
"/dev/null",
None,
CUE_ON.copy() if cfg is None else cfg,
False,
)
return played["n"], mic.flushed
try:
# No-pause: 6 loud + 15 silent. The cue is skipped and the peeked command frames
# (stopped at onset) are prepended, so capture sees all 21 and the mic is not flushed.
store: dict = {}
plays, flushed = run(_frames((4000, 6), (0, 15)), store)
check(
"no-pause command: cue skipped, no flush, peeked frames reach capture",
plays == 0 and flushed == 0 and store.get("frames") == 21,
f"plays={plays} flushed={flushed} capture_frames={store.get('frames')}",
)
# Pause: 6 + 4 + 12 = 22 frames with a silent onset. The cue plays, the post-cue
# flush runs, and the peeked room tone is dropped from capture.
store = {}
plays, flushed = run(_frames((0, 6), (4000, 4), (0, 12)), store)
check(
"pause: cue plays, mic flushed, peeked room tone dropped from capture",
plays == 1
and flushed == 1
and store.get("frames") == 22 - pipeline._CUE_PEEK_FRAMES,
f"plays={plays} flushed={flushed} capture_frames={store.get('frames')}",
)
# Cue failure on the pause branch: aplay raises, so no clean playback boundary
# was established and nothing is flushed. The buffered audio must be kept -- the
# peeked frames are fed back to capture, not dropped -- so capture still sees all
# 22 frames (issues #30, #55, #58).
store = {}
runtime_cfg = CUE_ON.copy()
plays, flushed = run(
_frames((0, 6), (4000, 4), (0, 12)),
store,
play_error=RuntimeError("aplay: no such device"),
cfg=runtime_cfg,
)
check(
"cue failure on a pause keeps the peeked frames and skips the flush",
plays == 1 and flushed == 0 and store.get("frames") == 22,
f"plays={plays} flushed={flushed} capture_frames={store.get('frames')}",
)
check(
"cue failure disables the optional cue for later turns",
runtime_cfg["wake_chime"] is False,
f"wake_chime={runtime_cfg['wake_chime']}",
)
# main() reuses runtime_cfg. The next turn must bypass _play_wav entirely,
# proving the assignment above changes behavior rather than only state.
store = {}
plays, flushed = run(
_frames((0, 6), (4000, 4), (0, 12)), store, cfg=runtime_cfg
)
check(
"the next turn uses no-chime semantics after a cue failure",
plays == 0 and flushed == 0 and store.get("frames") == 22,
f"plays={plays} flushed={flushed} capture_frames={store.get('frames')}",
)
finally:
(
live_driver.listen_for_wake,
live_driver._play_wav,
pipeline.capture_request,
pipeline._confirm_speech,
) = real
def main() -> int:
test_stdin_mic()
test_resolve_output_pcm()
test_abortable_aplay()
test_cue_pause_gate()
print("\n=== live_driver.run_turn: the hardware path honors the guard ===")
# A low-confidence transcript on the hardware path must be re-prompted, not
# dispatched: the brain is skipped and the re-prompt is spoken.
s = drive(GUARD_ON, Transcript("delete everything", -3.0, 0.1))
check(
"low-confidence transcript skips the brain and speaks the re-prompt",
s["ran"] is True
and not s["brain_called"]
and s["spoken"] == pipeline.STT_REPROMPT,
f"ran={s['ran']} brain_called={s['brain_called']} spoken={s['spoken']!r}",
)
# A confident transcript reaches the brain unchanged, and its reply is spoken.
s = drive(GUARD_ON, Transcript("file an issue", -0.3, 0.05))
check(
"confident transcript reaches the brain and speaks its reply",
s["ran"] is True
and s["brain_called"]
and s["brain_arg"] == "file an issue"
and s["spoken"] == "Brain answer.",
f"brain_called={s['brain_called']} brain_arg={s['brain_arg']!r} "
f"spoken={s['spoken']!r}",
)
check(
"the hardware path sends captured int16 PCM to transcription in memory",
isinstance(s["transcribe_arg"], np.ndarray)
and s["transcribe_arg"].dtype == np.int16
and s["transcribe_arg"].shape == (8 * FRAME_SIZE,),
f"type={type(s['transcribe_arg']).__name__} "
f"dtype={getattr(s['transcribe_arg'], 'dtype', None)} "
f"shape={getattr(s['transcribe_arg'], 'shape', None)}",
)
print("\n=== live_driver.run_turn: recover a fully consumed short command ===")
s = drive(
GUARD_ON,
Transcript("Computer, stop.", -0.3, 0.05),
empty_capture=True,
)
check(
"a fully consumed wake-prefixed command reaches the brain without its wake",
s["ran"] is True and s["brain_called"] and s["brain_arg"] == "stop.",
f"ran={s['ran']} brain_called={s['brain_called']} brain_arg={s['brain_arg']!r}",
)
s = drive(
GUARD_ON,
Transcript("Computer, stop.", -3.0, 0.05),
empty_capture=True,
)
check(
"a recovered low-confidence command is re-prompted, never dispatched",
s["ran"] is True
and not s["brain_called"]
and s["spoken"] == pipeline.STT_REPROMPT,
f"brain_called={s['brain_called']} spoken={s['spoken']!r}",
)
for transcript in ("Computer.", "please stop"):
s = drive(
GUARD_ON,
Transcript(transcript, -0.3, 0.05),
empty_capture=True,
)
check(
f"{transcript!r} dispatches nothing after an empty hardware capture",
s["ran"] is True and not s["brain_called"] and s["spoken"] is None,
f"ran={s['ran']} brain_called={s['brain_called']} spoken={s['spoken']!r}",
)
s = drive(
GUARD_ON,
Transcript("Computer, delete everything.", -0.3, 0.05),
empty_capture="vad_rejected",
)
check(
"VAD-rejected noise cannot enter hardware short-command recovery",
s["ran"] is True and not s["brain_called"] and s["transcribe_arg"] is None,
f"brain_called={s['brain_called']} transcribe_arg={s['transcribe_arg']!r}",
)
# With the guard disabled, even a low-confidence transcript dispatches (the knob
# is honored on the hardware path too).
s = drive(GUARD_OFF, Transcript("delete everything", -3.0, 0.99))
check(
"guard disabled: a low-confidence transcript still dispatches",
s["ran"] is True and s["brain_called"] and s["spoken"] == "Brain answer.",
f"brain_called={s['brain_called']} spoken={s['spoken']!r}",
)
print("\n=== live_driver.run_turn: the reply is played back (issue #11) ===")
# #11's core guarantee: a live turn must PLAY the reply that speak() produced, not
# just leave the WAV on disk, or the user never hears the answer. Pin that run_turn
# feeds the reply WAV to the player on the configured output device -- so a refactor
# that drops the playback call, or misroutes the device, fails here instead of
# silently going mute.
s = drive(
GUARD_ON,
Transcript("file an issue", -0.3, 0.05),
output_device="plughw:CARD=PowerConf,DEV=0",
)
check(
"the spoken reply WAV is played on the configured output device, after speak()",
s["played"] == [("/dev/null", "plughw:CARD=PowerConf,DEV=0")]
and s["events"] == ["speak", "play"],
f"played={s['played']} events={s['events']}",
)
# The re-prompt is a reply too: a guard-rejected turn must also PLAY what speak()
# produced, or the user is re-prompted silently and never knows to try again. Both
# the confident case above and this one share run_turn's single speak-then-play tail,
# so pin the rejected path on its own -- a refactor that moves _play_wav into the
# brain/success branch would leave STT_REPROMPT spoken but unplayed, and only this
# check would fail.
s = drive(
GUARD_ON,
Transcript("delete everything", -3.0, 0.1),
output_device="plughw:CARD=PowerConf,DEV=0",
)
check(
"a guard-rejected turn plays the re-prompt WAV on the configured device",
not s["brain_called"]
and s["spoken"] == pipeline.STT_REPROMPT
and s["played"] == [("/dev/null", "plughw:CARD=PowerConf,DEV=0")]
and s["events"] == ["speak", "play"],
f"brain_called={s['brain_called']} spoken={s['spoken']!r} "
f"played={s['played']} events={s['events']}",
)
# A dead or busy output device must degrade to a logged error, never crash: one
# failed aplay cannot take down the always-on loop (live_driver.run_turn wraps the
# playback in try/except). The turn still completes and the reply was still spoken.
s = drive(
GUARD_ON,
Transcript("file an issue", -0.3, 0.05),
play_error=RuntimeError("aplay: no such device"),
)
check(
"a playback failure degrades gracefully (the turn still completes)",
s["ran"] is True and s["spoken"] == "Brain answer." and len(s["played"]) == 1,
f"ran={s['ran']} spoken={s['spoken']!r} played={s['played']}",
)
n_pass = sum(1 for r in results if r[0] == PASS)
n_total = len(results)
print(f"\n=== {n_pass}/{n_total} checks passed ===")
return 0 if n_pass == n_total else 1
if __name__ == "__main__":
sys.exit(main())