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