a2a-source-edit: write tests/test_runtime_safety.py

tests/test_runtime_safety.py

@@ -1,60 +1,12 @@
-from __future__ import annotations
+"""Runtime safety checks for the current deterministic release validator."""
 
-import pytest
-from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode
-
-from agent import CustomerIntegrationEngineer, EndpointTest, RunAcceptanceRequest
+from agent import CustomerIntegrationEngineer
 
 
-def workspace():
-    return LocalWorkspaceClient(
-        {},
-        access=WorkspaceAccess.dynamic(
-            max_files=256,
-            allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
-            require_reason=False,
-        ),
-    )
+def test_no_llm_required_by_pricing():
+    assert CustomerIntegrationEngineer.pricing.caller_pays_llm is 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
+def test_workspace_modes_declared():
+    modes = {mode.value for mode in CustomerIntegrationEngineer.workspace_access.allowed_modes}
+    assert {"read_only", "read_write_overlay"} <= modes