test/acp/test_bridge.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
import sys
from pathlib import Path
import pytest
from acp import schema
from ag2.acp.bridge import BridgeState, make_bridge
from ag2.acp.config import ACPConfig
from ag2.events import BaseEvent, ModelMessageChunk, ModelReasoning
def _text(text: str) -> schema.TextContentBlock:
return schema.TextContentBlock(type="text", text=text)
class FakeContext:
def __init__(self) -> None:
self.sent: list[BaseEvent] = []
async def send(self, event: BaseEvent) -> None:
self.sent.append(event)
def _state(context: FakeContext, **cfg: object) -> BridgeState:
st = BridgeState(ACPConfig(**cfg)) # type: ignore[arg-type]
st.context = context # type: ignore[assignment]
st.begin_turn()
return st
@pytest.mark.asyncio
async def test_handle_update_sends_event_and_accumulates_text() -> None:
ctx = FakeContext()
st = _state(ctx)
await st.handle_update(schema.AgentMessageChunk(session_update="agent_message_chunk", content=_text("Hello ")))
await st.handle_update(schema.AgentMessageChunk(session_update="agent_message_chunk", content=_text("world")))
assert all(isinstance(e, ModelMessageChunk) for e in ctx.sent)
assert st.turn_text == "Hello world"
@pytest.mark.asyncio
async def test_thought_is_sent_but_not_in_turn_text() -> None:
ctx = FakeContext()
st = _state(ctx)
await st.handle_update(schema.AgentThoughtChunk(session_update="agent_thought_chunk", content=_text("thinking")))
assert isinstance(ctx.sent[0], ModelReasoning)
assert st.turn_text == ""
@pytest.mark.asyncio
async def test_unknown_update_sends_nothing() -> None:
ctx = FakeContext()
st = _state(ctx)
await st.handle_update(schema.SessionInfoUpdate(session_update="session_info_update", title="x"))
assert ctx.sent == []
@pytest.mark.asyncio
async def test_begin_turn_resets_buffer() -> None:
ctx = FakeContext()
st = _state(ctx)
await st.handle_update(schema.AgentMessageChunk(session_update="agent_message_chunk", content=_text("a")))
st.begin_turn()
assert st.turn_text == ""
def test_fs_write_then_read(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
st.write_text_file("hi\nthere\n", "notes.txt")
assert (tmp_path / "notes.txt").read_text() == "hi\nthere\n"
assert st.read_text_file("notes.txt") == "hi\nthere\n"
# absolute paths inside the root are equally fine
assert st.read_text_file(str(tmp_path / "notes.txt")) == "hi\nthere\n"
def test_fs_read_with_line_and_limit(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
st.write_text_file("l1\nl2\nl3\nl4\n", "f.txt")
assert st.read_text_file("f.txt", line=2, limit=2) == "l2\nl3\n"
def test_fs_rejects_escape(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
with pytest.raises(PermissionError):
st.read_text_file("../../etc/passwd")
with pytest.raises(PermissionError):
st.read_text_file(str(tmp_path / ".." / "etc" / "passwd"))
@pytest.mark.asyncio
async def test_permission_auto_returns_allow_id() -> None:
st = BridgeState(ACPConfig(permission_policy="auto"))
chosen = await st.resolve_permission(
[schema.PermissionOption(option_id="ok", kind="allow_once", name="Allow")],
schema.ToolCallUpdate(tool_call_id="tc1", title="Edit"),
)
assert chosen == "ok"
@pytest.mark.asyncio
async def test_request_permission_allows_selected_option() -> None:
bridge = make_bridge(ACPConfig(permission_policy="auto"))
resp = await bridge.request_permission(
options=[schema.PermissionOption(option_id="ok", kind="allow_once", name="Allow")],
session_id="s",
tool_call=schema.ToolCallUpdate(tool_call_id="tc1", title="Edit"),
)
assert resp.outcome == schema.AllowedOutcome(option_id="ok", outcome="selected")
@pytest.mark.asyncio
async def test_request_permission_denies_without_crashing() -> None:
# policy="deny" but no reject option -> resolve yields None -> the bridge must
# return a well-formed DeniedOutcome (regression: DeniedOutcome() was invalid).
bridge = make_bridge(ACPConfig(permission_policy="deny"))
resp = await bridge.request_permission(
options=[schema.PermissionOption(option_id="ok", kind="allow_once", name="Allow")],
session_id="s",
tool_call=schema.ToolCallUpdate(tool_call_id="tc1", title="Edit"),
)
assert resp.outcome == schema.DeniedOutcome(outcome="cancelled")
@pytest.mark.asyncio
async def test_terminal_runs_and_captures_output(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
tid = await st.terminals.create("printf", ["hello"])
exit_status = await st.terminals.wait(tid)
output, truncated, status = st.terminals.output(tid)
assert output == "hello"
assert truncated is False
assert exit_status == schema.TerminalExitStatus(exit_code=0, signal=None)
assert status == schema.TerminalExitStatus(exit_code=0, signal=None)
await st.terminals.release(tid)
@pytest.mark.asyncio
async def test_terminal_output_byte_limit_truncates(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
tid = await st.terminals.create("printf", ["abcdefghij"], output_byte_limit=4)
await st.terminals.wait(tid)
output, truncated, _ = st.terminals.output(tid)
assert output == "abcd"
assert truncated is True
await st.terminals.release(tid)
@pytest.mark.asyncio
async def test_terminal_cwd_confined(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
with pytest.raises(PermissionError):
await st.terminals.create("printf", ["x"], cwd="../../etc")
@pytest.mark.asyncio
async def test_terminal_kill(tmp_path: Path) -> None:
st = BridgeState(ACPConfig(cwd=str(tmp_path)))
tid = await st.terminals.create("sleep", ["30"])
await st.terminals.kill(tid)
exit_status = await st.terminals.wait(tid)
if sys.platform == "win32":
# Windows has no signals: TerminateProcess surfaces as a non-zero exit
# code, so exit_status() reports exit_code instead.
assert exit_status.exit_code is not None
else:
assert exit_status.signal is not None # terminated by signal
await st.terminals.release(tid)