| 1 | import shutil
|
| 2 | import tempfile
|
| 3 | from pathlib import Path
|
| 4 |
|
| 5 | import pytest
|
| 6 |
|
| 7 | from minisweagent.environments.extra.bubblewrap import BubblewrapEnvironment, BubblewrapEnvironmentConfig
|
| 8 |
|
| 9 |
|
| 10 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 11 | def test_bubblewrap_environment_basic_execution():
|
| 12 | """Test basic command execution in bubblewrap environment."""
|
| 13 | env = BubblewrapEnvironment()
|
| 14 |
|
| 15 | try:
|
| 16 | result = env.execute({"command": "echo 'hello world'"})
|
| 17 | print(f"test_bubblewrap_environment_basic_execution result: {result}")
|
| 18 | assert result["returncode"] == 0
|
| 19 | assert "hello world" in result["output"]
|
| 20 | finally:
|
| 21 | env.cleanup()
|
| 22 |
|
| 23 |
|
| 24 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 25 | def test_bubblewrap_environment_set_env_variables():
|
| 26 | """Test setting environment variables in the bubblewrap environment."""
|
| 27 | env = BubblewrapEnvironment(env={"TEST_VAR": "test_value", "ANOTHER_VAR": "another_value"})
|
| 28 |
|
| 29 | try:
|
| 30 | # Test single environment variable
|
| 31 | result = env.execute({"command": "echo $TEST_VAR"})
|
| 32 | print(f"test_bubblewrap_environment_set_env_variables result (single var): {result}")
|
| 33 | assert result["returncode"] == 0
|
| 34 | assert "test_value" in result["output"]
|
| 35 |
|
| 36 | # Test multiple environment variables
|
| 37 | result = env.execute({"command": "echo $TEST_VAR $ANOTHER_VAR"})
|
| 38 | print(f"test_bubblewrap_environment_set_env_variables result (multiple vars): {result}")
|
| 39 | assert result["returncode"] == 0
|
| 40 | assert "test_value another_value" in result["output"]
|
| 41 | finally:
|
| 42 | env.cleanup()
|
| 43 |
|
| 44 |
|
| 45 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 46 | def test_bubblewrap_environment_custom_cwd():
|
| 47 | """Test executing commands in a custom working directory."""
|
| 48 | with tempfile.TemporaryDirectory() as temp_dir:
|
| 49 | env = BubblewrapEnvironment(cwd=temp_dir)
|
| 50 |
|
| 51 | try:
|
| 52 | result = env.execute({"command": "pwd"})
|
| 53 | print(f"test_bubblewrap_environment_custom_cwd result: {result}")
|
| 54 | assert result["returncode"] == 0
|
| 55 | assert temp_dir in result["output"]
|
| 56 | finally:
|
| 57 | env.cleanup()
|
| 58 |
|
| 59 |
|
| 60 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 61 | def test_bubblewrap_environment_cwd_parameter_override():
|
| 62 | """Test that the cwd parameter in execute() overrides the config cwd."""
|
| 63 | with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
|
| 64 | env = BubblewrapEnvironment(cwd=temp_dir1)
|
| 65 |
|
| 66 | try:
|
| 67 | # Execute with different cwd parameter
|
| 68 | result = env.execute({"command": "pwd"}, cwd=temp_dir2)
|
| 69 | print(f"test_bubblewrap_environment_cwd_parameter_override result: {result}")
|
| 70 | assert result["returncode"] == 0
|
| 71 | assert temp_dir2 in result["output"]
|
| 72 | finally:
|
| 73 | env.cleanup()
|
| 74 |
|
| 75 |
|
| 76 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 77 | def test_bubblewrap_environment_command_failure():
|
| 78 | """Test that command failures are properly captured."""
|
| 79 | env = BubblewrapEnvironment()
|
| 80 |
|
| 81 | try:
|
| 82 | result = env.execute({"command": "exit 1"})
|
| 83 | print(f"test_bubblewrap_environment_command_failure result: {result}")
|
| 84 | assert result["returncode"] == 1
|
| 85 | assert result["output"] == ""
|
| 86 | finally:
|
| 87 | env.cleanup()
|
| 88 |
|
| 89 |
|
| 90 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 91 | def test_bubblewrap_environment_nonexistent_command():
|
| 92 | """Test execution of non-existent command."""
|
| 93 | env = BubblewrapEnvironment()
|
| 94 |
|
| 95 | try:
|
| 96 | result = env.execute({"command": "nonexistent_command_12345"})
|
| 97 | print(f"test_bubblewrap_environment_nonexistent_command result: {result}")
|
| 98 | assert result["returncode"] != 0
|
| 99 | assert "nonexistent_command_12345" in result["output"] or "command not found" in result["output"]
|
| 100 | finally:
|
| 101 | env.cleanup()
|
| 102 |
|
| 103 |
|
| 104 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 105 | def test_bubblewrap_environment_stderr_capture():
|
| 106 | """Test that stderr is properly captured."""
|
| 107 | env = BubblewrapEnvironment()
|
| 108 |
|
| 109 | try:
|
| 110 | result = env.execute({"command": "echo 'error message' >&2"})
|
| 111 | print(f"test_bubblewrap_environment_stderr_capture result: {result}")
|
| 112 | assert result["returncode"] == 0
|
| 113 | assert "error message" in result["output"]
|
| 114 | finally:
|
| 115 | env.cleanup()
|
| 116 |
|
| 117 |
|
| 118 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 119 | def test_bubblewrap_environment_timeout():
|
| 120 | """Test timeout functionality returns structured output instead of raising."""
|
| 121 | env = BubblewrapEnvironment(timeout=1)
|
| 122 |
|
| 123 | try:
|
| 124 | result = env.execute({"command": "sleep 2"})
|
| 125 | assert result["returncode"] == -1
|
| 126 | assert "timed out" in result["exception_info"]
|
| 127 | assert result["extra"]["exception_type"] == "TimeoutExpired"
|
| 128 | finally:
|
| 129 | env.cleanup()
|
| 130 |
|
| 131 |
|
| 132 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 133 | @pytest.mark.parametrize(
|
| 134 | ("command", "expected_returncode"),
|
| 135 | [
|
| 136 | ("echo 'test'", 0),
|
| 137 | ("exit 1", 1),
|
| 138 | ("exit 42", 42),
|
| 139 | ],
|
| 140 | )
|
| 141 | def test_bubblewrap_environment_return_codes(command, expected_returncode):
|
| 142 | """Test that various return codes are properly captured."""
|
| 143 | env = BubblewrapEnvironment()
|
| 144 |
|
| 145 | try:
|
| 146 | result = env.execute({"command": command})
|
| 147 | print(f"test_bubblewrap_environment_return_codes result (cmd: {command}): {result}")
|
| 148 | assert result["returncode"] == expected_returncode
|
| 149 | finally:
|
| 150 | env.cleanup()
|
| 151 |
|
| 152 |
|
| 153 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 154 | def test_bubblewrap_environment_multiline_output():
|
| 155 | """Test handling of multiline command output."""
|
| 156 | env = BubblewrapEnvironment()
|
| 157 |
|
| 158 | try:
|
| 159 | result = env.execute({"command": "echo -e 'line1\\nline2\\nline3'"})
|
| 160 | print(f"test_bubblewrap_environment_multiline_output result: {result}")
|
| 161 | assert result["returncode"] == 0
|
| 162 | output_lines = result["output"].strip().split("\n")
|
| 163 |
|
| 164 | assert len(output_lines) == 3
|
| 165 | assert "line1" in output_lines[0]
|
| 166 | assert "line2" in output_lines[1]
|
| 167 | assert "line3" in output_lines[2]
|
| 168 | finally:
|
| 169 | env.cleanup()
|
| 170 |
|
| 171 |
|
| 172 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 173 | def test_bubblewrap_environment_file_operations():
|
| 174 | """Test file operations in the bubblewrap environment."""
|
| 175 | with tempfile.TemporaryDirectory() as temp_dir:
|
| 176 | env = BubblewrapEnvironment(cwd=temp_dir)
|
| 177 |
|
| 178 | try:
|
| 179 | # Create a file
|
| 180 | result = env.execute({"command": "echo 'test content' > test.txt"})
|
| 181 | print(f"test_bubblewrap_environment_file_operations result (create file): {result}")
|
| 182 | assert result["returncode"] == 0
|
| 183 |
|
| 184 | # Read the file
|
| 185 | result = env.execute({"command": "cat test.txt"})
|
| 186 | print(f"test_bubblewrap_environment_file_operations result (read file): {result}")
|
| 187 | assert result["returncode"] == 0
|
| 188 | assert "test content" in result["output"]
|
| 189 |
|
| 190 | # Verify file exists (should be in the working directory)
|
| 191 | test_file = Path(temp_dir) / "test.txt"
|
| 192 |
|
| 193 | assert test_file.exists()
|
| 194 | assert test_file.read_text().strip() == "test content"
|
| 195 | finally:
|
| 196 | env.cleanup()
|
| 197 |
|
| 198 |
|
| 199 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 200 | def test_bubblewrap_environment_working_directory_creation():
|
| 201 | """Test that working directory is properly created."""
|
| 202 | env = BubblewrapEnvironment()
|
| 203 |
|
| 204 | try:
|
| 205 | assert env.working_dir.exists()
|
| 206 | assert env.working_dir.is_dir()
|
| 207 | finally:
|
| 208 | env.cleanup()
|
| 209 |
|
| 210 |
|
| 211 | @pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap not available")
|
| 212 | def test_bubblewrap_environment_cleanup():
|
| 213 | """Test that cleanup properly removes working directory."""
|
| 214 | env = BubblewrapEnvironment()
|
| 215 | working_dir = env.working_dir
|
| 216 |
|
| 217 | assert working_dir.exists()
|
| 218 |
|
| 219 | env.cleanup()
|
| 220 |
|
| 221 | assert not working_dir.exists()
|
| 222 |
|
| 223 |
|
| 224 | def test_bubblewrap_environment_custom_executable():
|
| 225 | """Test custom bubblewrap executable configuration."""
|
| 226 | config = BubblewrapEnvironmentConfig(executable="/custom/path/to/bwrap")
|
| 227 | env = BubblewrapEnvironment(**config.__dict__)
|
| 228 |
|
| 229 | try:
|
| 230 | assert env.config.executable == "/custom/path/to/bwrap"
|
| 231 | finally:
|
| 232 | env.cleanup()
|
| 233 |
|
| 234 |
|
| 235 | def test_bubblewrap_environment_custom_wrapper_args():
|
| 236 | """Test custom wrapper args configuration."""
|
| 237 | custom_args = ["--ro-bind", "/usr", "/usr", "--tmpfs", "/tmp"]
|
| 238 | config = BubblewrapEnvironmentConfig(wrapper_args=custom_args)
|
| 239 | env = BubblewrapEnvironment(**config.__dict__)
|
| 240 |
|
| 241 | try:
|
| 242 | assert env.config.wrapper_args == custom_args
|
| 243 | finally:
|
| 244 | env.cleanup()
|
| 245 |
|
| 246 |
|
| 247 | def test_bubblewrap_environment_get_template_vars():
|
| 248 | """Test get_template_vars method returns expected data."""
|
| 249 | env = BubblewrapEnvironment(env={"TEST_VAR": "test_value"})
|
| 250 |
|
| 251 | try:
|
| 252 | template_vars = env.get_template_vars()
|
| 253 | print(f"test_bubblewrap_environment_get_template_vars template_vars: {template_vars}")
|
| 254 |
|
| 255 | # Should contain config data
|
| 256 | assert "env" in template_vars
|
| 257 | assert template_vars["env"]["TEST_VAR"] == "test_value"
|
| 258 | assert "timeout" in template_vars
|
| 259 | assert template_vars["timeout"] == 30
|
| 260 |
|
| 261 | # Should contain platform info
|
| 262 | assert "system" in template_vars
|
| 263 | assert "machine" in template_vars
|
| 264 | finally:
|
| 265 | env.cleanup()
|
| 266 |
|