File size: 1,668 Bytes
517919c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """Tests for the pluggable agent runtime layer."""
from __future__ import annotations
from pi_test_support import ensure_agent_redact_paths
ensure_agent_redact_paths()
from agent_runtime import ( # noqa: E402
AgentStreamEvent,
PiAgentRuntime,
coerce_agent_runtime,
normalize_orchestrator,
orchestrator_label,
)
from pi_rpc_client import PiStreamEvent # noqa: E402
def test_normalize_orchestrator_defaults_to_pi(monkeypatch):
monkeypatch.delenv("AGENT_ORCHESTRATOR", raising=False)
assert normalize_orchestrator(None) == "pi"
assert normalize_orchestrator("langgraph") == "langgraph"
assert normalize_orchestrator("unknown") == "pi"
def test_orchestrator_label():
assert orchestrator_label("langgraph") == "LangGraph"
assert orchestrator_label("agentcore") == "Bedrock AgentCore Runtime"
assert orchestrator_label("agentcore-harness") == "Bedrock AgentCore Harness"
def test_pi_event_mapping():
from agent_runtime import _pi_event_to_agent_event
mapped = _pi_event_to_agent_event(
PiStreamEvent(kind="text_delta", text="hello", tool_name="bash")
)
assert mapped.kind == "text_delta"
assert mapped.text == "hello"
assert mapped.tool_name == "bash"
def test_coerce_agent_runtime_accepts_pi_adapter():
class _FakePi:
running = True
def drain_pending_ui_history(self):
return []
wrapped = PiAgentRuntime(_FakePi())
assert coerce_agent_runtime(wrapped) is wrapped
def test_agent_stream_event_fields():
event = AgentStreamEvent(kind="done", text="Agent finished.")
assert event.kind == "done"
assert event.is_error is False
|