loader Public

Watch 0 Fork 0 Star 0
Python · 10370 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_accepts_path_content_replacement(self, tool, sample_file):
112 result = await tool.execute(
113 path=str(sample_file),
114 content="Replacement file\nwith details\n",
115 )
116
117 assert not result.is_error
118 assert sample_file.read_text() == "Replacement file\nwith details\n"
119 assert result.metadata["structured_patch"]
120
121 @pytest.mark.asyncio
122 async def test_edit_nonexistent(self, tool, temp_dir):
123 result = await tool.execute(
124 file_path=str(temp_dir / "nonexistent.txt"),
125 old_string="foo",
126 new_string="bar",
127 )
128 assert result.is_error
129
130 @pytest.mark.asyncio
131 async def test_edit_string_not_found(self, tool, sample_file):
132 result = await tool.execute(
133 file_path=str(sample_file),
134 old_string="Not in file",
135 new_string="replacement",
136 )
137 assert result.is_error
138 assert "not found" in result.output.lower()
139
140
141 class TestGlobTool:
142 """Tests for GlobTool."""
143
144 @pytest.fixture
145 def tool(self):
146 return GlobTool()
147
148 @pytest.mark.asyncio
149 async def test_glob_finds_files(self, tool, temp_dir):
150 (temp_dir / "file1.py").write_text("# python")
151 (temp_dir / "file2.py").write_text("# python")
152 (temp_dir / "file3.txt").write_text("text")
153
154 result = await tool.execute(pattern="*.py", path=str(temp_dir))
155 assert not result.is_error
156 assert "file1.py" in result.output
157 assert "file2.py" in result.output
158 assert "file3.txt" not in result.output
159
160 @pytest.mark.asyncio
161 async def test_glob_no_matches(self, tool, temp_dir):
162 result = await tool.execute(pattern="*.xyz", path=str(temp_dir))
163 assert not result.is_error
164 assert "No files matching" in result.output
165
166 @pytest.mark.asyncio
167 async def test_glob_expands_home_prefixed_pattern(self, tool, monkeypatch, temp_dir):
168 home_dir = temp_dir / "fake-home"
169 animals_dir = home_dir / "Loader" / "animals"
170 animals_dir.mkdir(parents=True)
171 (animals_dir / "penguins.html").write_text("<h1>Penguins</h1>\n")
172 (animals_dir / "wolves.html").write_text("<h1>Wolves</h1>\n")
173
174 monkeypatch.setenv("HOME", str(home_dir))
175
176 result = await tool.execute(pattern="~/Loader/animals/*.html")
177
178 assert not result.is_error
179 assert "penguins.html" in result.output
180 assert "wolves.html" in result.output
181 assert result.metadata["base_path"] == str(animals_dir.resolve())
182 assert result.metadata["effective_pattern"] == "*.html"
183
184
185 class TestBashTool:
186 """Tests for BashTool."""
187
188 @pytest.fixture
189 def tool(self):
190 return BashTool()
191
192 @pytest.mark.asyncio
193 async def test_bash_simple_command(self, tool):
194 result = await tool.execute(command="echo 'hello world'")
195 assert not result.is_error
196 assert "hello world" in result.output
197
198 @pytest.mark.asyncio
199 async def test_bash_pwd(self, tool):
200 result = await tool.execute(command="pwd")
201 assert not result.is_error
202 assert "/" in result.output
203
204 @pytest.mark.asyncio
205 async def test_bash_failed_command(self, tool):
206 result = await tool.execute(command="exit 1")
207 assert result.is_error
208 assert "Exit code 1" in result.output
209
210 @pytest.mark.asyncio
211 async def test_bash_background_launch_and_wait(self, tool):
212 launch = await tool.execute(
213 command='python -c "import time; print(\'ready\'); time.sleep(0.1)"',
214 background=True,
215 )
216
217 assert not launch.is_error
218 job_id = launch.metadata["job_id"]
219 assert job_id.startswith("bash-")
220
221 await asyncio.sleep(0.05)
222 wait_result = await tool.manager.wait_for_job(job_id)
223
224 assert not wait_result.is_error
225 assert "ready" in wait_result.output
226 assert wait_result.metadata["status"] == "completed"
227
228 @pytest.mark.asyncio
229 async def test_bash_background_job_can_be_killed(self, tool):
230 launch = await tool.execute(
231 command='python -c "import time; print(\'server\'); time.sleep(30)"',
232 background=True,
233 )
234 job_id = launch.metadata["job_id"]
235
236 kill_result = await tool.manager.kill_job(job_id)
237
238 assert not kill_result.is_error
239 assert f"bash job {job_id}" in kill_result.output
240 assert kill_result.metadata["status"] == "killed"
241 assert kill_result.metadata["interrupted"] is False
242 assert kill_result.metadata["killed"] is True
243
244 @pytest.mark.asyncio
245 async def test_bash_rejects_long_running_foreground_command(self, tool):
246 result = await tool.execute(command="python -m http.server 8000")
247
248 assert result.is_error
249 assert "background=true" in result.output
250 assert result.metadata["suggest_background"] is True
251
252 def test_is_destructive(self, tool):
253 assert tool.is_destructive
254
255 def test_safe_command_no_confirmation(self, tool):
256 # ls is safe
257 tool.check_confirmation(skip_confirmation=False, command="ls -la")
258 # git status is safe
259 tool.check_confirmation(skip_confirmation=False, command="git status")
260
261 def test_unsafe_command_requires_confirmation(self, tool):
262 with pytest.raises(ConfirmationRequired):
263 tool.check_confirmation(skip_confirmation=False, command="rm -rf /tmp/test")
264
265
266 class TestGrepTool:
267 """Tests for GrepTool."""
268
269 @pytest.fixture
270 def tool(self):
271 return GrepTool()
272
273 @pytest.mark.asyncio
274 async def test_grep_finds_pattern(self, tool, sample_python_file):
275 result = await tool.execute(
276 pattern="def.*hello",
277 path=str(sample_python_file),
278 )
279 assert not result.is_error
280 assert "hello" in result.output
281
282 @pytest.mark.asyncio
283 async def test_grep_no_matches(self, tool, sample_file):
284 result = await tool.execute(
285 pattern="nonexistent_pattern",
286 path=str(sample_file),
287 )
288 assert not result.is_error
289 assert "No matches" in result.output
290
291
292 class TestToolRegistry:
293 """Tests for ToolRegistry."""
294
295 def test_create_default_registry(self):
296 registry = create_default_registry()
297 assert registry.get("read") is not None
298 assert registry.get("write") is not None
299 assert registry.get("edit") is not None
300 assert registry.get("glob") is not None
301 assert registry.get("bash") is not None
302 assert registry.get("grep") is not None
303
304 def test_unknown_tool(self):
305 registry = create_default_registry()
306 assert registry.get("nonexistent") is None
307
308 @pytest.mark.asyncio
309 async def test_execute_unknown_tool(self):
310 registry = create_default_registry()
311 result = await registry.execute("nonexistent")
312 assert result.is_error
313 assert "Unknown tool" in result.output
314
315 def test_skip_confirmation_flag(self):
316 registry = create_default_registry()
317 assert not registry.skip_confirmation
318 registry.skip_confirmation = True
319 assert registry.skip_confirmation