Spaces:
Running
Running
File size: 9,780 Bytes
05a6bc3 | 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 | """
Tests for the react package: cycle detection and trace schemas.
Cycle detection tests ported from langgraph-agentic-scaffold
(app/tests/unit/test_cycle_detection.py). Import path adjusted.
Schema tests verify Pydantic model instantiation and exception hierarchy.
"""
import pytest
from prompt_prix.react.cycle_detection import detect_cycle, detect_cycle_with_pattern
from prompt_prix.react.schemas import (
ToolCall,
ReActIteration,
ReActLoopTerminated,
MaxIterationsExceeded,
StagnationDetected,
)
# ─────────────────────────────────────────────────────────────────────
# CYCLE DETECTION (ported from LAS)
# ─────────────────────────────────────────────────────────────────────
class TestDetectCycle:
"""Test detect_cycle function."""
def test_single_item_repeated(self):
"""Period-1 cycle: A-A-A-A."""
history = ['a', 'a', 'a', 'a']
assert detect_cycle(history, min_repetitions=2) == 1
assert detect_cycle(history, min_repetitions=3) == 1
assert detect_cycle(history, min_repetitions=4) == 1
def test_two_step_cycle(self):
"""Period-2 cycle: A-B-A-B-A-B."""
history = ['a', 'b', 'a', 'b', 'a', 'b']
assert detect_cycle(history, min_repetitions=2) == 2
assert detect_cycle(history, min_repetitions=3) == 2
def test_four_step_cycle(self):
"""Period-4 cycle: A-B-C-D-A-B-C-D (batch of 4 files)."""
history = ['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd']
assert detect_cycle(history, min_repetitions=2) == 4
def test_four_step_cycle_with_more_repetitions(self):
"""Period-4 cycle repeated 3 times."""
history = ['a', 'b', 'c', 'd'] * 3
assert detect_cycle(history, min_repetitions=2) == 4
assert detect_cycle(history, min_repetitions=3) == 4
def test_no_cycle_short_history(self):
"""Not enough items to detect cycle."""
assert detect_cycle(['a'], min_repetitions=2) is None
assert detect_cycle(['a', 'b'], min_repetitions=2) is None
assert detect_cycle(['a', 'b', 'c'], min_repetitions=2) is None
def test_no_cycle_different_items(self):
"""No repeating pattern."""
history = ['a', 'b', 'c', 'd', 'e', 'f']
assert detect_cycle(history, min_repetitions=2) is None
def test_cycle_at_end_only(self):
"""Cycle detection should focus on the end of history."""
history = ['x', 'y', 'z', 'a', 'b', 'a', 'b']
assert detect_cycle(history, min_repetitions=2) == 2
def test_shortest_cycle_found_first(self):
"""When multiple cycles match, shortest period wins."""
history = ['a', 'a', 'a', 'a']
assert detect_cycle(history, min_repetitions=2) == 1
def test_max_period_limit(self):
"""Respect max_period parameter."""
history = ['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd']
assert detect_cycle(history, min_repetitions=2, max_period=3) is None
assert detect_cycle(history, min_repetitions=2, max_period=4) == 4
def test_tool_call_signatures(self):
"""Real-world tool call signature cycle (LAS Issue #78 scenario)."""
signatures = [
"read_file:path=sort_by_contents/c.txt",
"read_file:path=sort_by_contents/k.txt",
"read_file:path=sort_by_contents/s.txt",
"read_file:path=sort_by_contents/v.txt",
"read_file:path=sort_by_contents/c.txt",
"read_file:path=sort_by_contents/k.txt",
"read_file:path=sort_by_contents/s.txt",
"read_file:path=sort_by_contents/v.txt",
]
assert detect_cycle(signatures, min_repetitions=2) == 4
def test_empty_history(self):
"""Empty history returns None."""
assert detect_cycle([], min_repetitions=2) is None
class TestDetectCycleWithPattern:
"""Test detect_cycle_with_pattern function."""
def test_returns_pattern(self):
"""Should return both period and pattern."""
history = ['a', 'b', 'a', 'b']
period, pattern = detect_cycle_with_pattern(history, min_repetitions=2)
assert period == 2
assert pattern == ['a', 'b']
def test_four_item_pattern(self):
"""Four-item pattern from batch operation."""
history = ['c.txt', 'k.txt', 's.txt', 'v.txt', 'c.txt', 'k.txt', 's.txt', 'v.txt']
period, pattern = detect_cycle_with_pattern(history, min_repetitions=2)
assert period == 4
assert pattern == ['c.txt', 'k.txt', 's.txt', 'v.txt']
def test_no_cycle_returns_none(self):
"""No cycle returns (None, None)."""
history = ['a', 'b', 'c', 'd']
period, pattern = detect_cycle_with_pattern(history, min_repetitions=2)
assert period is None
assert pattern is None
def test_single_item_pattern(self):
"""Period-1 cycle has single-item pattern."""
history = ['a', 'a', 'a']
period, pattern = detect_cycle_with_pattern(history, min_repetitions=2)
assert period == 1
assert pattern == ['a']
class TestMinRepetitionsEdgeCases:
"""Test min_repetitions parameter edge cases."""
def test_min_repetitions_1_not_useful(self):
"""min_repetitions=1 would match everything, so not typically used."""
history = ['a', 'b', 'c']
assert detect_cycle(history, min_repetitions=1) == 1
def test_high_min_repetitions(self):
"""Need enough history for high min_repetitions."""
history = ['a', 'b'] * 5 # 10 items
assert detect_cycle(history, min_repetitions=5) == 2
assert detect_cycle(history, min_repetitions=6) is None # Would need 12 items
# ─────────────────────────────────────────────────────────────────────
# SCHEMA TESTS
# ─────────────────────────────────────────────────────────────────────
class TestToolCall:
"""Test ToolCall Pydantic model."""
def test_basic_instantiation(self):
tc = ToolCall(id="call_1", name="read_file", args={"path": "./1.txt"})
assert tc.id == "call_1"
assert tc.name == "read_file"
assert tc.args == {"path": "./1.txt"}
def test_default_args(self):
tc = ToolCall(id="call_2", name="list_directory")
assert tc.args == {}
def test_serialization_roundtrip(self):
tc = ToolCall(id="call_3", name="move_file", args={"src": "a.txt", "dst": "b/"})
data = tc.model_dump()
restored = ToolCall(**data)
assert restored == tc
class TestReActIteration:
"""Test ReActIteration Pydantic model."""
def test_successful_iteration(self):
tc = ToolCall(id="call_1", name="read_file", args={"path": "test.txt"})
step = ReActIteration(
iteration=1,
tool_call=tc,
observation="File contents here",
success=True,
thought="I should read this file first",
latency_ms=42.5,
)
assert step.iteration == 1
assert step.tool_call.name == "read_file"
assert step.success is True
assert step.thought == "I should read this file first"
assert step.latency_ms == 42.5
def test_failed_iteration(self):
tc = ToolCall(id="call_2", name="unknown_tool", args={})
step = ReActIteration(
iteration=3,
tool_call=tc,
observation="Error: No mock response for unknown_tool({})",
success=False,
)
assert step.success is False
assert step.thought is None
assert step.latency_ms == 0.0
def test_serialization_roundtrip(self):
tc = ToolCall(id="call_1", name="read_file", args={"path": "x"})
step = ReActIteration(
iteration=1, tool_call=tc,
observation="data", success=True, latency_ms=10.0,
)
data = step.model_dump()
restored = ReActIteration(**data)
assert restored == step
class TestExceptionHierarchy:
"""Test exception class hierarchy."""
def test_max_iterations_is_loop_terminated(self):
assert issubclass(MaxIterationsExceeded, ReActLoopTerminated)
def test_stagnation_is_loop_terminated(self):
assert issubclass(StagnationDetected, ReActLoopTerminated)
def test_loop_terminated_is_exception(self):
assert issubclass(ReActLoopTerminated, Exception)
def test_catch_base_catches_both(self):
"""Catching ReActLoopTerminated catches both subclasses."""
for exc_class in (MaxIterationsExceeded, StagnationDetected):
with pytest.raises(ReActLoopTerminated):
raise exc_class("test")
class TestPackageImports:
"""Test that package __init__.py exports work."""
def test_import_from_package(self):
from prompt_prix.react import (
detect_cycle,
detect_cycle_with_pattern,
ToolCall,
ReActIteration,
ReActLoopTerminated,
MaxIterationsExceeded,
StagnationDetected,
)
# Verify they're the actual classes, not None
assert callable(detect_cycle)
assert callable(detect_cycle_with_pattern)
assert ToolCall is not None
assert ReActIteration is not None
|