61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode
|
|
|
|
from agent import CustomerIntegrationEngineer, EndpointTest, RunAcceptanceRequest
|
|
|
|
|
|
def workspace():
|
|
return LocalWorkspaceClient(
|
|
{},
|
|
access=WorkspaceAccess.dynamic(
|
|
max_files=256,
|
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
|
require_reason=False,
|
|
),
|
|
)
|
|
|
|
|
|
def ctx():
|
|
return LocalRunContext(auth=NoAuth(), workspace=workspace(), task_id="test")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_dry_run_forces_skip_even_if_request_allows_network():
|
|
agent = CustomerIntegrationEngineer(config={"default_dry_run": True})
|
|
result = await agent.run_acceptance(
|
|
ctx(),
|
|
RunAcceptanceRequest(
|
|
integration_id="forced-dry-run",
|
|
endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events")],
|
|
external_host_allowlist=["hooks.example.com"],
|
|
allow_network=True,
|
|
dry_run=False,
|
|
),
|
|
)
|
|
assert result.status == "ok"
|
|
assert result.dry_run is True
|
|
assert result.network_calls_made == 0
|
|
assert result.step_results[0].status == "skipped"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_network_blocked_by_runtime_egress_policy():
|
|
agent = CustomerIntegrationEngineer(config={"default_dry_run": False})
|
|
result = await agent.run_acceptance(
|
|
ctx(),
|
|
RunAcceptanceRequest(
|
|
integration_id="live-blocked",
|
|
endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events")],
|
|
external_host_allowlist=["hooks.example.com"],
|
|
allow_network=True,
|
|
dry_run=False,
|
|
),
|
|
)
|
|
assert result.status == "blocked"
|
|
assert result.dry_run is False
|
|
assert result.network_calls_made == 0
|
|
assert result.step_results[0].status == "blocked"
|
|
assert "egress denies internet by default" in result.step_results[0].detail
|