loader Public

Watch 0 Fork 0 Star 0
Python · 9959 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 @pytest.mark.asyncio
156 async def test_glob_expands_home_prefixed_pattern(self, tool, monkeypatch, temp_dir):
157 home_dir = temp_dir / "fake-home"
158 animals_dir = home_dir / "Loader" / "animals"
159 animals_dir.mkdir(parents=True)
160 (animals_dir / "penguins.html").write_text("<h1>Penguins</h1>\n")
161 (animals_dir / "wolves.html").write_text("<h1>Wolves</h1>\n")
162
163 monkeypatch.setenv("HOME", str(home_dir))
164
165 result = await tool.execute(pattern="~/Loader/animals/*.html")
166
167 assert not result.is_error
168 assert "penguins.html" in result.output
169 assert "wolves.html" in result.output
170 assert result.metadata["base_path"] == str(animals_dir.resolve())
171 assert result.metadata["effective_pattern"] == "*.html"
172
173
174 class TestBashTool:
175 """Tests for BashTool."""
176
177 @pytest.fixture
178 def tool(self):
179 return BashTool()
180
181 @pytest.mark.asyncio
182 async def test_bash_simple_command(self, tool):
183 result = await tool.execute(command="echo 'hello world'")
184 assert not result.is_error
185 assert "hello world" in result.output
186
187 @pytest.mark.asyncio
188 async def test_bash_pwd(self, tool):
189 result = await tool.execute(command="pwd")
190 assert not result.is_error
191 assert "/" in result.output
192
193 @pytest.mark.asyncio
194 async def test_bash_failed_command(self, tool):
195 result = await tool.execute(command="exit 1")
196 assert result.is_error
197 assert "Exit code 1" in result.output
198
199 @pytest.mark.asyncio
200 async def test_bash_background_launch_and_wait(self, tool):
201 launch = await tool.execute(
202 command='python -c "import time; print(\'ready\'); time.sleep(0.1)"',
203 background=True,
204 )
205
206 assert not launch.is_error
207 job_id = launch.metadata["job_id"]
208 assert job_id.startswith("bash-")
209
210 await asyncio.sleep(0.05)
211 wait_result = await tool.manager.wait_for_job(job_id)
212
213 assert not wait_result.is_error
214 assert "ready" in wait_result.output
215 assert wait_result.metadata["status"] == "completed"
216
217 @pytest.mark.asyncio
218 async def test_bash_background_job_can_be_killed(self, tool):
219 launch = await tool.execute(
220 command='python -c "import time; print(\'server\'); time.sleep(30)"',
221 background=True,
222 )
223 job_id = launch.metadata["job_id"]
224
225 kill_result = await tool.manager.kill_job(job_id)
226
227 assert not kill_result.is_error
228 assert f"bash job {job_id}" in kill_result.output
229 assert kill_result.metadata["status"] == "killed"
230 assert kill_result.metadata["interrupted"] is False
231 assert kill_result.metadata["killed"] is True
232
233 @pytest.mark.asyncio
234 async def test_bash_rejects_long_running_foreground_command(self, tool):
235 result = await tool.execute(command="python -m http.server 8000")
236
237 assert result.is_error
238 assert "background=true" in result.output
239 assert result.metadata["suggest_background"] is True
240
241 def test_is_destructive(self, tool):
242 assert tool.is_destructive
243
244 def test_safe_command_no_confirmation(self, tool):
245 # ls is safe
246 tool.check_confirmation(skip_confirmation=False, command="ls -la")
247 # git status is safe
248 tool.check_confirmation(skip_confirmation=False, command="git status")
249
250 def test_unsafe_command_requires_confirmation(self, tool):
251 with pytest.raises(ConfirmationRequired):
252 tool.check_confirmation(skip_confirmation=False, command="rm -rf /tmp/test")
253
254
255 class TestGrepTool:
256 """Tests for GrepTool."""
257
258 @pytest.fixture
259 def tool(self):
260 return GrepTool()
261
262 @pytest.mark.asyncio
263 async def test_grep_finds_pattern(self, tool, sample_python_file):
264 result = await tool.execute(
265 pattern="def.*hello",
266 path=str(sample_python_file),
267 )
268 assert not result.is_error
269 assert "hello" in result.output
270
271 @pytest.mark.asyncio
272 async def test_grep_no_matches(self, tool, sample_file):
273 result = await tool.execute(
274 pattern="nonexistent_pattern",
275 path=str(sample_file),
276 )
277 assert not result.is_error
278 assert "No matches" in result.output
279
280
281 class TestToolRegistry:
282 """Tests for ToolRegistry."""
283
284 def test_create_default_registry(self):
285 registry = create_default_registry()
286 assert registry.get("read") is not None
287 assert registry.get("write") is not None
288 assert registry.get("edit") is not None
289 assert registry.get("glob") is not None
290 assert registry.get("bash") is not None
291 assert registry.get("grep") is not None
292
293 def test_unknown_tool(self):
294 registry = create_default_registry()
295 assert registry.get("nonexistent") is None
296
297 @pytest.mark.asyncio
298 async def test_execute_unknown_tool(self):
299 registry = create_default_registry()
300 result = await registry.execute("nonexistent")
301 assert result.is_error
302 assert "Unknown tool" in result.output
303
304 def test_skip_confirmation_flag(self):
305 registry = create_default_registry()
306 assert not registry.skip_confirmation
307 registry.skip_confirmation = True
308 assert registry.skip_confirmation