File size: 8,282 Bytes
dc893fb |
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 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 |
"""Test cases for Bash Tool."""
import asyncio
import pytest
from mini_agent.tools.bash_tool import BackgroundShellManager, BashKillTool, BashOutputTool, BashTool
@pytest.mark.asyncio
async def test_foreground_command():
"""Test executing a simple foreground command."""
print("\n=== Testing Foreground Command ===")
bash_tool = BashTool()
result = await bash_tool.execute(command="echo 'Hello from foreground'")
assert result.success
assert "Hello from foreground" in result.stdout
assert result.exit_code == 0
print(f"Output: {result.content}")
@pytest.mark.asyncio
async def test_foreground_command_with_stderr():
"""Test command that outputs to both stdout and stderr."""
print("\n=== Testing Stdout/Stderr Separation ===")
bash_tool = BashTool()
result = await bash_tool.execute(command="echo 'stdout message' && echo 'stderr message' >&2")
assert result.success
assert "stdout message" in result.stdout
assert "stderr message" in result.stderr
print(f"Stdout: {result.stdout}")
print(f"Stderr: {result.stderr}")
@pytest.mark.asyncio
async def test_command_failure():
"""Test command that fails with non-zero exit code."""
print("\n=== Testing Command Failure ===")
bash_tool = BashTool()
result = await bash_tool.execute(command="ls /nonexistent_directory_12345")
assert not result.success
assert result.exit_code != 0
assert result.error is not None
print(f"Error: {result.error}")
@pytest.mark.asyncio
async def test_command_timeout():
"""Test command timeout."""
print("\n=== Testing Command Timeout ===")
bash_tool = BashTool()
result = await bash_tool.execute(command="sleep 10", timeout=1)
assert not result.success
assert "timed out" in result.error.lower()
assert result.exit_code == -1
print(f"Timeout error: {result.error}")
@pytest.mark.asyncio
async def test_background_command():
"""Test running a command in the background."""
print("\n=== Testing Background Command ===")
bash_tool = BashTool()
result = await bash_tool.execute(
command="for i in 1 2 3; do echo 'Line '$i; sleep 0.5; done", run_in_background=True
)
assert result.success
assert result.bash_id is not None
assert "Background command started" in result.stdout
bash_id = result.bash_id
print(f"Background command started with ID: {bash_id}")
# Wait a bit for output
await asyncio.sleep(1)
# Check output
bash_output_tool = BashOutputTool()
output_result = await bash_output_tool.execute(bash_id=bash_id)
assert output_result.success
print(f"Output:\n{output_result.content}")
# Clean up - terminate the background process
bash_kill_tool = BashKillTool()
kill_result = await bash_kill_tool.execute(bash_id=bash_id)
assert kill_result.success
print("Background process terminated")
@pytest.mark.asyncio
async def test_bash_output_monitoring():
"""Test monitoring background command output."""
print("\n=== Testing Output Monitoring ===")
bash_tool = BashTool()
# Start background command
result = await bash_tool.execute(
command="for i in 1 2 3 4 5; do echo 'Line '$i; sleep 0.5; done", run_in_background=True
)
assert result.success
bash_id = result.bash_id
print(f"Started background command: {bash_id}")
bash_output_tool = BashOutputTool()
# Check output multiple times (incremental output)
for i in range(3):
await asyncio.sleep(1)
output_result = await bash_output_tool.execute(bash_id=bash_id)
assert output_result.success
print(f"\n--- Check #{i + 1} ---")
print(f"Output:\n{output_result.content}")
# Clean up
bash_kill_tool = BashKillTool()
await bash_kill_tool.execute(bash_id=bash_id)
@pytest.mark.asyncio
async def test_bash_output_with_filter():
"""Test bash_output with regex filter."""
print("\n=== Testing Output Filter ===")
bash_tool = BashTool()
# Start background command
result = await bash_tool.execute(
command="for i in 1 2 3 4 5; do echo 'Line '$i; sleep 0.3; done", run_in_background=True
)
assert result.success
bash_id = result.bash_id
# Wait for some output
await asyncio.sleep(2)
# Get filtered output (only lines with "Line 2" or "Line 4")
bash_output_tool = BashOutputTool()
output_result = await bash_output_tool.execute(bash_id=bash_id, filter_str="Line [24]")
assert output_result.success
lines = output_result.content
print(f"Filtered output:\n{output_result.content}")
# Clean up
bash_kill_tool = BashKillTool()
await bash_kill_tool.execute(bash_id=bash_id)
@pytest.mark.asyncio
async def test_bash_kill():
"""Test terminating a background command."""
print("\n=== Testing Bash Kill ===")
bash_tool = BashTool()
# Start a long-running background command
result = await bash_tool.execute(command="sleep 100", run_in_background=True)
assert result.success
bash_id = result.bash_id
print(f"Started long-running command: {bash_id}")
# Verify it's running
await asyncio.sleep(0.5)
bg_shell = BackgroundShellManager.get(bash_id)
assert bg_shell is not None
assert bg_shell.status == "running"
# Kill it
bash_kill_tool = BashKillTool()
kill_result = await bash_kill_tool.execute(bash_id=bash_id)
assert kill_result.success
# exit_code -15 means terminated by SIGTERM
assert kill_result.exit_code == -15 or kill_result.bash_id == bash_id
print(f"Kill result:\n{kill_result.content}")
# Verify it's removed from manager
bg_shell = BackgroundShellManager.get(bash_id)
assert bg_shell is None
@pytest.mark.asyncio
async def test_bash_kill_nonexistent():
"""Test killing a non-existent bash process."""
print("\n=== Testing Kill Non-existent Process ===")
bash_kill_tool = BashKillTool()
result = await bash_kill_tool.execute(bash_id="nonexistent123")
assert not result.success
assert "not found" in result.error.lower()
print(f"Expected error: {result.error}")
@pytest.mark.asyncio
async def test_bash_output_nonexistent():
"""Test getting output from non-existent bash process."""
print("\n=== Testing Output From Non-existent Process ===")
bash_output_tool = BashOutputTool()
result = await bash_output_tool.execute(bash_id="nonexistent123")
assert not result.success
assert "not found" in result.error.lower()
print(f"Expected error: {result.error}")
@pytest.mark.asyncio
async def test_multiple_background_commands():
"""Test running multiple background commands simultaneously."""
print("\n=== Testing Multiple Background Commands ===")
bash_tool = BashTool()
# Start multiple background commands
bash_ids = []
for i in range(3):
result = await bash_tool.execute(
command=f"for j in 1 2 3; do echo 'Command {i + 1} Line '$j; sleep 0.5; done", run_in_background=True
)
assert result.success
bash_ids.append(result.bash_id)
print(f"Started command {i + 1}: {result.bash_id}")
# Wait and check all commands
await asyncio.sleep(1)
bash_output_tool = BashOutputTool()
for bash_id in bash_ids:
output_result = await bash_output_tool.execute(bash_id=bash_id)
assert output_result.success
print(f"\nOutput for {bash_id}:\n{output_result.content[:100]}...")
# Clean up all
bash_kill_tool = BashKillTool()
for bash_id in bash_ids:
await bash_kill_tool.execute(bash_id=bash_id)
print("All background processes cleaned up")
@pytest.mark.asyncio
async def test_timeout_validation():
"""Test timeout parameter validation."""
print("\n=== Testing Timeout Validation ===")
bash_tool = BashTool()
# Test with timeout > 600 (should be capped to 600)
result = await bash_tool.execute(command="echo 'test'", timeout=1000)
assert result.success
print("Timeout > 600 handled correctly")
# Test with timeout < 1 (should be set to 120)
result = await bash_tool.execute(command="echo 'test'", timeout=0)
assert result.success
print("Timeout < 1 handled correctly")
|