From 2c2fb4485cbcf6c03e275293403c93f4531dfbb8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 00:07:38 +0000 Subject: [PATCH 1/2] Split oversized modules and gate file length at 500 lines No source or test file is allowed to exceed 500 lines, so split the worst offenders and add a check that enforces the cap going forward: - Extract the streaming orchestration (SourceOptions, StreamSession, validate_sources) out of commands/stream.py into streaming/session.py; the command keeps the flag signature and source dispatch. - Split tests/test_stream_command.py into command, session, and llm suites, and tests/test_setup.py into install vs remove/unit suites, sharing fakes via tests/setup_helpers.py. - Add scripts/max_file_length.py and wire it into scripts/check.sh. --- aai_cli/commands/stream.py | 225 +----------------- aai_cli/streaming/session.py | 222 +++++++++++++++++ scripts/check.sh | 6 + scripts/max_file_length.py | 49 ++++ tests/setup_helpers.py | 64 +++++ tests/test_setup.py | 394 +----------------------------- tests/test_setup_install.py | 355 +++++++++++++++++++++++++++ tests/test_stream_command.py | 448 ----------------------------------- tests/test_stream_llm.py | 117 +++++++++ tests/test_stream_session.py | 360 ++++++++++++++++++++++++++++ 10 files changed, 1186 insertions(+), 1054 deletions(-) create mode 100644 aai_cli/streaming/session.py create mode 100644 scripts/max_file_length.py create mode 100644 tests/setup_helpers.py create mode 100644 tests/test_setup_install.py create mode 100644 tests/test_stream_llm.py create mode 100644 tests/test_stream_session.py diff --git a/aai_cli/commands/stream.py b/aai_cli/commands/stream.py index 36bf64b2..ca739c6b 100644 --- a/aai_cli/commands/stream.py +++ b/aai_cli/commands/stream.py @@ -1,10 +1,6 @@ from __future__ import annotations -import queue import tempfile -import threading -from collections.abc import Callable, Iterable -from dataclasses import dataclass, field from pathlib import Path import typer @@ -22,228 +18,21 @@ youtube, ) from aai_cli.context import AppState, run_command -from aai_cli.errors import CLIError, UsageError +from aai_cli.errors import UsageError from aai_cli.follow import FollowRenderer from aai_cli.help_text import examples_epilog from aai_cli.microphone import MicrophoneSource from aai_cli.streaming.macos import MacSystemAudioSource from aai_cli.streaming.render import StreamRenderer +from aai_cli.streaming.session import SourceOptions, StreamSession, validate_sources from aai_cli.streaming.sources import TARGET_RATE, FileSource, StdinSource app = typer.Typer() DEFAULT_SPEECH_MODEL = SpeechModel.u3_rt_pro -# Sources that can be transcribed in parallel sessions: (label, audio chunks, sample rate). -_ParallelStreams = list[tuple[str, Iterable[bytes], int]] - -@dataclass(frozen=True) -class _SourceOptions: - """Where the audio comes from, distilled from the CLI flags. - - Centralizes the "which input?" predicates so the validation and dispatch helpers - below read off one object instead of re-deriving the same booleans. - """ - - source: str | None - sample: bool - sample_rate: int | None - device: int | None - system_audio: bool - system_audio_only: bool - - @property - def from_stdin(self) -> bool: - return self.source == "-" - - @property - def from_file(self) -> bool: - return bool(self.source) or self.sample - - @property - def from_system_audio(self) -> bool: - return self.system_audio or self.system_audio_only - - @property - def has_capture_overrides(self) -> bool: - """Whether a microphone-only flag (--sample-rate or --device) was given.""" - return self.sample_rate is not None or self.device is not None - - -def _validate_sources(opts: _SourceOptions, *, has_llm: bool, text_mode: bool) -> None: - """Reject flag combinations that can't be honored, before any audio is opened.""" - if opts.system_audio and opts.system_audio_only: - raise UsageError("Use either --system-audio or --system-audio-only, not both.") - _validate_input_source(opts) - if has_llm and text_mode: - raise UsageError( - "--llm renders a live panel (or NDJSON when piped); it can't be combined with -o text." - ) - - -def _validate_input_source(opts: _SourceOptions) -> None: - """Reject --sample-rate/--device/source combinations the chosen input can't accept.""" - if opts.from_system_audio: - if opts.from_file: - raise UsageError("--system-audio cannot be combined with an audio source or --sample.") - if opts.system_audio_only and opts.has_capture_overrides: - raise UsageError( - "--sample-rate and --device require microphone input; use --system-audio." - ) - elif opts.from_stdin: - if opts.device is not None: - raise UsageError("--device applies only to microphone input.") - elif opts.from_file and opts.has_capture_overrides: - raise UsageError("--sample-rate and --device apply only to microphone input.") - - -@dataclass -class _StreamSession: - """Owns one streaming run: the renderers, the LLM-chain state, and the audio - plumbing shared across single- and parallel-source streaming. - - Holding this as an object (rather than a nest of closures inside the command body) - keeps each step a small, independently readable method, and collapses the ~25 - per-call flags into one ``base_flags`` dict that only varies by sample rate. - """ - - api_key: str - base_flags: dict[str, object] - overrides: list[str] | None - config_file: str | Path | None - renderer: StreamRenderer - follow: FollowRenderer | None - llm_prompts: list[str] - model: str - max_tokens: int - transcript: list[str] = field(default_factory=list[str]) - _callback_lock: threading.RLock = field(default_factory=threading.RLock) - _listening_lock: threading.Lock = field(default_factory=threading.Lock) - _listening_started: bool = False - - @property - def on_open(self) -> Callable[[], None]: - """First-audio callback: announce "Listening…" once — unless the FollowRenderer - owns the screen in --llm mode, where the notice would clutter the live panel.""" - return (lambda: None) if self.follow is not None else self._listening_once - - def _listening_once(self) -> None: - with self._listening_lock: - if self._listening_started: - return - self._listening_started = True - self.renderer.listening() - - def on_turn(self, event: object, *, source_label: str | None = None) -> None: - with self._callback_lock: - if self.follow is None: - self.renderer.turn(event, source=source_label) - else: - self._refresh_answer(event, source_label) - - def _refresh_answer(self, event: object, source_label: str | None) -> None: - """Live --llm mode: re-run the prompt chain over the growing transcript on every - finalized turn, refreshing one evolving answer (partials are ignored).""" - follow = self.follow - if follow is None or not getattr(event, "end_of_turn", False): - return - text = getattr(event, "transcript", "") or "" - if not text: - return - if source_label is not None: - display_source = {"system": "System", "you": "You"}.get(source_label, source_label) - text = f"{display_source}: {text}" - self.transcript.append(text) - answer = llm.run_chain( - self.api_key, - self.llm_prompts, - transcript_text=" ".join(self.transcript), - model=self.model, - max_tokens=self.max_tokens, - ) - follow(answer, len(self.transcript)) - - def stream_one( - self, audio: Iterable[bytes], rate: int, *, source_label: str | None = None - ) -> None: - merged = config_builder.merge_streaming_params( - flags=self.base_flags | {"sample_rate": rate}, - overrides=self.overrides, - config_file=self.config_file, - ) - params = config_builder.construct_streaming_params(merged) - client.stream_audio( - self.api_key, - audio, - params=params, - on_begin=( - None - if self.follow is not None - else lambda event: self.renderer.begin(event, source=source_label) - ), - on_turn=lambda event: self.on_turn(event, source_label=source_label), - on_termination=( - None - if self.follow is not None - else lambda event: self.renderer.termination(event, source=source_label) - ), - ) - - def _guarded(self, work: Callable[[], None]) -> None: - """Run a streaming body with the shared lifecycle handling: enter the - FollowRenderer's live panel if present, treat Ctrl-C as a clean stop, exit 0 on - a closed downstream pipe, and always close the renderer.""" - try: - if self.follow is not None: - with self.follow: - work() - else: - work() - except KeyboardInterrupt: - # Ctrl-C is a normal "user stopped" signal -> exit 0. - if self.follow is None: - self.renderer.close() - self.renderer.stopped() - except BrokenPipeError: - # Downstream consumer (e.g. `| head`) closed the pipe; stop quietly. - raise typer.Exit(code=0) from None - finally: - if self.follow is None: - self.renderer.close() - - def run(self, audio: Iterable[bytes], rate: int, *, source_label: str | None = None) -> None: - self._guarded(lambda: self.stream_one(audio, rate, source_label=source_label)) - - def run_parallel(self, streams: _ParallelStreams) -> None: - self._guarded(lambda: self._drive(streams)) - - def _drive(self, streams: _ParallelStreams) -> None: - """Stream every source concurrently, surfacing the first worker error.""" - errors: queue.Queue[Exception] = queue.Queue() - - def worker(source_label: str, audio: Iterable[bytes], rate: int) -> None: - try: - self.stream_one(audio, rate, source_label=source_label) - except (CLIError, BrokenPipeError) as exc: - errors.put(exc) - - threads = [ - threading.Thread(target=worker, args=(label, audio, rate), daemon=True) - for label, audio, rate in streams - ] - for thread in threads: - thread.start() - while any(thread.is_alive() for thread in threads): - for thread in threads: - thread.join(timeout=0.1) - if not errors.empty(): - raise errors.get() - if not errors.empty(): - raise errors.get() - - -def _dispatch(session: _StreamSession, opts: _SourceOptions) -> None: +def _dispatch(session: StreamSession, opts: SourceOptions) -> None: """Open the right audio source(s) for the flags and stream them.""" if opts.from_system_audio: system = MacSystemAudioSource(on_open=session.on_open) @@ -274,7 +63,7 @@ def _dispatch(session: _StreamSession, opts: _SourceOptions) -> None: else: # Capture at the device's native rate (or --sample-rate override) and tell the # streaming API that rate, rather than forcing one the device may reject. - # "Listening…" is announced once the device is open (see _StreamSession.on_open), + # "Listening…" is announced once the device is open (see StreamSession.on_open), # not when the session opens — so early speech isn't lost in the gap. mic = MicrophoneSource( device=opts.device, capture_rate=opts.sample_rate, on_open=session.on_open @@ -522,7 +311,7 @@ def stream( def body(state: AppState, json_mode: bool) -> None: text_mode, json_mode = output.stream_output_modes(output_field, json_mode=json_mode) - opts = _SourceOptions( + opts = SourceOptions( source=source, sample=sample, sample_rate=sample_rate, @@ -573,10 +362,10 @@ def body(state: AppState, json_mode: bool) -> None: return api_key = config.resolve_api_key(profile=state.profile) - _validate_sources(opts, has_llm=bool(llm_prompt), text_mode=text_mode) + validate_sources(opts, has_llm=bool(llm_prompt), text_mode=text_mode) llm_prompts = list(llm_prompt or []) - session = _StreamSession( + session = StreamSession( api_key=api_key, base_flags=base_flags, overrides=config_kv, diff --git a/aai_cli/streaming/session.py b/aai_cli/streaming/session.py new file mode 100644 index 00000000..d4dd3d0b --- /dev/null +++ b/aai_cli/streaming/session.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import queue +import threading +from collections.abc import Callable, Iterable +from dataclasses import dataclass, field +from pathlib import Path + +import typer + +from aai_cli import client, config_builder, llm +from aai_cli.errors import CLIError, UsageError +from aai_cli.follow import FollowRenderer +from aai_cli.streaming.render import StreamRenderer + +# Sources that can be transcribed in parallel sessions: (label, audio chunks, sample rate). +_ParallelStreams = list[tuple[str, Iterable[bytes], int]] + + +@dataclass(frozen=True) +class SourceOptions: + """Where the audio comes from, distilled from the CLI flags. + + Centralizes the "which input?" predicates so the validation and dispatch helpers + in the command module read off one object instead of re-deriving the same booleans. + """ + + source: str | None + sample: bool + sample_rate: int | None + device: int | None + system_audio: bool + system_audio_only: bool + + @property + def from_stdin(self) -> bool: + return self.source == "-" + + @property + def from_file(self) -> bool: + return bool(self.source) or self.sample + + @property + def from_system_audio(self) -> bool: + return self.system_audio or self.system_audio_only + + @property + def has_capture_overrides(self) -> bool: + """Whether a microphone-only flag (--sample-rate or --device) was given.""" + return self.sample_rate is not None or self.device is not None + + +def validate_sources(opts: SourceOptions, *, has_llm: bool, text_mode: bool) -> None: + """Reject flag combinations that can't be honored, before any audio is opened.""" + if opts.system_audio and opts.system_audio_only: + raise UsageError("Use either --system-audio or --system-audio-only, not both.") + _validate_input_source(opts) + if has_llm and text_mode: + raise UsageError( + "--llm renders a live panel (or NDJSON when piped); it can't be combined with -o text." + ) + + +def _validate_input_source(opts: SourceOptions) -> None: + """Reject --sample-rate/--device/source combinations the chosen input can't accept.""" + if opts.from_system_audio: + if opts.from_file: + raise UsageError("--system-audio cannot be combined with an audio source or --sample.") + if opts.system_audio_only and opts.has_capture_overrides: + raise UsageError( + "--sample-rate and --device require microphone input; use --system-audio." + ) + elif opts.from_stdin: + if opts.device is not None: + raise UsageError("--device applies only to microphone input.") + elif opts.from_file and opts.has_capture_overrides: + raise UsageError("--sample-rate and --device apply only to microphone input.") + + +@dataclass +class StreamSession: + """Owns one streaming run: the renderers, the LLM-chain state, and the audio + plumbing shared across single- and parallel-source streaming. + + Holding this as an object (rather than a nest of closures inside the command body) + keeps each step a small, independently readable method, and collapses the ~25 + per-call flags into one ``base_flags`` dict that only varies by sample rate. + """ + + api_key: str + base_flags: dict[str, object] + overrides: list[str] | None + config_file: str | Path | None + renderer: StreamRenderer + follow: FollowRenderer | None + llm_prompts: list[str] + model: str + max_tokens: int + transcript: list[str] = field(default_factory=list[str]) + _callback_lock: threading.RLock = field(default_factory=threading.RLock) + _listening_lock: threading.Lock = field(default_factory=threading.Lock) + _listening_started: bool = False + + @property + def on_open(self) -> Callable[[], None]: + """First-audio callback: announce "Listening…" once — unless the FollowRenderer + owns the screen in --llm mode, where the notice would clutter the live panel.""" + return (lambda: None) if self.follow is not None else self._listening_once + + def _listening_once(self) -> None: + with self._listening_lock: + if self._listening_started: + return + self._listening_started = True + self.renderer.listening() + + def on_turn(self, event: object, *, source_label: str | None = None) -> None: + with self._callback_lock: + if self.follow is None: + self.renderer.turn(event, source=source_label) + else: + self._refresh_answer(event, source_label) + + def _refresh_answer(self, event: object, source_label: str | None) -> None: + """Live --llm mode: re-run the prompt chain over the growing transcript on every + finalized turn, refreshing one evolving answer (partials are ignored).""" + follow = self.follow + if follow is None or not getattr(event, "end_of_turn", False): + return + text = getattr(event, "transcript", "") or "" + if not text: + return + if source_label is not None: + display_source = {"system": "System", "you": "You"}.get(source_label, source_label) + text = f"{display_source}: {text}" + self.transcript.append(text) + answer = llm.run_chain( + self.api_key, + self.llm_prompts, + transcript_text=" ".join(self.transcript), + model=self.model, + max_tokens=self.max_tokens, + ) + follow(answer, len(self.transcript)) + + def stream_one( + self, audio: Iterable[bytes], rate: int, *, source_label: str | None = None + ) -> None: + merged = config_builder.merge_streaming_params( + flags=self.base_flags | {"sample_rate": rate}, + overrides=self.overrides, + config_file=self.config_file, + ) + params = config_builder.construct_streaming_params(merged) + client.stream_audio( + self.api_key, + audio, + params=params, + on_begin=( + None + if self.follow is not None + else lambda event: self.renderer.begin(event, source=source_label) + ), + on_turn=lambda event: self.on_turn(event, source_label=source_label), + on_termination=( + None + if self.follow is not None + else lambda event: self.renderer.termination(event, source=source_label) + ), + ) + + def _guarded(self, work: Callable[[], None]) -> None: + """Run a streaming body with the shared lifecycle handling: enter the + FollowRenderer's live panel if present, treat Ctrl-C as a clean stop, exit 0 on + a closed downstream pipe, and always close the renderer.""" + try: + if self.follow is not None: + with self.follow: + work() + else: + work() + except KeyboardInterrupt: + # Ctrl-C is a normal "user stopped" signal -> exit 0. + if self.follow is None: + self.renderer.close() + self.renderer.stopped() + except BrokenPipeError: + # Downstream consumer (e.g. `| head`) closed the pipe; stop quietly. + raise typer.Exit(code=0) from None + finally: + if self.follow is None: + self.renderer.close() + + def run(self, audio: Iterable[bytes], rate: int, *, source_label: str | None = None) -> None: + self._guarded(lambda: self.stream_one(audio, rate, source_label=source_label)) + + def run_parallel(self, streams: _ParallelStreams) -> None: + self._guarded(lambda: self._drive(streams)) + + def _drive(self, streams: _ParallelStreams) -> None: + """Stream every source concurrently, surfacing the first worker error.""" + errors: queue.Queue[Exception] = queue.Queue() + + def worker(source_label: str, audio: Iterable[bytes], rate: int) -> None: + try: + self.stream_one(audio, rate, source_label=source_label) + except (CLIError, BrokenPipeError) as exc: + errors.put(exc) + + threads = [ + threading.Thread(target=worker, args=(label, audio, rate), daemon=True) + for label, audio, rate in streams + ] + for thread in threads: + thread.start() + while any(thread.is_alive() for thread in threads): + for thread in threads: + thread.join(timeout=0.1) + if not errors.empty(): + raise errors.get() + if not errors.empty(): + raise errors.get() diff --git a/scripts/check.sh b/scripts/check.sh index 129bc38d..08ae507a 100755 --- a/scripts/check.sh +++ b/scripts/check.sh @@ -45,6 +45,12 @@ uv run deptry . echo "==> import-linter (architecture contracts)" uv run lint-imports +echo "==> max file length (500-line gate, src + tests + scripts)" +# Keep modules small enough for humans and AI coding agents to hold in context. +# Raising the cap is a deliberate edit to scripts/max_file_length.py, not a per-file +# exception. +uv run python scripts/max_file_length.py + echo "==> xenon (cyclomatic complexity gate, src only)" # Fail the build if any function gets too branchy. Grades map to cyclomatic # complexity: A=1-5, B=6-10, C=11-20, ... Thresholds: diff --git a/scripts/max_file_length.py b/scripts/max_file_length.py new file mode 100644 index 00000000..c7f732ad --- /dev/null +++ b/scripts/max_file_length.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import sys +from pathlib import Path + +# Files longer than this are hard for humans and AI coding agents alike to hold in +# context and navigate; split them into focused modules instead. Raising this limit +# should be a deliberate, reviewed decision — bump the number here, don't special-case. +MAX_LINES = 500 + +# First-party Python that the team reads and edits. Generated trees (docs/, dist/, +# build artifacts) and third-party code are out of scope. +ROOTS = ("aai_cli", "tests", "scripts") + + +def _line_count(path: Path) -> int: + # Count newlines; a trailing line without a newline still counts (+1) so the + # number matches `wc -l` on POSIX-clean files and never undercounts. + with path.open("rb") as handle: + data = handle.read() + if not data: + return 0 + return data.count(b"\n") + (0 if data.endswith(b"\n") else 1) + + +def _offenders() -> list[tuple[Path, int]]: + repo_root = Path(__file__).resolve().parent.parent + found: list[tuple[Path, int]] = [] + for root in ROOTS: + for path in sorted((repo_root / root).rglob("*.py")): + count = _line_count(path) + if count > MAX_LINES: + found.append((path.relative_to(repo_root), count)) + return found + + +def main() -> int: + offenders = _offenders() + if not offenders: + sys.stdout.write(f"All Python files are within {MAX_LINES} lines.\n") + return 0 + sys.stdout.write(f"Files over the {MAX_LINES}-line limit (split them into smaller modules):\n") + for path, count in offenders: + sys.stdout.write(f" {path}: {count} lines\n") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/setup_helpers.py b/tests/setup_helpers.py new file mode 100644 index 00000000..2f91211d --- /dev/null +++ b/tests/setup_helpers.py @@ -0,0 +1,64 @@ +"""Shared scaffolding for the `aai setup` test modules. + +Not a test module itself (no ``test_`` prefix, so pytest won't collect it): it +holds the fakes and helpers reused across ``test_setup.py`` and +``test_setup_install.py`` so neither file has to redeclare them. +""" + +import json +import shutil +import subprocess +from pathlib import Path + + +def _skill_path() -> Path: + return Path.home() / ".claude" / "skills" / "assemblyai" + + +def _cli_skill_path() -> Path: + return Path.home() / ".claude" / "skills" / "aai-cli" + + +class FakeRun: + """Records subprocess calls and returns canned CompletedProcess results. + + `returncodes` maps a command prefix tuple (the first N argv tokens) to a + return code; the longest matching prefix wins, default 0. To mimic the real + `skills` CLI, a successful `npx … add` materializes the assemblyai skill under + HOME (so `_install_skill`'s filesystem check passes) and `npx … remove` + deletes it — toggle with `creates_skill` / `removes_skill`. The aai-cli skill + is bundled and copied directly (no subprocess), so it never goes through here. + """ + + def __init__(self, returncodes=None, *, creates_skill=True, removes_skill=True): + self.calls = [] + self.returncodes = returncodes or {} + self.creates_skill = creates_skill + self.removes_skill = removes_skill + + def __call__(self, cmd, *args, **kwargs): + self.calls.append(cmd) + rc = 0 + best = -1 + for prefix, code in self.returncodes.items(): + n = len(prefix) + if tuple(cmd[:n]) == prefix and n > best: + rc, best = code, n + if rc == 0 and cmd[:1] == ["npx"]: + if "add" in cmd and self.creates_skill: + _skill_path().mkdir(parents=True, exist_ok=True) + (_skill_path() / "SKILL.md").write_text("# AssemblyAI") + elif "remove" in cmd and self.removes_skill: + shutil.rmtree(_skill_path(), ignore_errors=True) + return subprocess.CompletedProcess(args=cmd, returncode=rc, stdout="", stderr="boom") + + +def _all_tools_present(monkeypatch): + monkeypatch.setattr( + "aai_cli.commands.setup.shutil.which", + lambda tool: f"/usr/bin/{tool}", + ) + + +def _statuses(result): + return {s["name"]: s["status"] for s in json.loads(result.output)["steps"]} diff --git a/tests/test_setup.py b/tests/test_setup.py index cc6fecf6..47e6f13c 100644 --- a/tests/test_setup.py +++ b/tests/test_setup.py @@ -1,12 +1,10 @@ -import json -import shutil import subprocess -from pathlib import Path import pytest from typer.testing import CliRunner from aai_cli.main import app +from tests.setup_helpers import FakeRun, _all_tools_present, _cli_skill_path, _skill_path, _statuses runner = CliRunner() @@ -29,304 +27,7 @@ def test_proc_detail_prefers_stderr_then_falls_back_to_stdout(): assert setup._proc_detail(only_out) == "only out" -def _skill_path() -> Path: - return Path.home() / ".claude" / "skills" / "assemblyai" - - -def _cli_skill_path() -> Path: - return Path.home() / ".claude" / "skills" / "aai-cli" - - -class FakeRun: - """Records subprocess calls and returns canned CompletedProcess results. - - `returncodes` maps a command prefix tuple (the first N argv tokens) to a - return code; the longest matching prefix wins, default 0. To mimic the real - `skills` CLI, a successful `npx … add` materializes the assemblyai skill under - HOME (so `_install_skill`'s filesystem check passes) and `npx … remove` - deletes it — toggle with `creates_skill` / `removes_skill`. The aai-cli skill - is bundled and copied directly (no subprocess), so it never goes through here. - """ - - def __init__(self, returncodes=None, *, creates_skill=True, removes_skill=True): - self.calls = [] - self.returncodes = returncodes or {} - self.creates_skill = creates_skill - self.removes_skill = removes_skill - - def __call__(self, cmd, *args, **kwargs): - self.calls.append(cmd) - rc = 0 - best = -1 - for prefix, code in self.returncodes.items(): - n = len(prefix) - if tuple(cmd[:n]) == prefix and n > best: - rc, best = code, n - if rc == 0 and cmd[:1] == ["npx"]: - if "add" in cmd and self.creates_skill: - _skill_path().mkdir(parents=True, exist_ok=True) - (_skill_path() / "SKILL.md").write_text("# AssemblyAI") - elif "remove" in cmd and self.removes_skill: - shutil.rmtree(_skill_path(), ignore_errors=True) - return subprocess.CompletedProcess(args=cmd, returncode=rc, stdout="", stderr="boom") - - -def _all_tools_present(monkeypatch): - monkeypatch.setattr( - "aai_cli.commands.setup.shutil.which", - lambda tool: f"/usr/bin/{tool}", - ) - - -def _statuses(result): - return {s["name"]: s["status"] for s in json.loads(result.output)["steps"]} - - -# --- install: all three steps ------------------------------------------------ - - -def test_install_happy_path_runs_all_steps(monkeypatch): - _all_tools_present(monkeypatch) - # MCP not yet present -> `mcp get` returns non-zero. - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 - - statuses = _statuses(result) - assert statuses == {"mcp": "installed", "skill": "installed", "aai-cli skill": "installed"} - - assert [ - "claude", - "mcp", - "add", - "--transport", - "http", - "--scope", - "user", - "assemblyai-docs", - "https://mcp.assemblyai.com/docs", - ] in fake.calls - assert [ - "npx", - "-y", - "skills", - "add", - "AssemblyAI/assemblyai-skill", - "--global", - "--yes", - ] in fake.calls - # The bundled aai-cli skill was copied into HOME (no subprocess involved). - assert (_cli_skill_path() / "SKILL.md").exists() - - -def test_install_skill_failed_when_npx_succeeds_but_nothing_installed(monkeypatch): - # Regression: `install` must verify the skill landed, not trust npx's exit - # code — otherwise install says "installed" while status says "not_installed". - _all_tools_present(monkeypatch) - fake = FakeRun({("claude", "mcp", "get"): 1}, creates_skill=False) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 1 # skill step failed - assert _statuses(result)["skill"] == "failed" - - # And status agrees: still not installed. - status_result = runner.invoke(app, ["setup", "status"]) - assert _statuses(status_result)["skill"] == "not_installed" - - -def test_install_detaches_stdin_and_sets_timeout(monkeypatch): - """Regression: subprocess children must not inherit stdin, or an interactive - prompt (npx, claude) hangs the CLI forever. Each call must pass a timeout too.""" - _all_tools_present(monkeypatch) - seen = [] - - def record(cmd, *args, **kwargs): - seen.append(kwargs) - return subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="") - - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", record) - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code in (0, 1) - assert seen, "expected subprocess.run to be called" - for kwargs in seen: - assert kwargs.get("stdin") is subprocess.DEVNULL - assert kwargs.get("timeout") - - -def test_install_scope_passthrough(monkeypatch): - _all_tools_present(monkeypatch) - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--scope", "project"]) - assert result.exit_code == 0 - assert [ - "claude", - "mcp", - "add", - "--transport", - "http", - "--scope", - "project", - "assemblyai-docs", - "https://mcp.assemblyai.com/docs", - ] in fake.calls - - -def test_install_scope_local_passthrough(monkeypatch): - _all_tools_present(monkeypatch) - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--scope", "local"]) - assert result.exit_code == 0 - assert [ - "claude", - "mcp", - "add", - "--transport", - "http", - "--scope", - "local", - "assemblyai-docs", - "https://mcp.assemblyai.com/docs", - ] in fake.calls - - -def test_install_invalid_scope_exits_2(monkeypatch): - _all_tools_present(monkeypatch) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", FakeRun()) - result = runner.invoke(app, ["setup", "install", "--scope", "bogus"]) - assert result.exit_code == 2 - - -def test_install_idempotent_when_mcp_present(monkeypatch): - _all_tools_present(monkeypatch) - # `mcp get` returns 0 -> already registered. - fake = FakeRun({("claude", "mcp", "get"): 0}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 - assert _statuses(result)["mcp"] == "already" - # No `mcp add` should have run. - assert not any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) - - -def test_install_failure_exits_nonzero(monkeypatch): - _all_tools_present(monkeypatch) - # mcp not present, but `mcp add` fails. - fake = FakeRun({("claude", "mcp", "get"): 1, ("claude", "mcp", "add"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 1 - assert _statuses(result)["mcp"] == "failed" - - -def test_install_force_remove_failure_reports_failed(monkeypatch): - _all_tools_present(monkeypatch) - # present, but the forced remove fails - fake = FakeRun({("claude", "mcp", "get"): 0, ("claude", "mcp", "remove"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--force"]) - assert result.exit_code == 1 - assert _statuses(result)["mcp"] == "failed" - assert not any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) - - -def test_install_force_removes_then_adds(monkeypatch): - _all_tools_present(monkeypatch) - fake = FakeRun({("claude", "mcp", "get"): 0}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--force"]) - assert result.exit_code == 0 - assert ["claude", "mcp", "remove", "assemblyai-docs"] in fake.calls - assert any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) - - -def test_install_skips_mcp_when_claude_missing(monkeypatch): - monkeypatch.setattr( - "aai_cli.commands.setup.shutil.which", - lambda tool: None if tool == "claude" else f"/usr/bin/{tool}", - ) - fake = FakeRun() - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 # skip is not a failure - statuses = _statuses(result) - assert statuses["mcp"] == "skipped" - assert statuses["skill"] == "installed" - # The bundled aai-cli skill installs regardless of claude/npx. - assert statuses["aai-cli skill"] == "installed" - assert not any(c[0] == "claude" for c in fake.calls) - - -# --- assemblyai skill (npx-based) -------------------------------------------- - - -def test_install_skill_idempotent_when_present(monkeypatch): - # Regression: a repeat install must report the skill as `already` (like MCP), - # not re-run `npx skills add` and claim `installed` every time. - _all_tools_present(monkeypatch) - skill = _skill_path() - skill.mkdir(parents=True) - (skill / "SKILL.md").write_text("# AssemblyAI") - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 - assert _statuses(result)["skill"] == "already" - # No `npx … add` should have run — the skill was already present. - assert not any(c[0] == "npx" and "add" in c for c in fake.calls) - - -def test_install_force_reinstalls_skill(monkeypatch): - # --force must re-run `npx skills add` even when the skill is already present. - _all_tools_present(monkeypatch) - skill = _skill_path() - skill.mkdir(parents=True) - (skill / "SKILL.md").write_text("# AssemblyAI") - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--force"]) - assert result.exit_code == 0 - assert _statuses(result)["skill"] == "installed" - assert [ - "npx", - "-y", - "skills", - "add", - "AssemblyAI/assemblyai-skill", - "--global", - "--yes", - ] in fake.calls - - -def test_install_skips_skill_when_npx_missing(monkeypatch): - monkeypatch.setattr( - "aai_cli.commands.setup.shutil.which", - lambda tool: None if tool == "npx" else f"/usr/bin/{tool}", - ) - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 - statuses = _statuses(result) - assert statuses["skill"] == "skipped" - assert statuses["mcp"] == "installed" - # aai-cli skill copies in regardless (no npx needed). - assert statuses["aai-cli skill"] == "installed" - assert not any(c[0] == "npx" for c in fake.calls) +# --- remove: assemblyai skill ------------------------------------------------ def test_remove_skill_failure_reports_failed(monkeypatch): @@ -366,93 +67,7 @@ def test_remove_skill_skipped_when_npx_missing(monkeypatch): assert _statuses(result)["skill"] == "skipped" -# --- aai-cli skill (bundled, copied) ----------------------------------------- - - -def test_install_aai_cli_skill_idempotent_when_present(monkeypatch): - _all_tools_present(monkeypatch) - cli_skill = _cli_skill_path() - cli_skill.mkdir(parents=True) - (cli_skill / "SKILL.md").write_text("# old") - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install"]) - assert result.exit_code == 0 - assert _statuses(result)["aai-cli skill"] == "already" - # Not overwritten without --force. - assert (cli_skill / "SKILL.md").read_text() == "# old" - - -def test_install_aai_cli_skill_force_reinstalls(monkeypatch): - _all_tools_present(monkeypatch) - cli_skill = _cli_skill_path() - cli_skill.mkdir(parents=True) - (cli_skill / "SKILL.md").write_text("# old") - fake = FakeRun({("claude", "mcp", "get"): 1}) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) - - result = runner.invoke(app, ["setup", "install", "--force"]) - assert result.exit_code == 0 - assert _statuses(result)["aai-cli skill"] == "installed" - # Overwritten with the bundled copy (references/ exist; placeholder gone). - assert (cli_skill / "references").is_dir() - assert "# old" not in (cli_skill / "SKILL.md").read_text() - - -# --- status ------------------------------------------------------------------ - - -def test_status_reports_all_installed(monkeypatch, tmp_path): - _all_tools_present(monkeypatch) - for name in ("assemblyai", "aai-cli"): - d = tmp_path / ".claude" / "skills" / name - d.mkdir(parents=True) - (d / "SKILL.md").write_text("# x") - # `mcp get` returns 0 -> present. - monkeypatch.setattr( - "aai_cli.commands.setup.subprocess.run", - FakeRun({("claude", "mcp", "get"): 0}), - ) - - result = runner.invoke(app, ["setup", "status"]) - assert result.exit_code == 0 - assert _statuses(result) == { - "mcp": "installed", - "skill": "installed", - "aai-cli skill": "installed", - } - - -def test_status_reports_not_installed(monkeypatch): - _all_tools_present(monkeypatch) # no skill dirs created - monkeypatch.setattr( - "aai_cli.commands.setup.subprocess.run", - FakeRun({("claude", "mcp", "get"): 1}), - ) - - result = runner.invoke(app, ["setup", "status"]) - assert result.exit_code == 0 - assert _statuses(result) == { - "mcp": "not_installed", - "skill": "not_installed", - "aai-cli skill": "not_installed", - } - - -def test_status_mcp_unknown_when_claude_missing(monkeypatch): - monkeypatch.setattr( - "aai_cli.commands.setup.shutil.which", - lambda tool: None if tool == "claude" else f"/usr/bin/{tool}", - ) - monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", FakeRun()) - - result = runner.invoke(app, ["setup", "status"]) - assert result.exit_code == 0 - assert _statuses(result)["mcp"] == "unknown" - - -# --- remove ------------------------------------------------------------------ +# --- remove: all three steps ------------------------------------------------- def test_remove_unwinds_all(monkeypatch, tmp_path): @@ -530,6 +145,9 @@ def test_remove_mcp_failure_reports_failed(monkeypatch): assert _statuses(result)["mcp"] == "failed" +# --- aai-cli skill helpers ---------------------------------------------------- + + def test_copy_tree_skips_pycache_and_pyc(tmp_path): # _copy_tree must not copy compiled-Python detritus into the agent's skills dir. from aai_cli.commands import setup diff --git a/tests/test_setup_install.py b/tests/test_setup_install.py new file mode 100644 index 00000000..ff22d939 --- /dev/null +++ b/tests/test_setup_install.py @@ -0,0 +1,355 @@ +import subprocess + +import pytest +from typer.testing import CliRunner + +from aai_cli.main import app +from tests.setup_helpers import ( + FakeRun, + _all_tools_present, + _cli_skill_path, + _skill_path, + _statuses, +) + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _isolate_home(tmp_path, monkeypatch): + """Keep skill writes/reads inside a temp HOME so tests never touch ~/.claude.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + + +# --- install: all three steps ------------------------------------------------ + + +def test_install_happy_path_runs_all_steps(monkeypatch): + _all_tools_present(monkeypatch) + # MCP not yet present -> `mcp get` returns non-zero. + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 + + statuses = _statuses(result) + assert statuses == {"mcp": "installed", "skill": "installed", "aai-cli skill": "installed"} + + assert [ + "claude", + "mcp", + "add", + "--transport", + "http", + "--scope", + "user", + "assemblyai-docs", + "https://mcp.assemblyai.com/docs", + ] in fake.calls + assert [ + "npx", + "-y", + "skills", + "add", + "AssemblyAI/assemblyai-skill", + "--global", + "--yes", + ] in fake.calls + # The bundled aai-cli skill was copied into HOME (no subprocess involved). + assert (_cli_skill_path() / "SKILL.md").exists() + + +def test_install_skill_failed_when_npx_succeeds_but_nothing_installed(monkeypatch): + # Regression: `install` must verify the skill landed, not trust npx's exit + # code — otherwise install says "installed" while status says "not_installed". + _all_tools_present(monkeypatch) + fake = FakeRun({("claude", "mcp", "get"): 1}, creates_skill=False) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 1 # skill step failed + assert _statuses(result)["skill"] == "failed" + + # And status agrees: still not installed. + status_result = runner.invoke(app, ["setup", "status"]) + assert _statuses(status_result)["skill"] == "not_installed" + + +def test_install_detaches_stdin_and_sets_timeout(monkeypatch): + """Regression: subprocess children must not inherit stdin, or an interactive + prompt (npx, claude) hangs the CLI forever. Each call must pass a timeout too.""" + _all_tools_present(monkeypatch) + seen = [] + + def record(cmd, *args, **kwargs): + seen.append(kwargs) + return subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="") + + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", record) + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code in (0, 1) + assert seen, "expected subprocess.run to be called" + for kwargs in seen: + assert kwargs.get("stdin") is subprocess.DEVNULL + assert kwargs.get("timeout") + + +def test_install_scope_passthrough(monkeypatch): + _all_tools_present(monkeypatch) + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--scope", "project"]) + assert result.exit_code == 0 + assert [ + "claude", + "mcp", + "add", + "--transport", + "http", + "--scope", + "project", + "assemblyai-docs", + "https://mcp.assemblyai.com/docs", + ] in fake.calls + + +def test_install_scope_local_passthrough(monkeypatch): + _all_tools_present(monkeypatch) + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--scope", "local"]) + assert result.exit_code == 0 + assert [ + "claude", + "mcp", + "add", + "--transport", + "http", + "--scope", + "local", + "assemblyai-docs", + "https://mcp.assemblyai.com/docs", + ] in fake.calls + + +def test_install_invalid_scope_exits_2(monkeypatch): + _all_tools_present(monkeypatch) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", FakeRun()) + result = runner.invoke(app, ["setup", "install", "--scope", "bogus"]) + assert result.exit_code == 2 + + +def test_install_idempotent_when_mcp_present(monkeypatch): + _all_tools_present(monkeypatch) + # `mcp get` returns 0 -> already registered. + fake = FakeRun({("claude", "mcp", "get"): 0}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 + assert _statuses(result)["mcp"] == "already" + # No `mcp add` should have run. + assert not any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) + + +def test_install_failure_exits_nonzero(monkeypatch): + _all_tools_present(monkeypatch) + # mcp not present, but `mcp add` fails. + fake = FakeRun({("claude", "mcp", "get"): 1, ("claude", "mcp", "add"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 1 + assert _statuses(result)["mcp"] == "failed" + + +def test_install_force_remove_failure_reports_failed(monkeypatch): + _all_tools_present(monkeypatch) + # present, but the forced remove fails + fake = FakeRun({("claude", "mcp", "get"): 0, ("claude", "mcp", "remove"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--force"]) + assert result.exit_code == 1 + assert _statuses(result)["mcp"] == "failed" + assert not any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) + + +def test_install_force_removes_then_adds(monkeypatch): + _all_tools_present(monkeypatch) + fake = FakeRun({("claude", "mcp", "get"): 0}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--force"]) + assert result.exit_code == 0 + assert ["claude", "mcp", "remove", "assemblyai-docs"] in fake.calls + assert any(c[:3] == ["claude", "mcp", "add"] for c in fake.calls) + + +def test_install_skips_mcp_when_claude_missing(monkeypatch): + monkeypatch.setattr( + "aai_cli.commands.setup.shutil.which", + lambda tool: None if tool == "claude" else f"/usr/bin/{tool}", + ) + fake = FakeRun() + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 # skip is not a failure + statuses = _statuses(result) + assert statuses["mcp"] == "skipped" + assert statuses["skill"] == "installed" + # The bundled aai-cli skill installs regardless of claude/npx. + assert statuses["aai-cli skill"] == "installed" + assert not any(c[0] == "claude" for c in fake.calls) + + +# --- assemblyai skill (npx-based) -------------------------------------------- + + +def test_install_skill_idempotent_when_present(monkeypatch): + # Regression: a repeat install must report the skill as `already` (like MCP), + # not re-run `npx skills add` and claim `installed` every time. + _all_tools_present(monkeypatch) + skill = _skill_path() + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text("# AssemblyAI") + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 + assert _statuses(result)["skill"] == "already" + # No `npx … add` should have run — the skill was already present. + assert not any(c[0] == "npx" and "add" in c for c in fake.calls) + + +def test_install_force_reinstalls_skill(monkeypatch): + # --force must re-run `npx skills add` even when the skill is already present. + _all_tools_present(monkeypatch) + skill = _skill_path() + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text("# AssemblyAI") + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--force"]) + assert result.exit_code == 0 + assert _statuses(result)["skill"] == "installed" + assert [ + "npx", + "-y", + "skills", + "add", + "AssemblyAI/assemblyai-skill", + "--global", + "--yes", + ] in fake.calls + + +def test_install_skips_skill_when_npx_missing(monkeypatch): + monkeypatch.setattr( + "aai_cli.commands.setup.shutil.which", + lambda tool: None if tool == "npx" else f"/usr/bin/{tool}", + ) + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 + statuses = _statuses(result) + assert statuses["skill"] == "skipped" + assert statuses["mcp"] == "installed" + # aai-cli skill copies in regardless (no npx needed). + assert statuses["aai-cli skill"] == "installed" + assert not any(c[0] == "npx" for c in fake.calls) + + +# --- aai-cli skill (bundled, copied) ----------------------------------------- + + +def test_install_aai_cli_skill_idempotent_when_present(monkeypatch): + _all_tools_present(monkeypatch) + cli_skill = _cli_skill_path() + cli_skill.mkdir(parents=True) + (cli_skill / "SKILL.md").write_text("# old") + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install"]) + assert result.exit_code == 0 + assert _statuses(result)["aai-cli skill"] == "already" + # Not overwritten without --force. + assert (cli_skill / "SKILL.md").read_text() == "# old" + + +def test_install_aai_cli_skill_force_reinstalls(monkeypatch): + _all_tools_present(monkeypatch) + cli_skill = _cli_skill_path() + cli_skill.mkdir(parents=True) + (cli_skill / "SKILL.md").write_text("# old") + fake = FakeRun({("claude", "mcp", "get"): 1}) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", fake) + + result = runner.invoke(app, ["setup", "install", "--force"]) + assert result.exit_code == 0 + assert _statuses(result)["aai-cli skill"] == "installed" + # Overwritten with the bundled copy (references/ exist; placeholder gone). + assert (cli_skill / "references").is_dir() + assert "# old" not in (cli_skill / "SKILL.md").read_text() + + +# --- status ------------------------------------------------------------------ + + +def test_status_reports_all_installed(monkeypatch, tmp_path): + _all_tools_present(monkeypatch) + for name in ("assemblyai", "aai-cli"): + d = tmp_path / ".claude" / "skills" / name + d.mkdir(parents=True) + (d / "SKILL.md").write_text("# x") + # `mcp get` returns 0 -> present. + monkeypatch.setattr( + "aai_cli.commands.setup.subprocess.run", + FakeRun({("claude", "mcp", "get"): 0}), + ) + + result = runner.invoke(app, ["setup", "status"]) + assert result.exit_code == 0 + assert _statuses(result) == { + "mcp": "installed", + "skill": "installed", + "aai-cli skill": "installed", + } + + +def test_status_reports_not_installed(monkeypatch): + _all_tools_present(monkeypatch) # no skill dirs created + monkeypatch.setattr( + "aai_cli.commands.setup.subprocess.run", + FakeRun({("claude", "mcp", "get"): 1}), + ) + + result = runner.invoke(app, ["setup", "status"]) + assert result.exit_code == 0 + assert _statuses(result) == { + "mcp": "not_installed", + "skill": "not_installed", + "aai-cli skill": "not_installed", + } + + +def test_status_mcp_unknown_when_claude_missing(monkeypatch): + monkeypatch.setattr( + "aai_cli.commands.setup.shutil.which", + lambda tool: None if tool == "claude" else f"/usr/bin/{tool}", + ) + monkeypatch.setattr("aai_cli.commands.setup.subprocess.run", FakeRun()) + + result = runner.invoke(app, ["setup", "status"]) + assert result.exit_code == 0 + assert _statuses(result)["mcp"] == "unknown" diff --git a/tests/test_stream_command.py b/tests/test_stream_command.py index a14e8cf7..fbc06ef9 100644 --- a/tests/test_stream_command.py +++ b/tests/test_stream_command.py @@ -1,7 +1,6 @@ import json import time import types -from collections.abc import Callable from typer.testing import CliRunner @@ -29,68 +28,6 @@ def _login_result(): ) -def test_stream_session_listening_notice_latches(monkeypatch): - # _listening_once must announce "Listening…" exactly once even if the first-audio - # callback fires repeatedly (pins the `self._listening_started = True` latch). - import io - - from aai_cli.commands.stream import _StreamSession - from aai_cli.streaming.render import StreamRenderer - - renderer = StreamRenderer(json_mode=False, out=io.StringIO()) - calls = {"n": 0} - monkeypatch.setattr(renderer, "listening", lambda: calls.__setitem__("n", calls["n"] + 1)) - session = _StreamSession( - api_key="sk", - base_flags={}, - overrides=None, - config_file=None, - renderer=renderer, - follow=None, - llm_prompts=[], - model="m", - max_tokens=1, - ) - session._listening_once() - session._listening_once() - assert calls["n"] == 1 - - -def test_stream_session_closes_renderer_on_error(monkeypatch): - # When streaming raises mid-run, the live region must still be torn down (pins the - # `if self.follow is None: self.renderer.close()` in the finally block). - import io - - import pytest - - from aai_cli.commands.stream import _StreamSession - from aai_cli.errors import CLIError - from aai_cli.streaming.render import StreamRenderer - - renderer = StreamRenderer(json_mode=False, out=io.StringIO()) - closed = {"n": 0} - monkeypatch.setattr(renderer, "close", lambda: closed.__setitem__("n", closed["n"] + 1)) - - def boom(*_args, **_kwargs): - raise CLIError("stream blew up") - - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", boom) - session = _StreamSession( - api_key="sk", - base_flags={}, - overrides=None, - config_file=None, - renderer=renderer, - follow=None, - llm_prompts=[], - model="m", - max_tokens=1, - ) - with pytest.raises(CLIError): - session.run([b"\x00"], 16000) - assert closed["n"] >= 1 - - def test_stream_help_lists_command(): result = runner.invoke(app, ["stream", "--help"]) assert result.exit_code == 0 @@ -322,102 +259,6 @@ def fake( assert {"type": "termination", "audio_duration_seconds": 2.0} in lines -def test_stream_llm_refreshes_live_over_growing_transcript(monkeypatch): - config.set_api_key("default", "sk_live") - seen = {"texts": []} - - def fake(api_key, source, *, params, on_turn=None, **kwargs): - if on_turn: - on_turn(types.SimpleNamespace(transcript="hola", end_of_turn=True)) - on_turn(types.SimpleNamespace(transcript="mundo", end_of_turn=True)) - on_turn(types.SimpleNamespace(transcript="partial", end_of_turn=False)) # ignored - on_turn(types.SimpleNamespace(transcript="no-eot")) # missing flag -> not final - - def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): - seen["texts"].append(transcript_text) - seen["prompts"] = prompts - seen["model"] = model - seen["max_tokens"] = max_tokens - return f"answer:{transcript_text}" - - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) - monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) - result = runner.invoke( - app, - [ - "stream", - "--llm", - "translate to english", - "--model", - "gpt-4.1", - "--max-tokens", - "50", - "--json", - ], - ) - assert result.exit_code == 0 - # One refresh per finalized turn, over the growing transcript (partials ignored). - assert seen["texts"] == ["hola", "hola mundo"] - assert seen["prompts"] == ["translate to english"] - assert seen["model"] == "gpt-4.1" - assert seen["max_tokens"] == 50 - lines = [json.loads(x) for x in result.output.splitlines() if x.strip()] - assert {"turns": 1, "output": "answer:hola"} in lines - assert {"turns": 2, "output": "answer:hola mundo"} in lines - # Live mode replaces the raw turn envelopes; only follow refreshes reach stdout. - assert '"type"' not in result.output - - -def test_stream_llm_chains_multiple_prompts(monkeypatch): - config.set_api_key("default", "sk_live") - seen = {} - - def fake(api_key, source, *, params, on_turn=None, **kwargs): - if on_turn: - on_turn(types.SimpleNamespace(transcript="hi", end_of_turn=True)) - - def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): - seen["prompts"] = prompts - return "done" - - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) - monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) - result = runner.invoke( - app, ["stream", "--llm", "summarize", "--llm", "translate to french", "--json"] - ) - assert result.exit_code == 0 - assert seen["prompts"] == ["summarize", "translate to french"] - - -def test_stream_llm_rejects_output_text(monkeypatch): - config.set_api_key("default", "sk_live") - monkeypatch.setattr( - "aai_cli.commands.stream.client.stream_audio", - lambda *a, **k: (_ for _ in ()).throw(AssertionError("must not stream")), - ) - result = runner.invoke(app, ["stream", "--llm", "summarize", "-o", "text"]) - assert result.exit_code == 2 # --llm renders a panel/NDJSON; -o text is contradictory - - -def test_stream_without_prompt_does_not_transform(monkeypatch): - config.set_api_key("default", "sk_live") - called = {"ran": False} - - def fake(api_key, source, *, params, on_turn=None, **kwargs): - if on_turn: - on_turn(types.SimpleNamespace(transcript="hi", end_of_turn=True)) - - def fake_run_chain(*a, **k): - called["ran"] = True - return "x" - - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) - monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) - result = runner.invoke(app, ["stream", "--json"]) - assert result.exit_code == 0 - assert called["ran"] is False # no --llm -> no gateway call - - def test_stream_prompt_biases_speech_model(monkeypatch): config.set_api_key("default", "sk_live") seen = {} @@ -582,164 +423,6 @@ def test_stream_stdin_rejects_device(monkeypatch): assert result.exit_code == 2 # --device applies only to the microphone -def test_stream_system_audio_uses_macos_source(monkeypatch): - config.set_api_key("default", "sk_live") - source_types: list[str] = [] - rates: list[int] = [] - mic_target_rate: list[int | None] = [None] - system_on_open: list[Callable[[], None] | None] = [None] - mic_on_open: list[Callable[[], None] | None] = [None] - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - system_on_open[0] = on_open - self.sample_rate = 16000 - - def __iter__(self): - if system_on_open[0] is not None: - system_on_open[0]() - return iter([b"system"]) - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - mic_target_rate[0] = target_rate - mic_on_open[0] = on_open - self.sample_rate = 16000 - - def __iter__(self): - if mic_on_open[0] is not None: - mic_on_open[0]() - return iter([b"mic"]) - - def fake_stream_audio(api_key, source, *, params, on_begin=None, on_turn=None, **_kwargs): - source_type = type(source).__name__ - source_types.append(source_type) - rates.append(params.sample_rate) - if on_begin: - on_begin(types.SimpleNamespace(id=source_type)) - list(source) - if on_turn: - on_turn(types.SimpleNamespace(transcript=source_type, end_of_turn=True)) - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) - result = runner.invoke(app, ["stream", "--system-audio", "--json"]) - assert result.exit_code == 0 - assert set(source_types) == {"FakeSystemAudio", "FakeMic"} - assert rates == [16000, 16000] - assert mic_target_rate[0] == 16000 - lines = [json.loads(x) for x in result.output.splitlines() if x.strip()] - assert { - "type": "turn", - "transcript": "FakeSystemAudio", - "end_of_turn": True, - "source": "system", - } in lines - assert {"type": "turn", "transcript": "FakeMic", "end_of_turn": True, "source": "you"} in lines - - -def test_stream_system_audio_only_disables_mic(monkeypatch): - config.set_api_key("default", "sk_live") - seen = {} - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - def __iter__(self): - return iter([b"\x00\x00"]) - - def fail_mic(**_kwargs): - raise AssertionError("system-audio-only must not open the microphone") - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", fail_mic) - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", _capture_source(seen)) - result = runner.invoke(app, ["stream", "--system-audio-only", "--json"]) - assert result.exit_code == 0 - assert type(seen["source"]).__name__ == "FakeSystemAudio" - - -def test_stream_system_audio_rejects_other_sources(): - config.set_api_key("default", "sk_live") - result = runner.invoke(app, ["stream", "--system-audio", "--sample"]) - assert result.exit_code == 2 - assert "cannot be combined" in result.output - - -def test_stream_system_audio_forwards_mic_device_flags(monkeypatch): - config.set_api_key("default", "sk_live") - seen = {} - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - def __iter__(self): - return iter([b"system"]) - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - seen["target_rate"] = target_rate - seen["device"] = device - seen["capture_rate"] = capture_rate - self.sample_rate = target_rate - - def __iter__(self): - return iter([b"mic"]) - - def fake_stream_audio(api_key, source, *, params, **_kwargs): - list(source) - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) - result = runner.invoke( - app, - ["stream", "--system-audio", "--device", "2", "--sample-rate", "44100", "--json"], - ) - assert result.exit_code == 0 - assert seen == {"target_rate": 16000, "device": 2, "capture_rate": 44100} - - -def test_stream_system_audio_llm_prefixes_sources(monkeypatch): - config.set_api_key("default", "sk_live") - transcript_inputs = [] - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - def __iter__(self): - return iter([b"system"]) - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - self.sample_rate = target_rate - - def __iter__(self): - return iter([b"mic"]) - - def fake_stream_audio(api_key, source, *, params, on_turn=None, **_kwargs): - if on_turn: - on_turn(types.SimpleNamespace(transcript="", end_of_turn=True)) - on_turn(types.SimpleNamespace(transcript=type(source).__name__, end_of_turn=True)) - - def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): - transcript_inputs.append(transcript_text) - return "summary" - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) - monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) - result = runner.invoke(app, ["stream", "--system-audio", "--llm", "summarize", "--json"]) - assert result.exit_code == 0 - assert any("System: FakeSystemAudio" in value for value in transcript_inputs) - assert any("You: FakeMic" in value for value in transcript_inputs) - - def test_stream_system_audio_parallel_worker_error_surfaces(monkeypatch): config.set_api_key("default", "sk_live") @@ -770,125 +453,6 @@ def fake_stream_audio(api_key, source, *, params, **_kwargs): assert "mic failed" in result.output -def test_stream_system_audio_parallel_final_worker_error_surfaces(monkeypatch): - config.set_api_key("default", "sk_live") - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - def __iter__(self): - return iter([b"system"]) - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - self.sample_rate = target_rate - - def __iter__(self): - return iter([b"mic"]) - - class ImmediateThread: - def __init__(self, *, target, args, daemon): - self._target = target - self._args = args - - def start(self): - self._target(*self._args) - - def is_alive(self): - return False - - def join(self, timeout=None): - return None - - def fake_stream_audio(api_key, source, *, params, **_kwargs): - raise APIError(f"{type(source).__name__} failed") - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) - monkeypatch.setattr("aai_cli.commands.stream.threading.Thread", ImmediateThread) - result = runner.invoke(app, ["stream", "--system-audio", "--json"]) - assert result.exit_code == 1 - assert "failed" in result.output - - -def test_stream_system_audio_parallel_keyboard_interrupt_exits_cleanly(monkeypatch): - config.set_api_key("default", "sk_live") - monkeypatch.setattr("aai_cli.output.resolve_json", lambda *, explicit: False) - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - self.sample_rate = target_rate - - class InterruptingThread: - def __init__(self, *, target, args, daemon): - pass - - def start(self): - raise KeyboardInterrupt - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.threading.Thread", InterruptingThread) - result = runner.invoke(app, ["stream", "--system-audio"]) - assert result.exit_code == 0 - assert "Stopped." in result.output - - -def test_stream_system_audio_parallel_broken_pipe_exits_zero(monkeypatch): - config.set_api_key("default", "sk_live") - - class FakeSystemAudio: - def __init__(self, *, on_open=None): - self.sample_rate = 16000 - - class FakeMic: - def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): - self.sample_rate = target_rate - - class BrokenPipeThread: - def __init__(self, *, target, args, daemon): - pass - - def start(self): - raise BrokenPipeError - - monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) - monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) - monkeypatch.setattr("aai_cli.commands.stream.threading.Thread", BrokenPipeThread) - result = runner.invoke(app, ["stream", "--system-audio"]) - assert result.exit_code == 0 - - -def test_stream_system_audio_only_rejects_mic_device_flags(): - config.set_api_key("default", "sk_live") - result = runner.invoke(app, ["stream", "--system-audio-only", "--device", "2"]) - assert result.exit_code == 2 - assert "--device" in result.output - - result = runner.invoke(app, ["stream", "--system-audio-only", "--sample-rate", "44100"]) - assert result.exit_code == 2 - assert "--sample-rate" in result.output - - -def test_stream_system_audio_rejects_both_modes(): - config.set_api_key("default", "sk_live") - result = runner.invoke(app, ["stream", "--system-audio", "--system-audio-only"]) - assert result.exit_code == 2 - assert "either --system-audio" in result.output - - -def test_stream_show_code_rejects_system_audio(): - result = runner.invoke(app, ["stream", "--system-audio", "--show-code"]) - assert result.exit_code == 2 - assert "--show-code" in result.output - - def test_stream_output_text_emits_plain_finalized_turns(monkeypatch): # `-o text` -> only finalized transcripts as plain stdout lines (pipe into aai llm). config.set_api_key("default", "sk_live") @@ -904,15 +468,3 @@ def fake_stream_audio(api_key, source, *, params, on_begin=None, on_turn=None, * # Final turn only, plain text; partials and JSON envelopes are not on stdout. assert result.output.strip() == "hello world" assert '"type"' not in result.output - - -def test_stream_show_code_with_llm_emits_follow_loop(monkeypatch): - def _boom(*a, **k): - raise AssertionError("must not stream") - - monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", _boom) - result = runner.invoke(app, ["stream", "--llm", "summarize", "--show-code"]) - assert result.exit_code == 0 - assert "from openai import OpenAI" in result.output - assert "summarize" in result.output - assert "run_chain" in result.output # the live transcribe->LLM-per-turn loop diff --git a/tests/test_stream_llm.py b/tests/test_stream_llm.py new file mode 100644 index 00000000..3e8058fa --- /dev/null +++ b/tests/test_stream_llm.py @@ -0,0 +1,117 @@ +import json +import types + +from typer.testing import CliRunner + +from aai_cli import config +from aai_cli.main import app + +runner = CliRunner() + + +def test_stream_llm_refreshes_live_over_growing_transcript(monkeypatch): + config.set_api_key("default", "sk_live") + seen = {"texts": []} + + def fake(api_key, source, *, params, on_turn=None, **kwargs): + if on_turn: + on_turn(types.SimpleNamespace(transcript="hola", end_of_turn=True)) + on_turn(types.SimpleNamespace(transcript="mundo", end_of_turn=True)) + on_turn(types.SimpleNamespace(transcript="partial", end_of_turn=False)) # ignored + on_turn(types.SimpleNamespace(transcript="no-eot")) # missing flag -> not final + + def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): + seen["texts"].append(transcript_text) + seen["prompts"] = prompts + seen["model"] = model + seen["max_tokens"] = max_tokens + return f"answer:{transcript_text}" + + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) + monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) + result = runner.invoke( + app, + [ + "stream", + "--llm", + "translate to english", + "--model", + "gpt-4.1", + "--max-tokens", + "50", + "--json", + ], + ) + assert result.exit_code == 0 + # One refresh per finalized turn, over the growing transcript (partials ignored). + assert seen["texts"] == ["hola", "hola mundo"] + assert seen["prompts"] == ["translate to english"] + assert seen["model"] == "gpt-4.1" + assert seen["max_tokens"] == 50 + lines = [json.loads(x) for x in result.output.splitlines() if x.strip()] + assert {"turns": 1, "output": "answer:hola"} in lines + assert {"turns": 2, "output": "answer:hola mundo"} in lines + # Live mode replaces the raw turn envelopes; only follow refreshes reach stdout. + assert '"type"' not in result.output + + +def test_stream_llm_chains_multiple_prompts(monkeypatch): + config.set_api_key("default", "sk_live") + seen = {} + + def fake(api_key, source, *, params, on_turn=None, **kwargs): + if on_turn: + on_turn(types.SimpleNamespace(transcript="hi", end_of_turn=True)) + + def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): + seen["prompts"] = prompts + return "done" + + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) + monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) + result = runner.invoke( + app, ["stream", "--llm", "summarize", "--llm", "translate to french", "--json"] + ) + assert result.exit_code == 0 + assert seen["prompts"] == ["summarize", "translate to french"] + + +def test_stream_llm_rejects_output_text(monkeypatch): + config.set_api_key("default", "sk_live") + monkeypatch.setattr( + "aai_cli.commands.stream.client.stream_audio", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("must not stream")), + ) + result = runner.invoke(app, ["stream", "--llm", "summarize", "-o", "text"]) + assert result.exit_code == 2 # --llm renders a panel/NDJSON; -o text is contradictory + + +def test_stream_without_prompt_does_not_transform(monkeypatch): + config.set_api_key("default", "sk_live") + called = {"ran": False} + + def fake(api_key, source, *, params, on_turn=None, **kwargs): + if on_turn: + on_turn(types.SimpleNamespace(transcript="hi", end_of_turn=True)) + + def fake_run_chain(*a, **k): + called["ran"] = True + return "x" + + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake) + monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) + result = runner.invoke(app, ["stream", "--json"]) + assert result.exit_code == 0 + assert called["ran"] is False # no --llm -> no gateway call + + +def test_stream_show_code_with_llm_emits_follow_loop(monkeypatch): + def _boom(*a, **k): + raise AssertionError("must not stream") + + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", _boom) + result = runner.invoke(app, ["stream", "--llm", "summarize", "--show-code"]) + assert result.exit_code == 0 + assert "from openai import OpenAI" in result.output + assert "summarize" in result.output + assert "run_chain" in result.output # the live transcribe->LLM-per-turn loop diff --git a/tests/test_stream_session.py b/tests/test_stream_session.py new file mode 100644 index 00000000..6b043c98 --- /dev/null +++ b/tests/test_stream_session.py @@ -0,0 +1,360 @@ +import json +import types +from collections.abc import Callable + +from typer.testing import CliRunner + +from aai_cli import config +from aai_cli.errors import APIError +from aai_cli.main import app + +runner = CliRunner() + + +def _capture_source(seen): + def fake( + api_key, source, *, params, on_begin=None, on_turn=None, on_termination=None, **_kwargs + ): + seen["source"] = source + seen["rate"] = params.sample_rate + + return fake + + +def test_stream_session_listening_notice_latches(monkeypatch): + # _listening_once must announce "Listening…" exactly once even if the first-audio + # callback fires repeatedly (pins the `self._listening_started = True` latch). + import io + + from aai_cli.commands.stream import StreamSession + from aai_cli.streaming.render import StreamRenderer + + renderer = StreamRenderer(json_mode=False, out=io.StringIO()) + calls = {"n": 0} + monkeypatch.setattr(renderer, "listening", lambda: calls.__setitem__("n", calls["n"] + 1)) + session = StreamSession( + api_key="sk", + base_flags={}, + overrides=None, + config_file=None, + renderer=renderer, + follow=None, + llm_prompts=[], + model="m", + max_tokens=1, + ) + session._listening_once() + session._listening_once() + assert calls["n"] == 1 + + +def test_stream_session_closes_renderer_on_error(monkeypatch): + # When streaming raises mid-run, the live region must still be torn down (pins the + # `if self.follow is None: self.renderer.close()` in the finally block). + import io + + import pytest + + from aai_cli.commands.stream import StreamSession + from aai_cli.errors import CLIError + from aai_cli.streaming.render import StreamRenderer + + renderer = StreamRenderer(json_mode=False, out=io.StringIO()) + closed = {"n": 0} + monkeypatch.setattr(renderer, "close", lambda: closed.__setitem__("n", closed["n"] + 1)) + + def boom(*_args, **_kwargs): + raise CLIError("stream blew up") + + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", boom) + session = StreamSession( + api_key="sk", + base_flags={}, + overrides=None, + config_file=None, + renderer=renderer, + follow=None, + llm_prompts=[], + model="m", + max_tokens=1, + ) + with pytest.raises(CLIError): + session.run([b"\x00"], 16000) + assert closed["n"] >= 1 + + +def test_stream_system_audio_uses_macos_source(monkeypatch): + config.set_api_key("default", "sk_live") + source_types: list[str] = [] + rates: list[int] = [] + mic_target_rate: list[int | None] = [None] + system_on_open: list[Callable[[], None] | None] = [None] + mic_on_open: list[Callable[[], None] | None] = [None] + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + system_on_open[0] = on_open + self.sample_rate = 16000 + + def __iter__(self): + if system_on_open[0] is not None: + system_on_open[0]() + return iter([b"system"]) + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + mic_target_rate[0] = target_rate + mic_on_open[0] = on_open + self.sample_rate = 16000 + + def __iter__(self): + if mic_on_open[0] is not None: + mic_on_open[0]() + return iter([b"mic"]) + + def fake_stream_audio(api_key, source, *, params, on_begin=None, on_turn=None, **_kwargs): + source_type = type(source).__name__ + source_types.append(source_type) + rates.append(params.sample_rate) + if on_begin: + on_begin(types.SimpleNamespace(id=source_type)) + list(source) + if on_turn: + on_turn(types.SimpleNamespace(transcript=source_type, end_of_turn=True)) + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) + result = runner.invoke(app, ["stream", "--system-audio", "--json"]) + assert result.exit_code == 0 + assert set(source_types) == {"FakeSystemAudio", "FakeMic"} + assert rates == [16000, 16000] + assert mic_target_rate[0] == 16000 + lines = [json.loads(x) for x in result.output.splitlines() if x.strip()] + assert { + "type": "turn", + "transcript": "FakeSystemAudio", + "end_of_turn": True, + "source": "system", + } in lines + assert {"type": "turn", "transcript": "FakeMic", "end_of_turn": True, "source": "you"} in lines + + +def test_stream_system_audio_only_disables_mic(monkeypatch): + config.set_api_key("default", "sk_live") + seen = {} + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + def __iter__(self): + return iter([b"\x00\x00"]) + + def fail_mic(**_kwargs): + raise AssertionError("system-audio-only must not open the microphone") + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", fail_mic) + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", _capture_source(seen)) + result = runner.invoke(app, ["stream", "--system-audio-only", "--json"]) + assert result.exit_code == 0 + assert type(seen["source"]).__name__ == "FakeSystemAudio" + + +def test_stream_system_audio_rejects_other_sources(): + config.set_api_key("default", "sk_live") + result = runner.invoke(app, ["stream", "--system-audio", "--sample"]) + assert result.exit_code == 2 + assert "cannot be combined" in result.output + + +def test_stream_system_audio_forwards_mic_device_flags(monkeypatch): + config.set_api_key("default", "sk_live") + seen = {} + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + def __iter__(self): + return iter([b"system"]) + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + seen["target_rate"] = target_rate + seen["device"] = device + seen["capture_rate"] = capture_rate + self.sample_rate = target_rate + + def __iter__(self): + return iter([b"mic"]) + + def fake_stream_audio(api_key, source, *, params, **_kwargs): + list(source) + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) + result = runner.invoke( + app, + ["stream", "--system-audio", "--device", "2", "--sample-rate", "44100", "--json"], + ) + assert result.exit_code == 0 + assert seen == {"target_rate": 16000, "device": 2, "capture_rate": 44100} + + +def test_stream_system_audio_llm_prefixes_sources(monkeypatch): + config.set_api_key("default", "sk_live") + transcript_inputs = [] + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + def __iter__(self): + return iter([b"system"]) + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + self.sample_rate = target_rate + + def __iter__(self): + return iter([b"mic"]) + + def fake_stream_audio(api_key, source, *, params, on_turn=None, **_kwargs): + if on_turn: + on_turn(types.SimpleNamespace(transcript="", end_of_turn=True)) + on_turn(types.SimpleNamespace(transcript=type(source).__name__, end_of_turn=True)) + + def fake_run_chain(api_key, prompts, *, transcript_text, model, max_tokens): + transcript_inputs.append(transcript_text) + return "summary" + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) + monkeypatch.setattr("aai_cli.commands.stream.llm.run_chain", fake_run_chain) + result = runner.invoke(app, ["stream", "--system-audio", "--llm", "summarize", "--json"]) + assert result.exit_code == 0 + assert any("System: FakeSystemAudio" in value for value in transcript_inputs) + assert any("You: FakeMic" in value for value in transcript_inputs) + + +def test_stream_system_audio_parallel_final_worker_error_surfaces(monkeypatch): + config.set_api_key("default", "sk_live") + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + def __iter__(self): + return iter([b"system"]) + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + self.sample_rate = target_rate + + def __iter__(self): + return iter([b"mic"]) + + class ImmediateThread: + def __init__(self, *, target, args, daemon): + self._target = target + self._args = args + + def start(self): + self._target(*self._args) + + def is_alive(self): + return False + + def join(self, timeout=None): + return None + + def fake_stream_audio(api_key, source, *, params, **_kwargs): + raise APIError(f"{type(source).__name__} failed") + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.commands.stream.client.stream_audio", fake_stream_audio) + monkeypatch.setattr("aai_cli.streaming.session.threading.Thread", ImmediateThread) + result = runner.invoke(app, ["stream", "--system-audio", "--json"]) + assert result.exit_code == 1 + assert "failed" in result.output + + +def test_stream_system_audio_parallel_keyboard_interrupt_exits_cleanly(monkeypatch): + config.set_api_key("default", "sk_live") + monkeypatch.setattr("aai_cli.output.resolve_json", lambda *, explicit: False) + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + self.sample_rate = target_rate + + class InterruptingThread: + def __init__(self, *, target, args, daemon): + pass + + def start(self): + raise KeyboardInterrupt + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.streaming.session.threading.Thread", InterruptingThread) + result = runner.invoke(app, ["stream", "--system-audio"]) + assert result.exit_code == 0 + assert "Stopped." in result.output + + +def test_stream_system_audio_parallel_broken_pipe_exits_zero(monkeypatch): + config.set_api_key("default", "sk_live") + + class FakeSystemAudio: + def __init__(self, *, on_open=None): + self.sample_rate = 16000 + + class FakeMic: + def __init__(self, *, target_rate=None, device=None, capture_rate=None, on_open=None): + self.sample_rate = target_rate + + class BrokenPipeThread: + def __init__(self, *, target, args, daemon): + pass + + def start(self): + raise BrokenPipeError + + monkeypatch.setattr("aai_cli.commands.stream.MacSystemAudioSource", FakeSystemAudio) + monkeypatch.setattr("aai_cli.commands.stream.MicrophoneSource", FakeMic) + monkeypatch.setattr("aai_cli.streaming.session.threading.Thread", BrokenPipeThread) + result = runner.invoke(app, ["stream", "--system-audio"]) + assert result.exit_code == 0 + + +def test_stream_system_audio_only_rejects_mic_device_flags(): + config.set_api_key("default", "sk_live") + result = runner.invoke(app, ["stream", "--system-audio-only", "--device", "2"]) + assert result.exit_code == 2 + assert "--device" in result.output + + result = runner.invoke(app, ["stream", "--system-audio-only", "--sample-rate", "44100"]) + assert result.exit_code == 2 + assert "--sample-rate" in result.output + + +def test_stream_system_audio_rejects_both_modes(): + config.set_api_key("default", "sk_live") + result = runner.invoke(app, ["stream", "--system-audio", "--system-audio-only"]) + assert result.exit_code == 2 + assert "either --system-audio" in result.output + + +def test_stream_show_code_rejects_system_audio(): + result = runner.invoke(app, ["stream", "--system-audio", "--show-code"]) + assert result.exit_code == 2 + assert "--show-code" in result.output From e2dba3eaf572ba6d734758094e248f37401d89b3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 00:09:20 +0000 Subject: [PATCH 2/2] Make path-validation tests width-independent Typer renders option errors in a bordered Rich panel; at narrow terminal widths the message wraps and the box borders split "does not exist" across lines, so the plain substring check failed depending on terminal width. Flatten the panel (drop borders, collapse whitespace) before matching. --- tests/test_path_validation.py | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/tests/test_path_validation.py b/tests/test_path_validation.py index a4f077fe..7d84b3ee 100644 --- a/tests/test_path_validation.py +++ b/tests/test_path_validation.py @@ -7,6 +7,8 @@ from __future__ import annotations +import re + from typer.testing import CliRunner from aai_cli.main import app @@ -14,12 +16,23 @@ runner = CliRunner() +def _flatten(output: str) -> str: + """Collapse a Rich error panel into one line for width-independent matching. + + Typer renders option errors in a bordered panel; at narrow terminal widths the + message wraps and the box borders ("│") split a phrase like "does not exist" + across lines. Drop the borders and collapse whitespace so the substring check + doesn't depend on the sandbox's terminal width. + """ + return re.sub(r"\s+", " ", output.replace("│", " ")) + + def test_transcribe_config_file_must_exist(tmp_path): result = runner.invoke( app, ["transcribe", "x.mp3", "--config-file", str(tmp_path / "nope.json")] ) assert result.exit_code == 2 - assert "does not exist" in result.output + assert "does not exist" in _flatten(result.output) def test_transcribe_custom_spelling_file_must_exist(tmp_path): @@ -27,16 +40,16 @@ def test_transcribe_custom_spelling_file_must_exist(tmp_path): app, ["transcribe", "x.mp3", "--custom-spelling-file", str(tmp_path / "nope.json")] ) assert result.exit_code == 2 - assert "does not exist" in result.output + assert "does not exist" in _flatten(result.output) def test_stream_config_file_must_exist(tmp_path): result = runner.invoke(app, ["stream", "--config-file", str(tmp_path / "nope.json")]) assert result.exit_code == 2 - assert "does not exist" in result.output + assert "does not exist" in _flatten(result.output) def test_agent_system_prompt_file_must_exist(tmp_path): result = runner.invoke(app, ["agent", "--system-prompt-file", str(tmp_path / "nope.txt")]) assert result.exit_code == 2 - assert "does not exist" in result.output + assert "does not exist" in _flatten(result.output)