File size: 4,710 Bytes
31c8075 | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """Tests for the standalone Thread inspector seam."""
from dataclasses import dataclass, field
import pytest
from arena.thread_inspector import (
AgentRunFailed,
FileAccessFailed,
ThreadInspector,
ThreadNotFound,
)
class FakeResult:
def __init__(self, *, error: str | None = None, entries=None, file_data=None) -> None:
self.error = error
self.entries = entries
self.file_data = file_data
class FakeBackend:
def __init__(self) -> None:
self.listed: list[str] = []
self.reads: list[str] = []
def ls(self, path: str) -> FakeResult:
self.listed.append(path)
return FakeResult(entries=[{"path": f"{path}/hi.txt", "is_dir": False}])
def read(self, path: str) -> FakeResult:
self.reads.append(path)
return FakeResult(file_data={"content": f"contents of {path}"})
class FakeAgent:
def __init__(self) -> None:
self.invocations = 0
def invoke(self, payload, config):
self.invocations += 1
return {"messages": [{"role": "assistant", "content": f"reply #{self.invocations}"}]}
@dataclass
class FakeSession:
thread_id: str
provider: str
backend: FakeBackend
agent: FakeAgent
messages: list[dict[str, str]] = field(default_factory=list)
closed: bool = False
def close(self) -> None:
self.closed = True
def make_inspector():
backend = FakeBackend()
agent = FakeAgent()
def factory(thread_id: str | None = None):
return FakeSession(
thread_id=thread_id or "thread-1",
provider="local",
backend=backend,
agent=agent,
)
return ThreadInspector(session_factory=factory), backend, agent
def test_inspector_create_thread_records_session() -> None:
inspector, _, _ = make_inspector()
response = inspector.create_thread()
assert response.thread_id == "thread-1"
assert response.sandbox_provider == "local"
assert "thread-1" in inspector.sessions
def test_inspector_send_message_appends_user_and_assistant_turns() -> None:
inspector, _, _ = make_inspector()
inspector.create_thread()
response = inspector.send_message("thread-1", "hello")
session = inspector.sessions["thread-1"]
assert session.messages[0]["role"] == "user"
assert session.messages[0]["content"] == "hello"
assert session.messages[1]["role"] == "assistant"
assert response.content == "reply #1"
def test_inspector_send_message_wraps_agent_failures() -> None:
inspector, _, _ = make_inspector()
inspector.create_thread()
original_invoke = inspector.sessions["thread-1"].agent.invoke
def boom(payload, config):
original_invoke(payload, config)
raise RuntimeError("kaboom")
inspector.sessions["thread-1"].agent.invoke = boom
with pytest.raises(AgentRunFailed):
inspector.send_message("thread-1", "hello")
def test_inspector_list_files_uses_backend_path() -> None:
inspector, backend, _ = make_inspector()
inspector.create_thread()
entries = inspector.list_files("thread-1", path="/work")
assert [entry.path for entry in entries] == ["/work/hi.txt"]
assert backend.listed == ["/work"]
def test_inspector_read_file_returns_decoded_content() -> None:
inspector, backend, _ = make_inspector()
inspector.create_thread()
payload = inspector.read_file("thread-1", "/work/hi.txt")
assert payload == {"path": "/work/hi.txt", "content": "contents of /work/hi.txt"}
assert backend.reads == ["/work/hi.txt"]
def test_inspector_list_files_wraps_backend_error() -> None:
inspector, _, _ = make_inspector()
inspector.create_thread()
inspector.sessions["thread-1"].backend.ls = lambda path: FakeResult(error="nope")
with pytest.raises(FileAccessFailed):
inspector.list_files("thread-1", path="/work")
def test_inspector_delete_thread_drops_session_and_closes_backend() -> None:
inspector, _, _ = make_inspector()
inspector.create_thread()
session = inspector.sessions["thread-1"]
result = inspector.delete_thread("thread-1")
assert result == {"deleted": True}
assert "thread-1" not in inspector.sessions
assert session.closed is True
def test_inspector_unknown_thread_raises_thread_not_found() -> None:
inspector, _, _ = make_inspector()
with pytest.raises(ThreadNotFound):
inspector.send_message("missing", "hello")
with pytest.raises(ThreadNotFound):
inspector.list_files("missing", path="/")
with pytest.raises(ThreadNotFound):
inspector.read_file("missing", "/x")
with pytest.raises(ThreadNotFound):
inspector.delete_thread("missing")
|