server/tests/test_approval.py
Rama main · réplica de lectura del repositorio original.
import asyncio
from pathlib import Path
import pytest
from bumblehive.protocols import ToolCall
from bumblehive.protocols.tool_approval import ToolApprovalRequest
from bumblehive.tools import ToolManager
from bumblehive_server.chat.approval import ToolApprovals, approval_details
def request(workspace, name, **arguments):
return ToolApprovalRequest("call-1", name, arguments, workspace.resolve())
@pytest.mark.parametrize(
"name",
["read_file", "list_dir", "find_files", "grep", "list_exec_sessions", "sub_agent"],
)
def test_reads_and_readonly_subagent_are_automatic(tmp_path, name):
assert approval_details(request(tmp_path, name, path="/outside")) is None
@pytest.mark.parametrize("name", ["write_file", "edit_file"])
def test_file_scope_matches_resolved_paths(tmp_path, name):
workspace = tmp_path / "workspace"
workspace.mkdir()
assert approval_details(request(workspace, name, path="inside.txt")) is None
assert (
approval_details(request(workspace, name, path=str(workspace / "inside.txt")))
is None
)
outside = tmp_path / "outside.txt"
(workspace / "link").symlink_to(outside)
for path in ["../outside.txt", str(outside), "link"]:
details = approval_details(request(workspace, name, path=path))
assert details["outside_paths"] == [str(outside)]
# Keep whitespace and follow the platform's path resolution semantics.
raw_path = " ../outside.txt "
resolved = (workspace / raw_path).resolve()
details = approval_details(request(workspace, name, path=raw_path))
if resolved.is_relative_to(workspace.resolve()):
assert details is None
else:
assert details["outside_paths"] == [str(resolved)]
def test_patch_strips_paths_and_approves_dry_run(tmp_path):
workspace = tmp_path / "workspace"
workspace.mkdir()
edits = [{"path": "inside.txt"}, {"path": " ../outside.txt "}]
details = approval_details(request(workspace, "apply_patch", edits=edits))
assert details["paths"] == [
str(workspace / "inside.txt"),
str(tmp_path / "outside.txt"),
]
assert details["outside_paths"] == [str(tmp_path / "outside.txt")]
assert (
approval_details(request(workspace, "apply_patch", edits=edits, dry_run=True))
is None
)
@pytest.mark.parametrize("working_dir", [None, "", ".", ".."])
def test_shell_always_asks_and_displays_effective_directory(tmp_path, working_dir):
details = approval_details(
request(tmp_path, "exec", command="pwd", working_dir=working_dir)
)
assert details["command"] == "pwd"
assert Path(details["working_dir"]) == (tmp_path / (working_dir or ".")).resolve()
@pytest.mark.parametrize(
"arguments,asks",
[
({}, False),
({"chars": ""}, False),
({"wait_for": "done"}, False),
({"terminate": True}, False),
({"chars": "\n"}, True),
({"close_stdin": True}, True),
({"terminate": True, "chars": "exit\n"}, True),
({"terminate": True, "close_stdin": True}, True),
],
)
def test_stdin_rules(tmp_path, arguments, asks):
details = approval_details(
request(tmp_path, "write_stdin", session_id="exec-1", **arguments)
)
assert (details is not None) == asks
def test_unknown_tools_ask(tmp_path):
assert approval_details(request(tmp_path, "mcp_update")) == {"reason": "tool"}
@pytest.mark.asyncio
@pytest.mark.parametrize("approved", [False, True])
@pytest.mark.parametrize("patch", [False, True])
async def test_real_write_waits_for_decision(tmp_path, approved, patch):
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside.txt"
frames = asyncio.Queue()
approvals = ToolApprovals("session-1", frames.put)
call = (
ToolCall(
"call-1",
"apply_patch",
{
"edits": [
{"path": "inside.txt", "action": "add", "new_text": "hello"},
{"path": " ../outside.txt ", "action": "add", "new_text": "hello"},
]
},
)
if patch
else ToolCall(
"call-1", "write_file", {"path": "../outside.txt", "content": "hello"}
)
)
async with ToolManager() as tools:
tools.register_builtin_tools()
task = asyncio.create_task(
tools.execute_call(call, workspace=workspace, approval_handler=approvals)
)
frame = await asyncio.wait_for(frames.get(), 2)
assert frame["type"] == "approval_request"
assert frame["workspace"] == str(workspace)
assert not task.done() and not outside.exists()
assert not (workspace / "inside.txt").exists()
await approvals.decide(frame["approval_id"], approved)
assert (await frames.get())["type"] == "approval_resolved"
result = await asyncio.wait_for(task, 2)
assert outside.exists() is approved
if patch:
assert (workspace / "inside.txt").exists() is approved
if approved:
assert result.error is None
assert outside.read_text() == ("hello\n" if patch else "hello")
else:
assert result.error.code == "tool_approval_denied"
assert result.error.message == "The user rejected this tool call."
await approvals.decide(frame["approval_id"], not approved)
assert frames.empty() and not approvals.pending
@pytest.mark.asyncio
async def test_parallel_approvals_are_isolated_and_cleared(tmp_path):
frames = asyncio.Queue()
first = ToolApprovals("first", frames.put)
second = ToolApprovals("second", frames.put)
tasks = [
asyncio.create_task(manager(request(tmp_path, "exec", command="pwd")))
for manager in (first, first, second)
]
requests = [await asyncio.wait_for(frames.get(), 2) for _ in tasks]
own = [frame for frame in requests if frame["session_id"] == "first"]
foreign = next(frame for frame in requests if frame["session_id"] == "second")
await first.decide(foreign["approval_id"], True)
assert all(not task.done() for task in tasks)
await first.decide(own[1]["approval_id"], False)
assert not (await tasks[1]).approved
first.clear()
second.clear()
results = await asyncio.gather(*tasks, return_exceptions=True)
assert isinstance(results[0], asyncio.CancelledError)
assert isinstance(results[2], asyncio.CancelledError)
assert not first.pending and not second.pending
@pytest.mark.asyncio
async def test_full_access_skips_confirmation(tmp_path):
frames = asyncio.Queue()
approvals = ToolApprovals("session-1", frames.put, "full_access")
for name, arguments in [
("write_file", {"path": "../outside.txt"}),
("exec", {"command": "pwd"}),
("write_stdin", {"chars": "exit\n"}),
("custom_tool", {}),
]:
assert (await approvals(request(tmp_path, name, **arguments))).approved
assert frames.empty() and not approvals.pending
def test_chat_approval_mode_defaults_and_validation():
from pydantic import ValidationError
from bumblehive_server.schemas import ChatRequest
assert ChatRequest(content="hello").approval_mode == "request"
assert ChatRequest(content="hello", approval_mode="full_access").approval_mode == "full_access"
with pytest.raises(ValidationError):
ChatRequest(content="hello", approval_mode="unknown")