File size: 11,407 Bytes
246d201 |
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 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 |
import json
from openhands.events.observation.commands import (
CMD_OUTPUT_METADATA_PS1_REGEX,
CMD_OUTPUT_PS1_BEGIN,
CMD_OUTPUT_PS1_END,
CmdOutputMetadata,
CmdOutputObservation,
)
def test_ps1_metadata_format():
"""Test that PS1 prompt has correct format markers"""
prompt = CmdOutputMetadata.to_ps1_prompt()
print(prompt)
assert prompt.startswith('\n###PS1JSON###\n')
assert prompt.endswith('\n###PS1END###\n')
assert r'\"exit_code\"' in prompt, 'PS1 prompt should contain escaped double quotes'
def test_ps1_metadata_json_structure():
"""Test that PS1 prompt contains valid JSON with expected fields"""
prompt = CmdOutputMetadata.to_ps1_prompt()
# Extract JSON content between markers
json_str = prompt.replace('###PS1JSON###\n', '').replace('\n###PS1END###\n', '')
# Remove escaping before parsing
json_str = json_str.replace(r'\"', '"')
# Remove any trailing content after the JSON
json_str = json_str.split('###PS1END###')[0].strip()
data = json.loads(json_str)
# Check required fields
expected_fields = {
'pid',
'exit_code',
'username',
'hostname',
'working_dir',
'py_interpreter_path',
}
assert set(data.keys()) == expected_fields
def test_ps1_metadata_parsing():
"""Test parsing PS1 output into CmdOutputMetadata"""
test_data = {
'exit_code': 0,
'username': 'testuser',
'hostname': 'localhost',
'working_dir': '/home/testuser',
'py_interpreter_path': '/usr/bin/python',
}
ps1_str = f"""###PS1JSON###
{json.dumps(test_data, indent=2)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == test_data['exit_code']
assert metadata.username == test_data['username']
assert metadata.hostname == test_data['hostname']
assert metadata.working_dir == test_data['working_dir']
assert metadata.py_interpreter_path == test_data['py_interpreter_path']
def test_ps1_metadata_parsing_string():
"""Test parsing PS1 output into CmdOutputMetadata"""
ps1_str = r"""###PS1JSON###
{
"exit_code": "0",
"username": "myname",
"hostname": "myhostname",
"working_dir": "~/mydir",
"py_interpreter_path": "/my/python/path"
}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.username == 'myname'
assert metadata.hostname == 'myhostname'
assert metadata.working_dir == '~/mydir'
assert metadata.py_interpreter_path == '/my/python/path'
def test_ps1_metadata_parsing_string_real_example():
"""Test parsing PS1 output into CmdOutputMetadata"""
ps1_str = r"""
###PS1JSON###
{
"pid": "",
"exit_code": "0",
"username": "runner",
"hostname": "fv-az1055-610",
"working_dir": "/home/runner/work/OpenHands/OpenHands",
"py_interpreter_path": "/home/runner/.cache/pypoetry/virtualenvs/openhands-ai-ULPBlkAi-py3.12/bin/python"
}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.username == 'runner'
assert metadata.hostname == 'fv-az1055-610'
assert metadata.working_dir == '/home/runner/work/OpenHands/OpenHands'
assert (
metadata.py_interpreter_path
== '/home/runner/.cache/pypoetry/virtualenvs/openhands-ai-ULPBlkAi-py3.12/bin/python'
)
def test_ps1_metadata_parsing_additional_prefix():
"""Test parsing PS1 output into CmdOutputMetadata"""
test_data = {
'exit_code': 0,
'username': 'testuser',
'hostname': 'localhost',
'working_dir': '/home/testuser',
'py_interpreter_path': '/usr/bin/python',
}
ps1_str = f"""
This is something that not part of the PS1 prompt
###PS1JSON###
{json.dumps(test_data, indent=2)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == test_data['exit_code']
assert metadata.username == test_data['username']
assert metadata.hostname == test_data['hostname']
assert metadata.working_dir == test_data['working_dir']
assert metadata.py_interpreter_path == test_data['py_interpreter_path']
def test_ps1_metadata_parsing_invalid():
"""Test parsing invalid PS1 output returns default metadata"""
# Test with invalid JSON
invalid_json = """###PS1JSON###
{invalid json}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(invalid_json)
assert len(matches) == 0 # No matches should be found for invalid JSON
# Test with missing markers
invalid_format = """NOT A VALID PS1 PROMPT"""
matches = CmdOutputMetadata.matches_ps1_metadata(invalid_format)
assert len(matches) == 0
# Test with empty PS1 metadata
empty_metadata = """###PS1JSON###
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(empty_metadata)
assert len(matches) == 0 # No matches should be found for empty metadata
# Test with whitespace in PS1 metadata
whitespace_metadata = """###PS1JSON###
{
"exit_code": "0",
"pid": "123",
"username": "test",
"hostname": "localhost",
"working_dir": "/home/test",
"py_interpreter_path": "/usr/bin/python"
}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(whitespace_metadata)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.pid == 123
def test_ps1_metadata_missing_fields():
"""Test handling of missing fields in PS1 metadata"""
# Test with only required fields
minimal_data = {'exit_code': 0, 'pid': 123}
ps1_str = f"""###PS1JSON###
{json.dumps(minimal_data)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.pid == 123
assert metadata.username is None
assert metadata.hostname is None
assert metadata.working_dir is None
assert metadata.py_interpreter_path is None
# Test with missing exit_code but valid pid
no_exit_code = {'pid': 123, 'username': 'test'}
ps1_str = f"""###PS1JSON###
{json.dumps(no_exit_code)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == -1 # default value
assert metadata.pid == 123
assert metadata.username == 'test'
def test_ps1_metadata_multiple_blocks():
"""Test handling multiple PS1 metadata blocks"""
test_data = {
'exit_code': 0,
'username': 'testuser',
'hostname': 'localhost',
'working_dir': '/home/testuser',
'py_interpreter_path': '/usr/bin/python',
}
ps1_str = f"""###PS1JSON###
{json.dumps(test_data, indent=2)}
###PS1END###
Some other content
###PS1JSON###
{json.dumps(test_data, indent=2)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 2 # Should find both blocks
# Both blocks should parse successfully
metadata1 = CmdOutputMetadata.from_ps1_match(matches[0])
metadata2 = CmdOutputMetadata.from_ps1_match(matches[1])
assert metadata1.exit_code == test_data['exit_code']
assert metadata2.exit_code == test_data['exit_code']
def test_ps1_metadata_regex_pattern():
"""Test the regex pattern used to extract PS1 metadata"""
# Test basic pattern matching
test_str = f'{CMD_OUTPUT_PS1_BEGIN}test\n{CMD_OUTPUT_PS1_END}'
matches = CMD_OUTPUT_METADATA_PS1_REGEX.finditer(test_str)
match = next(matches)
assert match.group(1).strip() == 'test'
# Test with content before and after
test_str = f'prefix\n{CMD_OUTPUT_PS1_BEGIN}test\n{CMD_OUTPUT_PS1_END}suffix'
matches = CMD_OUTPUT_METADATA_PS1_REGEX.finditer(test_str)
match = next(matches)
assert match.group(1).strip() == 'test'
# Test with multiline content
test_str = f'{CMD_OUTPUT_PS1_BEGIN}line1\nline2\nline3\n{CMD_OUTPUT_PS1_END}'
matches = CMD_OUTPUT_METADATA_PS1_REGEX.finditer(test_str)
match = next(matches)
assert match.group(1).strip() == 'line1\nline2\nline3'
def test_cmd_output_observation_properties():
"""Test CmdOutputObservation class properties"""
# Test with successful command
metadata = CmdOutputMetadata(exit_code=0, pid=123)
obs = CmdOutputObservation(command='ls', content='file1\nfile2', metadata=metadata)
assert obs.command_id == 123
assert obs.exit_code == 0
assert not obs.error
assert 'exit code 0' in obs.message
assert 'ls' in obs.message
assert 'file1' in str(obs)
assert 'file2' in str(obs)
assert 'metadata' in str(obs)
# Test with failed command
metadata = CmdOutputMetadata(exit_code=1, pid=456)
obs = CmdOutputObservation(command='invalid', content='error', metadata=metadata)
assert obs.command_id == 456
assert obs.exit_code == 1
assert obs.error
assert 'exit code 1' in obs.message
assert 'invalid' in obs.message
assert 'error' in str(obs)
def test_ps1_metadata_empty_fields():
"""Test handling of empty fields in PS1 metadata"""
# Test with empty strings
empty_data = {
'exit_code': 0,
'pid': 123,
'username': '',
'hostname': '',
'working_dir': '',
'py_interpreter_path': '',
}
ps1_str = f"""###PS1JSON###
{json.dumps(empty_data)}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(ps1_str)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.pid == 123
assert metadata.username == ''
assert metadata.hostname == ''
assert metadata.working_dir == ''
assert metadata.py_interpreter_path == ''
# Test with malformed but valid JSON
malformed_json = """###PS1JSON###
{
"exit_code":0,
"pid" : 123,
"username": "test" ,
"hostname": "host",
"working_dir" :"dir",
"py_interpreter_path":"path"
}
###PS1END###
"""
matches = CmdOutputMetadata.matches_ps1_metadata(malformed_json)
assert len(matches) == 1
metadata = CmdOutputMetadata.from_ps1_match(matches[0])
assert metadata.exit_code == 0
assert metadata.pid == 123
assert metadata.username == 'test'
assert metadata.hostname == 'host'
assert metadata.working_dir == 'dir'
assert metadata.py_interpreter_path == 'path'
|