loader Public

Watch 0 Fork 0 Star 0
Python · 9172 bytes Raw Blame History
1 """Tests for tool implementations."""
2
3 import asyncio
4
5 import pytest
6
7 from loader.tools import (
8 BashTool,
9 ConfirmationRequired,
10 EditTool,
11 GlobTool,
12 GrepTool,
13 ReadTool,
14 WriteTool,
15 )
16 from loader.tools.base import create_default_registry
17
18
19 class TestReadTool:
20 """Tests for ReadTool."""
21
22 @pytest.fixture
23 def tool(self):
24 return ReadTool()
25
26 @pytest.mark.asyncio
27 async def test_read_file(self, tool, sample_file):
28 result = await tool.execute(file_path=str(sample_file))
29 assert not result.is_error
30 assert "Line 1" in result.output
31 assert "Line 2" in result.output
32
33 @pytest.mark.asyncio
34 async def test_read_nonexistent(self, tool, temp_dir):
35 result = await tool.execute(file_path=str(temp_dir / "nonexistent.txt"))
36 assert result.is_error
37 assert "not found" in result.output.lower()
38
39 @pytest.mark.asyncio
40 async def test_read_with_offset(self, tool, sample_file):
41 result = await tool.execute(file_path=str(sample_file), offset=2, limit=1)
42 assert not result.is_error
43 assert "Line 2" in result.output
44 assert "Line 1" not in result.output
45
46 def test_is_not_destructive(self, tool):
47 assert not tool.is_destructive
48
49
50 class TestWriteTool:
51 """Tests for WriteTool."""
52
53 @pytest.fixture
54 def tool(self):
55 return WriteTool()
56
57 @pytest.mark.asyncio
58 async def test_write_file(self, tool, temp_dir):
59 file_path = temp_dir / "new_file.txt"
60 result = await tool.execute(file_path=str(file_path), content="Hello, world!")
61 assert not result.is_error
62 assert file_path.exists()
63 assert file_path.read_text() == "Hello, world!"
64
65 @pytest.mark.asyncio
66 async def test_write_creates_parents(self, tool, temp_dir):
67 file_path = temp_dir / "subdir" / "deep" / "file.txt"
68 result = await tool.execute(file_path=str(file_path), content="nested")
69 assert not result.is_error
70 assert file_path.exists()
71
72 def test_is_destructive(self, tool):
73 assert tool.is_destructive
74
75 def test_requires_confirmation(self, tool):
76 with pytest.raises(ConfirmationRequired) as exc_info:
77 tool.check_confirmation(
78 skip_confirmation=False,
79 file_path="/tmp/test.txt",
80 content="test",
81 )
82 assert "Write to file" in exc_info.value.message
83
84 def test_skip_confirmation(self, tool):
85 # Should not raise
86 tool.check_confirmation(
87 skip_confirmation=True,
88 file_path="/tmp/test.txt",
89 content="test",
90 )
91
92
93 class TestEditTool:
94 """Tests for EditTool."""
95
96 @pytest.fixture
97 def tool(self):
98 return EditTool()
99
100 @pytest.mark.asyncio
101 async def test_edit_file(self, tool, sample_file):
102 result = await tool.execute(
103 file_path=str(sample_file),
104 old_string="Line 2",
105 new_string="Modified Line 2",
106 )
107 assert not result.is_error
108 assert "Modified Line 2" in sample_file.read_text()
109
110 @pytest.mark.asyncio
111 async def test_edit_nonexistent(self, tool, temp_dir):
112 result = await tool.execute(
113 file_path=str(temp_dir / "nonexistent.txt"),
114 old_string="foo",
115 new_string="bar",
116 )
117 assert result.is_error
118
119 @pytest.mark.asyncio
120 async def test_edit_string_not_found(self, tool, sample_file):
121 result = await tool.execute(
122 file_path=str(sample_file),
123 old_string="Not in file",
124 new_string="replacement",
125 )
126 assert result.is_error
127 assert "not found" in result.output.lower()
128
129
130 class TestGlobTool:
131 """Tests for GlobTool."""
132
133 @pytest.fixture
134 def tool(self):
135 return GlobTool()
136
137 @pytest.mark.asyncio
138 async def test_glob_finds_files(self, tool, temp_dir):
139 (temp_dir / "file1.py").write_text("# python")
140 (temp_dir / "file2.py").write_text("# python")
141 (temp_dir / "file3.txt").write_text("text")
142
143 result = await tool.execute(pattern="*.py", path=str(temp_dir))
144 assert not result.is_error
145 assert "file1.py" in result.output
146 assert "file2.py" in result.output
147 assert "file3.txt" not in result.output
148
149 @pytest.mark.asyncio
150 async def test_glob_no_matches(self, tool, temp_dir):
151 result = await tool.execute(pattern="*.xyz", path=str(temp_dir))
152 assert not result.is_error
153 assert "No files matching" in result.output
154
155
156 class TestBashTool:
157 """Tests for BashTool."""
158
159 @pytest.fixture
160 def tool(self):
161 return BashTool()
162
163 @pytest.mark.asyncio
164 async def test_bash_simple_command(self, tool):
165 result = await tool.execute(command="echo 'hello world'")
166 assert not result.is_error
167 assert "hello world" in result.output
168
169 @pytest.mark.asyncio
170 async def test_bash_pwd(self, tool):
171 result = await tool.execute(command="pwd")
172 assert not result.is_error
173 assert "/" in result.output
174
175 @pytest.mark.asyncio
176 async def test_bash_failed_command(self, tool):
177 result = await tool.execute(command="exit 1")
178 assert result.is_error
179 assert "Exit code 1" in result.output
180
181 @pytest.mark.asyncio
182 async def test_bash_background_launch_and_wait(self, tool):
183 launch = await tool.execute(
184 command='python -c "import time; print(\'ready\'); time.sleep(0.1)"',
185 background=True,
186 )
187
188 assert not launch.is_error
189 job_id = launch.metadata["job_id"]
190 assert job_id.startswith("bash-")
191
192 await asyncio.sleep(0.05)
193 wait_result = await tool.manager.wait_for_job(job_id)
194
195 assert not wait_result.is_error
196 assert "ready" in wait_result.output
197 assert wait_result.metadata["status"] == "completed"
198
199 @pytest.mark.asyncio
200 async def test_bash_background_job_can_be_killed(self, tool):
201 launch = await tool.execute(
202 command='python -c "import time; print(\'server\'); time.sleep(30)"',
203 background=True,
204 )
205 job_id = launch.metadata["job_id"]
206
207 kill_result = await tool.manager.kill_job(job_id)
208
209 assert not kill_result.is_error
210 assert f"bash job {job_id}" in kill_result.output
211 assert kill_result.metadata["status"] == "killed"
212 assert kill_result.metadata["interrupted"] is False
213 assert kill_result.metadata["killed"] is True
214
215 @pytest.mark.asyncio
216 async def test_bash_rejects_long_running_foreground_command(self, tool):
217 result = await tool.execute(command="python -m http.server 8000")
218
219 assert result.is_error
220 assert "background=true" in result.output
221 assert result.metadata["suggest_background"] is True
222
223 def test_is_destructive(self, tool):
224 assert tool.is_destructive
225
226 def test_safe_command_no_confirmation(self, tool):
227 # ls is safe
228 tool.check_confirmation(skip_confirmation=False, command="ls -la")
229 # git status is safe
230 tool.check_confirmation(skip_confirmation=False, command="git status")
231
232 def test_unsafe_command_requires_confirmation(self, tool):
233 with pytest.raises(ConfirmationRequired):
234 tool.check_confirmation(skip_confirmation=False, command="rm -rf /tmp/test")
235
236
237 class TestGrepTool:
238 """Tests for GrepTool."""
239
240 @pytest.fixture
241 def tool(self):
242 return GrepTool()
243
244 @pytest.mark.asyncio
245 async def test_grep_finds_pattern(self, tool, sample_python_file):
246 result = await tool.execute(
247 pattern="def.*hello",
248 path=str(sample_python_file),
249 )
250 assert not result.is_error
251 assert "hello" in result.output
252
253 @pytest.mark.asyncio
254 async def test_grep_no_matches(self, tool, sample_file):
255 result = await tool.execute(
256 pattern="nonexistent_pattern",
257 path=str(sample_file),
258 )
259 assert not result.is_error
260 assert "No matches" in result.output
261
262
263 class TestToolRegistry:
264 """Tests for ToolRegistry."""
265
266 def test_create_default_registry(self):
267 registry = create_default_registry()
268 assert registry.get("read") is not None
269 assert registry.get("write") is not None
270 assert registry.get("edit") is not None
271 assert registry.get("glob") is not None
272 assert registry.get("bash") is not None
273 assert registry.get("grep") is not None
274
275 def test_unknown_tool(self):
276 registry = create_default_registry()
277 assert registry.get("nonexistent") is None
278
279 @pytest.mark.asyncio
280 async def test_execute_unknown_tool(self):
281 registry = create_default_registry()
282 result = await registry.execute("nonexistent")
283 assert result.is_error
284 assert "Unknown tool" in result.output
285
286 def test_skip_confirmation_flag(self):
287 registry = create_default_registry()
288 assert not registry.skip_confirmation
289 registry.skip_confirmation = True
290 assert registry.skip_confirmation