25481d4286
Restore the ACP editor-integration implementation that was present on the original PR branch but did not actually land in main. Includes: - acp_adapter/ server, session manager, event bridge, auth, permissions, and tool helpers - hermes acp subcommand and hermes-acp entry point - hermes-acp curated toolset - ACP registry manifest, setup guide, and ACP test suite - jupyter-live-kernel data science skill from the original branch Also updates the revived ACP code for current main by: - resolving runtime providers through the modern shared provider router - binding ACP sessions to per-session cwd task overrides - tracking duplicate same-name tool calls with FIFO IDs - restoring terminal approval callbacks after prompts - normalizing supporting docs/skill metadata Validated with tests/acp and the full pytest suite (-n0).
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
"""Tests for acp_adapter.permissions — ACP approval bridging."""
|
|
|
|
import asyncio
|
|
from concurrent.futures import Future
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from acp.schema import (
|
|
AllowedOutcome,
|
|
DeniedOutcome,
|
|
RequestPermissionResponse,
|
|
)
|
|
from acp_adapter.permissions import make_approval_callback
|
|
|
|
|
|
def _make_response(outcome):
|
|
"""Helper to build a RequestPermissionResponse with the given outcome."""
|
|
return RequestPermissionResponse(outcome=outcome)
|
|
|
|
|
|
def _setup_callback(outcome, timeout=60.0):
|
|
"""
|
|
Create a callback wired to a mock request_permission coroutine
|
|
that resolves to the given outcome.
|
|
|
|
Returns:
|
|
(callback, mock_request_permission_fn)
|
|
"""
|
|
loop = MagicMock(spec=asyncio.AbstractEventLoop)
|
|
mock_rp = MagicMock(name="request_permission")
|
|
|
|
response = _make_response(outcome)
|
|
|
|
# Patch asyncio.run_coroutine_threadsafe so it returns a future
|
|
# that immediately yields the response.
|
|
future = MagicMock(spec=Future)
|
|
future.result.return_value = response
|
|
|
|
with patch("acp_adapter.permissions.asyncio.run_coroutine_threadsafe", return_value=future):
|
|
cb = make_approval_callback(mock_rp, loop, session_id="s1", timeout=timeout)
|
|
result = cb("rm -rf /", "dangerous command")
|
|
|
|
return result
|
|
|
|
|
|
class TestApprovalMapping:
|
|
def test_approval_allow_once_maps_correctly(self):
|
|
outcome = AllowedOutcome(option_id="allow_once", outcome="selected")
|
|
result = _setup_callback(outcome)
|
|
assert result == "once"
|
|
|
|
def test_approval_allow_always_maps_correctly(self):
|
|
outcome = AllowedOutcome(option_id="allow_always", outcome="selected")
|
|
result = _setup_callback(outcome)
|
|
assert result == "always"
|
|
|
|
def test_approval_deny_maps_correctly(self):
|
|
outcome = DeniedOutcome(outcome="cancelled")
|
|
result = _setup_callback(outcome)
|
|
assert result == "deny"
|
|
|
|
def test_approval_timeout_returns_deny(self):
|
|
"""When the future times out, the callback should return 'deny'."""
|
|
loop = MagicMock(spec=asyncio.AbstractEventLoop)
|
|
mock_rp = MagicMock(name="request_permission")
|
|
|
|
future = MagicMock(spec=Future)
|
|
future.result.side_effect = TimeoutError("timed out")
|
|
|
|
with patch("acp_adapter.permissions.asyncio.run_coroutine_threadsafe", return_value=future):
|
|
cb = make_approval_callback(mock_rp, loop, session_id="s1", timeout=0.01)
|
|
result = cb("rm -rf /", "dangerous")
|
|
|
|
assert result == "deny"
|