Files
browser-use__browser-use/tests/ci/test_native_tool_router.py
2026-05-31 12:56:22 -07:00

698 lines
27 KiB
Python

import json
import pytest
from pydantic import BaseModel
from browser_use.agent.runtime import (
BrowserAgentSession,
BrowserRuntimeEventTypes,
ClickCoordinatesInput,
NativeToolCall,
NativeToolRouter,
click_coordinates_as_click_arguments,
)
from browser_use.agent.runtime.tools import _TOOL_CONTEXT_BUILTIN_ACTIONS
from browser_use.filesystem.file_system import FileSystem
from browser_use.tools.service import Tools
from browser_use.tools.views import StructuredOutputAction
def test_native_tool_router_exposes_api_safe_names() -> None:
router = NativeToolRouter.from_tools(Tools())
navigate = router.resolve('browser.navigate')
click = router.resolve('browser_click')
assert navigate.api_name == 'browser_navigate'
assert click.name == 'browser.click'
assert 'browser.get_state' in router.definitions
assert 'browser.cdp' in router.definitions
assert 'browser.html' in router.definitions
assert 'browser.get_html' in router.definitions
assert 'browser.markdown' in router.definitions
assert 'browser.accessibility' in router.definitions
assert 'browser.get_accessibility_tree' in router.definitions
assert 'browser.inspect_element' in router.definitions
assert 'browser.network' in router.definitions
assert 'browser.http_fetch' in router.definitions
assert 'browser.fetch' in router.definitions
assert 'file.read' in router.definitions
assert 'file.write' in router.definitions
assert any(tool['function']['name'] == 'browser_navigate' for tool in router.tool_schemas())
assert 'file.read/file.write for workspace files' in router.guidance()
def test_native_tool_router_workspace_tools_are_opt_in() -> None:
default_router = NativeToolRouter.from_tools(Tools())
workspace_router = NativeToolRouter.from_tools(Tools(), include_workspace_tools=True)
assert 'workspace.read_file' not in default_router.definitions
assert 'workspace.read_file' in workspace_router.definitions
assert 'workspace.import_artifacts' in workspace_router.definitions
assert workspace_router.resolve('shell_run').name == 'shell.run'
def test_default_native_builtin_actions_are_direct_or_tool_context_backed() -> None:
dedicated_direct_actions = {'search', 'navigate', 'go_back', 'wait', 'done', 'scroll'}
default_actions = set(Tools().registry.registry.actions)
assert default_actions <= _TOOL_CONTEXT_BUILTIN_ACTIONS | dedicated_direct_actions
def test_native_tool_router_validates_with_existing_pydantic_models() -> None:
router = NativeToolRouter.from_tools(Tools())
call = NativeToolCall(tool_name='browser_navigate', arguments={'url': 'https://example.com', 'new_tab': False})
params = router.validate_call(call)
assert params.model_dump() == {'url': 'https://example.com', 'new_tab': False}
with pytest.raises(ValueError):
router.validate_call(NativeToolCall(tool_name='browser.navigate', arguments={'new_tab': False}))
def test_native_tool_router_uses_native_structured_done_input() -> None:
class MyOutput(BaseModel):
answer: str
tools = Tools(output_model=MyOutput)
router = NativeToolRouter.from_tools(tools)
done_definition = router.resolve('browser.done')
assert done_definition.source_action == 'done'
assert done_definition.input_model is not tools.registry.registry.actions['done'].param_model
assert done_definition.input_model is not StructuredOutputAction[MyOutput]
assert 'StructuredDoneInput' in done_definition.input_model.__name__
params = router.validate_call(NativeToolCall(tool_name='browser.done', arguments={'data': {'answer': 'ok'}, 'success': True}))
assert params.model_dump(mode='json') == {'success': True, 'data': {'answer': 'ok'}, 'files_to_display': []}
def test_native_tool_router_filters_page_specific_actions() -> None:
tools = Tools()
@tools.action('Only available on example.com', domains=['example.com'])
async def example_only() -> str:
return 'ok'
assert 'browser.example_only' not in NativeToolRouter.from_tools(tools).definitions
assert 'browser.example_only' in NativeToolRouter.from_tools(tools, page_url='https://example.com').definitions
@pytest.mark.asyncio
async def test_native_tool_router_executes_existing_action_without_fake_action_model() -> None:
tools = Tools()
session = BrowserAgentSession.create(task='Wait briefly')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, action_timeout=5)
router = NativeToolRouter.from_tools(tools)
result = await router.execute(NativeToolCall(tool_name='browser.wait', arguments={'seconds': 1}, call_id='call-1'), context)
assert result.call_id == 'call-1'
assert result.is_error is False
assert result.content == 'Waited for 1 seconds'
assert result.structured_content == {'seconds': 1, 'actual_seconds': 0, 'direct_service': True}
assert result.to_context_item().render().startswith('<tool_result name="browser.wait" id="call-1">')
assert [event.event_type for event in session.event_stream.events] == ['turn.started', 'tool.started', 'tool.completed']
@pytest.mark.asyncio
async def test_native_tool_router_executes_done_without_registered_action_adapter(monkeypatch, tmp_path) -> None:
tools = Tools()
async def fail_execute_action(*args, **kwargs):
raise AssertionError('browser.done should not use the registered action adapter in native mode')
monkeypatch.setattr(tools.registry, 'execute_action', fail_execute_action)
session = BrowserAgentSession.create(task='Finish directly')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, file_system=FileSystem(str(tmp_path)), action_timeout=5)
router = NativeToolRouter.from_tools(tools)
result = await router.execute(
NativeToolCall(
tool_name='browser.done',
arguments={'text': 'native done ok', 'success': True},
call_id='done-call',
),
context,
)
assert result.call_id == 'done-call'
assert result.is_error is False
assert result.content == 'native done ok'
assert result.structured_content['is_done'] is True
assert result.structured_content['success'] is True
assert [event.event_type for event in session.event_stream.events] == ['turn.started', 'tool.started', 'tool.completed']
@pytest.mark.asyncio
async def test_native_tool_router_executes_structured_done_without_registered_action_adapter(
monkeypatch, tmp_path, browser_session
) -> None:
class MyOutput(BaseModel):
answer: str
tools = Tools(output_model=MyOutput)
async def fail_execute_action(*args, **kwargs):
raise AssertionError('structured browser.done should not use the registered action adapter in native mode')
monkeypatch.setattr(tools.registry, 'execute_action', fail_execute_action)
session = BrowserAgentSession.create(task='Finish with structured output')
turn = session.start_turn(step_index=0)
context = session.tool_context(
turn,
tools=tools,
browser_session=browser_session,
file_system=FileSystem(str(tmp_path)),
action_timeout=5,
)
router = NativeToolRouter.from_tools(tools)
result = await router.execute(
NativeToolCall(
tool_name='browser.done',
arguments={'data': {'answer': 'structured ok'}, 'success': True},
call_id='structured-done-call',
),
context,
)
assert result.call_id == 'structured-done-call'
assert result.is_error is False
assert result.content is not None
assert json.loads(result.content) == {'answer': 'structured ok'}
assert result.structured_content['is_done'] is True
assert result.structured_content['success'] is True
@pytest.mark.asyncio
async def test_native_tool_router_executes_builtin_file_tool_without_registered_action_adapter(monkeypatch, tmp_path) -> None:
tools = Tools()
async def fail_execute_action(*args, **kwargs):
raise AssertionError('built-in native file tools should not use the registered action adapter')
monkeypatch.setattr(tools.registry, 'execute_action', fail_execute_action)
session = BrowserAgentSession.create(task='Write a file directly')
turn = session.start_turn(step_index=0)
file_system = FileSystem(str(tmp_path))
context = session.tool_context(turn, tools=tools, file_system=file_system, action_timeout=5)
router = NativeToolRouter.from_tools(tools)
result = await router.execute(
NativeToolCall(
tool_name='browser.write_file',
arguments={'file_name': 'native-file.txt', 'content': 'native-file-ok'},
call_id='write-call',
),
context,
)
assert result.call_id == 'write-call'
assert result.is_error is False
assert (file_system.get_dir() / 'native-file.txt').read_text() == 'native-file-ok\n'
assert result.structured_content['long_term_memory']
@pytest.mark.asyncio
async def test_native_tool_router_can_drive_simple_browser_task(monkeypatch, browser_session, httpserver) -> None:
httpserver.expect_request('/native-tool').respond_with_data(
"""<!doctype html>
<html>
<body>
<button id="reveal" onclick="document.getElementById('answer').textContent = 'native-tool-ok'">Reveal</button>
<p id="answer">hidden</p>
</body>
</html>""",
content_type='text/html',
)
tools = Tools()
async def fail_execute_action(*args, **kwargs):
raise AssertionError('built-in native browser tools should not use the registered action adapter')
def fail_dispatch(*args, **kwargs):
raise AssertionError('built-in native browser tools should not require event dispatch')
monkeypatch.setattr(tools.registry, 'execute_action', fail_execute_action)
monkeypatch.setattr(browser_session.event_bus, 'dispatch', fail_dispatch)
router = NativeToolRouter.from_tools(tools)
session = BrowserAgentSession.create(task='Reveal the answer')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, browser_session=browser_session, action_timeout=30)
navigate_result = await router.execute(
NativeToolCall(tool_name='browser.navigate', arguments={'url': httpserver.url_for('/native-tool')}), context
)
assert navigate_result.is_error is False
state = await browser_session.get_browser_state_summary(include_screenshot=False)
button_index = next(
idx
for idx, element in state.dom_state.selector_map.items()
if getattr(element, 'tag_name', None) == 'button' and getattr(element, 'attributes', {}).get('id') == 'reveal'
)
click_result = await router.execute(NativeToolCall(tool_name='browser.click', arguments={'index': button_index}), context)
assert click_result.is_error is False
cdp_session = await browser_session.get_or_create_cdp_session()
readback = await cdp_session.cdp_client.send.Runtime.evaluate(
params={'expression': "document.getElementById('answer').textContent", 'returnByValue': True},
session_id=cdp_session.session_id,
)
assert readback.get('result', {}).get('value') == 'native-tool-ok'
@pytest.mark.asyncio
async def test_native_tool_router_executes_direct_navigation_go_back_and_page_scroll(browser_session, httpserver) -> None:
httpserver.expect_request('/first').respond_with_data(
"""<!doctype html>
<html>
<body>
<a href="/second">Second</a>
<div style="height: 2400px">First page</div>
</body>
</html>""",
content_type='text/html',
)
httpserver.expect_request('/second').respond_with_data(
"""<!doctype html>
<html>
<body>
<div style="height: 2400px">Second page</div>
</body>
</html>""",
content_type='text/html',
)
tools = Tools()
router = NativeToolRouter.from_tools(tools)
session = BrowserAgentSession.create(task='Use direct navigation services')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, browser_session=browser_session, action_timeout=30)
first_url = httpserver.url_for('/first')
second_url = httpserver.url_for('/second')
first_result = await router.execute(NativeToolCall(tool_name='browser.navigate', arguments={'url': first_url}), context)
second_result = await router.execute(NativeToolCall(tool_name='browser.navigate', arguments={'url': second_url}), context)
back_result = await router.execute(NativeToolCall(tool_name='browser.go_back', arguments={}), context)
assert first_result.is_error is False
assert first_result.structured_content['direct_service'] is True
assert second_result.is_error is False
assert back_result.is_error is False
assert back_result.structured_content['url'].endswith('/first')
scroll_result = await router.execute(
NativeToolCall(tool_name='browser.scroll', arguments={'down': True, 'pages': 1.0, 'index': 0}), context
)
assert scroll_result.is_error is False
assert scroll_result.structured_content['direct_service'] is True
assert scroll_result.structured_content['pixels'] > 0
cdp_session = await browser_session.get_or_create_cdp_session()
readback = await cdp_session.cdp_client.send.Runtime.evaluate(
params={'expression': 'Math.round(window.scrollY)', 'returnByValue': True},
session_id=cdp_session.session_id,
)
assert readback.get('result', {}).get('value') > 0
@pytest.mark.asyncio
async def test_native_tool_router_executes_get_state_and_raw_cdp(browser_session, httpserver) -> None:
httpserver.expect_request('/native-state').respond_with_data(
"""<!doctype html>
<html>
<head><title>Native State</title></head>
<body>
<button id="state-button">State Button</button>
</body>
</html>""",
content_type='text/html',
)
tools = Tools()
session = BrowserAgentSession.create(task='Inspect state')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, browser_session=browser_session, action_timeout=30)
router = NativeToolRouter.from_tools(tools)
navigate_result = await router.execute(
NativeToolCall(tool_name='browser.navigate', arguments={'url': httpserver.url_for('/native-state')}), context
)
assert navigate_result.is_error is False
state_result = await router.execute(
NativeToolCall(
tool_name='browser.get_state',
arguments={'include_screenshot': False, 'include_dom': True},
call_id='state-call',
),
context,
)
assert state_result.is_error is False
assert state_result.call_id == 'state-call'
assert state_result.structured_content['url'].endswith('/native-state')
assert 'State Button' in state_result.structured_content['dom']
assert len(state_result.structured_content['runtime_handles']['current_target_id']) > 4
assert len(state_result.structured_content['runtime_handles']['current_session_id']) > 4
assert len(state_result.structured_content['tabs'][0]['target_id']) > 4
cdp_result = await router.execute(
NativeToolCall(
tool_name='browser.cdp',
arguments={
'method': 'Runtime.evaluate',
'params': {'expression': 'document.title', 'returnByValue': True},
},
call_id='cdp-call',
),
context,
)
assert cdp_result.is_error is False
assert cdp_result.call_id == 'cdp-call'
assert cdp_result.structured_content['response']['result']['value'] == 'Native State'
assert len(cdp_result.structured_content['target_id']) > 4
assert len(cdp_result.structured_content['session_id']) > 4
assert BrowserRuntimeEventTypes.BROWSER_STATE_REFRESHED in [event.event_type for event in session.event_stream.events]
@pytest.mark.asyncio
async def test_native_tool_router_executes_coordinate_click(browser_session, httpserver) -> None:
httpserver.expect_request('/native-coordinate').respond_with_data(
"""<!doctype html>
<html>
<body>
<button id="reveal" style="margin: 40px; width: 160px; height: 60px;"
onclick="document.getElementById('answer').textContent = 'coordinate-ok'">
Reveal
</button>
<p id="answer">hidden</p>
</body>
</html>""",
content_type='text/html',
)
tools = Tools()
router = NativeToolRouter.from_tools(tools)
session = BrowserAgentSession.create(task='Click by coordinates')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, browser_session=browser_session, action_timeout=30)
navigate_result = await router.execute(
NativeToolCall(tool_name='browser.navigate', arguments={'url': httpserver.url_for('/native-coordinate')}), context
)
assert navigate_result.is_error is False
rect_result = await router.execute(
NativeToolCall(
tool_name='browser.cdp',
arguments={
'method': 'Runtime.evaluate',
'params': {
'expression': """
const rect = document.getElementById('reveal').getBoundingClientRect();
JSON.stringify({ x: Math.floor(rect.left + rect.width / 2), y: Math.floor(rect.top + rect.height / 2) });
""",
'returnByValue': True,
},
},
),
context,
)
assert rect_result.is_error is False
coords = json.loads(rect_result.structured_content['response']['result']['value'])
click_result = await router.execute(
NativeToolCall(
tool_name='browser.click_coordinates',
arguments={'coordinate_x': coords['x'], 'coordinate_y': coords['y']},
call_id='coordinate-click',
),
context,
)
assert click_result.is_error is False
assert click_result.call_id == 'coordinate-click'
readback = await router.execute(
NativeToolCall(
tool_name='browser.cdp',
arguments={
'method': 'Runtime.evaluate',
'params': {'expression': "document.getElementById('answer').textContent", 'returnByValue': True},
},
),
context,
)
assert readback.structured_content['response']['result']['value'] == 'coordinate-ok'
@pytest.mark.asyncio
async def test_native_tool_router_executes_read_and_inspect_escape_hatches(browser_session, httpserver) -> None:
httpserver.expect_request('/api/page-data').respond_with_json({'page': 'loaded'})
httpserver.expect_request('/api/tool-data').respond_with_json({'tool': 'fetch-ok'})
httpserver.expect_request('/native-read').respond_with_data(
"""<!doctype html>
<html>
<head><title>Read Tools</title></head>
<body>
<main>
<h1>Read Tools</h1>
<button id="inspect" aria-label="Inspect Button" data-kind="escape">Inspect Me</button>
</main>
<script>fetch('/api/page-data')</script>
</body>
</html>""",
content_type='text/html',
)
tools = Tools()
router = NativeToolRouter.from_tools(tools)
session = BrowserAgentSession.create(task='Read page in multiple ways')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools, browser_session=browser_session, action_timeout=30)
navigate_result = await router.execute(
NativeToolCall(tool_name='browser.navigate', arguments={'url': httpserver.url_for('/native-read')}), context
)
assert navigate_result.is_error is False
html_result = await router.execute(
NativeToolCall(tool_name='browser.html', arguments={'selector': '#inspect', 'max_chars': 2000}), context
)
assert html_result.is_error is False
assert 'Inspect Me' in html_result.structured_content['html']
get_html_result = await router.execute(
NativeToolCall(tool_name='browser.get_html', arguments={'selector': '#inspect', 'max_chars': 2000}), context
)
assert get_html_result.is_error is False
assert 'Inspect Me' in get_html_result.structured_content['html']
markdown_result = await router.execute(NativeToolCall(tool_name='browser.markdown', arguments={'max_chars': 2000}), context)
assert markdown_result.is_error is False
assert 'Read Tools' in markdown_result.structured_content['markdown']
accessibility_result = await router.execute(
NativeToolCall(tool_name='browser.accessibility', arguments={'max_nodes': 50}), context
)
assert accessibility_result.is_error is False
assert accessibility_result.structured_content['returned_nodes'] > 0
assert len(accessibility_result.structured_content['session_id']) > 4
get_accessibility_result = await router.execute(
NativeToolCall(tool_name='browser.get_accessibility_tree', arguments={'max_nodes': 50}), context
)
assert get_accessibility_result.is_error is False
assert get_accessibility_result.structured_content['returned_nodes'] > 0
state = await browser_session.get_browser_state_summary(include_screenshot=False)
button_index = next(
idx for idx, element in state.dom_state.selector_map.items() if getattr(element, 'attributes', {}).get('id') == 'inspect'
)
inspect_result = await router.execute(
NativeToolCall(tool_name='browser.inspect_element', arguments={'index': button_index}), context
)
assert inspect_result.is_error is False
assert inspect_result.structured_content['backend_node_id'] == button_index
assert inspect_result.structured_content['target_id']
assert inspect_result.structured_content['html']['text'].startswith('<button')
network_result = await router.execute(NativeToolCall(tool_name='browser.network', arguments={'max_entries': 20}), context)
assert network_result.is_error is False
assert 'pending_requests' in network_result.structured_content
assert 'performance_entries' in network_result.structured_content
fetch_result = await router.execute(
NativeToolCall(tool_name='browser.http_fetch', arguments={'url': httpserver.url_for('/api/tool-data')}), context
)
assert fetch_result.is_error is False
assert fetch_result.structured_content['status'] == 200
assert 'fetch-ok' in fetch_result.structured_content['body']
browser_fetch_result = await router.execute(
NativeToolCall(tool_name='browser.fetch', arguments={'url': httpserver.url_for('/api/tool-data')}), context
)
assert browser_fetch_result.is_error is False
assert browser_fetch_result.structured_content['status'] == 200
assert 'fetch-ok' in browser_fetch_result.structured_content['body']
@pytest.mark.asyncio
async def test_native_tool_router_returns_structured_error_without_browser_session() -> None:
tools = Tools()
session = BrowserAgentSession.create(task='Inspect state')
turn = session.start_turn(step_index=0)
context = session.tool_context(turn, tools=tools)
router = NativeToolRouter.from_tools(tools)
result = await router.execute(NativeToolCall(tool_name='browser.get_state', arguments={}), context)
assert result.is_error is True
assert 'requires ToolContext.browser_session' in (result.content or '')
@pytest.mark.asyncio
async def test_native_tool_router_executes_agent_workspace_file_aliases(tmp_path) -> None:
tools = Tools()
router = NativeToolRouter.from_tools(tools)
session = BrowserAgentSession.create(task='Use file aliases')
turn = session.start_turn(step_index=0)
file_system = FileSystem(tmp_path / 'agent-files', create_default_files=False)
context = session.tool_context(turn, tools=tools, file_system=file_system)
write_result = await router.execute(
NativeToolCall(
tool_name='file.write',
arguments={'path': 'reports/notes.txt', 'content': 'file-ok', 'create_parent_dirs': True},
),
context,
)
assert write_result.is_error is False
read_result = await router.execute(NativeToolCall(tool_name='file.read', arguments={'path': 'reports/notes.txt'}), context)
assert read_result.is_error is False
assert read_result.structured_content['content'] == 'file-ok'
assert read_result.artifact_ids
escape_result = await router.execute(NativeToolCall(tool_name='file.read', arguments={'path': '../outside.txt'}), context)
assert escape_result.is_error is True
assert 'escapes workspace root' in (escape_result.content or '')
@pytest.mark.asyncio
async def test_native_tool_router_executes_permission_gated_workspace_tools(tmp_path) -> None:
class BrowserSessionWithDownloads:
downloaded_files: list[str]
def __init__(self, downloaded_files: list[str]) -> None:
self.downloaded_files = downloaded_files
tools = Tools()
router = NativeToolRouter.from_tools(tools, include_workspace_tools=True)
session = BrowserAgentSession.create(task='Use workspace tools')
turn = session.start_turn(step_index=0)
workspace_root = tmp_path / 'workspace'
source_dir = tmp_path / 'source'
source_dir.mkdir()
available_csv = source_dir / 'available.csv'
available_csv.write_text('name,value\nalpha,1\n', encoding='utf-8')
downloaded_html = source_dir / 'downloaded.html'
downloaded_html.write_text('<html><body>downloaded artifact</body></html>', encoding='utf-8')
generated_json = source_dir / 'generated.json'
generated_json.write_text('{"generated": true}', encoding='utf-8')
session.artifact_store.add(kind='generated', path=generated_json, name='generated.json', media_type='application/json')
denied_context = session.tool_context(turn, tools=tools, metadata={'workspace_root': str(workspace_root)})
denied = await router.execute(
NativeToolCall(tool_name='workspace.write_file', arguments={'path': 'notes.txt', 'content': 'blocked'}),
denied_context,
)
assert denied.is_error is True
assert 'allow_file_tools' in (denied.content or '')
context = session.tool_context(
turn,
tools=tools,
browser_session=BrowserSessionWithDownloads([str(downloaded_html)]),
available_file_paths=[str(available_csv)],
file_system=FileSystem(tmp_path / 'file-system', create_default_files=False),
metadata={
'workspace_root': str(workspace_root),
'allow_file_tools': True,
'allow_shell_tools': True,
'allowed_shell_commands': ['/bin/echo'],
},
)
write_result = await router.execute(
NativeToolCall(
tool_name='workspace.write_file',
arguments={'path': 'reports/notes.txt', 'content': 'workspace-ok', 'create_parent_dirs': True},
),
context,
)
assert write_result.is_error is False
assert (workspace_root / 'reports' / 'notes.txt').read_text() == 'workspace-ok'
read_result = await router.execute(
NativeToolCall(tool_name='workspace.read_file', arguments={'path': 'reports/notes.txt'}), context
)
assert read_result.is_error is False
assert 'workspace-ok' in read_result.structured_content['content']
assert read_result.artifact_ids
list_result = await router.execute(
NativeToolCall(tool_name='workspace.list_files', arguments={'path': '.', 'recursive': True}), context
)
assert list_result.is_error is False
assert any(entry['path'] == 'reports/notes.txt' for entry in list_result.structured_content['entries'])
import_result = await router.execute(
NativeToolCall(tool_name='workspace.import_artifacts', arguments={'destination_dir': 'artifacts'}), context
)
assert import_result.is_error is False
imported_paths = {item['workspace_path'] for item in import_result.structured_content['imported']}
assert {'artifacts/available.csv', 'artifacts/downloaded.html', 'artifacts/generated.json'} <= imported_paths
assert len(import_result.artifact_ids) == 3
assert len(context.artifact_store.artifacts) >= 4
assert BrowserRuntimeEventTypes.ARTIFACT_CREATED in [event.event_type for event in session.event_stream.events]
imported_read = await router.execute(
NativeToolCall(tool_name='workspace.read_file', arguments={'path': 'artifacts/available.csv'}), context
)
assert imported_read.is_error is False
assert 'alpha,1' in imported_read.structured_content['content']
escape_result = await router.execute(
NativeToolCall(tool_name='workspace.read_file', arguments={'path': '../outside.txt'}), context
)
assert escape_result.is_error is True
assert 'escapes workspace root' in (escape_result.content or '')
shell_result = await router.execute(
NativeToolCall(tool_name='shell.run', arguments={'command': ['/bin/echo', 'shell-ok']}), context
)
assert shell_result.is_error is False
assert shell_result.structured_content['stdout'].strip() == 'shell-ok'
blocked_shell = await router.execute(NativeToolCall(tool_name='shell.run', arguments={'command': ['/bin/pwd']}), context)
assert blocked_shell.is_error is True
assert 'not allowed' in (blocked_shell.content or '')
def test_click_coordinates_translate_to_current_click_shape() -> None:
arguments = click_coordinates_as_click_arguments(ClickCoordinatesInput(coordinate_x=10, coordinate_y=20))
assert arguments == {'coordinate_x': 10, 'coordinate_y': 20}