vh/ratatoskr
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
v0.20.13
ratatoskr/tests/test_sessions.py
T
vh e643d38f58 fix: persona_state SET body → canonical {pad:{pleasure,arousal,dominance}} + re-vendor Tier-3 prose (v0.19.7) 
worldtree-dev landed the Tier-3 persona/memory/persona_state prose docs
(c9e59ec) — shapes that serialize as freeform Any in the OpenAPI, so the
prose markdown is their source of truth. Re-vendored docs/conversation-api-spec.md
(tolerate_drift markdown pin; worldtree-spec-rev 879cefe→c9e59ec).

Consumer alignment: --set-persona-pad / _set_persona_probe was building
{pad:[list]}, but the canonical POST /sessions/{id}/persona_state body (#317)
is {pad:{pleasure,arousal,dominance}} (named dict). Aligned the probe to the
named dict + a len!=3 guard; updated contract #2's note, the set_persona_state
docstring, and the tests. The set_persona_state wrapper was already correct
(freeform pass-through) — only the CLI probe's body construction drifted.

Suite 602 green. (Also this session: heid-code-review on the #347 slice
returned unanimous zero drift across all three panel arms.)
2026-07-06 09:56:46 -07:00

1381 lines
58 KiB
Python
Raw Permalink Blame History

"""Tests for ratatoskr.sessions per docs/contracts/issues/2.contract.md."""
import httpx
import pytest
import respx
from ratatoskr.sessions import (
AgentInfo,
AgentNotAvailable,
AgentNotFound,
AuthoredHistoryUnavailable,
AuthScopeDenied,
BifrostBinding,
BifrostConsumerKeyMissing,
BifrostHandshakeFailed,
InvalidCursor,
PersonaNotConfigured,
SessionApiFailed,
SessionPage,
create_character,
create_session,
delete_character,
endpoint_for_plane,
get_capabilities,
get_character_state,
get_me,
get_persona_state,
get_session_bifrost,
get_session_messages,
get_session_tools,
list_agents,
list_character_models,
list_sessions,
set_persona_state,
write_authored_history,
)
class TestCreateSession:
@respx.mock
async def test_happy_create(self) -> None:
"""happy_create [happy,tracer]: full 201 body -> SessionInfo with create-origin defaults."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "550e8400-e29b-41d4-a716-446655440000",
"agent_id": "mimir",
"message_count": 0,
"created_at": "2026-04-15T12:00:00+00:00",
"last_active": "2026-04-15T12:00:00+00:00",
"metadata": {},
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
info = await create_session(client, "mimir")
assert info.session_id == "550e8400-e29b-41d4-a716-446655440000"
assert info.agent_id == "mimir"
assert info.created_at == "2026-04-15T12:00:00+00:00"
assert info.last_active == "2026-04-15T12:00:00+00:00"
assert info.metadata == {}
assert info.message_count == 0
# INV-001 create-origin fixed defaults
assert info.name is None
assert info.archived is False
assert info.tags == []
@respx.mock
async def test_happy_create_with_metadata(self) -> None:
"""happy_create_with_metadata: response carries metadata -> SessionInfo.metadata matches."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s1",
"agent_id": "mimir",
"message_count": 0,
"created_at": "2026-04-15T12:00:00+00:00",
"last_active": "2026-04-15T12:00:00+00:00",
"metadata": {"model": "glm5-turbo"},
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
info = await create_session(client, "mimir")
assert info.metadata == {"model": "glm5-turbo"}
@respx.mock
async def test_request_body_shape(self) -> None:
"""request_body_shape [trace]: outbound JSON is exactly {"agent_id": <arg>}."""
import json as _json
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s1",
"agent_id": "mimir",
"message_count": 0,
"created_at": "2026-04-15T12:00:00+00:00",
"last_active": "2026-04-15T12:00:00+00:00",
"metadata": {},
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await create_session(client, "mimir")
body = _json.loads(route.calls[0].request.content)
assert body == {"agent_id": "mimir"}
@respx.mock
async def test_unknown_agent_id(self) -> None:
"""unknown_agent_id: 404 -> AgentNotFound(agent_id=<arg>)."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AgentNotFound) as exc_info:
await create_session(client, "mimir")
assert exc_info.value.agent_id == "mimir"
@respx.mock
async def test_validation_failed(self) -> None:
"""validation_failed: 422 -> SessionApiFailed(status=422); body truncated."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
422,
json={"error_code": "validation_failed", "message": "missing agent_id"},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await create_session(client, "mimir")
assert exc_info.value.status == 422
assert len(exc_info.value.body) <= 1024
@respx.mock
async def test_unexpected_status_truncates(self) -> None:
"""unexpected_status_truncates: 500 + 5000-byte body -> SessionApiFailed; body == 1024."""
big = b"x" * 5000
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(500, content=big)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await create_session(client, "mimir")
assert exc_info.value.status == 500
assert exc_info.value.body == big[:1024]
@respx.mock
async def test_empty_agent_id(self) -> None:
"""empty_agent_id [adversarial]: '' -> AssertionError; no HTTP issued."""
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(201, content=b"{}")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await create_session(client, "")
assert route.call_count == 0
@respx.mock
async def test_happy_create_with_end_user_id(self) -> None:
"""happy_create_with_end_user_id [happy]: body carries both keys (issue #5)."""
import json as _json
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s-new",
"agent_id": "lofn",
"message_count": 0,
"created_at": "2026-05-22T12:00:00+00:00",
"last_active": "2026-05-22T12:00:00+00:00",
"metadata": {},
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
info = await create_session(client, "lofn", end_user_id="alice")
body = _json.loads(route.calls[0].request.content)
# INV: body MUST be exactly {"agent_id": ..., "end_user_id": ...} — byte-for-byte
assert body == {"agent_id": "lofn", "end_user_id": "alice"}
assert info.session_id == "s-new"
assert info.agent_id == "lofn"
@respx.mock
async def test_default_omits_end_user_id(self) -> None:
"""default_omits_end_user_id [trace]: omit kwarg → body has no end_user_id (INV-002)."""
import json as _json
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s-new",
"agent_id": "mimir",
"message_count": 0,
"created_at": "2026-05-22T12:00:00+00:00",
"last_active": "2026-05-22T12:00:00+00:00",
"metadata": {},
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await create_session(client, "mimir")
body = _json.loads(route.calls[0].request.content)
# Exact equality — no end_user_id key in the body when the kwarg is omitted
assert body == {"agent_id": "mimir"}
assert "end_user_id" not in body
@respx.mock
async def test_empty_end_user_id(self) -> None:
"""empty_end_user_id [adversarial]: '' → AssertionError before HTTP (PRE-003)."""
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(201, content=b"{}")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await create_session(client, "mimir", end_user_id="")
assert route.call_count == 0
class TestCreateSessionBifrostBind:
"""Issue #17 slice 1 — the create_session Bifrost-bind primitive."""
@respx.mock
async def test_bind_happy_consumer_key_and_body(self) -> None:
"""bind_happy [tracer]: a bifrost binding makes the body carry the
`bifrost` field AND overrides the bearer to the consumer key (NOT the
client's canary default), 201 → SessionInfo. Proves the bind path
end-to-end (FN create_session STEPS 1-2, POST-001, INV-001)."""
import json as _json
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s-bound",
"agent_id": "ratatoskr:sindra",
"message_count": 0,
"created_at": "2026-06-18T12:00:00+00:00",
"last_active": "2026-06-18T12:00:00+00:00",
"metadata": {},
},
)
)
binding = BifrostBinding(endpoint_url="http://10.100.10.50:8391")
async with httpx.AsyncClient(
base_url="https://w.example",
headers={"Authorization": "Bearer canary-key"},
) as client:
info = await create_session(
client,
"ratatoskr:sindra",
end_user_id="smoke-user",
bifrost=binding,
consumer_key="consumer-key",
)
req = route.calls[0].request
body = _json.loads(req.content)
# body carries the bifrost field alongside agent_id/end_user_id
assert body == {
"agent_id": "ratatoskr:sindra",
"end_user_id": "smoke-user",
"bifrost": {
"endpoint_url": "http://10.100.10.50:8391",
"scope": None,
},
}
# bearer overridden to the consumer key (INV-001: never the canary default)
assert req.headers["Authorization"] == "Bearer consumer-key"
assert info.session_id == "s-bound"
assert info.agent_id == "ratatoskr:sindra"
@respx.mock
async def test_bind_without_consumer_key_raises_before_http(self) -> None:
"""missing_key [adversarial]: bifrost set but consumer_key None →
BifrostConsumerKeyMissing BEFORE any HTTP (PRE-001, INV-001: never fall
back to the canary key)."""
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(201, content=b"{}")
)
binding = BifrostBinding(endpoint_url="http://10.100.10.50:8391")
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(BifrostConsumerKeyMissing):
await create_session(client, "ratatoskr:sindra", bifrost=binding)
assert route.call_count == 0
@respx.mock
async def test_bind_with_empty_consumer_key_raises_before_http(self) -> None:
"""empty_key [adversarial]: empty-string consumer_key is also rejected
before HTTP (PRE-001 requires a NON-EMPTY str)."""
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(201, content=b"{}")
)
binding = BifrostBinding(endpoint_url="http://10.100.10.50:8391")
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(BifrostConsumerKeyMissing):
await create_session(
client, "ratatoskr:sindra", bifrost=binding, consumer_key=""
)
assert route.call_count == 0
@respx.mock
async def test_bind_handshake_failure_maps_to_502(self) -> None:
"""handshake_502 [adversarial]: a bound create that 502s with
detail.bifrost_error → BifrostHandshakeFailed carrying the bifrost_error
+ raw body (POST-002, INV-002 bind-time failure). 'bifrost.auth_rejected'
is the canary-key-instead-of-consumer-key tell."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
502,
json={
"error_code": "bifrost_handshake_failed",
"detail": {"bifrost_error": "bifrost.auth_rejected"},
},
)
)
binding = BifrostBinding(endpoint_url="http://10.100.10.50:8391")
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(BifrostHandshakeFailed) as exc_info:
await create_session(
client, "ratatoskr:sindra", bifrost=binding, consumer_key="ck"
)
assert exc_info.value.bifrost_error == "bifrost.auth_rejected"
# the raw 502 body is carried for debugging
assert exc_info.value.body
@respx.mock
async def test_bind_ephemeral_rejection_is_session_api_failed(self) -> None:
"""ephemeral_422 [boundary]: 422 ephemeral_does_not_accept_bifrost is a
generic create failure → SessionApiFailed, NOT a distinct exception
(POST-003 — deliberate, an operator config error)."""
respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
422, json={"error_code": "ephemeral_does_not_accept_bifrost"}
)
)
binding = BifrostBinding(endpoint_url="http://10.100.10.50:8391")
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await create_session(
client, "echo", bifrost=binding, consumer_key="ck"
)
assert exc_info.value.status == 422
@respx.mock
async def test_unbound_create_unchanged_no_auth_override(self) -> None:
"""unbound_unchanged [regression]: with no bifrost, the body is the
pre-#17 shape AND create_session sends NO per-request Authorization
override — the client's default canary bearer governs (INV-001: the two
call sites never cross)."""
import json as _json
route = respx.post("https://w.example/sessions").mock(
return_value=httpx.Response(
201,
json={
"session_id": "s1",
"agent_id": "mimir",
"message_count": 0,
"created_at": "2026-04-15T12:00:00+00:00",
"last_active": "2026-04-15T12:00:00+00:00",
"metadata": {},
},
)
)
async with httpx.AsyncClient(
base_url="https://w.example",
headers={"Authorization": "Bearer canary-key"},
) as client:
await create_session(client, "mimir")
req = route.calls[0].request
body = _json.loads(req.content)
assert body == {"agent_id": "mimir"}
# the client default bearer is used unchanged — no consumer-key override
assert req.headers["Authorization"] == "Bearer canary-key"
class TestEndpointForPlane:
"""Issue #17 — endpoint_for_plane: plane name → Worldtree-visible base URL."""
def test_memory_plane_maps_to_8391(self) -> None:
"""memory [tracer]: 'memory' → http://<host>:8391 (POST-001)."""
assert (
endpoint_for_plane("memory", "10.100.10.50")
== "http://10.100.10.50:8391"
)
def test_affect_plane_maps_to_8390(self) -> None:
"""affect: 'affect' → http://<host>:8390 (POST-001)."""
assert (
endpoint_for_plane("affect", "10.100.10.50")
== "http://10.100.10.50:8390"
)
def test_combined_plane_maps_to_8392(self) -> None:
"""combined [#18 composite]: 'combined' → http://<host>:8392 (POST-001)."""
assert (
endpoint_for_plane("combined", "10.100.10.50")
== "http://10.100.10.50:8392"
)
def test_unknown_plane_raises_value_error(self) -> None:
"""unknown_plane [adversarial]: any other plane → ValueError (PRE-001)."""
with pytest.raises(ValueError):
endpoint_for_plane("persona", "10.100.10.50")
def _list_item(
*,
session_id: str = "s1",
agent_id: str = "mimir",
created_at: str = "2026-04-15T12:00:00+00:00",
last_active: str = "2026-04-15T12:05:00+00:00",
metadata: dict[str, object] | None = None,
name: str | None = "Research session",
archived: bool = False,
tags: list[str] | None = None,
) -> dict[str, object]:
"""Build a GET /sessions list-item body for tests."""
item: dict[str, object] = {
"session_id": session_id,
"agent_id": agent_id,
"created_at": created_at,
"last_active": last_active,
"metadata": metadata if metadata is not None else {},
"name": name,
"archived": archived,
"tags": tags if tags is not None else ["work"],
}
return item
class TestListSessions:
@respx.mock
async def test_explicit_null_list_defaults(self) -> None:
"""INV-002: explicit-null archived -> False; explicit-null tags -> []."""
raw_item = {
"session_id": "s1",
"agent_id": "mimir",
"created_at": "2026-04-15T12:00:00+00:00",
"last_active": "2026-04-15T12:05:00+00:00",
"metadata": {},
"name": None,
"archived": None,
"tags": None,
}
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200, json={"items": [raw_item], "next_cursor": None}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
page = await list_sessions(client)
info = page.items[0]
assert info.archived is False, "explicit-null archived must default to False"
assert info.tags == [], "explicit-null tags must default to []"
assert info.name is None
@respx.mock
async def test_happy_first_page(self) -> None:
"""happy_first_page [happy,tracer]: one item + next_cursor -> SessionPage shape."""
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200,
json={
"items": [_list_item()],
"next_cursor": "v1.eyJhYmMifQ",
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
page = await list_sessions(client)
assert isinstance(page, SessionPage)
assert len(page.items) == 1
assert page.next_cursor == "v1.eyJhYmMifQ"
info = page.items[0]
assert info.session_id == "s1"
assert info.message_count is None # INV-002: not in list response
assert info.name == "Research session"
assert info.archived is False
assert info.tags == ["work"]
@respx.mock
async def test_happy_last_page(self) -> None:
"""happy_last_page: next_cursor=null -> SessionPage(next_cursor=None)."""
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200,
json={"items": [_list_item()], "next_cursor": None},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
page = await list_sessions(client)
assert page.next_cursor is None
@respx.mock
async def test_empty_results(self) -> None:
"""empty_results: {items: [], next_cursor: null} -> SessionPage([], None)."""
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
page = await list_sessions(client)
assert page == SessionPage(items=[], next_cursor=None)
@respx.mock
async def test_include_archived_query(self) -> None:
"""include_archived_query: True -> has param; default -> NO param at all."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200, json={"items": [], "next_cursor": None}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await list_sessions(client, include_archived=True)
await list_sessions(client) # default
url_with = str(route.calls[0].request.url)
url_default = str(route.calls[1].request.url)
assert "include_archived=true" in url_with
assert "include_archived" not in url_default
@respx.mock
async def test_cursor_threaded(self) -> None:
"""cursor_threaded: cursor=opaque -> URL has cursor=opaque."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200, json={"items": [], "next_cursor": None}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await list_sessions(client, cursor="opaque-from-prev-page")
assert "cursor=opaque-from-prev-page" in str(route.calls[0].request.url)
@respx.mock
async def test_limit_query(self) -> None:
"""limit_query: limit=10 -> URL has limit=10."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
200, json={"items": [], "next_cursor": None}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await list_sessions(client, limit=10)
assert "limit=10" in str(route.calls[0].request.url)
@respx.mock
async def test_invalid_cursor_server(self) -> None:
"""invalid_cursor_server: 422 cursor_invalid -> InvalidCursor(raw=<passed cursor>)."""
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
422,
json={"error_code": "cursor_invalid", "message": "bad cursor"},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(InvalidCursor) as exc_info:
await list_sessions(client, cursor="bogus")
assert exc_info.value.raw == "bogus"
@respx.mock
async def test_other_validation_failed(self) -> None:
"""other_validation_failed: 422 other error_code -> SessionApiFailed(422); truncated."""
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(
422,
json={"error_code": "validation_failed", "message": "limit out of range"},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await list_sessions(client)
assert exc_info.value.status == 422
assert len(exc_info.value.body) <= 1024
@respx.mock
async def test_unexpected_status_truncates(self) -> None:
"""unexpected_status_truncates: 500 + 5000-byte body -> SessionApiFailed; body == 1024."""
big = b"x" * 5000
respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(500, content=big)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await list_sessions(client)
assert exc_info.value.status == 500
assert exc_info.value.body == big[:1024]
@respx.mock
async def test_limit_below_one(self) -> None:
"""limit_below_one [adversarial]: limit=0 -> AssertionError; no HTTP."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await list_sessions(client, limit=0)
assert route.call_count == 0
@respx.mock
async def test_limit_above_max(self) -> None:
"""limit_above_max [adversarial]: limit=300 -> AssertionError; no HTTP."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await list_sessions(client, limit=300)
assert route.call_count == 0
@respx.mock
async def test_empty_cursor(self) -> None:
"""empty_cursor [adversarial]: cursor='' -> AssertionError; no HTTP."""
route = respx.get("https://w.example/sessions").mock(
return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await list_sessions(client, cursor="")
assert route.call_count == 0
# ---- Issue #8: list_agents + AgentInfo --------------------------------------
class TestListAgents:
@respx.mock
async def test_happy_full_shape(self) -> None:
"""happy_full_shape [happy,tracer]: spec full-shape mimir example → all fields."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(
200,
json=[
{
"agent_id": "mimir",
"name": "Mimir",
"description": "Keeper of the Well of Knowledge.",
"version": "0.2.0",
"capabilities": ["knowledge_base", "semantic_search"],
"supported_models": ["default", "heavy"],
"persona_traits": {
"ocean": {
"openness": 0.7,
"conscientiousness": 0.9,
"extraversion": 0.1,
"agreeableness": 0.5,
"neuroticism": 0.3,
},
"vibe": "contemplative",
},
"ui_hints": {"icon": "well", "color_hint": "#5b8aa3"},
}
],
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
agents = await list_agents(client)
assert len(agents) == 1
a = agents[0]
assert isinstance(a, AgentInfo)
assert a.agent_id == "mimir"
assert a.name == "Mimir"
assert a.description == "Keeper of the Well of Knowledge."
assert a.version == "0.2.0"
assert a.capabilities == ["knowledge_base", "semantic_search"]
assert a.supported_models == ["default", "heavy"]
assert a.persona_traits["vibe"] == "contemplative"
assert a.ui_hints["icon"] == "well"
@respx.mock
async def test_happy_minimum_shape(self) -> None:
"""happy_minimum_shape: required-only agent → optional fields default."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(
200,
json=[
{
"agent_id": "minimal",
"name": "Minimal Agent",
"description": "Just a sketch.",
}
],
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
agents = await list_agents(client)
a = agents[0]
assert a.agent_id == "minimal"
assert a.version is None
assert a.capabilities == []
assert a.supported_models == []
assert a.persona_traits == {}
assert a.ui_hints == {}
@respx.mock
async def test_happy_multi_agent(self) -> None:
"""happy_multi_agent: 3 agents preserve order."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(
200,
json=[
{"agent_id": "a", "name": "A", "description": "x"},
{"agent_id": "b", "name": "B", "description": "y"},
{"agent_id": "c", "name": "C", "description": "z"},
],
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
agents = await list_agents(client)
assert [a.agent_id for a in agents] == ["a", "b", "c"]
@respx.mock
async def test_happy_empty(self) -> None:
"""happy_empty: 200 with [] returns empty list (no error)."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(200, json=[])
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
agents = await list_agents(client)
assert agents == []
@respx.mock
async def test_omit_capabilities_empty_list(self) -> None:
"""omit_capabilities_empty: explicit [] from server still defaults to []."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(
200,
json=[
{
"agent_id": "a",
"name": "A",
"description": "x",
"capabilities": [],
}
],
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
agents = await list_agents(client)
assert agents[0].capabilities == []
@respx.mock
async def test_500_raises_session_api_failed(self) -> None:
"""500 → SessionApiFailed with status=500."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(500, content=b"oops")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as excinfo:
await list_agents(client)
assert excinfo.value.status == 500
@respx.mock
async def test_401_raises_session_api_failed(self) -> None:
"""401 → SessionApiFailed with status=401."""
respx.get("https://w.example/agents").mock(
return_value=httpx.Response(401, content=b'{"error":"unauthorized"}')
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as excinfo:
await list_agents(client)
assert excinfo.value.status == 401
class TestGetPersonaState:
"""Worldtree #204 / v0.28.0 — GET /agents/{agent_id}/persona_state.
Bootstrap read for the persona snapshot — same shape as `affect_update`'s
`current` snapshot. Auth via `persona.read` scope (user-tier default).
"""
@respx.mock
async def test_happy_full_snapshot(self) -> None:
"""happy_full_snapshot [happy,tracer]: 200 → snapshot dict with pad +
dominant_emotion + emotions_active + baseline_pad + mood_drift.
"""
snapshot = {
"agent_id": "mimir",
"pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
"dominant_emotion": "curiosity",
"emotions_active": [
{"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7}
],
"baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50},
"mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07},
"last_updated_at": "2026-05-25T22:30:18+00:00",
}
respx.get("https://w.example/agents/mimir/persona_state").mock(
return_value=httpx.Response(200, json=snapshot)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await get_persona_state(client, "mimir")
assert result == snapshot
@respx.mock
async def test_persona_not_configured_404(self) -> None:
"""persona_not_configured_404 [error]: 404 with error_code
persona_not_configured → PersonaNotConfigured. Agent exists but has
no persona surface (e.g. domari, muninn, Tier 3).
"""
respx.get("https://w.example/agents/domari/persona_state").mock(
return_value=httpx.Response(
404, json={"error_code": "persona_not_configured", "message": "no persona"}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(PersonaNotConfigured) as exc_info:
await get_persona_state(client, "domari")
assert exc_info.value.agent_id == "domari"
@respx.mock
async def test_agent_not_available_404(self) -> None:
"""agent_not_available_404 [error]: 404 with error_code
agent_not_available → AgentNotAvailable. Distinct from
persona_not_configured — the agent_id itself is unknown.
"""
respx.get("https://w.example/agents/bogus/persona_state").mock(
return_value=httpx.Response(
404, json={"error_code": "agent_not_available", "message": "unknown agent"}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AgentNotAvailable) as exc_info:
await get_persona_state(client, "bogus")
assert exc_info.value.agent_id == "bogus"
@respx.mock
async def test_auth_scope_denied_403(self) -> None:
"""auth_scope_denied_403 [error]: 403 with error_code auth_scope_denied
→ AuthScopeDenied. Key lacks `persona.read` scope.
"""
respx.get("https://w.example/agents/mimir/persona_state").mock(
return_value=httpx.Response(
403,
json={"error_code": "auth_scope_denied", "message": "missing persona.read"},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AuthScopeDenied) as exc_info:
await get_persona_state(client, "mimir")
assert exc_info.value.scope == "persona.read"
@respx.mock
async def test_404_unknown_error_code_falls_through(self) -> None:
"""404_unknown_error_code_falls_through [adversarial]: 404 without the
two known error codes → SessionApiFailed (don't swallow novel failure
modes as something more specific than they are).
"""
respx.get("https://w.example/agents/mimir/persona_state").mock(
return_value=httpx.Response(404, json={"error_code": "novel_404"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await get_persona_state(client, "mimir")
assert exc_info.value.status == 404
@respx.mock
async def test_500_unexpected_status(self) -> None:
"""500_unexpected_status [error]: 5xx → SessionApiFailed (matches the
list_agents / list_sessions / create_session precedent)."""
respx.get("https://w.example/agents/mimir/persona_state").mock(
return_value=httpx.Response(500, content=b"boom")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc_info:
await get_persona_state(client, "mimir")
assert exc_info.value.status == 500
@respx.mock
async def test_auth_scope_denied_detail_envelope(self) -> None:
"""auth_scope_denied_detail_envelope [regression]: real Worldtree
returns `{"detail": {"error_code": "auth_scope_denied", …}}`
(FastAPI default), not flat `{"error_code": …}`. Smoke against
personal:8081 2026-05-28 surfaced this — pre-fix the response
fell through to SessionApiFailed(403) instead of AuthScopeDenied.
"""
respx.get("https://w.example/agents/mimir/persona_state").mock(
return_value=httpx.Response(
403,
json={
"detail": {
"error_code": "auth_scope_denied",
"message": "Missing required scope: persona.read",
}
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AuthScopeDenied) as exc_info:
await get_persona_state(client, "mimir")
assert exc_info.value.scope == "persona.read"
@respx.mock
async def test_persona_not_configured_detail_envelope(self) -> None:
"""persona_not_configured_detail_envelope [regression]: same
envelope-shape unwrap on 404 + persona_not_configured.
"""
respx.get("https://w.example/agents/domari/persona_state").mock(
return_value=httpx.Response(
404,
json={"detail": {"error_code": "persona_not_configured"}},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(PersonaNotConfigured) as exc_info:
await get_persona_state(client, "domari")
assert exc_info.value.agent_id == "domari"
class TestGetMe:
"""docs/contracts/issues/2.contract.md FN get_me (slice: capabilities+me)."""
@respx.mock
async def test_happy_authenticated(self) -> None:
"""happy_authenticated [happy,tracer]: 200 → parsed identity dict verbatim."""
respx.get("https://w.example/me").mock(
return_value=httpx.Response(
200,
json={
"user_id": "alice",
"scopes": ["conversations.read", "conversations.write"],
"tier": "user",
"key_id": "a1b2c3d4",
"key_label": "alice phone",
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
me = await get_me(client)
assert me["user_id"] == "alice"
assert me["tier"] == "user"
assert me["key_id"] == "a1b2c3d4"
assert me["scopes"] == ["conversations.read", "conversations.write"]
@respx.mock
async def test_anonymous_dev_mode(self) -> None:
"""anonymous_dev_mode: 200 anonymous shape → dict with tier=anonymous."""
respx.get("https://w.example/me").mock(
return_value=httpx.Response(
200,
json={
"user_id": "anonymous",
"scopes": ["conversations.read"],
"tier": "anonymous",
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
me = await get_me(client)
assert me["tier"] == "anonymous"
assert "key_id" not in me # optional fields omitted, not null
@respx.mock
async def test_401_raises_session_api_failed(self) -> None:
"""401_raises [error]: bad/absent key → SessionApiFailed(status=401)."""
respx.get("https://w.example/me").mock(
return_value=httpx.Response(401, json={"detail": "auth_invalid"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_me(client)
assert exc.value.status == 401
class TestGetCapabilities:
"""docs/contracts/issues/2.contract.md FN get_capabilities (slice: capabilities+me)."""
@respx.mock
async def test_happy(self) -> None:
"""happy [happy]: 200 → ephemeral_templates dict verbatim."""
respx.get("https://w.example/capabilities").mock(
return_value=httpx.Response(
200,
json={
"ephemeral_templates": {
"echo": {
"allowed_models": ["glm5-turbo", "glm4.7"],
"default_model": "glm5-turbo",
"system_prompt_max_bytes": 32768,
}
}
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
caps = await get_capabilities(client)
echo = caps["ephemeral_templates"]["echo"]
assert echo["default_model"] == "glm5-turbo"
assert echo["system_prompt_max_bytes"] == 32768
@respx.mock
async def test_non_200_raises(self) -> None:
"""non_200_raises [error]: 500 → SessionApiFailed(status=500)."""
respx.get("https://w.example/capabilities").mock(
return_value=httpx.Response(500, content=b"boom")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_capabilities(client)
assert exc.value.status == 500
class TestGetSessionTools:
"""docs/contracts/issues/2.contract.md — get_session_tools (GET /sessions/{id}/tools, #183)."""
@respx.mock
async def test_happy(self) -> None:
"""happy [happy,tracer]: 200 → merged tool inventory dict verbatim."""
respx.get("https://w.example/sessions/s1/tools").mock(
return_value=httpx.Response(
200,
json={
"agent_id": "alice:wizard",
"builtin_tools": [],
"bifrost_tools": [
{"name": "bifrost.alice.set_field", "description": "d", "parameters": {}}
],
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
tools = await get_session_tools(client, "s1")
assert tools["agent_id"] == "alice:wizard"
assert tools["builtin_tools"] == []
assert tools["bifrost_tools"][0]["name"] == "bifrost.alice.set_field"
@respx.mock
async def test_cross_owner_404_raises(self) -> None:
"""cross_owner_404 [error]: 404 session_not_found → SessionApiFailed(404)."""
respx.get("https://w.example/sessions/s1/tools").mock(
return_value=httpx.Response(404, json={"error_code": "session_not_found"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_session_tools(client, "s1")
assert exc.value.status == 404
@respx.mock
async def test_empty_session_id_asserts(self) -> None:
"""empty_session_id [adversarial]: '' → AssertionError; no HTTP issued."""
route = respx.get("https://w.example/sessions//tools").mock(
return_value=httpx.Response(200, json={})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await get_session_tools(client, "")
assert route.call_count == 0
class TestGetSessionBifrost:
"""#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176)."""
@respx.mock
async def test_happy_uses_admin_bearer(self) -> None:
"""happy [happy,tracer]: 200 → binding dict; request carries the ADMIN bearer (override)."""
route = respx.get("https://w.example/admin/sessions/s1/bifrost").mock(
return_value=httpx.Response(
200,
json={
"endpoint_url": "https://bifrost.example/mcp",
"consumer_id": "alice",
"connected": True,
"capabilities_granted": ["tools:call", "tools:read"],
"tools": [{"name": "bifrost.alice.echo", "description": "echo"}],
},
)
)
async with httpx.AsyncClient(
base_url="https://w.example",
headers={"Authorization": "Bearer consumer-key"},
) as client:
state = await get_session_bifrost(client, "s1", admin_key="admin-xyz")
assert state["connected"] is True
assert state["tools"][0]["name"] == "bifrost.alice.echo"
# the request overrode the client's default consumer bearer with the admin key
assert route.calls[0].request.headers["Authorization"] == "Bearer admin-xyz"
@respx.mock
async def test_403_scope_denied(self) -> None:
"""403 [error]: admin key lacks admin.sessions.read → SessionApiFailed(403)."""
respx.get("https://w.example/admin/sessions/s1/bifrost").mock(
return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_session_bifrost(client, "s1", admin_key="k")
assert exc.value.status == 403
@respx.mock
async def test_404_not_bound(self) -> None:
"""404 [error]: session_not_bifrost_bound → SessionApiFailed(404)."""
respx.get("https://w.example/admin/sessions/s1/bifrost").mock(
return_value=httpx.Response(404, json={"error_code": "session_not_bifrost_bound"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_session_bifrost(client, "s1", admin_key="k")
assert exc.value.status == 404
@respx.mock
async def test_empty_admin_key_asserts(self) -> None:
"""empty_admin_key [adversarial]: '' → AssertionError; no HTTP issued."""
route = respx.get("https://w.example/admin/sessions/s1/bifrost").mock(
return_value=httpx.Response(200, json={})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await get_session_bifrost(client, "s1", admin_key="")
assert route.call_count == 0
class TestTransientCharacters:
"""docs/contracts/issues/2.contract.md — transient-character wrappers (#161)."""
@respx.mock
async def test_list_models(self) -> None:
"""list_models [happy,tracer]: 200 → {items:[...]} verbatim."""
respx.get("https://w.example/models/available-for-characters").mock(
return_value=httpx.Response(200, json={"items": [{"name": "fast", "thinking": False}]})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
models = await list_character_models(client)
assert models["items"][0]["name"] == "fast"
@respx.mock
async def test_create_body_and_response(self) -> None:
"""create [happy]: body is {character, state}; 201 → {character_id, ttl_expires_at}."""
import json as _json
route = respx.post("https://w.example/characters").mock(
return_value=httpx.Response(201, json={"character_id": "char_x", "ttl_expires_at": "t"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
out = await create_character(client, {"schema_version": "1", "name": "H"})
assert out["character_id"] == "char_x"
body = _json.loads(route.calls[0].request.content)
assert body == {"character": {"schema_version": "1", "name": "H"}, "state": None}
@respx.mock
async def test_get_state(self) -> None:
"""get_state [happy]: 200 → live PAD/emotions snapshot."""
respx.get("https://w.example/characters/char_x/state").mock(
return_value=httpx.Response(200, json={"schema_version": "1", "pad": [0.4, 0.1, -0.2]})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
state = await get_character_state(client, "char_x")
assert state["pad"] == [0.4, 0.1, -0.2]
@respx.mock
async def test_delete_204(self) -> None:
"""delete [happy]: 204 → None."""
respx.delete("https://w.example/characters/char_x").mock(
return_value=httpx.Response(204)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
assert await delete_character(client, "char_x") is None
@respx.mock
async def test_create_403_scope(self) -> None:
"""create_403 [error]: key lacks character.write → SessionApiFailed(403)."""
respx.post("https://w.example/characters").mock(
return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await create_character(client, {"name": "H"})
assert exc.value.status == 403
class TestSetPersonaState:
"""#2 contract — set_persona_state (POST /sessions/{id}/persona_state)."""
@respx.mock
async def test_happy_204(self) -> None:
"""happy [happy,tracer]: freeform snapshot body; 204 → None."""
import json as _json
route = respx.post("https://w.example/sessions/s1/persona_state").mock(
return_value=httpx.Response(204)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await set_persona_state(
client, "s1", {"pad": {"pleasure": 0.4, "arousal": 0.1, "dominance": -0.2}}
)
assert result is None
assert _json.loads(route.calls[0].request.content) == {
"pad": {"pleasure": 0.4, "arousal": 0.1, "dominance": -0.2}
}
@respx.mock
async def test_non_204_raises(self) -> None:
"""non_204 [error]: 422 (bad snapshot shape) → SessionApiFailed(422)."""
respx.post("https://w.example/sessions/s1/persona_state").mock(
return_value=httpx.Response(422, json={"error_code": "validation_failed"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await set_persona_state(client, "s1", {"pad": [1, 2, 3]})
assert exc.value.status == 422
_AUTHORED_ACK = {
"author": "assistant",
"content_chars": 5,
"injected_at": "2026-07-06T12:00:00+00:00",
"phase": "seeded",
"seq": 0,
"session_id": "s1",
"turn_id": "t1",
}
class TestWriteAuthoredHistory:
"""write_authored_history — #347 POST /sessions/{id}/history (contract #2 amendment)."""
@respx.mock
async def test_happy_fresh_201(self) -> None:
"""happy_fresh_201 [happy,tracer]: 201 → ack verbatim; minimal body."""
import json as _json
route = respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(201, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await write_authored_history(
client, "s1", content="hello", idempotency_key="k1"
)
assert result == _AUTHORED_ACK
assert _json.loads(route.calls[0].request.content) == {
"author": "assistant",
"content": "hello",
"idempotency_key": "k1",
}
@respx.mock
async def test_happy_replay_200(self) -> None:
"""happy_replay_200 [happy]: 200 replay (byte-identical body) → dict verbatim."""
respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(200, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await write_authored_history(
client, "s1", content="hello", idempotency_key="k1"
)
assert result == _AUTHORED_ACK
@respx.mock
async def test_body_includes_effects(self) -> None:
"""body_includes_effects [trace]: effects + claimed_original_at appear iff non-None."""
import json as _json
route = respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(201, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await write_authored_history(
client,
"s1",
content="hi",
idempotency_key="k1",
effects="none",
claimed_original_at="2020-01-01T00:00:00Z",
)
assert _json.loads(route.calls[0].request.content) == {
"author": "assistant",
"content": "hi",
"idempotency_key": "k1",
"effects": "none",
"claimed_original_at": "2020-01-01T00:00:00Z",
}
@respx.mock
async def test_hide_existence_404(self) -> None:
"""hide_existence_404 [error]: 404 → AuthoredHistoryUnavailable (NOT SessionApiFailed)."""
respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(404, json={"error_code": "session_not_found"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AuthoredHistoryUnavailable) as exc:
await write_authored_history(client, "s1", content="hi", idempotency_key="k1")
assert exc.value.session_id == "s1"
@respx.mock
async def test_generation_active_409(self) -> None:
"""generation_active_409 [error]: 409 → SessionApiFailed(409)."""
respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(409, json={"error_code": "generation_active"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await write_authored_history(client, "s1", content="hi", idempotency_key="k1")
assert exc.value.status == 409
@respx.mock
async def test_content_too_long_422(self) -> None:
"""content_too_long_422 [error]: 422 → SessionApiFailed(422)."""
respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(422, json={"error_code": "content_too_long"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await write_authored_history(client, "s1", content="x", idempotency_key="k1")
assert exc.value.status == 422
@respx.mock
async def test_empty_content(self) -> None:
"""empty_content [adversarial]: content="" → AssertionError; no HTTP issued."""
route = respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(201, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await write_authored_history(client, "s1", content="", idempotency_key="k1")
assert not route.called
@respx.mock
async def test_empty_idempotency_key(self) -> None:
"""empty_idempotency_key [adversarial]: key="" → AssertionError; no HTTP issued."""
route = respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(201, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await write_authored_history(client, "s1", content="hi", idempotency_key="")
assert not route.called
@respx.mock
async def test_empty_session_id(self) -> None:
"""empty_session_id [adversarial]: session_id="" → AssertionError; no HTTP issued."""
route = respx.post("https://w.example/sessions/s1/history").mock(
return_value=httpx.Response(201, json=_AUTHORED_ACK)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await write_authored_history(client, "", content="hi", idempotency_key="k1")
assert not route.called
class TestGetSessionMessages:
"""#2 contract (amendment 2026-07-06) — get_session_messages (GET /sessions/{id}/messages)."""
@respx.mock
async def test_happy(self) -> None:
"""happy [happy,tracer]: 200 {session_id, items, next_cursor} → dict verbatim."""
payload = {
"session_id": "s1",
"items": [{"seq": 0, "role": "assistant", "content": "hello there"}],
"next_cursor": None,
}
respx.get("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(200, json=payload)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await get_session_messages(client, "s1")
assert result == payload
@respx.mock
async def test_not_found_404(self) -> None:
"""not_found_404 [error]: 404 → SessionApiFailed(404)."""
respx.get("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(404, json={"error_code": "session_not_found"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await get_session_messages(client, "s1")
assert exc.value.status == 404
@respx.mock
async def test_empty_session_id(self) -> None:
"""empty_session_id [adversarial]: "" → AssertionError; no HTTP issued."""
route = respx.get("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(200, json={})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await get_session_messages(client, "")
assert not route.called
Reference in New Issue View Git Blame Copy Permalink