diff --git a/aai_cli/commands/transcribe.py b/aai_cli/commands/transcribe.py index 239a50d4..43501f38 100644 --- a/aai_cli/commands/transcribe.py +++ b/aai_cli/commands/transcribe.py @@ -4,7 +4,6 @@ import assemblyai as aai import typer -from rich.markup import escape from aai_cli import ( choices, @@ -17,14 +16,12 @@ options, output, transcribe_exec, - transcribe_render, ) # The package attribute `code_gen.transcribe` is the wrapper function, so the module's # render() (which also takes the -o output field) is imported from the submodule itself. from aai_cli.code_gen.transcribe import render as render_transcribe_code from aai_cli.context import AppState, run_command -from aai_cli.errors import UsageError from aai_cli.help_text import examples_epilog app = typer.Typer() @@ -398,12 +395,7 @@ def body(state: AppState, json_mode: bool) -> None: } flags.update(config_builder.auth_header_flags(webhook_auth_header)) - if out is not None and llm_prompt: - # --out captures the transcript itself; an LLM transform is a separate step. - raise UsageError( - "--out can't be combined with --llm.", - suggestion='Pipe the transform instead, e.g. -o text | assembly llm -f "…".', - ) + transcribe_exec.validate_out_with_llm(out, llm_prompt) merged = config_builder.merge_transcribe_config( flags=flags, overrides=config_kv, config_file=config_file @@ -447,43 +439,16 @@ def body(state: AppState, json_mode: bool) -> None: download_sections=list(download_sections or []), ) - if out is not None: - # Write a clean file artifact and confirm on stderr; stdout stays empty. - if ".." in out.parts: # reject path-traversal segments in --out - raise UsageError(f"--out path can't contain '..': {out}") - out.write_text( - transcribe_exec.out_payload(transcript, output_field, json_mode=json_mode) + "\n" - ) - if not state.quiet: - output.error_console.print(output.success(f"Saved to {escape(str(out))}")) - return - - if output_field is not None: - # Raw single-field output for pipelines (overrides --json and analysis render). - output.emit_text(client.select_transcript_field(transcript, output_field)) - return - - if llm_prompt: - # Chain the prompts: the first runs over the transcript (injected server-side - # via transcript_id); each subsequent prompt runs over the prior response. - steps = llm.run_chain_steps( - api_key, - llm_prompt, - transcript_id=transcript.id, - model=model, - max_tokens=max_tokens, - ) - output.emit( - client.transcript_summary(transcript) - | {"transform": {"model": model, "steps": steps}}, - transcribe_exec.render_transform_steps, - json_mode=json_mode, - ) - return - - if json_mode: - output.emit(client.transcript_json_payload(transcript), lambda d: d, json_mode=True) - else: - transcribe_render.render_transcript_result(transcript, output.console) + transcribe_exec.deliver_result( + transcript, + api_key=api_key, + out=out, + output_field=output_field, + transform=transcribe_exec.TransformOptions( + prompts=list(llm_prompt or []), model=model, max_tokens=max_tokens + ), + json_mode=json_mode, + quiet=state.quiet, + ) run_command(ctx, body, json=json_out) diff --git a/aai_cli/transcribe_exec.py b/aai_cli/transcribe_exec.py index 03c3a5a3..800b8998 100644 --- a/aai_cli/transcribe_exec.py +++ b/aai_cli/transcribe_exec.py @@ -1,4 +1,4 @@ -"""Transcription execution and non-Rich output shaping for the ``transcribe`` command. +"""Transcription execution and result delivery for the ``transcribe`` command. Kept out of ``commands/transcribe.py`` so the command stays a thin option surface, and so ``run_transcription`` lives in a core module that ``onboard`` can import directly @@ -10,11 +10,12 @@ import json import tempfile from pathlib import Path -from typing import Any +from typing import Any, NamedTuple import assemblyai as aai +from rich.markup import escape -from aai_cli import choices, client, stdio, youtube +from aai_cli import choices, client, llm, output, stdio, transcribe_render, youtube from aai_cli.errors import UsageError # The PII policy strings the SDK accepts, validated client-side so a typo'd @@ -47,6 +48,15 @@ def validate_speakers_expected(merged: dict[str, object]) -> None: ) +def validate_out_with_llm(out: Path | None, llm_prompts: list[str] | None) -> None: + if out is not None and llm_prompts: + # --out captures the transcript itself; an LLM transform is a separate step. + raise UsageError( + "--out can't be combined with --llm.", + suggestion='Pipe the transform instead, e.g. -o text | assembly llm -f "…".', + ) + + def render_transform_steps(d: dict[str, Any]) -> str: """Human view of chained LLM-Gateway steps: the lone output, or each step labeled.""" steps = d["transform"]["steps"] @@ -105,3 +115,61 @@ def run_transcription( local = youtube.download_audio(audio, Path(td), download_sections=download_sections) return client.transcribe(api_key, str(local), config=transcription_config) return client.transcribe(api_key, audio, config=transcription_config) + + +class TransformOptions(NamedTuple): + """The ``--llm`` chain options: the prompts plus the gateway model settings.""" + + prompts: list[str] + model: str + max_tokens: int + + +def deliver_result( + transcript: aai.Transcript, + *, + api_key: str, + out: Path | None, + output_field: choices.TranscriptOutput | None, + transform: TransformOptions, + json_mode: bool, + quiet: bool, +) -> None: + """Route the finished transcript: ``--out`` file, single ``-o`` field, ``--llm`` + transform chain, or the default JSON/human render — first match wins.""" + if out is not None: + # Write a clean file artifact and confirm on stderr; stdout stays empty. + if ".." in out.parts: # reject path-traversal segments in --out + raise UsageError(f"--out path can't contain '..': {out}") + out.write_text(out_payload(transcript, output_field, json_mode=json_mode) + "\n") + if not quiet: + output.error_console.print(output.success(f"Saved to {escape(str(out))}")) + return + + if output_field is not None: + # Raw single-field output for pipelines (overrides --json and analysis render). + output.emit_text(client.select_transcript_field(transcript, output_field)) + return + + if transform.prompts: + # Chain the prompts: the first runs over the transcript (injected server-side + # via transcript_id); each subsequent prompt runs over the prior response. + steps = llm.run_chain_steps( + api_key, + transform.prompts, + transcript_id=transcript.id, + model=transform.model, + max_tokens=transform.max_tokens, + ) + output.emit( + client.transcript_summary(transcript) + | {"transform": {"model": transform.model, "steps": steps}}, + render_transform_steps, + json_mode=json_mode, + ) + return + + if json_mode: + output.emit(client.transcript_json_payload(transcript), lambda d: d, json_mode=True) + else: + transcribe_render.render_transcript_result(transcript, output.console) diff --git a/tests/test_code_gen.py b/tests/test_code_gen.py index fce952c4..c97e3a11 100644 --- a/tests/test_code_gen.py +++ b/tests/test_code_gen.py @@ -1,6 +1,7 @@ -"""Example-based code_gen tests: serializers, snippets, rendered scripts. +"""Example-based code_gen tests: serializers, snippets, transcribe rendering. -Hypothesis fuzz/property tests live in test_code_gen_fuzz.py. +Stream/agent scaffold tests live in test_code_gen_stream_agent.py; hypothesis +fuzz/property tests live in test_code_gen_fuzz.py. """ from __future__ import annotations @@ -182,55 +183,6 @@ def test_transcribe_render_plain_url_is_not_downloaded(): assert "transcriber.transcribe('https://assembly.ai/wildfires.mp3')" in code -def test_stream_render_parses_and_is_runnable_shape(): - from assemblyai.streaming.v3 import SpeechModel - - code = code_gen.stream( - {"sample_rate": 16000, "format_turns": True, "speech_model": SpeechModel.u3_rt_pro} - ) - ast.parse(code) - assert "StreamingClient(" in code - assert "StreamingParameters(" in code - assert "SpeechModel.u3_rt_pro" in code - assert "MicrophoneStream" in code - assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code - - -def test_stream_render_mic_rate_matches_params(): - code = code_gen.stream({"sample_rate": 8000}) - ast.parse(code) - assert "StreamingParameters(\n sample_rate=8000," in code - assert "MicrophoneStream(sample_rate=8000)" in code - - -def test_stream_render_empty_is_clean_and_has_no_speechmodel_import(): - code = code_gen.stream({}) - ast.parse(code) - assert "StreamingParameters()" in code - assert " SpeechModel," not in code # not imported when unused (keeps script lint-clean) - assert "MicrophoneStream(sample_rate=16000)" in code # default rate - - -def test_agent_render_parses_and_injects_session_fields(): - code = code_gen.agent(voice="ivy", system_prompt="Be terse.", greeting="Hi there") - ast.parse(code) - assert '"voice": "ivy"' in code - assert "Be terse." in code - assert "Hi there" in code - assert "agents.assemblyai.com" in code - assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code - - -def test_agent_render_escapes_quotes_in_prompt(): - import json as _json - - tricky = 'Say "hi"\nand stop' - code = code_gen.agent(voice="ivy", system_prompt=tricky, greeting="Hello") - ast.parse(code) # valid Python despite embedded quotes/newlines - # The prompt is injected via json.dumps, so its escaped form appears verbatim. - assert _json.dumps(tricky) in code - - # --------------------------------------------------------------------------- # Validity & fidelity checks (the exhaustive hypothesis harness — Task 10 — # lives in test_code_gen_fuzz.py). @@ -378,77 +330,6 @@ def test_transcribe_show_code_without_gateway_has_no_openai_import(): assert "transcript.utterances" in code # normal result handling instead -def test_agent_show_code_uses_single_full_duplex_stream(): - # ONE sd.RawStream (mic+speaker); two separate streams fail on macOS CoreAudio. - code = code_gen.agent(voice="ivy", system_prompt="p", greeting="g") - ast.parse(code) - assert "sd.RawStream(" in code - assert "samplerate=RATE" in code # opens at the agent's native 24 kHz, no resampling - assert "RawInputStream" not in code - assert "RawOutputStream" not in code - # No audioop: it's deprecated and removed in Python 3.13, so the script stays portable. - assert "audioop" not in code - - -def test_stream_show_code_includes_llm_follow_loop(): - code = code_gen.stream( - {"speech_model": "universal_streaming"}, - llm={ - "prompts": ["summarize", "translate to french"], - "model": "claude-haiku-4-5-20251001", - "max_tokens": 500, - "interval": 30.0, - }, - ) - ast.parse(code) - assert "from openai import OpenAI" in code - assert "llm-gateway.assemblyai.com" in code - # Both prompts appear, in order, for the chain. - assert code.index("summarize") < code.index("translate to french") - # Still streams from the mic, refreshing the answer on the interval. - assert "MicrophoneStream" in code - assert "end_of_turn" in code - assert "claude-haiku-4-5-20251001" in code - # The generated loop mirrors --llm-interval: a baked-in throttle plus a closing flush. - assert "LLM_INTERVAL = 30.0" in code - assert "now - _last_summary < LLM_INTERVAL" in code - assert "summarize(final=True)" in code - - -def test_gateway_options_defaults_interval_to_per_turn(): - # Called without an explicit interval (transcribe's path), the baked-in cadence is - # per-turn (0.0); pins the default so it can't drift. - opts = code_gen.gateway_options(["summarize"], "m", 100) - assert opts is not None - assert opts["interval"] == 0.0 - - -def test_stream_show_code_defaults_interval_when_absent(): - # An llm dict with no "interval" key falls back to per-turn (LLM_INTERVAL = 0.0). - code = code_gen.stream({}, llm={"prompts": ["s"], "model": "m", "max_tokens": 1}) - ast.parse(code) - assert "LLM_INTERVAL = 0.0" in code - - -def test_stream_show_code_llm_interval_zero_is_per_turn(): - # --llm-interval 0 bakes in the legacy per-turn cadence (LLM_INTERVAL = 0.0 makes the - # throttle a no-op), while still emitting the closing flush. - code = code_gen.stream( - {}, - llm=code_gen.gateway_options(["summarize"], "m", 100, interval=0.0), - ) - ast.parse(code) - assert "LLM_INTERVAL = 0.0" in code - assert "summarize(final=True)" in code - - -def test_stream_show_code_without_llm_is_plain_scaffold(): - code = code_gen.stream({}) - ast.parse(code) - assert "from openai import OpenAI" not in code # no gateway when --llm absent - assert "MicrophoneStream" in code - - def test_generated_code_targets_active_environment(): # --show-code embeds hosts from the active environment, so a sandbox user's # generated script talks to the sandbox that minted their key, not production. diff --git a/tests/test_code_gen_stream_agent.py b/tests/test_code_gen_stream_agent.py new file mode 100644 index 00000000..d93b94f9 --- /dev/null +++ b/tests/test_code_gen_stream_agent.py @@ -0,0 +1,131 @@ +"""Example-based code_gen tests for the stream and agent scaffolds. + +Split from test_code_gen.py (serializers, snippets, transcribe rendering); +hypothesis fuzz/property tests live in test_code_gen_fuzz.py. +""" + +from __future__ import annotations + +import ast + +from aai_cli import code_gen + + +def test_stream_render_parses_and_is_runnable_shape(): + from assemblyai.streaming.v3 import SpeechModel + + code = code_gen.stream( + {"sample_rate": 16000, "format_turns": True, "speech_model": SpeechModel.u3_rt_pro} + ) + ast.parse(code) + assert "StreamingClient(" in code + assert "StreamingParameters(" in code + assert "SpeechModel.u3_rt_pro" in code + assert "MicrophoneStream" in code + assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code + + +def test_stream_render_mic_rate_matches_params(): + code = code_gen.stream({"sample_rate": 8000}) + ast.parse(code) + assert "StreamingParameters(\n sample_rate=8000," in code + assert "MicrophoneStream(sample_rate=8000)" in code + + +def test_stream_render_empty_is_clean_and_has_no_speechmodel_import(): + code = code_gen.stream({}) + ast.parse(code) + assert "StreamingParameters()" in code + assert " SpeechModel," not in code # not imported when unused (keeps script lint-clean) + assert "MicrophoneStream(sample_rate=16000)" in code # default rate + + +def test_agent_render_parses_and_injects_session_fields(): + code = code_gen.agent(voice="ivy", system_prompt="Be terse.", greeting="Hi there") + ast.parse(code) + assert '"voice": "ivy"' in code + assert "Be terse." in code + assert "Hi there" in code + assert "agents.assemblyai.com" in code + assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code + + +def test_agent_render_escapes_quotes_in_prompt(): + import json as _json + + tricky = 'Say "hi"\nand stop' + code = code_gen.agent(voice="ivy", system_prompt=tricky, greeting="Hello") + ast.parse(code) # valid Python despite embedded quotes/newlines + # The prompt is injected via json.dumps, so its escaped form appears verbatim. + assert _json.dumps(tricky) in code + + +def test_agent_show_code_uses_single_full_duplex_stream(): + # ONE sd.RawStream (mic+speaker); two separate streams fail on macOS CoreAudio. + code = code_gen.agent(voice="ivy", system_prompt="p", greeting="g") + ast.parse(code) + assert "sd.RawStream(" in code + assert "samplerate=RATE" in code # opens at the agent's native 24 kHz, no resampling + assert "RawInputStream" not in code + assert "RawOutputStream" not in code + # No audioop: it's deprecated and removed in Python 3.13, so the script stays portable. + assert "audioop" not in code + + +def test_stream_show_code_includes_llm_follow_loop(): + code = code_gen.stream( + {"speech_model": "universal_streaming"}, + llm={ + "prompts": ["summarize", "translate to french"], + "model": "claude-haiku-4-5-20251001", + "max_tokens": 500, + "interval": 30.0, + }, + ) + ast.parse(code) + assert "from openai import OpenAI" in code + assert "llm-gateway.assemblyai.com" in code + # Both prompts appear, in order, for the chain. + assert code.index("summarize") < code.index("translate to french") + # Still streams from the mic, refreshing the answer on the interval. + assert "MicrophoneStream" in code + assert "end_of_turn" in code + assert "claude-haiku-4-5-20251001" in code + # The generated loop mirrors --llm-interval: a baked-in throttle plus a closing flush. + assert "LLM_INTERVAL = 30.0" in code + assert "now - _last_summary < LLM_INTERVAL" in code + assert "summarize(final=True)" in code + + +def test_gateway_options_defaults_interval_to_per_turn(): + # Called without an explicit interval (transcribe's path), the baked-in cadence is + # per-turn (0.0); pins the default so it can't drift. + opts = code_gen.gateway_options(["summarize"], "m", 100) + assert opts is not None + assert opts["interval"] == 0.0 + + +def test_stream_show_code_defaults_interval_when_absent(): + # An llm dict with no "interval" key falls back to per-turn (LLM_INTERVAL = 0.0). + code = code_gen.stream({}, llm={"prompts": ["s"], "model": "m", "max_tokens": 1}) + ast.parse(code) + assert "LLM_INTERVAL = 0.0" in code + + +def test_stream_show_code_llm_interval_zero_is_per_turn(): + # --llm-interval 0 bakes in the legacy per-turn cadence (LLM_INTERVAL = 0.0 makes the + # throttle a no-op), while still emitting the closing flush. + code = code_gen.stream( + {}, + llm=code_gen.gateway_options(["summarize"], "m", 100, interval=0.0), + ) + ast.parse(code) + assert "LLM_INTERVAL = 0.0" in code + assert "summarize(final=True)" in code + + +def test_stream_show_code_without_llm_is_plain_scaffold(): + code = code_gen.stream({}) + ast.parse(code) + assert "from openai import OpenAI" not in code # no gateway when --llm absent + assert "MicrophoneStream" in code diff --git a/tests/test_stream_command.py b/tests/test_stream_command.py index cfea68ef..1c003e08 100644 --- a/tests/test_stream_command.py +++ b/tests/test_stream_command.py @@ -1,7 +1,8 @@ -"""`assembly stream` source/streaming behavior and --show-code tests. +"""`assembly stream` source/streaming behavior. Flag-to-params mapping and conflicting-flag validation live in -test_stream_command_flags.py. +test_stream_command_flags.py; --show-code print-only behavior lives in +test_stream_show_code.py. """ import json @@ -308,104 +309,6 @@ def fake_stream(api_key, source, *, params, **kwargs): assert seen["src"] == str(fake) -def test_stream_show_code_prints_without_streaming(monkeypatch): - # Print-only: emits the mic-streaming script, never opens audio or streams, no auth. - called = [] - monkeypatch.setattr( - "aai_cli.commands.stream.client.stream_audio", - lambda *a, **k: called.append(True), - ) - result = runner.invoke(app, ["stream", "--show-code"]) - assert result.exit_code == 0 - assert called == [] # never streamed - assert "StreamingClient(" in result.output - assert "MicrophoneStream(sample_rate=16000)" in result.output - assert 'os.environ["ASSEMBLYAI_API_KEY"]' in result.output - - -def test_stream_show_code_file_source_streams_that_file(): - # A file source must generate file-streaming code, not silently emit mic code. - # The file need not exist: generating code for it is legitimate (check_local=False). - result = runner.invoke(app, ["stream", "rec.wav", "--show-code"]) - assert result.exit_code == 0 - assert "client.stream(file_chunks())" in result.output - assert "'rec.wav'" in result.output # the ffmpeg input is the file passed - assert "MicrophoneStream" not in result.output - compile(result.output, "", "exec") # the printed script is runnable - - -def test_stream_show_code_stdin_source_reads_stdin(): - result = runner.invoke(app, ["stream", "-", "--show-code"]) - assert result.exit_code == 0 - assert "client.stream(stdin_chunks())" in result.output - assert "sys.stdin.buffer" in result.output - assert "MicrophoneStream" not in result.output - compile(result.output, "", "exec") - - -def test_stream_show_code_sample_streams_hosted_clip(): - result = runner.invoke(app, ["stream", "--sample", "--show-code"]) - assert result.exit_code == 0 - assert "wildfires.mp3" in result.output - assert "file_chunks()" in result.output - assert "MicrophoneStream" not in result.output - - -def test_stream_show_code_honors_sample_rate_flag(): - result = runner.invoke(app, ["stream", "--sample-rate", "8000", "--show-code"]) - assert result.exit_code == 0 - assert "MicrophoneStream(sample_rate=8000)" in result.output - assert "sample_rate=8000," in result.output # params match the capture rate - - -def test_stream_show_code_honors_config_sample_rate(): - # An explicit `--config sample_rate=…` must not be overridden by the 16 kHz default. - result = runner.invoke(app, ["stream", "--config", "sample_rate=8000", "--show-code"]) - assert result.exit_code == 0 - assert "MicrophoneStream(sample_rate=8000)" in result.output - assert "sample_rate=8000," in result.output - - -def test_stream_show_code_sample_rate_flag_beats_config(): - result = runner.invoke( - app, ["stream", "--sample-rate", "8000", "--config", "sample_rate=44100", "--show-code"] - ) - assert result.exit_code == 0 - assert "MicrophoneStream(sample_rate=8000)" in result.output - assert "44100" not in result.output - - -def test_stream_show_code_file_with_mic_flags_rejected(): - # --show-code applies the same source validation as a real run, so the - # file + --sample-rate conflict errors instead of generating mic code. - result = runner.invoke(app, ["stream", "rec.wav", "--sample-rate", "8000", "--show-code"]) - assert result.exit_code == 2 - assert "--sample-rate" in result.output - - -def test_stream_show_code_rejects_downloaded_sources(): - # YouTube and podcast-page URLs both route through the download path, which the - # generated streaming script doesn't model yet. - for url in ("https://youtu.be/abc", "https://www.spreaker.com/episode/12345"): - result = runner.invoke(app, ["stream", url, "--show-code"]) - assert result.exit_code == 2 - assert "downloaded sources" in result.output - assert "YouTube" in result.output - - -def test_stream_show_code_ignores_json_flag(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", "--show-code", "--json"]) - assert result.exit_code == 0 - assert "StreamingClient(" in result.output - - def test_stream_reads_raw_pcm_from_stdin(monkeypatch): config.set_api_key("default", "sk_live") seen = {} diff --git a/tests/test_stream_show_code.py b/tests/test_stream_show_code.py new file mode 100644 index 00000000..c2f2f675 --- /dev/null +++ b/tests/test_stream_show_code.py @@ -0,0 +1,109 @@ +"""`assembly stream --show-code` print-only behavior. + +Split from test_stream_command.py (source/streaming behavior); flag-to-params +mapping and conflicting-flag validation live in test_stream_command_flags.py. +""" + +from typer.testing import CliRunner + +from aai_cli.main import app + +runner = CliRunner() + + +def test_stream_show_code_prints_without_streaming(monkeypatch): + # Print-only: emits the mic-streaming script, never opens audio or streams, no auth. + called = [] + monkeypatch.setattr( + "aai_cli.commands.stream.client.stream_audio", + lambda *a, **k: called.append(True), + ) + result = runner.invoke(app, ["stream", "--show-code"]) + assert result.exit_code == 0 + assert called == [] # never streamed + assert "StreamingClient(" in result.output + assert "MicrophoneStream(sample_rate=16000)" in result.output + assert 'os.environ["ASSEMBLYAI_API_KEY"]' in result.output + + +def test_stream_show_code_file_source_streams_that_file(): + # A file source must generate file-streaming code, not silently emit mic code. + # The file need not exist: generating code for it is legitimate (check_local=False). + result = runner.invoke(app, ["stream", "rec.wav", "--show-code"]) + assert result.exit_code == 0 + assert "client.stream(file_chunks())" in result.output + assert "'rec.wav'" in result.output # the ffmpeg input is the file passed + assert "MicrophoneStream" not in result.output + compile(result.output, "", "exec") # the printed script is runnable + + +def test_stream_show_code_stdin_source_reads_stdin(): + result = runner.invoke(app, ["stream", "-", "--show-code"]) + assert result.exit_code == 0 + assert "client.stream(stdin_chunks())" in result.output + assert "sys.stdin.buffer" in result.output + assert "MicrophoneStream" not in result.output + compile(result.output, "", "exec") + + +def test_stream_show_code_sample_streams_hosted_clip(): + result = runner.invoke(app, ["stream", "--sample", "--show-code"]) + assert result.exit_code == 0 + assert "wildfires.mp3" in result.output + assert "file_chunks()" in result.output + assert "MicrophoneStream" not in result.output + + +def test_stream_show_code_honors_sample_rate_flag(): + result = runner.invoke(app, ["stream", "--sample-rate", "8000", "--show-code"]) + assert result.exit_code == 0 + assert "MicrophoneStream(sample_rate=8000)" in result.output + assert "sample_rate=8000," in result.output # params match the capture rate + + +def test_stream_show_code_honors_config_sample_rate(): + # An explicit `--config sample_rate=…` must not be overridden by the 16 kHz default. + result = runner.invoke(app, ["stream", "--config", "sample_rate=8000", "--show-code"]) + assert result.exit_code == 0 + assert "MicrophoneStream(sample_rate=8000)" in result.output + assert "sample_rate=8000," in result.output + + +def test_stream_show_code_sample_rate_flag_beats_config(): + result = runner.invoke( + app, ["stream", "--sample-rate", "8000", "--config", "sample_rate=44100", "--show-code"] + ) + assert result.exit_code == 0 + assert "MicrophoneStream(sample_rate=8000)" in result.output + assert "44100" not in result.output + + +def test_stream_show_code_file_with_mic_flags_rejected(): + # --show-code applies the same source validation as a real run, so the + # file + --sample-rate conflict errors instead of generating mic code. + result = runner.invoke(app, ["stream", "rec.wav", "--sample-rate", "8000", "--show-code"]) + assert result.exit_code == 2 + assert "--sample-rate" in result.output + + +def test_stream_show_code_rejects_downloaded_sources(): + # YouTube and podcast-page URLs both route through the download path, which the + # generated streaming script doesn't model yet. + for url in ("https://youtu.be/abc", "https://www.spreaker.com/episode/12345"): + result = runner.invoke(app, ["stream", url, "--show-code"]) + assert result.exit_code == 2 + assert "downloaded sources" in result.output + assert "YouTube" in result.output + + +def test_stream_show_code_ignores_json_flag(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", "--show-code", "--json"]) + assert result.exit_code == 0 + assert "StreamingClient(" in result.output diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index 4d5596f1..dcd00251 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -1,6 +1,7 @@ -"""`assembly transcribe` behavior: output rendering, LLM transforms, sources, --show-code. +"""`assembly transcribe` behavior: output rendering, LLM transforms, sources. -Flag-to-config mapping and flag validation live in test_transcribe_flags.py. +Flag-to-config mapping and flag validation live in test_transcribe_flags.py; +--show-code print-only behavior lives in test_transcribe_show_code.py. """ import json @@ -362,108 +363,6 @@ def _no_download(url, d, *, download_sections=None): assert tx.call_args.args[1] == "https://example.com/episode.mp3" -def test_transcribe_show_code_prints_without_transcribing(monkeypatch): - # Print-only: emits code, never calls the API, needs no auth. - called = [] - monkeypatch.setattr( - "aai_cli.commands.transcribe.client.transcribe", - lambda *a, **k: called.append(True), - ) - result = runner.invoke(app, ["transcribe", "--sample", "--speaker-labels", "--show-code"]) - assert result.exit_code == 0 - assert called == [] # never transcribed - assert "import assemblyai as aai" in result.output - assert "TranscriptionConfig(" in result.output - assert 'os.environ["ASSEMBLYAI_API_KEY"]' in result.output - # --sample resolves to the hosted sample URL (not the no-source placeholder). - assert "wildfires" in result.output - assert "your-audio-file.mp3" not in result.output - - -def test_transcribe_show_code_includes_download_sections(monkeypatch): - # --show-code reflects --download-sections in the generated yt-dlp download block. - def _boom(*a, **k): - raise AssertionError("must not transcribe") - - monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) - result = runner.invoke( - app, - ["transcribe", "https://youtu.be/abc", "--download-sections", "*0:00-5:00", "--show-code"], - ) - assert result.exit_code == 0 - assert "download_range_func([], [(0.0, 300.0)], False)" in result.output - assert '"force_keyframes_at_cuts": True,' in result.output - - -def test_transcribe_show_code_without_source_uses_placeholder(monkeypatch): - # --show-code never transcribes, so it must not demand a source/--sample; it emits - # runnable code with a clearly-marked placeholder path instead of erroring. - def _boom(*a, **k): - raise AssertionError("must not transcribe") - - monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) - result = runner.invoke(app, ["transcribe", "--show-code"]) - assert result.exit_code == 0 - assert "import assemblyai as aai" in result.output - assert "your-audio-file.mp3" in result.output - - -def test_transcribe_show_code_ignores_json_flag(monkeypatch): - # --show-code is print-only; --json does not suppress or wrap it. - def _boom(*a, **k): - raise AssertionError("must not transcribe") - - monkeypatch.setattr( - "aai_cli.commands.transcribe.client.transcribe", - _boom, - ) - result = runner.invoke(app, ["transcribe", "--sample", "--show-code", "--json"]) - assert result.exit_code == 0 - assert "import assemblyai as aai" in result.output - - -def test_transcribe_show_code_includes_llm_gateway_without_running(monkeypatch): - # --llm must be reflected in the generated code, still without - # transcribing or calling the gateway. - def _boom(*a, **k): - raise AssertionError("must not call the API") - - monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) - monkeypatch.setattr("aai_cli.commands.transcribe.llm.transform_transcript", _boom) - result = runner.invoke( - app, - ["transcribe", "--sample", "--llm", "translate to spanish", "--show-code"], - ) - assert result.exit_code == 0 - assert "llm-gateway.assemblyai.com" in result.output - assert "translate to spanish" in result.output - assert '"transcript_id": transcript.id' in result.output - - -def test_transcribe_show_code_output_srt_generates_export(monkeypatch): - # -o srt must be reflected in the generated code (not silently dropped). - def _boom(*a, **k): - raise AssertionError("must not transcribe") - - monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) - result = runner.invoke(app, ["transcribe", "--sample", "-o", "srt", "--show-code"]) - assert result.exit_code == 0 - compile(result.output, "", "exec") # the emitted script is runnable - assert "print(transcript.export_subtitles_srt())" in result.output - assert "print(transcript.text)" not in result.output - - -def test_transcribe_show_code_output_utterances_generates_loop(monkeypatch): - def _boom(*a, **k): - raise AssertionError("must not transcribe") - - monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) - result = runner.invoke(app, ["transcribe", "--sample", "-o", "utterances", "--show-code"]) - assert result.exit_code == 0 - compile(result.output, "", "exec") - assert 'print(f"Speaker {utt.speaker}: {utt.text}")' in result.output - - def test_transcribe_renders_summary_human(monkeypatch, mocker): _auth() monkeypatch.setattr("aai_cli.output.resolve_json", lambda *, explicit: False) diff --git a/tests/test_transcribe_show_code.py b/tests/test_transcribe_show_code.py new file mode 100644 index 00000000..d28e1aaf --- /dev/null +++ b/tests/test_transcribe_show_code.py @@ -0,0 +1,113 @@ +"""`assembly transcribe --show-code` print-only behavior. + +Split from test_transcribe.py (output rendering, LLM transforms, sources); +flag-to-config mapping and flag validation live in test_transcribe_flags.py. +""" + +from typer.testing import CliRunner + +from aai_cli.main import app + +runner = CliRunner() + + +def test_transcribe_show_code_prints_without_transcribing(monkeypatch): + # Print-only: emits code, never calls the API, needs no auth. + called = [] + monkeypatch.setattr( + "aai_cli.commands.transcribe.client.transcribe", + lambda *a, **k: called.append(True), + ) + result = runner.invoke(app, ["transcribe", "--sample", "--speaker-labels", "--show-code"]) + assert result.exit_code == 0 + assert called == [] # never transcribed + assert "import assemblyai as aai" in result.output + assert "TranscriptionConfig(" in result.output + assert 'os.environ["ASSEMBLYAI_API_KEY"]' in result.output + # --sample resolves to the hosted sample URL (not the no-source placeholder). + assert "wildfires" in result.output + assert "your-audio-file.mp3" not in result.output + + +def test_transcribe_show_code_includes_download_sections(monkeypatch): + # --show-code reflects --download-sections in the generated yt-dlp download block. + def _boom(*a, **k): + raise AssertionError("must not transcribe") + + monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) + result = runner.invoke( + app, + ["transcribe", "https://youtu.be/abc", "--download-sections", "*0:00-5:00", "--show-code"], + ) + assert result.exit_code == 0 + assert "download_range_func([], [(0.0, 300.0)], False)" in result.output + assert '"force_keyframes_at_cuts": True,' in result.output + + +def test_transcribe_show_code_without_source_uses_placeholder(monkeypatch): + # --show-code never transcribes, so it must not demand a source/--sample; it emits + # runnable code with a clearly-marked placeholder path instead of erroring. + def _boom(*a, **k): + raise AssertionError("must not transcribe") + + monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) + result = runner.invoke(app, ["transcribe", "--show-code"]) + assert result.exit_code == 0 + assert "import assemblyai as aai" in result.output + assert "your-audio-file.mp3" in result.output + + +def test_transcribe_show_code_ignores_json_flag(monkeypatch): + # --show-code is print-only; --json does not suppress or wrap it. + def _boom(*a, **k): + raise AssertionError("must not transcribe") + + monkeypatch.setattr( + "aai_cli.commands.transcribe.client.transcribe", + _boom, + ) + result = runner.invoke(app, ["transcribe", "--sample", "--show-code", "--json"]) + assert result.exit_code == 0 + assert "import assemblyai as aai" in result.output + + +def test_transcribe_show_code_includes_llm_gateway_without_running(monkeypatch): + # --llm must be reflected in the generated code, still without + # transcribing or calling the gateway. + def _boom(*a, **k): + raise AssertionError("must not call the API") + + monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) + monkeypatch.setattr("aai_cli.commands.transcribe.llm.transform_transcript", _boom) + result = runner.invoke( + app, + ["transcribe", "--sample", "--llm", "translate to spanish", "--show-code"], + ) + assert result.exit_code == 0 + assert "llm-gateway.assemblyai.com" in result.output + assert "translate to spanish" in result.output + assert '"transcript_id": transcript.id' in result.output + + +def test_transcribe_show_code_output_srt_generates_export(monkeypatch): + # -o srt must be reflected in the generated code (not silently dropped). + def _boom(*a, **k): + raise AssertionError("must not transcribe") + + monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) + result = runner.invoke(app, ["transcribe", "--sample", "-o", "srt", "--show-code"]) + assert result.exit_code == 0 + compile(result.output, "", "exec") # the emitted script is runnable + assert "print(transcript.export_subtitles_srt())" in result.output + assert "print(transcript.text)" not in result.output + + +def test_transcribe_show_code_output_utterances_generates_loop(monkeypatch): + def _boom(*a, **k): + raise AssertionError("must not transcribe") + + monkeypatch.setattr("aai_cli.commands.transcribe.client.transcribe", _boom) + result = runner.invoke(app, ["transcribe", "--sample", "-o", "utterances", "--show-code"]) + assert result.exit_code == 0 + compile(result.output, "", "exec") + assert 'print(f"Speaker {utt.speaker}: {utt.text}")' in result.output