0bebad74ad
Slice 3a of issue #17 — the CLI surface of the bind trigger (TUI + web follow, INV-008 lockstep). ratatoskr can now self-drive a bound session from the CLI: - New flags: --bifrost-plane {memory,affect} (dev shortcut -> endpoint_for_plane over --bifrost-host / RATATOSKR_PROVIDER_VISIBLE_HOST) and --bifrost-url (the direct HTTPS/prod endpoint, bypassing the plane shortcut). Mutually exclusive; a binding is a session-CREATE concern (forbidden with --session). - Consumer key resolved from RATATOSKR_BIFROST_CONSUMER_KEY only (the privileged handshake identity — never a CLI flag, distinct from the canary WORLDTREE_API_KEY). - _amain threads bifrost + consumer_key into create_session and routes the bind failures: BifrostConsumerKeyMissing -> exit 22; BifrostHandshakeFailed -> exit 23 with the 401-scoping hint ("use the consumer key, not WORLDTREE_API_KEY") keyed on bifrost_error == bifrost.auth_rejected. - Bound-state indicator on success: ". bifrost: status=bound plane=... endpoint=..." — shows WHICH identity/endpoint bound, not a bare boolean. Also fixes a pre-existing test-isolation bug: test_no_textual_import did a live importlib.reload(ratatoskr.cli) that mutated the shared module in place, breaking class identity (isinstance / pytest.raises) for every test after it. The real check is the static source grep; the reload was vestigial and is removed. 9 new CLI bind tests; full suite 462 green; ruff clean (no new mypy errors).
1543 lines
63 KiB
Python
1543 lines
63 KiB
Python
"""Tests for ratatoskr.cli per docs/contracts/issues/3.contract.md."""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from ratatoskr import cli as cli_mod
|
|
from ratatoskr.cli import (
|
|
ParsedArgs,
|
|
UsageError,
|
|
_amain,
|
|
_AuthError,
|
|
_cancel_and_log,
|
|
_parse_args,
|
|
_run_turn,
|
|
main,
|
|
)
|
|
from ratatoskr.sessions import BifrostBinding
|
|
from ratatoskr.sse_client import (
|
|
Cancelled,
|
|
Done,
|
|
Error,
|
|
SseId,
|
|
Text,
|
|
TextBoundary,
|
|
Thinking,
|
|
ToolResult,
|
|
ToolStart,
|
|
WorkerPhase,
|
|
)
|
|
|
|
|
|
class _FlushCountingIO(io.StringIO):
|
|
"""StringIO subclass that counts flush() calls — used to verify INV-010."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.flush_count = 0
|
|
|
|
def flush(self) -> None:
|
|
self.flush_count += 1
|
|
super().flush()
|
|
|
|
|
|
def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
|
|
return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()
|
|
|
|
|
|
def _sse_resp(body: bytes | httpx.AsyncByteStream) -> httpx.Response:
|
|
"""Wrap an SSE response body (bytes or stream) with the right content-type."""
|
|
headers = {"content-type": "text/event-stream"}
|
|
if isinstance(body, bytes):
|
|
return httpx.Response(200, headers=headers, content=body)
|
|
return httpx.Response(200, headers=headers, stream=body)
|
|
|
|
|
|
_DONE_BODY = {
|
|
"type": "done",
|
|
"phase": "succeeded",
|
|
"response": "hello",
|
|
"model": "m",
|
|
"duration_ms": 1,
|
|
"usage": {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
}
|
|
_CANCELLED_BODY = {
|
|
"type": "cancelled",
|
|
"phase": "cancelled",
|
|
"turn_id": 42,
|
|
"reason": "user_cancel",
|
|
"partial_message_id": None,
|
|
}
|
|
_CANCEL_OK_RESP = {"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Tests assert env-resolution behavior; default to a clean slate per test."""
|
|
monkeypatch.delenv("WORLDTREE_API_KEY", raising=False)
|
|
monkeypatch.delenv("WORLDTREE_API_URL", raising=False)
|
|
monkeypatch.delenv("RATATOSKR_END_USER_ID", raising=False)
|
|
monkeypatch.delenv("RATATOSKR_BIFROST_CONSUMER_KEY", raising=False)
|
|
monkeypatch.delenv("RATATOSKR_PROVIDER_VISIBLE_HOST", raising=False)
|
|
|
|
|
|
class TestParseArgs:
|
|
def test_happy_new(self) -> None:
|
|
"""happy_new [happy,tracer]: --send --new --agent --api-key → full ParsedArgs."""
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"])
|
|
assert args == ParsedArgs(
|
|
send_content="hi",
|
|
session_id=None,
|
|
new=True,
|
|
agent_id="mimir",
|
|
api_key="k",
|
|
server_url="http://localhost:8000",
|
|
raw=False,
|
|
)
|
|
|
|
def test_happy_existing_session(self) -> None:
|
|
"""happy_existing_session: --send --session --api-key → ParsedArgs with session_id."""
|
|
args = _parse_args(["--send", "hi", "--session", "s-1", "--api-key", "k"])
|
|
assert args == ParsedArgs(
|
|
send_content="hi",
|
|
session_id="s-1",
|
|
new=False,
|
|
agent_id=None,
|
|
api_key="k",
|
|
server_url="http://localhost:8000",
|
|
raw=False,
|
|
)
|
|
|
|
def test_api_key_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""api_key_from_env: WORLDTREE_API_KEY env var fills in when --api-key omitted."""
|
|
monkeypatch.setenv("WORLDTREE_API_KEY", "from-env")
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "mimir"])
|
|
assert args.api_key == "from-env"
|
|
|
|
def test_api_key_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""api_key_flag_beats_env: explicit --api-key wins over WORLDTREE_API_KEY."""
|
|
monkeypatch.setenv("WORLDTREE_API_KEY", "env")
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "flag"])
|
|
assert args.api_key == "flag"
|
|
|
|
def test_server_default(self) -> None:
|
|
"""server_default: no --server, no WORLDTREE_API_URL → http://localhost:8000."""
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert args.server_url == "http://localhost:8000"
|
|
|
|
def test_server_env_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""server_env_fallback: WORLDTREE_API_URL fills in when --server omitted."""
|
|
monkeypatch.setenv("WORLDTREE_API_URL", "http://t.local:9000")
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert args.server_url == "http://t.local:9000"
|
|
|
|
def test_server_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""server_flag_beats_env: explicit --server wins over WORLDTREE_API_URL."""
|
|
monkeypatch.setenv("WORLDTREE_API_URL", "env")
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--server", "flag"]
|
|
)
|
|
assert args.server_url == "flag"
|
|
|
|
def test_no_send_marks_tui_mode(self) -> None:
|
|
"""no_send_marks_tui_mode: missing --send → send_content=None (TUI marker)."""
|
|
args = _parse_args(["--new", "--agent", "mimir", "--api-key", "k"])
|
|
assert args.send_content is None
|
|
# Other fields still populate normally
|
|
assert args.new is True
|
|
assert args.agent_id == "mimir"
|
|
|
|
def test_raw_flag_default_false(self) -> None:
|
|
"""raw_flag_default_false: --raw absent → ParsedArgs.raw == False."""
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert args.raw is False
|
|
|
|
def test_raw_flag_set(self) -> None:
|
|
"""raw_flag_set: --raw → ParsedArgs.raw == True."""
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--raw"])
|
|
assert args.raw is True
|
|
|
|
def test_usage_both_session_and_new(self) -> None:
|
|
"""usage_both_session_and_new: --session AND --new → UsageError('mutually exclusive')."""
|
|
with pytest.raises(UsageError, match="mutually exclusive"):
|
|
_parse_args(
|
|
["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"]
|
|
)
|
|
|
|
def test_usage_neither_session_nor_new(self) -> None:
|
|
"""usage_neither_session_nor_new: neither flag → UsageError('pass exactly one')."""
|
|
with pytest.raises(UsageError, match="pass exactly one"):
|
|
_parse_args(["--send", "hi", "--api-key", "k"])
|
|
|
|
def test_usage_send_new_without_agent(self) -> None:
|
|
"""send_new_without_agent (issue #8): --send --new without --agent → UsageError.
|
|
|
|
--send mode is non-interactive — cannot prompt; --agent stays required.
|
|
"""
|
|
with pytest.raises(UsageError, match="--agent is required when --new"):
|
|
_parse_args(["--send", "hi", "--new", "--api-key", "k"])
|
|
|
|
def test_parse_bare_new_without_agent_accepted(self) -> None:
|
|
"""bare_new_without_agent (issue #8): --new without --send or --agent → agent_id=None.
|
|
|
|
TUI mode CAN prompt; startup picker handles the choice. _parse_args
|
|
accepts None here and the TUI's _resolve_then_run drives the picker.
|
|
"""
|
|
args = _parse_args(["--new", "--api-key", "k"])
|
|
assert args.new is True
|
|
assert args.agent_id is None
|
|
assert args.send_content is None
|
|
|
|
def test_parse_bare_new_with_agent_accepted(self) -> None:
|
|
"""bare_new_with_agent (issue #8): --new --agent mimir (no --send) → picker skipped.
|
|
|
|
Existing TUI launch path with an explicit agent_id continues to work
|
|
— _resolve_then_run sees `args.agent_id is not None` and skips the
|
|
picker entirely.
|
|
"""
|
|
args = _parse_args(["--new", "--agent", "mimir", "--api-key", "k"])
|
|
assert args.new is True
|
|
assert args.agent_id == "mimir"
|
|
assert args.send_content is None
|
|
|
|
def test_usage_session_with_agent(self) -> None:
|
|
"""usage_session_with_agent: --session AND --agent → UsageError."""
|
|
with pytest.raises(UsageError, match="forbidden with --session"):
|
|
_parse_args(["--send", "hi", "--session", "s-1", "--agent", "x", "--api-key", "k"])
|
|
|
|
def test_auth_missing(self) -> None:
|
|
"""auth_missing: no --api-key and no env → _AuthError."""
|
|
with pytest.raises(_AuthError, match="no API key"):
|
|
_parse_args(["--send", "hi", "--new", "--agent", "mimir"])
|
|
|
|
def test_empty_send(self) -> None:
|
|
"""empty_send: --send '' → UsageError (non-empty enforced)."""
|
|
with pytest.raises(UsageError):
|
|
_parse_args(["--send", "", "--new", "--agent", "x", "--api-key", "k"])
|
|
|
|
def test_happy_new_with_end_user_id(self) -> None:
|
|
"""happy_new_with_end_user_id [happy]: --end-user-id alice → end_user_id='alice'."""
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "lofn", "--api-key", "k",
|
|
"--end-user-id", "alice"]
|
|
)
|
|
assert args.end_user_id == "alice"
|
|
assert args.agent_id == "lofn"
|
|
|
|
def test_end_user_id_default_none(self) -> None:
|
|
"""end_user_id_default_none [trace]: omit --end-user-id → ParsedArgs.end_user_id is None."""
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert args.end_user_id is None
|
|
|
|
def test_empty_end_user_id(self) -> None:
|
|
"""empty_end_user_id [adversarial]: --end-user-id '' → UsageError (mirrors empty --send)."""
|
|
with pytest.raises(UsageError, match="--end-user-id"):
|
|
_parse_args(
|
|
["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--end-user-id", ""]
|
|
)
|
|
|
|
def test_end_user_id_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""end_user_id_from_env [trace]: $RATATOSKR_END_USER_ID fills when flag omitted."""
|
|
monkeypatch.setenv("RATATOSKR_END_USER_ID", "ratatoskr-tui")
|
|
args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert args.end_user_id == "ratatoskr-tui"
|
|
|
|
def test_end_user_id_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""end_user_id_flag_beats_env [trace]: explicit --end-user-id wins over env."""
|
|
monkeypatch.setenv("RATATOSKR_END_USER_ID", "from-env")
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "m", "--api-key", "k",
|
|
"--end-user-id", "from-flag"]
|
|
)
|
|
assert args.end_user_id == "from-flag"
|
|
|
|
|
|
SID = SseId(42, 5)
|
|
|
|
|
|
# Issue #12 — presenter contract semantics amendment.
|
|
# CliPresenterState replaces the stateless _render_event with a stateful per-turn
|
|
# presenter that coalesces thinking runs and demotes telemetry events.
|
|
# (Pre-amendment TestRenderEvent class and `_render_event` function have been
|
|
# removed under the project's no-backwards-compatibility rule.)
|
|
|
|
SID42 = SseId(42, 1)
|
|
|
|
|
|
class TestCliPresenterState:
|
|
"""Tests for the new CliPresenterState — per issue #12 contract."""
|
|
|
|
def test_thinking_coalesce_single_run(self) -> None:
|
|
"""thinking_coalesce_single_run [happy,tracer]:
|
|
Thinking("hello") + Thinking(" world") + Done →
|
|
stderr has ". thinking: hello world\\n" followed by the [done] line.
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Thinking(sse_id=SID42, content="hello"), stdout=stdout, stderr=stderr)
|
|
# After first delta: stderr has the open prefix + content, no \n yet.
|
|
assert stderr.getvalue() == ". thinking: hello"
|
|
state.render(Thinking(sse_id=SID42, content=" world"), stdout=stdout, stderr=stderr)
|
|
# After second delta: still the same growing logical line, still no \n.
|
|
assert stderr.getvalue() == ". thinking: hello world"
|
|
# Now a Done event closes the thinking run with \n then writes the terminal label.
|
|
done = Done(
|
|
sse_id=SID42,
|
|
phase="succeeded",
|
|
response="hi",
|
|
model="m",
|
|
duration_ms=1,
|
|
usage={
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
},
|
|
)
|
|
state.render(done, stdout=stdout, stderr=stderr)
|
|
captured = stderr.getvalue()
|
|
# Thinking run closed with \n; terminal label landed; no demotion prefix on [done].
|
|
assert captured.startswith(". thinking: hello world\n")
|
|
assert "[done]" in captured
|
|
# stdout untouched (no Text events were rendered)
|
|
assert stdout.getvalue() == ""
|
|
|
|
def test_thinking_closes_on_first_non_thinking_event(self) -> None:
|
|
"""thinking_closes_on_first_non_thinking_event [happy]:
|
|
Thinking → WorkerPhase → stderr has ". thinking: ...\\n" then ". worker_phase: ..."
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
|
state.render(
|
|
WorkerPhase(sse_id=SID42, phase="streaming", turn_id=42),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
out = stderr.getvalue()
|
|
# Thinking run closed; worker_phase rendered with demotion prefix.
|
|
assert ". thinking: x\n" in out
|
|
assert ". worker_phase:" in out
|
|
# `[worker_phase]` (bracketed, pre-amendment shape) MUST NOT appear.
|
|
assert "[worker_phase]" not in out
|
|
|
|
def test_thinking_closes_on_error(self) -> None:
|
|
"""thinking_closes_on_error [error]:
|
|
Thinking → Error → thinking line closes with \\n, then [error] line rendered
|
|
(partial thinking content is NOT discarded — observability requirement).
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
|
state.render(
|
|
Error(sse_id=SID42, phase="failed", message="boom", error_code="bad"),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
out = stderr.getvalue()
|
|
# Partial thinking preserved with closing \n; error rendered without demotion prefix.
|
|
assert ". thinking: x\n" in out
|
|
assert "[error]" in out
|
|
# Demotion prefix MUST NOT precede [error]: it's load-bearing.
|
|
assert ". [error]" not in out
|
|
|
|
def test_cancelled_mid_thinking(self) -> None:
|
|
"""cancelled_mid_thinking [scenario]: Thinking → Cancelled → thinking closes with \\n;
|
|
then [cancelled] (no demotion prefix, partial thinking preserved).
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
|
state.render(
|
|
Cancelled(
|
|
sse_id=SID42, phase="cancelled", turn_id=42, reason="user", partial_message_id=None
|
|
),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
out = stderr.getvalue()
|
|
assert ". thinking: x\n" in out
|
|
assert "[cancelled]" in out
|
|
assert ". [cancelled]" not in out
|
|
|
|
def test_text_then_done_newline_boundary(self) -> None:
|
|
"""text_then_done_newline_boundary [trace]: Text("answer") → Done;
|
|
stdout receives "answer\\n" (the \\n is the INV-005 boundary), stderr has "[done] ...".
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
|
|
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
|
# INV-005: text without trailing \n → exactly one \n gets injected before terminal label
|
|
assert stdout.getvalue() == "answer\n"
|
|
assert "[done]" in stderr.getvalue()
|
|
|
|
def test_no_text_then_done_no_extra_newline(self) -> None:
|
|
"""no_text_then_done_no_extra_newline [trace]: Done with no preceding Text →
|
|
stdout untouched; stderr receives only "[done] ...".
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
|
# INV-005 boundary fires ONLY when text was written; no text → no \n injection.
|
|
assert stdout.getvalue() == ""
|
|
assert "[done]" in stderr.getvalue()
|
|
|
|
def test_newline_terminated_text_then_done(self) -> None:
|
|
"""newline_terminated_text_then_done [trace]: Text("answer\\n") → Done;
|
|
stdout receives "answer\\n" exactly ONCE (no double-newline before [done]).
|
|
Tests the F4 Volva fix: text_written_since_newline tracks last-char-was-\\n.
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Text(sse_id=SID42, content="answer\n"), stdout=stdout, stderr=stderr)
|
|
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
|
# POST-003: content ends with \n → state.text_written_since_newline = False
|
|
# → INV-005 does NOT inject an extra \n before [done].
|
|
assert stdout.getvalue() == "answer\n"
|
|
|
|
def test_multiple_thinking_runs(self) -> None:
|
|
"""multiple_thinking_runs [scenario]: Thinking → Text → Thinking → Done →
|
|
TWO separate ". thinking: ..." runs in stderr; stdout has the text + INV-005 boundary.
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(Thinking(sse_id=SID42, content="first"), stdout=stdout, stderr=stderr)
|
|
state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
|
|
state.render(Thinking(sse_id=SID42, content="second"), stdout=stdout, stderr=stderr)
|
|
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
|
err = stderr.getvalue()
|
|
# Each thinking RUN gets its own ". thinking: " prefix.
|
|
assert err.count(". thinking: ") == 2
|
|
assert ". thinking: first" in err
|
|
assert ". thinking: second" in err
|
|
assert stdout.getvalue() == "answer\n"
|
|
assert "[done]" in err
|
|
|
|
|
|
def test_tool_start_demoted(self) -> None:
|
|
"""tool_start_demoted [trace]: ToolStart → stderr line starts with ". tool_start:" """
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(
|
|
ToolStart(sse_id=SID42, name="read_file", arguments={"path": "/x"}),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
line = stderr.getvalue()
|
|
assert line.startswith(". tool_start:")
|
|
assert "[tool_start]" not in line
|
|
|
|
def test_tool_result_truncated(self) -> None:
|
|
"""tool_result_truncated [trace]: long result repr truncates to ≤200 chars."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(
|
|
ToolResult(sse_id=SID42, name="x", result="b" * 500, duration_ms=42),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
line = stderr.getvalue()
|
|
assert line.startswith(". tool_result:")
|
|
# Full 500-char result MUST NOT fit; truncation applied.
|
|
assert "b" * 500 not in line
|
|
|
|
def test_text_boundary_demoted(self) -> None:
|
|
"""text_boundary_demoted [trace]: TextBoundary → stderr ". text_boundary:" prefix."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(
|
|
TextBoundary(sse_id=SID42, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z"),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
line = stderr.getvalue()
|
|
assert line.startswith(". text_boundary:")
|
|
assert "[text_boundary]" not in line
|
|
|
|
def test_duration_format_seconds(self) -> None:
|
|
"""duration_format_seconds [trace]: Done(duration_ms=5467) → "duration=5.5s"."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(_make_done(duration_ms=5467), stdout=io.StringIO(), stderr=stderr)
|
|
assert "duration=5.5s" in stderr.getvalue()
|
|
assert "duration_ms=5467" not in stderr.getvalue()
|
|
|
|
def test_duration_format_subsecond(self) -> None:
|
|
"""duration_format_subsecond [trace]: Done(duration_ms=347) → "duration=347ms"."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(_make_done(duration_ms=347), stdout=io.StringIO(), stderr=stderr)
|
|
assert "duration=347ms" in stderr.getvalue()
|
|
|
|
def test_duration_format_minutes(self) -> None:
|
|
"""duration_format_minutes [trace]: Done(duration_ms=72000) → "duration=1.2m"."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(_make_done(duration_ms=72000), stdout=io.StringIO(), stderr=stderr)
|
|
assert "duration=1.2m" in stderr.getvalue()
|
|
|
|
def test_usage_format_ascii_arrow(self) -> None:
|
|
"""usage_format_ascii_arrow [trace]: stderr label contains the natural-language
|
|
usage shape with ASCII arrow (-> not →) for CLI scriptability.
|
|
"""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
stderr = io.StringIO()
|
|
state = CliPresenterState()
|
|
state.render(
|
|
_make_done(
|
|
usage={
|
|
"prompt_tokens": 6756,
|
|
"completion_tokens": 126,
|
|
"total_tokens": 6882,
|
|
"cached_input_tokens": 0,
|
|
}
|
|
),
|
|
stdout=io.StringIO(),
|
|
stderr=stderr,
|
|
)
|
|
out = stderr.getvalue()
|
|
assert "usage 6756 in -> 126 out (6882 total, 0 cached)" in out
|
|
# Raw dict shape MUST NOT leak through.
|
|
assert "'prompt_tokens'" not in out
|
|
|
|
def test_state_reset_per_amain(self) -> None:
|
|
"""state_reset_per_amain [trace]: fresh CliPresenterState() starts with no thinking open."""
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
# Simulate two _amain calls by constructing two independent states.
|
|
s1 = CliPresenterState()
|
|
s2 = CliPresenterState()
|
|
# Run thinking into s1 — it should NOT bleed into s2.
|
|
s1.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=io.StringIO())
|
|
assert s1.thinking_open is True
|
|
assert s2.thinking_open is False
|
|
# s2's first render produces its own ". thinking: " prefix.
|
|
e2 = io.StringIO()
|
|
s2.render(Thinking(sse_id=SID42, content="y"), stdout=io.StringIO(), stderr=e2)
|
|
assert e2.getvalue() == ". thinking: y"
|
|
|
|
|
|
class TestFormatDurationMs:
|
|
"""Unit tests for _format_duration_ms per INV-006."""
|
|
|
|
def test_subsecond(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(347) == "347ms"
|
|
|
|
def test_exact_one_second(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(1000) == "1.0s"
|
|
|
|
def test_fractional_seconds(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(5467) == "5.5s"
|
|
|
|
def test_exact_one_minute(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(60000) == "1.0m"
|
|
|
|
def test_fractional_minutes(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(72000) == "1.2m"
|
|
|
|
def test_zero(self) -> None:
|
|
from ratatoskr.cli import _format_duration_ms
|
|
assert _format_duration_ms(0) == "0ms"
|
|
|
|
|
|
class TestFormatUsage:
|
|
"""Unit tests for _format_usage per INV-007."""
|
|
|
|
def test_ascii_arrow(self) -> None:
|
|
from ratatoskr.cli import _format_usage
|
|
usage = {
|
|
"prompt_tokens": 6756,
|
|
"completion_tokens": 126,
|
|
"total_tokens": 6882,
|
|
"cached_input_tokens": 0,
|
|
}
|
|
assert (
|
|
_format_usage(usage, arrow="->")
|
|
== "6756 in -> 126 out (6882 total, 0 cached)"
|
|
)
|
|
|
|
def test_unicode_arrow(self) -> None:
|
|
from ratatoskr.cli import _format_usage
|
|
usage = {
|
|
"prompt_tokens": 6756,
|
|
"completion_tokens": 126,
|
|
"total_tokens": 6882,
|
|
"cached_input_tokens": 0,
|
|
}
|
|
assert (
|
|
_format_usage(usage, arrow="→")
|
|
== "6756 in → 126 out (6882 total, 0 cached)"
|
|
)
|
|
|
|
|
|
_USAGE_ZERO: dict[str, int] = {
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
}
|
|
|
|
|
|
def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done:
|
|
return Done(
|
|
sse_id=SID42,
|
|
phase="succeeded",
|
|
response="r",
|
|
model="m",
|
|
duration_ms=duration_ms,
|
|
usage=usage if usage is not None else _USAGE_ZERO,
|
|
)
|
|
|
|
|
|
class TestCancelAndLog:
|
|
@respx.mock
|
|
async def test_happy_cancel(self) -> None:
|
|
"""happy_cancel [happy,tracer]: 200 OK → returns None; stderr empty."""
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None},
|
|
)
|
|
)
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
result = await _cancel_and_log(client, "s-1", 42, stderr=stderr)
|
|
assert result is None
|
|
assert stderr.getvalue() == ""
|
|
|
|
@respx.mock
|
|
async def test_cancel_failed_500(self) -> None:
|
|
"""cancel_failed_500 [error]: …"""
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(500, content=b"boom")
|
|
)
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await _cancel_and_log(client, "s-1", 42, stderr=stderr)
|
|
out = stderr.getvalue()
|
|
assert "[cancel_failed]" in out
|
|
assert "CancelFailed" in out
|
|
|
|
@respx.mock
|
|
async def test_cancel_already_completed(self) -> None:
|
|
"""cancel_already_completed [scenario]: …"""
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(409)
|
|
)
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await _cancel_and_log(client, "s-1", 42, stderr=stderr)
|
|
out = stderr.getvalue()
|
|
assert "[cancel_failed]" in out
|
|
assert "CancelAlreadyCompleted" in out
|
|
|
|
@respx.mock
|
|
async def test_cancel_turn_not_found(self) -> None:
|
|
"""cancel_turn_not_found [scenario]: 404 → returns None; stderr CancelTurnNotFound."""
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
return_value=httpx.Response(404)
|
|
)
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await _cancel_and_log(client, "s-1", 42, stderr=stderr)
|
|
out = stderr.getvalue()
|
|
assert "[cancel_failed]" in out
|
|
assert "CancelTurnNotFound" in out
|
|
|
|
@respx.mock
|
|
async def test_transport_error_swallowed(self) -> None:
|
|
"""transport_error_swallowed [error]: …"""
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
side_effect=httpx.ConnectError("network down")
|
|
)
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
await _cancel_and_log(client, "s-1", 42, stderr=stderr)
|
|
out = stderr.getvalue()
|
|
assert "[cancel_failed]" in out
|
|
assert "ConnectError" in out
|
|
|
|
|
|
class _GatedStream(httpx.AsyncByteStream):
|
|
"""SSE byte stream: list of (bytes | asyncio.Event); Event entries pause until set."""
|
|
|
|
def __init__(self, items: list[bytes | asyncio.Event]) -> None:
|
|
self._items = items
|
|
|
|
async def __aiter__(self): # type: ignore[no-untyped-def]
|
|
for item in self._items:
|
|
if isinstance(item, asyncio.Event):
|
|
await item.wait()
|
|
else:
|
|
yield item
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
|
|
class TestRunTurn:
|
|
@respx.mock
|
|
async def test_happy_text_then_done(self) -> None:
|
|
"""happy_text_then_done [happy,tracer]: …"""
|
|
stream = _sse_chunk(
|
|
"42:1", {"type": "text", "content": "hello"}
|
|
) + _sse_chunk("42:2", _DONE_BODY)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 0
|
|
assert stdout.getvalue() == "hello\n"
|
|
assert "[done]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_error_terminal(self) -> None:
|
|
"""error_terminal: text + error → exit 2; stderr has [error]."""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
|
"42:2",
|
|
{
|
|
"type": "error",
|
|
"phase": "failed",
|
|
"error_code": "llm_output_invalid",
|
|
"message": "boom",
|
|
},
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 2
|
|
assert "[error]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_cancelled_terminal_server(self) -> None:
|
|
"""cancelled_terminal_server: text + cancelled → exit 3; stderr has [cancelled]."""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
|
"42:2", _CANCELLED_BODY
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 3
|
|
assert "[cancelled]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_sse_connect_failed_404(self) -> None:
|
|
"""sse_connect_failed_404 [error]: 404 → exit 20; stderr [sse_connect_failed] status=404."""
|
|
respx.post("https://w.example/sessions/missing/messages").mock(
|
|
return_value=httpx.Response(404, json={"error": "session_not_found"})
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(
|
|
client, "missing", "hi", sigint, stdout=stdout, stderr=stderr
|
|
)
|
|
assert exit_code == 20
|
|
out = stderr.getvalue()
|
|
assert "[sse_connect_failed]" in out
|
|
assert "status=404" in out
|
|
|
|
@respx.mock
|
|
async def test_connection_dropped(self) -> None:
|
|
"""connection_dropped [error]: RemoteProtocolError mid-stream → exit 21."""
|
|
|
|
class _DropAfter(httpx.AsyncByteStream):
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self._chunks = chunks
|
|
|
|
async def __aiter__(self): # type: ignore[no-untyped-def]
|
|
for c in self._chunks:
|
|
yield c
|
|
raise httpx.RemoteProtocolError("simulated mid-stream drop")
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
first = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, stream=_DropAfter([first])
|
|
)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 21
|
|
assert "[connection_dropped]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_malformed_sse_id(self) -> None:
|
|
"""malformed_sse_id [error]: id without seq → exit 22; stderr [malformed_sse_id]."""
|
|
stream = b"id: 42\ndata: {\"type\": \"text\", \"content\": \"x\"}\n\n"
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 22
|
|
assert "[malformed_sse_id]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_malformed_sse_data(self) -> None:
|
|
"""malformed_sse_data [error]: text + non-JSON → exit 22; [malformed_sse_data] raw='...'."""
|
|
stream = (
|
|
_sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
+ b"id: 42:2\ndata: not-json\n\n"
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 22
|
|
out = stderr.getvalue()
|
|
# Exact label + `raw='X'` shape — the stderr format the contract specifies
|
|
assert "[malformed_sse_data] raw='not-json'" in out
|
|
|
|
@respx.mock
|
|
async def test_malformed_sse_data_truncation(self) -> None:
|
|
"""malformed_sse_data_truncation [security]: 5000-char bad data → raw truncated."""
|
|
huge_bad = "x" * 5000
|
|
stream = (
|
|
_sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
+ f"id: 42:2\ndata: {huge_bad}\n\n".encode()
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 22
|
|
out = stderr.getvalue()
|
|
assert "[malformed_sse_data]" in out
|
|
# MalformedSseData.raw was truncated to 200 chars at the exception layer;
|
|
# presenter's `repr()` rendering of that 200-char string carries through.
|
|
# Full 5000-char payload MUST NOT appear in stderr.
|
|
assert "x" * 5000 not in out
|
|
assert "x" * 200 in out # the truncated form IS in the rendered raw='...'
|
|
|
|
@respx.mock
|
|
async def test_turn_id_flip(self) -> None:
|
|
"""turn_id_flip [error]: …"""
|
|
stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
|
"99:2", {"type": "text", "content": "y"}
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 22
|
|
out = stderr.getvalue()
|
|
assert "[turn_id_flip]" in out
|
|
assert "expected=42" in out
|
|
assert "got=99" in out
|
|
|
|
@respx.mock
|
|
async def test_sigint_before_first_event(self) -> None:
|
|
"""sigint_before_first_event [scenario]: …"""
|
|
gate = asyncio.Event()
|
|
# Stream never yields anything until gate (the gate is never set; the test exits via sigint)
|
|
stream = _GatedStream([gate])
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
cancel_route = respx.post("https://w.example/sessions/s-1/turns").mock(
|
|
return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
|
|
)
|
|
sigint = asyncio.Event()
|
|
sigint.set() # SIGINT before _run_turn even starts
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await asyncio.wait_for(
|
|
_run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr), timeout=2.0
|
|
)
|
|
assert exit_code == 3
|
|
assert "[cancelled] (before any event arrived)" in stderr.getvalue()
|
|
assert cancel_route.call_count == 0
|
|
|
|
@respx.mock
|
|
async def test_sigint_mid_stream_drains_to_cancelled(self) -> None:
|
|
"""sigint_mid_stream_drains_to_cancelled [scenario,tracer]: …"""
|
|
cancel_observed = asyncio.Event()
|
|
|
|
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
|
cancel_observed.set()
|
|
return httpx.Response(200, json=_CANCEL_OK_RESP)
|
|
|
|
cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
side_effect=cancel_handler
|
|
)
|
|
text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
|
|
stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
task = asyncio.create_task(
|
|
_run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
)
|
|
# Wait for the first event to flush to stdout (signals last_turn_id is set)
|
|
for _ in range(50):
|
|
if "x" in stdout.getvalue():
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
else:
|
|
task.cancel()
|
|
pytest.fail("text event never reached stdout")
|
|
sigint.set()
|
|
exit_code = await asyncio.wait_for(task, timeout=2.0)
|
|
assert exit_code == 3
|
|
assert cancel_route.call_count == 1
|
|
|
|
@respx.mock
|
|
async def test_sigint_twice_issues_one_cancel(self) -> None:
|
|
"""sigint_twice_issues_one_cancel [scenario]: sigint set twice → one cancel POST."""
|
|
cancel_observed = asyncio.Event()
|
|
|
|
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
|
cancel_observed.set()
|
|
return httpx.Response(200, json=_CANCEL_OK_RESP)
|
|
|
|
cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
side_effect=cancel_handler
|
|
)
|
|
text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
|
|
stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
task = asyncio.create_task(
|
|
_run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
)
|
|
for _ in range(50):
|
|
if "x" in stdout.getvalue():
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
sigint.set()
|
|
# Set again — should be no-op (event is already set; idempotent)
|
|
sigint.set()
|
|
exit_code = await asyncio.wait_for(task, timeout=2.0)
|
|
assert exit_code == 3
|
|
assert cancel_route.call_count == 1
|
|
|
|
@respx.mock
|
|
async def test_no_busy_loop_after_cancel(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""no_busy_loop_after_cancel [trace]: only one sigint_event.wait()-task created."""
|
|
cancel_observed = asyncio.Event()
|
|
|
|
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
|
cancel_observed.set()
|
|
return httpx.Response(200, json=_CANCEL_OK_RESP)
|
|
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
side_effect=cancel_handler
|
|
)
|
|
text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
|
|
stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
|
|
sigint = asyncio.Event()
|
|
wait_call_count = 0
|
|
original_wait = sigint.wait
|
|
|
|
async def counting_wait() -> bool:
|
|
nonlocal wait_call_count
|
|
wait_call_count += 1
|
|
return await original_wait()
|
|
|
|
monkeypatch.setattr(sigint, "wait", counting_wait)
|
|
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
task = asyncio.create_task(
|
|
_run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
)
|
|
for _ in range(50):
|
|
if "x" in stdout.getvalue():
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
sigint.set()
|
|
exit_code = await asyncio.wait_for(task, timeout=2.0)
|
|
assert exit_code == 3
|
|
# INV-007 + busy-loop fix: sigint.wait() created at most once per pre-cancelling
|
|
# iteration. For text → sigint → cancelled, that's iter 1 (raced w/ text) and
|
|
# iter 2 (raced w/ sigint; flipped cancelling=True). Iter 3+ MUST skip wait()
|
|
# creation entirely — the busy-loop bug would make this number grow unbounded.
|
|
assert wait_call_count == 2
|
|
|
|
@respx.mock
|
|
async def test_cancel_failed_drains_anyway(self) -> None:
|
|
"""cancel_failed_drains_anyway [scenario]: …"""
|
|
cancel_observed = asyncio.Event()
|
|
|
|
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
|
cancel_observed.set()
|
|
return httpx.Response(500, content=b"boom")
|
|
|
|
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
|
side_effect=cancel_handler
|
|
)
|
|
text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
|
cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
|
|
stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(stream)
|
|
)
|
|
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
task = asyncio.create_task(
|
|
_run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
)
|
|
for _ in range(50):
|
|
if "x" in stdout.getvalue():
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
sigint.set()
|
|
exit_code = await asyncio.wait_for(task, timeout=2.0)
|
|
assert exit_code == 3
|
|
# INV-009: cancel POST failed but stream still drained to cancelled terminal
|
|
assert "[cancel_failed]" in stderr.getvalue()
|
|
assert "[cancelled]" in stderr.getvalue()
|
|
|
|
@respx.mock
|
|
async def test_render_called_once_per_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""render_called_once_per_event [trace]: spy on render; call_count == event count.
|
|
|
|
Per issue #12: rendering went from stateless `_render_event` to
|
|
`CliPresenterState.render`; the spy moves accordingly.
|
|
"""
|
|
chunks = [
|
|
_sse_chunk("42:1", {"type": "worker_phase", "phase": "streaming", "turn_id": 42}),
|
|
_sse_chunk("42:2", {"type": "text", "content": "hi"}),
|
|
_sse_chunk("42:3", _DONE_BODY),
|
|
]
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=b"".join(chunks)
|
|
)
|
|
)
|
|
|
|
from ratatoskr.cli import CliPresenterState
|
|
|
|
call_count = 0
|
|
original = CliPresenterState.render
|
|
|
|
def spy(self, event, **kw): # type: ignore[no-untyped-def]
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return original(self, event, **kw)
|
|
|
|
monkeypatch.setattr(CliPresenterState, "render", spy)
|
|
|
|
sigint = asyncio.Event()
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
|
|
assert exit_code == 0
|
|
assert call_count == 3
|
|
|
|
|
|
_PARSED_NEW = ParsedArgs(
|
|
send_content="hi",
|
|
session_id=None,
|
|
new=True,
|
|
agent_id="mimir",
|
|
api_key="k",
|
|
server_url="https://w.example",
|
|
raw=False,
|
|
)
|
|
_PARSED_NEW_WITH_END_USER = ParsedArgs(
|
|
send_content="hi",
|
|
session_id=None,
|
|
new=True,
|
|
agent_id="mimir",
|
|
api_key="k",
|
|
server_url="https://w.example",
|
|
raw=False,
|
|
end_user_id="alice",
|
|
)
|
|
_PARSED_EXISTING = ParsedArgs(
|
|
send_content="hi",
|
|
session_id="s-1",
|
|
new=False,
|
|
agent_id=None,
|
|
api_key="k",
|
|
server_url="https://w.example",
|
|
raw=False,
|
|
)
|
|
_CREATE_OK_RESP = {
|
|
"session_id": "s-new",
|
|
"agent_id": "mimir",
|
|
"message_count": 0,
|
|
"created_at": "2026-05-21T00:00:00+00:00",
|
|
"last_active": "2026-05-21T00:00:00+00:00",
|
|
"metadata": {},
|
|
}
|
|
|
|
|
|
class TestAmain:
|
|
@respx.mock
|
|
async def test_user_agent_header_sent(self) -> None:
|
|
"""user_agent_header_sent [trace]: outbound requests carry the ratatoskr User-Agent.
|
|
|
|
Worldtree-dev (althing 2026-05-23) requested consumers send User-Agent
|
|
so server logs can distinguish ratatoskr traffic.
|
|
"""
|
|
sessions_route = respx.post("https://w.example/sessions").mock(
|
|
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
|
)
|
|
sse_body = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk(
|
|
"42:2", _DONE_BODY
|
|
)
|
|
respx.post("https://w.example/sessions/s-new/messages").mock(
|
|
return_value=_sse_resp(sse_body)
|
|
)
|
|
await _amain(_PARSED_NEW)
|
|
ua = sessions_route.calls[0].request.headers["User-Agent"]
|
|
assert ua.startswith("ratatoskr/")
|
|
assert "vh@phasefinal.com" in ua
|
|
|
|
@respx.mock
|
|
async def test_happy_new_session_then_stream(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""happy_new_session_then_stream [happy,tracer]: …"""
|
|
respx.post("https://w.example/sessions").mock(
|
|
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
|
)
|
|
sse_body = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
|
|
"42:2", _DONE_BODY
|
|
)
|
|
respx.post("https://w.example/sessions/s-new/messages").mock(
|
|
return_value=_sse_resp(sse_body)
|
|
)
|
|
exit_code = await _amain(_PARSED_NEW)
|
|
assert exit_code == 0
|
|
captured = capsys.readouterr()
|
|
err = captured.err
|
|
# Per issue #12: [create_session] lifecycle line demoted to `. create_session:`.
|
|
assert ". create_session:" in err
|
|
assert "[create_session]" not in err # pre-amendment shape forbidden
|
|
assert "[done]" in err
|
|
assert err.index(". create_session:") < err.index("[done]")
|
|
|
|
@respx.mock
|
|
async def test_happy_existing_session(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""happy_existing_session: --session, no create POST; just SSE stream → exit 0."""
|
|
sessions_route = respx.post("https://w.example/sessions").mock(
|
|
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
|
)
|
|
sse_body = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk(
|
|
"42:2", _DONE_BODY
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(sse_body)
|
|
)
|
|
exit_code = await _amain(_PARSED_EXISTING)
|
|
assert exit_code == 0
|
|
assert sessions_route.call_count == 0
|
|
|
|
@respx.mock
|
|
async def test_agent_not_found_exits_12(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""agent_not_found_exits_12 [error]: POST /sessions → 404 → exit 12; no stream_turn."""
|
|
respx.post("https://w.example/sessions").mock(
|
|
return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
|
|
)
|
|
stream_route = respx.post("https://w.example/sessions/s-new/messages").mock(
|
|
return_value=httpx.Response(200)
|
|
)
|
|
exit_code = await _amain(_PARSED_NEW)
|
|
assert exit_code == 12
|
|
assert "[agent_not_found]" in capsys.readouterr().err
|
|
assert stream_route.call_count == 0
|
|
|
|
@respx.mock
|
|
async def test_session_api_failed_exits_20(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""session_api_failed_exits_20: POST /sessions → 500 → exit 20; [session_api_failed]."""
|
|
respx.post("https://w.example/sessions").mock(
|
|
return_value=httpx.Response(500, content=b"server error")
|
|
)
|
|
exit_code = await _amain(_PARSED_NEW)
|
|
assert exit_code == 20
|
|
err = capsys.readouterr().err
|
|
assert "[session_api_failed]" in err
|
|
assert "status=500" in err
|
|
|
|
@respx.mock
|
|
async def test_connect_error_exits_21(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""connect_error_exits_21: httpx.ConnectError on POST /sessions → exit 21."""
|
|
respx.post("https://w.example/sessions").mock(side_effect=httpx.ConnectError("down"))
|
|
exit_code = await _amain(_PARSED_NEW)
|
|
assert exit_code == 21
|
|
assert "[network_error]" in capsys.readouterr().err
|
|
|
|
@respx.mock
|
|
async def test_sigint_handler_installed_and_removed(self) -> None:
|
|
"""sigint_handler_installed_and_removed [trace]: signal handler add/remove paired."""
|
|
sse_body = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
|
|
"42:2", _DONE_BODY
|
|
)
|
|
respx.post("https://w.example/sessions/s-1/messages").mock(
|
|
return_value=_sse_resp(sse_body)
|
|
)
|
|
loop = asyncio.get_running_loop()
|
|
original_add = loop.add_signal_handler
|
|
original_remove = loop.remove_signal_handler
|
|
add_calls: list[int] = []
|
|
remove_calls: list[int] = []
|
|
|
|
def add_spy(sig, callback, *args): # type: ignore[no-untyped-def]
|
|
add_calls.append(sig)
|
|
return original_add(sig, callback, *args)
|
|
|
|
def remove_spy(sig): # type: ignore[no-untyped-def]
|
|
remove_calls.append(sig)
|
|
return original_remove(sig)
|
|
|
|
loop.add_signal_handler = add_spy # type: ignore[method-assign]
|
|
loop.remove_signal_handler = remove_spy # type: ignore[method-assign]
|
|
try:
|
|
exit_code = await _amain(_PARSED_EXISTING)
|
|
finally:
|
|
loop.add_signal_handler = original_add # type: ignore[method-assign]
|
|
loop.remove_signal_handler = original_remove # type: ignore[method-assign]
|
|
import signal as _sig
|
|
|
|
assert exit_code == 0
|
|
assert add_calls == [_sig.SIGINT]
|
|
assert remove_calls == [_sig.SIGINT]
|
|
|
|
def test_no_textual_import(self) -> None:
|
|
"""no_textual_import [scenario]: …"""
|
|
# INV-001 import-only boundary: cli.py must not import textual/rich at the
|
|
# source level. The load-bearing check is a static source grep (NOT a live
|
|
# `importlib.reload`, which would mutate the shared module in place and break
|
|
# class identity — isinstance / pytest.raises — for every later test).
|
|
import pathlib
|
|
|
|
src = pathlib.Path(__file__).parent.parent / "src" / "ratatoskr" / "cli.py"
|
|
text = src.read_text()
|
|
for forbidden in ("import textual", "from textual", "import rich", "from rich"):
|
|
assert forbidden not in text, f"INV-001 violation: cli.py contains '{forbidden}'"
|
|
|
|
|
|
class TestMain:
|
|
def test_happy_returns_amain_exit_code(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""happy_returns_amain_exit_code [happy,tracer]: …"""
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
assert args.send_content == "hi"
|
|
return 0
|
|
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert rc == 0
|
|
|
|
def test_usage_error_no_send(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""usage_error_no_send: empty argv → exit 10; stderr [usage_error]; _amain never called."""
|
|
amain_calls: list[int] = []
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
amain_calls.append(1)
|
|
return 0
|
|
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main([])
|
|
assert rc == 10
|
|
assert "[usage_error]" in capsys.readouterr().err
|
|
assert amain_calls == []
|
|
|
|
def test_usage_error_both_session_and_new(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""usage_error_both_session_and_new: both flags → exit 10; [usage_error]."""
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
return 0
|
|
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main(["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"])
|
|
assert rc == 10
|
|
assert "[usage_error]" in capsys.readouterr().err
|
|
|
|
def test_auth_error_missing_key(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""auth_error_missing_key: …"""
|
|
# _clear_env fixture has already deleted WORLDTREE_API_KEY
|
|
rc = main(["--send", "hi", "--new", "--agent", "m"])
|
|
assert rc == 11
|
|
assert "[auth_error]" in capsys.readouterr().err
|
|
|
|
def test_no_argv_uses_sys_argv(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""no_argv_uses_sys_argv [trace]: argv=None → _parse_args reads sys.argv[1:]."""
|
|
monkeypatch.setattr(
|
|
"sys.argv",
|
|
["ratatoskr", "--send", "hi", "--new", "--agent", "m", "--api-key", "k"],
|
|
)
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
assert args.send_content == "hi"
|
|
assert args.agent_id == "m"
|
|
return 0
|
|
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main(None)
|
|
assert rc == 0
|
|
|
|
def test_help_exits_cleanly(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""help_exits_cleanly [happy]: --help → main returns 0; _amain never called."""
|
|
amain_calls: list[int] = []
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
amain_calls.append(1)
|
|
return 0
|
|
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main(["--help"])
|
|
assert rc == 0
|
|
assert amain_calls == []
|
|
# argparse prints help text to stdout
|
|
assert "ratatoskr" in capsys.readouterr().out
|
|
|
|
def test_no_send_dispatches_to_tui(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""no_send_dispatches_to_tui: --send omitted → main calls run_tui, NOT _amain."""
|
|
from ratatoskr import tui as tui_mod
|
|
|
|
tui_calls: list[ParsedArgs] = []
|
|
amain_calls: list[int] = []
|
|
|
|
def fake_run_tui(args: ParsedArgs) -> int:
|
|
tui_calls.append(args)
|
|
return 0
|
|
|
|
async def fake_amain(args: ParsedArgs) -> int:
|
|
amain_calls.append(1)
|
|
return 0
|
|
|
|
monkeypatch.setattr(tui_mod, "run_tui", fake_run_tui)
|
|
monkeypatch.setattr(cli_mod, "_amain", fake_amain)
|
|
rc = main(["--session", "s-1", "--api-key", "k"])
|
|
assert rc == 0
|
|
assert len(tui_calls) == 1
|
|
assert tui_calls[0].send_content is None
|
|
assert tui_calls[0].session_id == "s-1"
|
|
assert amain_calls == []
|
|
|
|
|
|
class TestBifrostBindCli:
|
|
"""Issue #17 slice 3a — the CLI Bifrost-bind trigger (INV-008, one of three)."""
|
|
|
|
def test_plane_and_host_build_binding(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""tracer: --bifrost-plane + --bifrost-host resolve a BifrostBinding via
|
|
endpoint_for_plane; the consumer key comes from the env."""
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
args = _parse_args(
|
|
[
|
|
"--send", "hi", "--new", "--agent", "ratatoskr:sindra", "--api-key", "k",
|
|
"--bifrost-plane", "memory", "--bifrost-host", "10.100.10.50",
|
|
]
|
|
)
|
|
assert args.bifrost == BifrostBinding(endpoint_url="http://10.100.10.50:8391")
|
|
assert args.bifrost_plane == "memory"
|
|
assert args.consumer_key == "ck"
|
|
|
|
def test_host_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--bifrost-host falls back to RATATOSKR_PROVIDER_VISIBLE_HOST."""
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
monkeypatch.setenv("RATATOSKR_PROVIDER_VISIBLE_HOST", "10.0.0.9")
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
|
|
"--bifrost-plane", "affect"]
|
|
)
|
|
assert args.bifrost == BifrostBinding(endpoint_url="http://10.0.0.9:8390")
|
|
|
|
def test_direct_url_bypasses_plane(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--bifrost-url is the direct (HTTPS/prod) endpoint, bypassing the plane
|
|
shortcut; no plane label."""
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
|
|
"--bifrost-url", "https://prov.example:8391"]
|
|
)
|
|
assert args.bifrost == BifrostBinding(endpoint_url="https://prov.example:8391")
|
|
assert args.bifrost_plane is None
|
|
|
|
def test_no_bifrost_flags_leaves_binding_none(self) -> None:
|
|
"""regression: no bifrost flags → bifrost/consumer_key None (pre-#17 path)."""
|
|
args = _parse_args(
|
|
["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"]
|
|
)
|
|
assert args.bifrost is None
|
|
assert args.consumer_key is None
|
|
|
|
def test_plane_and_url_mutually_exclusive(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
with pytest.raises(UsageError):
|
|
_parse_args(
|
|
["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
|
|
"--bifrost-plane", "memory", "--bifrost-host", "h",
|
|
"--bifrost-url", "https://x:8391"]
|
|
)
|
|
|
|
def test_plane_without_host_is_usage_error(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
with pytest.raises(UsageError):
|
|
_parse_args(
|
|
["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
|
|
"--bifrost-plane", "memory"]
|
|
)
|
|
|
|
def test_bind_with_existing_session_is_usage_error(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A binding is a session-CREATE concern; --session (existing) + bind is
|
|
a usage error."""
|
|
monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
|
|
with pytest.raises(UsageError):
|
|
_parse_args(
|
|
["--send", "hi", "--session", "s-1", "--api-key", "k",
|
|
"--bifrost-plane", "memory", "--bifrost-host", "h"]
|
|
)
|
|
|
|
@respx.mock
|
|
async def test_amain_bound_create_carries_binding_and_routes_502(
|
|
self, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""_amain on a bound create sends the bifrost body + the consumer-key
|
|
bearer; a 502 auth_rejected routes to BifrostHandshakeFailed with the
|
|
consumer-key-mismatch hint (INV-001/002, 401-message scoping)."""
|
|
route = respx.post("http://w/sessions").mock(
|
|
return_value=httpx.Response(
|
|
502,
|
|
json={
|
|
"error_code": "bifrost_handshake_failed",
|
|
"detail": {"bifrost_error": "bifrost.auth_rejected"},
|
|
},
|
|
)
|
|
)
|
|
args = ParsedArgs(
|
|
send_content="hi", session_id=None, new=True, agent_id="ratatoskr:sindra",
|
|
api_key="canary", server_url="http://w", raw=False, end_user_id="smoke-user",
|
|
bifrost=BifrostBinding(endpoint_url="http://10.100.10.50:8391"),
|
|
bifrost_plane="memory", consumer_key="ck",
|
|
)
|
|
rc = await _amain(args)
|
|
assert rc == 23
|
|
body = json.loads(route.calls[0].request.content)
|
|
assert body["bifrost"] == {
|
|
"endpoint_url": "http://10.100.10.50:8391", "scope": None
|
|
}
|
|
assert route.calls[0].request.headers["Authorization"] == "Bearer ck"
|
|
err = capsys.readouterr().err
|
|
assert "bifrost.auth_rejected" in err
|
|
assert "consumer key" in err # the 401-scoping hint
|
|
|
|
async def test_amain_bind_without_consumer_key_exits(self) -> None:
|
|
"""_amain on a bind with no consumer key raises BifrostConsumerKeyMissing
|
|
(before HTTP) → a clean exit code, never a canary fallback."""
|
|
args = ParsedArgs(
|
|
send_content="hi", session_id=None, new=True, agent_id="a",
|
|
api_key="canary", server_url="http://w", raw=False, end_user_id=None,
|
|
bifrost=BifrostBinding(endpoint_url="http://x:8391"),
|
|
bifrost_plane="memory", consumer_key=None,
|
|
)
|
|
rc = await _amain(args)
|
|
assert rc == 22
|