Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions tests/test_agent_render.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,25 @@ def flush(self):
AgentRenderer(json_mode=True, out=FlakyOut()).connected() # non-pipe errors are non-fatal


# --- text mode (plain stdout + stderr status) ------------------------------
def test_text_connected_status_goes_to_stderr():
out, err = io.StringIO(), io.StringIO()
r = AgentRenderer(json_mode=False, text_mode=True, out=out, err=err)
r.connected()
# The "start talking" prompt is a status notice -> stderr, keeping stdout clean
# for the piped transcript.
assert "start talking" in err.getvalue().lower()
assert out.getvalue() == ""


def test_text_notice_goes_to_stderr():
out, err = io.StringIO(), io.StringIO()
r = AgentRenderer(json_mode=False, text_mode=True, out=out, err=err)
r.notice("Half-duplex note.\n")
assert "Half-duplex note." in err.getvalue()
assert out.getvalue() == ""


# --- human mode (Rich) -----------------------------------------------------
def test_human_partial_then_final():
r, buf = _human()
Expand Down
27 changes: 27 additions & 0 deletions tests/test_auth_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,33 @@ def test_find_or_create_creates_when_existing_cli_token_disabled(monkeypatch):
assert flow.find_or_create_cli_key(1, "jwt") == "sk_fresh"


def test_find_or_create_raises_when_no_projects(monkeypatch):
# An account with zero projects can't hold a key; surface a clean APIError.
monkeypatch.setattr(flow.ams, "list_projects", lambda acct, jwt: [])
monkeypatch.setattr(flow.ams, "create_token", lambda *a, **k: pytest.fail("should not create"))
with pytest.raises(APIError) as exc:
flow.find_or_create_cli_key(1, "jwt")
assert "no project" in exc.value.message


def test_capture_delegates_to_loopback(monkeypatch):
sentinel = loopback.CallbackResult(token="tok", token_type="discovery_oauth")
monkeypatch.setattr(flow.loopback, "capture_callback", lambda: sentinel)
assert flow._capture() is sentinel


def test_run_login_flow_rejects_wrong_token_type(monkeypatch):
monkeypatch.setattr(flow, "_open_browser", lambda url: None)
monkeypatch.setattr(
flow,
"_capture",
lambda: loopback.CallbackResult(token="tok", token_type="something_else"),
)
with pytest.raises(APIError) as exc:
flow.run_login_flow()
assert "valid OAuth token" in exc.value.message


def test_run_login_flow_happy_path(monkeypatch):
opened = {}
monkeypatch.setattr(flow, "_open_browser", lambda url: opened.setdefault("url", url))
Expand Down
18 changes: 18 additions & 0 deletions tests/test_auth_loopback.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,24 @@ def run():
assert result.error is None


def test_capture_ignores_unknown_paths():
# A request to a non-callback path gets a 404 and the server keeps waiting; the
# real callback that follows still completes the capture.
result_box = {}

def run():
result_box["result"] = loopback.capture_callback(timeout=5.0)

t = threading.Thread(target=run)
t.start()
_hit("/favicon.ico") # unknown path -> 404, capture stays open
_hit("/callback?stytch_token_type=discovery_oauth&token=tok_late")
t.join(timeout=5)

result = result_box["result"]
assert result.token == "tok_late"


def test_capture_times_out_without_callback():
result = loopback.capture_callback(timeout=0.3)
assert result.error == "timeout"
Expand Down
129 changes: 128 additions & 1 deletion tests/test_init_command.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,25 @@
import io
import subprocess
import sys
import types

import pytest
import typer
from typer.testing import CliRunner

from aai_cli.commands import init as init_cmd
from aai_cli.errors import CLIError
from aai_cli.main import app

runner = CliRunner()
TEMPLATE = "audio-transcription"


class _Tty(io.StringIO):
def isatty(self) -> bool:
return True


def test_init_scaffold_only_creates_project(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["init", TEMPLATE, "myapp", "--no-install"])
Expand All @@ -24,7 +38,6 @@ def test_init_writes_key_from_env(tmp_path, monkeypatch):
def test_init_logged_out_installs_but_skips_launch_with_hint(tmp_path, monkeypatch):
# Logged out + install: deps install, but the server can't start without a key —
# the report must say so (and must not launch) rather than exiting silently.
import subprocess

monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
Expand Down Expand Up @@ -141,3 +154,117 @@ def test_init_unregistered_template_errors_cleanly(tmp_path, monkeypatch):
result = runner.invoke(app, ["init", "llm", "x", "--no-install"])
assert result.exit_code == 1
assert "llm" in result.output


def _fake_questionary(choice):
"""A minimal stand-in for the questionary module's select(...).ask() chain."""

class _Choice:
def __init__(self, title, value):
self.value = value

class _Select:
def ask(self):
return choice

return types.SimpleNamespace(Choice=_Choice, select=lambda *a, **k: _Select())


def test_pick_template_interactive_returns_choice(monkeypatch):
monkeypatch.setattr("sys.stdin", _Tty())
monkeypatch.setattr("sys.stdout", _Tty())
monkeypatch.setitem(sys.modules, "questionary", _fake_questionary(TEMPLATE))
assert init_cmd._pick_template() == TEMPLATE


def test_pick_template_ctrl_c_exits_130(monkeypatch):
# questionary returns None when the user presses Ctrl-C at the prompt.
monkeypatch.setattr("sys.stdin", _Tty())
monkeypatch.setattr("sys.stdout", _Tty())
monkeypatch.setitem(sys.modules, "questionary", _fake_questionary(None))
with pytest.raises(typer.Exit) as exc:
init_cmd._pick_template()
assert exc.value.exit_code == 130


def test_pick_template_missing_questionary_errors(monkeypatch):
# A broken/stale install missing the declared 'questionary' dep.
monkeypatch.setattr("sys.stdin", _Tty())
monkeypatch.setattr("sys.stdout", _Tty())
monkeypatch.setitem(sys.modules, "questionary", None) # makes `import questionary` raise
with pytest.raises(CLIError) as exc:
init_cmd._pick_template()
assert exc.value.error_type == "missing_dependency"


def test_init_install_failure_reports_and_exits(tmp_path, monkeypatch):
# A failing dependency install is reported and exits non-zero (no launch).
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
"aai_cli.init.runner.run_setup",
lambda *a, **k: subprocess.CompletedProcess([], 1, "", "pip exploded"),
)
launched = {"v": False}
monkeypatch.setattr(
"aai_cli.init.runner.launch_and_open",
lambda *a, **k: launched.__setitem__("v", True) or 0,
)
result = runner.invoke(app, ["init", TEMPLATE, "app", "--json"])
assert result.exit_code == 1
assert launched["v"] is False
assert "pip exploded" in result.output


def test_init_launches_when_key_present(tmp_path, monkeypatch):
# Key present + install succeeds -> the server is launched and the browser opens.
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("ASSEMBLYAI_API_KEY", "sk-real-key")
monkeypatch.setattr("aai_cli.output._is_agentic", lambda: False) # exercise human banner
monkeypatch.setattr(
"aai_cli.init.runner.run_setup",
lambda *a, **k: subprocess.CompletedProcess([], 0, "ok", ""),
)
monkeypatch.setattr("aai_cli.init.runner.find_free_port", lambda preferred: 4321)
captured = {}
monkeypatch.setattr(
"aai_cli.init.runner.launch_and_open",
lambda target, **k: captured.update(k) or 0,
)
result = runner.invoke(app, ["init", TEMPLATE, "app"])
assert result.exit_code == 0, result.output
assert captured["port"] == 4321
assert captured["open_browser"] is True
assert "Starting" in result.output
assert "4321" in result.output


def test_init_no_open_keeps_browser_closed(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("ASSEMBLYAI_API_KEY", "sk-real-key")
monkeypatch.setattr(
"aai_cli.init.runner.run_setup",
lambda *a, **k: subprocess.CompletedProcess([], 0, "ok", ""),
)
monkeypatch.setattr("aai_cli.init.runner.find_free_port", lambda preferred: 4321)
captured = {}
monkeypatch.setattr(
"aai_cli.init.runner.launch_and_open",
lambda target, **k: captured.update(k) or 0,
)
result = runner.invoke(app, ["init", TEMPLATE, "app", "--no-open"])
assert result.exit_code == 0, result.output
assert captured["open_browser"] is False


def test_init_launch_failure_propagates_exit_code(tmp_path, monkeypatch):
# A non-zero server exit code is surfaced as the command's exit code.
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("ASSEMBLYAI_API_KEY", "sk-real-key")
monkeypatch.setattr(
"aai_cli.init.runner.run_setup",
lambda *a, **k: subprocess.CompletedProcess([], 0, "ok", ""),
)
monkeypatch.setattr("aai_cli.init.runner.find_free_port", lambda preferred: 4321)
monkeypatch.setattr("aai_cli.init.runner.launch_and_open", lambda *a, **k: 7)
result = runner.invoke(app, ["init", TEMPLATE, "app"])
assert result.exit_code == 7
8 changes: 8 additions & 0 deletions tests/test_init_scaffold.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,14 @@ def test_scaffold_unknown_template_raises(tmp_path):
scaffold.scaffold("nope", tmp_path / "app", api_key=None)


def test_scaffold_registered_but_missing_files_raises(tmp_path, monkeypatch):
# Defense in depth: the registry lists a template whose on-disk dir is gone.
monkeypatch.setattr("aai_cli.init.templates.is_template", lambda _t: True)
with pytest.raises(CLIError) as exc:
scaffold.scaffold("ghost-template", tmp_path / "app", api_key=None)
assert exc.value.error_type == "template_missing"


def test_target_conflict_detects_nonempty_dir(tmp_path):
empty = tmp_path / "empty"
empty.mkdir()
Expand Down
25 changes: 25 additions & 0 deletions tests/test_init_template_transcribe.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,3 +185,28 @@ def test_status_surfaces_sdk_exception_as_502_not_500(monkeypatch):
client = TestClient(app)
resp = client.get("/api/status/t-x")
assert resp.status_code == 502


def test_submit_returns_502_when_no_id(monkeypatch):
# The SDK returned a transcript object without an id — surface a clean 502.
app, fake, _api = _load_app(monkeypatch)
fake.Transcriber.return_value.submit.return_value = MagicMock(id=None)
client = TestClient(app)
resp = client.post("/api/transcribe-url", json={"url": "https://example.com/a.mp3"})
assert resp.status_code == 502
assert "did not return an id" in resp.json()["detail"]


def test_ask_surfaces_gateway_error_as_502(monkeypatch):
# A gateway failure (e.g. no plan access) becomes a clean 502, not a 500 traceback.
fake_openai = MagicMock()
fake_openai.OpenAI.return_value.chat.completions.create.side_effect = RuntimeError(
"no plan access"
)
monkeypatch.setitem(sys.modules, "openai", fake_openai)

app, _aai, _api = _load_app(monkeypatch)
client = TestClient(app)
resp = client.post("/api/ask", json={"transcript_id": "t-1", "question": "what?"})
assert resp.status_code == 502
assert "LLM Gateway error" in resp.json()["detail"]
52 changes: 52 additions & 0 deletions tests/test_keys.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,3 +142,55 @@ def test_keys_rename_calls_ams():
result = runner.invoke(app, ["keys", "rename", "10", "prod"])
assert result.exit_code == 0
rename.assert_called_once_with(42, 10, "prod", "jwt")


def test_keys_list_renders_human_table():
# Human-readable (non-JSON) render path: the table includes names, project, and
# the disabled flag rendered as yes/no.
_auth()
projects = [
{
"project": {"id": 1, "name": "Default"},
"tokens": [
{"id": 10, "name": "ci", "api_key": "sk_abcdef1234", "is_disabled": False},
{"id": 11, "name": "old", "api_key": "sk_zzzzzz9999", "is_disabled": True},
],
}
]
with (
patch("aai_cli.output._is_agentic", return_value=False),
patch("aai_cli.commands.keys.ams.list_projects", return_value=projects),
):
result = runner.invoke(app, ["keys", "list"])
assert result.exit_code == 0
assert "ci" in result.output
assert "Default" in result.output
assert "yes" in result.output # the disabled key
assert "no" in result.output # the enabled key
assert "sk_abcdef1234" not in result.output # masked


def test_keys_create_rejects_empty_project_list():
# No projects at all -> clean APIError, create_token never called.
_auth()
with (
patch("aai_cli.commands.keys.ams.list_projects", return_value=[]),
patch("aai_cli.commands.keys.ams.create_token") as create,
):
result = runner.invoke(app, ["keys", "create", "--name", "ci"])
assert result.exit_code == 1
create.assert_not_called()


def test_keys_create_with_explicit_project_skips_lookup():
# Passing --project bypasses the default-project lookup entirely.
_auth()
created = {"id": 5, "name": "ci", "api_key": "sk_explicit999", "is_disabled": False}
with (
patch("aai_cli.commands.keys.ams.list_projects") as list_projects,
patch("aai_cli.commands.keys.ams.create_token", return_value=created) as create,
):
result = runner.invoke(app, ["keys", "create", "--name", "ci", "--project", "9"])
assert result.exit_code == 0
list_projects.assert_not_called()
create.assert_called_once_with(42, 9, "ci", "jwt")
19 changes: 19 additions & 0 deletions tests/test_llm_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -252,6 +252,25 @@ def test_llm_output_with_follow_is_rejected(monkeypatch):
assert "one-shot" in result.output


def test_llm_follow_requires_a_prompt(monkeypatch):
# --follow re-runs a prompt over each turn; with no prompt there's nothing to run.
_auth()
monkeypatch.setattr("aai_cli.commands.llm.gateway.complete", lambda *a, **k: _payload())
result = runner.invoke(app, ["llm", "--follow", "--json"], input="x\n")
assert result.exit_code == 2
assert "prompt" in result.output.lower()


def test_llm_follow_requires_piped_stdin(monkeypatch):
# Interactively (no pipe) --follow would block forever; reject it with guidance.
_auth()
monkeypatch.setattr("aai_cli.commands.llm.stdio.stdin_is_piped", lambda: False)
monkeypatch.setattr("aai_cli.commands.llm.gateway.complete", lambda *a, **k: _payload())
result = runner.invoke(app, ["llm", "summarize", "--follow", "--json"])
assert result.exit_code == 2
assert "stdin" in result.output.lower()


def test_llm_follow_stops_cleanly_on_interrupt(monkeypatch):
_auth()
calls = []
Expand Down
Loading
Loading