From 282c2d7f5491fad9556b6269954ff213e8d5877e Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Mon, 15 Jun 2026 10:00:52 +0800 Subject: [PATCH] test(tools): cover runtime hardening edge cases --- .../pkg/provider/tools/loaders/mcp_stdio.py | 7 +- .../provider/test_mcp_box_integration.py | 28 ++++ .../provider/test_tool_manager_native.py | 132 ++++++++++++++++++ 3 files changed, 163 insertions(+), 4 deletions(-) diff --git a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py index b74b077b..dcfbb913 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py @@ -6,6 +6,7 @@ import os import shutil import shlex import threading +from contextlib import suppress from typing import TYPE_CHECKING, Any import pydantic @@ -285,11 +286,9 @@ class BoxStdioSessionRuntime: if os.path.isdir(path) and not os.path.islink(path): shutil.rmtree(path, ignore_errors=True) else: - try: + # The entry may disappear between listdir and unlink if cleanup races us. + with suppress(FileNotFoundError): os.unlink(path) - except FileNotFoundError: - # The entry may disappear between listdir and unlink if cleanup races us. - pass shutil.copytree( source_path, process_host_workspace, diff --git a/tests/unit_tests/provider/test_mcp_box_integration.py b/tests/unit_tests/provider/test_mcp_box_integration.py index 598f7dce..74cd2487 100644 --- a/tests/unit_tests/provider/test_mcp_box_integration.py +++ b/tests/unit_tests/provider/test_mcp_box_integration.py @@ -543,6 +543,34 @@ class TestPythonWorkspacePreparation: assert (workspace / '.venv' / 'bin' / 'python').exists() assert (workspace / '.langbot' / 'python-env.lock').is_dir() + def test_staging_refresh_ignores_unlink_race(self, mcp_module, tmp_path, monkeypatch): + mcp_stdio_module = sys.modules['langbot.pkg.provider.tools.loaders.mcp_stdio'] + + source = tmp_path / 'source' + source.mkdir() + (source / 'server.py').write_text('print("new")\n', encoding='utf-8') + + process_root = tmp_path / 'shared' / '.mcp' / 'u1' + workspace = process_root / 'workspace' + workspace.mkdir(parents=True) + stale_file = workspace / 'removed.py' + stale_file.write_text('stale\n', encoding='utf-8') + + real_unlink = os.unlink + + def unlink_with_race(path): + if os.fspath(path) == str(stale_file): + real_unlink(path) + raise FileNotFoundError(path) + real_unlink(path) + + monkeypatch.setattr(mcp_stdio_module.os, 'unlink', unlink_with_race) + + mcp_module.BoxStdioSessionRuntime._copy_workspace_tree(str(source), str(process_root), str(workspace)) + + assert not stale_file.exists() + assert (workspace / 'server.py').read_text(encoding='utf-8') == 'print("new")\n' + # ── get_runtime_info_dict ─────────────────────────────────────────── diff --git a/tests/unit_tests/provider/test_tool_manager_native.py b/tests/unit_tests/provider/test_tool_manager_native.py index 117a20fd..17901e79 100644 --- a/tests/unit_tests/provider/test_tool_manager_native.py +++ b/tests/unit_tests/provider/test_tool_manager_native.py @@ -248,3 +248,135 @@ async def test_path_escape_blocked(): with pytest.raises(ValueError, match='escapes'): await loader.invoke_tool('read', {'path': '/workspace/../../etc/passwd'}, _make_query()) + + +@pytest.mark.asyncio +async def test_box_availability_helper_handles_unavailable_and_errors(): + from langbot.pkg.provider.tools.loaders.availability import is_box_backend_available + + assert await is_box_backend_available(SimpleNamespace()) is False + assert await is_box_backend_available(SimpleNamespace(box_service=SimpleNamespace(available=False))) is False + + unavailable_backend = SimpleNamespace( + available=True, + get_status=AsyncMock(return_value={'backend': {'available': False}}), + ) + assert await is_box_backend_available(SimpleNamespace(box_service=unavailable_backend)) is False + + failing_backend = SimpleNamespace( + available=True, + get_status=AsyncMock(side_effect=RuntimeError('box unavailable')), + ) + assert await is_box_backend_available(SimpleNamespace(box_service=failing_backend)) is False + + +@pytest.mark.asyncio +async def test_read_file_supports_offset_limit_and_truncation_metadata(): + with tempfile.TemporaryDirectory() as tmpdir: + loader, _ = _make_loader_with_workspace(tmpdir) + with open(os.path.join(tmpdir, 'lines.txt'), 'w', encoding='utf-8') as f: + f.write('one\ntwo\nthree\nfour\n') + + result = await loader.invoke_tool( + 'read', + {'path': '/workspace/lines.txt', 'offset': 2, 'limit': 2}, + _make_query(), + ) + + assert result == { + 'ok': True, + 'content': 'two\nthree', + 'truncated': True, + 'truncated_by': 'lines', + 'start_line': 2, + 'end_line': 3, + 'next_offset': 4, + 'max_lines': 2, + 'max_bytes': 50 * 1024, + } + + +@pytest.mark.asyncio +async def test_read_file_handles_line_larger_than_byte_limit(): + with tempfile.TemporaryDirectory() as tmpdir: + loader, _ = _make_loader_with_workspace(tmpdir) + with open(os.path.join(tmpdir, 'long-line.txt'), 'w', encoding='utf-8') as f: + f.write('abcdef\n') + + result = await loader.invoke_tool( + 'read', + {'path': '/workspace/long-line.txt', 'max_bytes': 3}, + _make_query(), + ) + + assert result['ok'] is True + assert result['truncated'] is True + assert result['truncated_by'] == 'bytes' + assert result['next_offset'] == 1 + assert 'exceeds the 3B read limit' in result['content'] + + +@pytest.mark.asyncio +async def test_exec_result_is_capped_and_exposes_preview_metadata(): + with tempfile.TemporaryDirectory() as tmpdir: + box_service = SimpleNamespace( + available=True, + default_workspace=tmpdir, + execute_tool=AsyncMock( + return_value={ + 'ok': True, + 'stdout': 'a' * 60000, + 'stderr': 'b' * 60000, + 'exit_code': 0, + } + ), + ) + loader = NativeToolLoader(SimpleNamespace(box_service=box_service, logger=Mock())) + + result = await loader.invoke_tool('exec', {'command': 'python -V'}, _make_query()) + + assert result['ok'] is True + assert len(result['stdout'].encode('utf-8')) == 50 * 1024 + assert len(result['stderr'].encode('utf-8')) == 50 * 1024 + assert len(result['preview'].encode('utf-8')) == 50 * 1024 + assert result['stdout_truncated'] is True + assert result['stderr_truncated'] is True + assert result['truncated'] is True + assert result['truncated_by'] == 'bytes' + + +@pytest.mark.asyncio +async def test_glob_caps_match_count_and_returns_preview(): + with tempfile.TemporaryDirectory() as tmpdir: + loader, _ = _make_loader_with_workspace(tmpdir) + for index in range(105): + with open(os.path.join(tmpdir, f'file-{index:03d}.txt'), 'w', encoding='utf-8') as f: + f.write(str(index)) + + result = await loader.invoke_tool('glob', {'path': '/workspace', 'pattern': '*.txt'}, _make_query()) + + assert result['ok'] is True + assert result['total'] == 105 + assert len(result['matches']) == 100 + assert result['preview'] == '\n'.join(result['matches']) + assert result['truncated'] is True + assert result['truncated_by'] == 'matches' + + +@pytest.mark.asyncio +async def test_grep_reports_invalid_regex_and_truncates_long_matching_lines(): + with tempfile.TemporaryDirectory() as tmpdir: + loader, _ = _make_loader_with_workspace(tmpdir) + with open(os.path.join(tmpdir, 'data.txt'), 'w', encoding='utf-8') as f: + f.write('needle ' + ('x' * 600) + '\n') + + invalid = await loader.invoke_tool('grep', {'path': '/workspace', 'pattern': '['}, _make_query()) + result = await loader.invoke_tool('grep', {'path': '/workspace', 'pattern': 'needle'}, _make_query()) + + assert invalid['ok'] is False + assert 'Invalid regex' in invalid['error'] + assert result['ok'] is True + assert result['truncated'] is True + assert result['truncated_by'] == 'line' + assert result['matches'][0]['file'] == '/workspace/data.txt' + assert result['matches'][0]['content'].endswith('... [truncated]')