From 9e9d29af7ccbd877ca31c6307b5b8fe8128efa24 Mon Sep 17 00:00:00 2001
From: a2a-cloud <noreply@a2acloud.io>
Date: Mon, 13 Jul 2026 03:18:42 +0000
Subject: [PATCH] a2a-source-edit: write tests/test_runtime_safety.py

---
 tests/test_runtime_safety.py | 60 ++++++++++++++++++++++++++++++++++++
 1 file changed, 60 insertions(+)
 create mode 100644 tests/test_runtime_safety.py

diff --git a/tests/test_runtime_safety.py b/tests/test_runtime_safety.py
new file mode 100644
index 0000000..f02d473
--- /dev/null
+++ b/tests/test_runtime_safety.py
@@ -0,0 +1,60 @@
+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