File size: 2,638 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 |
from unittest.mock import Mock
import pytest
import requests
from openhands.core.exceptions import (
AgentRuntimeDisconnectedError,
AgentRuntimeTimeoutError,
)
from openhands.events.action import CmdRunAction
from openhands.runtime.base import Runtime
@pytest.fixture
def mock_session():
return Mock()
@pytest.fixture
def runtime(mock_session):
runtime = Mock(spec=Runtime)
runtime.session = mock_session
runtime.send_action_for_execution = Mock()
return runtime
def test_runtime_timeout_error(runtime, mock_session):
# Create a command action
action = CmdRunAction(command='test command')
action.set_hard_timeout(120)
# Mock the runtime to raise a timeout error
runtime.send_action_for_execution.side_effect = AgentRuntimeTimeoutError(
'Runtime failed to return execute_action before the requested timeout of 120s'
)
# Verify that the error message indicates a timeout
with pytest.raises(AgentRuntimeTimeoutError) as exc_info:
runtime.send_action_for_execution(action)
assert (
str(exc_info.value)
== 'Runtime failed to return execute_action before the requested timeout of 120s'
)
@pytest.mark.parametrize(
'status_code,expected_message',
[
(404, 'Runtime is not responding. This may be temporary, please try again.'),
(
502,
'Runtime is temporarily unavailable. This may be due to a restart or network issue, please try again.',
),
],
)
def test_runtime_disconnected_error(
runtime, mock_session, status_code, expected_message
):
# Mock the request to return the specified status code
mock_response = Mock()
mock_response.status_code = status_code
mock_response.raise_for_status = Mock(
side_effect=requests.HTTPError(response=mock_response)
)
mock_response.json = Mock(
return_value={
'observation': 'run',
'content': 'test',
'extras': {'command_id': 'test_id', 'command': 'test command'},
}
)
# Mock the runtime to raise the error
runtime.send_action_for_execution.side_effect = AgentRuntimeDisconnectedError(
expected_message
)
# Create a command action
action = CmdRunAction(command='test command')
action.set_hard_timeout(120)
# Verify that the error message is correct
with pytest.raises(AgentRuntimeDisconnectedError) as exc_info:
runtime.send_action_for_execution(action)
assert str(exc_info.value) == expected_message
|