Spaces:
Running
Running
File size: 11,355 Bytes
0a70295 | 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 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 | """Tests for react_step() MCP tool — single ReAct iteration primitive.
Per ADR-006: MCP tool tests mock the adapter layer.
react_step() calls complete_stream(), so we mock that.
"""
import json
import pytest
from unittest.mock import patch
from prompt_prix.react.schemas import ToolCall, ReActIteration
# ─────────────────────────────────────────────────────────────────────
# HELPERS
# ─────────────────────────────────────────────────────────────────────
def _tool_call_sentinel(name: str, arguments: dict) -> str:
"""Build a __TOOL_CALLS__ sentinel string."""
return f"__TOOL_CALLS__:{json.dumps([{'name': name, 'arguments': json.dumps(arguments)}])}"
def _make_stream(*chunks):
"""Create an async generator that yields chunks then a latency sentinel."""
async def stream(**kwargs):
for chunk in chunks:
yield chunk
yield "__LATENCY_MS__:100"
return stream
# ─────────────────────────────────────────────────────────────────────
# MOCK TOOL DISPATCH TESTS
# ─────────────────────────────────────────────────────────────────────
class TestDispatchMock:
"""Test dispatch_mock() resolution logic."""
def test_exact_args_match(self):
from prompt_prix.mcp.tools.react_step import dispatch_mock
mock_tools = {
"read_file": {
json.dumps({"path": "./1.txt"}, sort_keys=True): "File contents here"
}
}
result = dispatch_mock("read_file", {"path": "./1.txt"}, mock_tools)
assert result == "File contents here"
def test_first_arg_value_match(self):
from prompt_prix.mcp.tools.react_step import dispatch_mock
mock_tools = {
"read_file": {
"./1.txt": "File contents here"
}
}
result = dispatch_mock("read_file", {"path": "./1.txt"}, mock_tools)
assert result == "File contents here"
def test_default_fallback(self):
from prompt_prix.mcp.tools.react_step import dispatch_mock
mock_tools = {
"move_file": {
"_default": "File moved"
}
}
result = dispatch_mock("move_file", {"src": "a.txt", "dst": "b/"}, mock_tools)
assert result == "File moved"
def test_no_match_returns_error(self):
from prompt_prix.mcp.tools.react_step import dispatch_mock
mock_tools = {
"read_file": {
"./known.txt": "Known content"
}
}
result = dispatch_mock("read_file", {"path": "./unknown.txt"}, mock_tools)
assert "Error" in result
assert "read_file" in result
def test_unknown_tool_returns_error(self):
from prompt_prix.mcp.tools.react_step import dispatch_mock
mock_tools = {}
result = dispatch_mock("nonexistent", {"arg": "val"}, mock_tools)
assert "Error" in result
# ─────────────────────────────────────────────────────────────────────
# MESSAGE BUILDING TESTS
# ─────────────────────────────────────────────────────────────────────
class TestBuildMessages:
"""Test build_react_messages() trace serialization."""
def test_empty_trace(self):
from prompt_prix.mcp.tools.react_step import build_react_messages
msgs = build_react_messages("You are helpful.", "Do the thing.", [])
assert len(msgs) == 2
assert msgs[0]["role"] == "system"
assert msgs[1]["role"] == "user"
def test_trace_produces_assistant_tool_pairs(self):
from prompt_prix.mcp.tools.react_step import build_react_messages
trace = [
ReActIteration(
iteration=1,
tool_call=ToolCall(id="call_1", name="read_file", args={"path": "x"}),
observation="file data",
success=True,
thought="Let me read this file",
)
]
msgs = build_react_messages("sys", "goal", trace)
assert len(msgs) == 4 # system, user, assistant, tool
assert msgs[2]["role"] == "assistant"
assert msgs[2]["tool_calls"][0]["function"]["name"] == "read_file"
assert msgs[3]["role"] == "tool"
assert msgs[3]["content"] == "file data"
assert msgs[3]["tool_call_id"] == "call_1"
# ─────────────────────────────────────────────────────────────────────
# REACT_STEP SINGLE-ITERATION TESTS
# ─────────────────────────────────────────────────────────────────────
class TestReactStep:
"""Tests for react_step() — one model call, one result."""
@pytest.mark.asyncio
async def test_model_completes_with_text(self):
"""Model responds with text only (no tool calls) → completed."""
async def mock_stream(**kwargs):
yield "The answer is 42."
yield "__LATENCY_MS__:50"
with patch("prompt_prix.mcp.tools.react_step.complete_stream", side_effect=mock_stream):
from prompt_prix.mcp.tools.react_step import react_step
result = await react_step(
model_id="test-model",
system_prompt="sys",
initial_message="What is the answer?",
trace=[],
mock_tools={},
tools=[],
)
assert result["completed"] is True
assert result["final_response"] == "The answer is 42."
assert result["new_iterations"] == []
assert result["latency_ms"] == 50.0
@pytest.mark.asyncio
async def test_model_makes_tool_call(self):
"""Model makes a tool call → returns new iteration, not completed."""
sentinel = _tool_call_sentinel("read_file", {"path": "./1.txt"})
async def mock_stream(**kwargs):
yield "I'll read the file."
yield sentinel
yield "__LATENCY_MS__:80"
with patch("prompt_prix.mcp.tools.react_step.complete_stream", side_effect=mock_stream):
from prompt_prix.mcp.tools.react_step import react_step
result = await react_step(
model_id="test-model",
system_prompt="sys",
initial_message="Read the file",
trace=[],
mock_tools={"read_file": {"./1.txt": "File contents here"}},
tools=[{"type": "function", "function": {"name": "read_file"}}],
)
assert result["completed"] is False
assert result["final_response"] is None
assert len(result["new_iterations"]) == 1
iteration = result["new_iterations"][0]
assert isinstance(iteration, ReActIteration)
assert iteration.tool_call.name == "read_file"
assert iteration.observation == "File contents here"
assert iteration.success is True
assert iteration.thought == "I'll read the file."
@pytest.mark.asyncio
async def test_garbled_tool_args(self):
"""Model produces unparseable tool args → invalid iteration."""
async def mock_stream(**kwargs):
yield "I'll read the file."
yield '__TOOL_CALLS__:[{"name":"read_file","arguments":"not valid json"}]'
yield "__LATENCY_MS__:50"
with patch("prompt_prix.mcp.tools.react_step.complete_stream", side_effect=mock_stream):
from prompt_prix.mcp.tools.react_step import react_step
result = await react_step(
model_id="test-model",
system_prompt="sys",
initial_message="Read it",
trace=[],
mock_tools={"read_file": {"./1.txt": "data"}},
tools=[{"type": "function", "function": {"name": "read_file"}}],
)
assert result["completed"] is False
assert len(result["new_iterations"]) == 1
assert result["new_iterations"][0].success is False
assert "Error" in result["new_iterations"][0].observation
@pytest.mark.asyncio
async def test_call_counter_threads_through(self):
"""call_counter increments and returns updated value."""
sentinel = _tool_call_sentinel("read_file", {"path": "./1.txt"})
async def mock_stream(**kwargs):
yield sentinel
yield "__LATENCY_MS__:50"
with patch("prompt_prix.mcp.tools.react_step.complete_stream", side_effect=mock_stream):
from prompt_prix.mcp.tools.react_step import react_step
result = await react_step(
model_id="test-model",
system_prompt="sys",
initial_message="Read",
trace=[],
mock_tools={"read_file": {"_default": "data"}},
tools=[{"type": "function", "function": {"name": "read_file"}}],
call_counter=5,
)
assert result["call_counter"] == 6
assert result["new_iterations"][0].tool_call.id == "call_6"
@pytest.mark.asyncio
async def test_trace_passed_to_message_builder(self):
"""Previous trace entries are included in messages sent to model."""
existing_trace = [
ReActIteration(
iteration=1,
tool_call=ToolCall(id="call_1", name="list_dir", args={"path": "."}),
observation="file1.txt\nfile2.txt",
success=True,
)
]
async def mock_stream(**kwargs):
# Verify trace was included in messages
messages = kwargs.get("messages", [])
assert len(messages) == 4 # system, user, assistant, tool
assert messages[2]["role"] == "assistant"
assert messages[3]["role"] == "tool"
yield "All done."
yield "__LATENCY_MS__:30"
with patch("prompt_prix.mcp.tools.react_step.complete_stream", side_effect=mock_stream):
from prompt_prix.mcp.tools.react_step import react_step
result = await react_step(
model_id="test-model",
system_prompt="sys",
initial_message="List and report",
trace=existing_trace,
mock_tools={},
tools=[],
call_counter=1,
)
assert result["completed"] is True
assert result["final_response"] == "All done."
|