| 227 | |
| 228 | @pytest.mark.asyncio |
| 229 | async def test_timeout(self, workspace, tool_context_confirmed): |
| 230 | tool = bash_tool.ExecuteBashTool(workspace=workspace) |
| 231 | mock_process = mock.AsyncMock() |
| 232 | mock_process.pid = 12345 |
| 233 | mock_process.communicate.return_value = (b"", b"") |
| 234 | with ( |
| 235 | mock.patch.object( |
| 236 | asyncio, |
| 237 | "create_subprocess_exec", |
| 238 | autospec=True, |
| 239 | return_value=mock_process, |
| 240 | ), |
| 241 | mock.patch.object( |
| 242 | asyncio, "wait_for", autospec=True, side_effect=asyncio.TimeoutError |
| 243 | ), |
| 244 | mock.patch("os.killpg") as mock_killpg, |
| 245 | ): |
| 246 | result = await tool.run_async( |
| 247 | args={"command": "python scripts/do_thing.py"}, |
| 248 | tool_context=tool_context_confirmed, |
| 249 | ) |
| 250 | mock_killpg.assert_called_with(12345, signal.SIGKILL) |
| 251 | assert "error" in result |
| 252 | assert "timed out" in result["error"].lower() |
| 253 | |
| 254 | @pytest.mark.asyncio |
| 255 | async def test_cwd_is_workspace(self, workspace, tool_context_confirmed): |