fix(supervise): restore egress proposal tools
lint / lint (push) Successful in 1m35s
test / unit (pull_request) Successful in 30s
test / integration (pull_request) Successful in 16s

This commit is contained in:
2026-06-23 01:24:28 -04:00
parent 3ea35ba5d2
commit 4c39b45e34
6 changed files with 164 additions and 34 deletions
+10 -3
View File
@@ -317,15 +317,22 @@ class TestToolConstants(unittest.TestCase):
def test_tools_tuple_matches_individual_constants(self):
self.assertEqual(
(
supervise.TOOL_ALLOW,
TOOL_CAPABILITY_BLOCK,
supervise.TOOL_EGRESS_BLOCK,
supervise.TOOL_LIST_EGRESS_ROUTES,
),
supervise.TOOLS,
)
def test_component_map_has_no_entries(self):
# egress-block removed in issue #198; capability-block never had one.
self.assertEqual({}, supervise.COMPONENT_FOR_TOOL)
def test_component_map_has_egress_entries(self):
self.assertEqual(
{
supervise.TOOL_ALLOW: "egress",
supervise.TOOL_EGRESS_BLOCK: "egress",
},
supervise.COMPONENT_FOR_TOOL,
)
class _StubSupervise(supervise.Supervise):
+10 -3
View File
@@ -2,9 +2,6 @@
The curses TUI itself isn't exercised here — these tests cover the
discovery + approve/reject paths that the TUI's key handlers call into.
egress-block (add_route) was removed in issue #198; the TestEgressApplyWiring
class and all stubs for add_route have been dropped accordingly.
"""
import os
@@ -33,6 +30,8 @@ FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal(slug: str = "dev", tool: str = TOOL_CAPABILITY_BLOCK) -> Proposal:
payloads = {
TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n",
supervise.TOOL_ALLOW: "routes:\n - host: example.com\n",
supervise.TOOL_EGRESS_BLOCK: "routes:\n - host: example.com\n",
}
payload = payloads.get(tool, "")
return Proposal.new(
@@ -154,6 +153,14 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
supervise_cli.approve(qp)
self.assertEqual([], read_audit_entries("egress", "dev"))
def test_approve_egress_block_writes_audit_log(self):
qp = self._enqueue(tool=supervise.TOOL_EGRESS_BLOCK)
supervise_cli.approve(qp)
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual(STATUS_APPROVED, entries[0].operator_action)
self.assertEqual("needed for dev", entries[0].justification)
# class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
# # DISABLED — capability_apply functionality is currently commented out.
+45 -5
View File
@@ -54,13 +54,19 @@ class TestValidation(unittest.TestCase):
)
def test_empty_proposed_file_rejected_for_tools_with_file_field(self):
# egress-block has structured input (validated in
# _validate_and_bundle_egress_route, not here) and
# list-egress-routes takes no input. Only capability-block
# goes through `validate_proposed_file`.
with self.assertRaises(_RpcError):
validate_proposed_file(_sv.TOOL_CAPABILITY_BLOCK, " \n\t")
def test_egress_routes_yaml_is_validated(self):
validate_proposed_file(
_sv.TOOL_ALLOW,
"routes:\n - host: example.com\n",
)
def test_invalid_egress_routes_yaml_rejected(self):
with self.assertRaises(_RpcError):
validate_proposed_file(_sv.TOOL_EGRESS_BLOCK, "routes: nope\n")
# --- JSON-RPC parsing ------------------------------------------------------
@@ -141,7 +147,9 @@ class TestHandleToolsList(unittest.TestCase):
names = [t["name"] for t in result["tools"]] # type: ignore[index]
self.assertEqual(
sorted([
_sv.TOOL_ALLOW,
_sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_LIST_EGRESS_ROUTES,
]),
sorted(names),
@@ -172,6 +180,17 @@ class TestHandleToolsList(unittest.TestCase):
# No `required` array because no inputs are required.
self.assertNotIn("required", schema) # type: ignore[operator]
def test_egress_tools_take_routes_yaml_and_justification(self):
for tool_name in (_sv.TOOL_ALLOW, _sv.TOOL_EGRESS_BLOCK):
with self.subTest(tool_name=tool_name):
tool = next(t for t in TOOL_DEFINITIONS if t["name"] == tool_name)
schema = tool["inputSchema"]
self.assertEqual("object", schema["type"]) # type: ignore[index]
self.assertEqual(
["routes_yaml", "justification"],
schema["required"], # type: ignore[index]
)
class TestHandleToolsCall(unittest.TestCase):
def setUp(self):
@@ -220,6 +239,26 @@ class TestHandleToolsCall(unittest.TestCase):
self.assertIn("status: approved", text)
self.assertIn("notes: lgtm", text)
def test_allow_round_trips_through_queue(self):
responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED, notes="ok")
try:
result = handle_tools_call(
{
"name": _sv.TOOL_ALLOW,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"justification": "need example.com",
},
},
self.config,
)
finally:
responder.join()
self.assertFalse(result["isError"]) # type: ignore[index]
text = result["content"][0]["text"] # type: ignore[index]
self.assertIn("status: approved", text)
self.assertIn("notes: ok", text)
def test_rejected_response_sets_isError(self):
responder = self._respond_when_proposal_appears(_sv.STATUS_REJECTED, notes="nope")
try:
@@ -412,7 +451,8 @@ class TestHttpEndToEnd(unittest.TestCase):
self.assertEqual(1, result["id"])
names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index]
self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
self.assertNotIn("egress-block", names)
self.assertIn(_sv.TOOL_ALLOW, names)
self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
def test_unknown_method_returns_jsonrpc_error(self):
result = self._post_jsonrpc(