From 385cca72a7cc128346c24680c38caa609d3b3704 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 03:04:56 +0000 Subject: [PATCH] test: satisfy pyright strict annotations on the new supervise RPC tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add parameter/return annotations to the fake resolver's propose_supervise / poll_supervise, annotate the test helpers and the shared _ARGS payload, and narrow the None-able poll result — no behavior change. pylint 9.83, pyright clean. Co-Authored-By: Claude Opus 4.8 --- tests/unit/test_egress_addon_request_flow.py | 13 +++++---- tests/unit/test_orchestrator_control_plane.py | 12 +++++---- tests/unit/test_policy_resolver.py | 1 + tests/unit/test_supervise_server.py | 27 ++++++++++++------- 4 files changed, 33 insertions(+), 20 deletions(-) diff --git a/tests/unit/test_egress_addon_request_flow.py b/tests/unit/test_egress_addon_request_flow.py index 3de75b9..e9223ad 100644 --- a/tests/unit/test_egress_addon_request_flow.py +++ b/tests/unit/test_egress_addon_request_flow.py @@ -279,14 +279,15 @@ class _SuperviseRpcFake: propose_error: bool = False @property - def propose_calls(self) -> list: + def propose_calls(self) -> list[dict[str, str]]: if not hasattr(self, "_propose_calls"): - self._propose_calls: list = [] + self._propose_calls: list[dict[str, str]] = [] return self._propose_calls def propose_supervise( - self, source_ip, identity_token, *, tool, proposed_file, justification, - ): + self, source_ip: str, identity_token: str, *, + tool: str, proposed_file: str, justification: str, + ) -> str: del proposed_file, justification self.propose_calls.append( {"source_ip": source_ip, "identity_token": identity_token, "tool": tool} @@ -295,7 +296,9 @@ class _SuperviseRpcFake: raise PolicyResolveError("orchestrator down") return "prop-1" - def poll_supervise(self, source_ip, identity_token, proposal_id): + def poll_supervise( + self, source_ip: str, identity_token: str, proposal_id: str, + ) -> dict[str, object]: del source_ip, identity_token, proposal_id if self.supervise_status is None: return {"status": "pending"} diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index fe445b2..2e98eaf 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -21,7 +21,7 @@ from unittest.mock import patch from bot_bottle.orchestrator.broker import StubBroker from bot_bottle.orchestrator.control_plane import dispatch, make_server -from bot_bottle.orchestrator.registry import RegistryStore +from bot_bottle.orchestrator.registry import BottleRecord, RegistryStore from bot_bottle.orchestrator.service import Orchestrator from bot_bottle.store_manager import StoreManager from bot_bottle.supervise import ( @@ -455,13 +455,13 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase): return self.store.register( source_ip, metadata=json.dumps({"slug": slug}), policy="routes: []\n") - def _propose(self, rec, proposed: str = "routes:\n - host: g.com\n"): + def _propose(self, rec: BottleRecord, proposed: str = "routes:\n - host: g.com\n"): return dispatch(self.orch, "POST", "/supervise/propose", _body({ "source_ip": rec.source_ip, "identity_token": rec.identity_token, "tool": TOOL_EGRESS_ALLOW, "proposed_file": proposed, "justification": "need it", })) - def _poll(self, rec, proposal_id: str): + def _poll(self, rec: BottleRecord, proposal_id: str): return dispatch(self.orch, "POST", "/supervise/poll", _body({ "source_ip": rec.source_ip, "identity_token": rec.identity_token, "proposal_id": proposal_id, @@ -474,9 +474,11 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase): pid = payload["proposal_id"] assert isinstance(pid, str) and pid _, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"") - self.assertEqual(pid, listing["proposals"][0]["id"]) + proposals = listing["proposals"] + assert isinstance(proposals, list) + self.assertEqual(pid, proposals[0]["id"]) # Queued under the orchestrator-resolved bottle id, never a caller slug. - self.assertEqual(rec.bottle_id, listing["proposals"][0]["bottle_slug"]) + self.assertEqual(rec.bottle_id, proposals[0]["bottle_slug"]) def test_propose_unattributed_is_403(self) -> None: status, payload = dispatch(self.orch, "POST", "/supervise/propose", _body({ diff --git a/tests/unit/test_policy_resolver.py b/tests/unit/test_policy_resolver.py index 6ab5d1c..cc5bde0 100644 --- a/tests/unit/test_policy_resolver.py +++ b/tests/unit/test_policy_resolver.py @@ -144,6 +144,7 @@ class TestPolicyResolver(unittest.TestCase): {"status": "approved", "notes": "ok", "final_file": None}) ) as m: result = self.r.poll_supervise("10.243.0.7", "tok", "p-7") + assert result is not None self.assertEqual("approved", result["status"]) req = m.call_args.args[0] self.assertTrue(req.full_url.endswith("/supervise/poll")) diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index 725943f..26b8ed6 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -15,7 +15,6 @@ import time import types import unittest from pathlib import Path -from unittest.mock import patch from tests.unit import use_bottle_root @@ -72,8 +71,9 @@ class _FakeSuperviseResolver: self._policy = policy def propose_supervise( - self, source_ip, identity_token, *, tool, proposed_file, justification, - ): + self, source_ip: str, identity_token: str, *, + tool: str, proposed_file: str, justification: str, + ) -> str | None: del source_ip, identity_token if self.raises: raise supervise_server.PolicyResolveError("orchestrator down") @@ -86,7 +86,9 @@ class _FakeSuperviseResolver: _sv.write_proposal(proposal) return proposal.id - def poll_supervise(self, source_ip, identity_token, proposal_id): + def poll_supervise( + self, source_ip: str, identity_token: str, proposal_id: str, + ) -> dict[str, object] | None: del source_ip, identity_token if self.raises: raise supervise_server.PolicyResolveError("orchestrator down") @@ -107,23 +109,28 @@ class _FakeSuperviseResolver: } # Used by list-egress-routes (`_resolved_routes_payload`), unchanged path. - def resolve_policy_and_bottle_id(self, source_ip, identity_token=""): + def resolve_policy_and_bottle_id( + self, source_ip: str, identity_token: str = "", + ) -> tuple[str | None, str | None, dict[str, str]]: del source_ip, identity_token if self.raises: raise supervise_server.PolicyResolveError("orchestrator down") return self._policy, self.bottle_id, {} -def _tools_call(resolver, params, config=None): +def _tools_call( + resolver: object, params: dict[str, object], + config: "ServerConfig | None" = None, +) -> dict[str, object]: return handle_tools_call( params, config or ServerConfig(), - resolver=resolver, source_ip=_SRC, identity_token=_TOK, + resolver=resolver, source_ip=_SRC, identity_token=_TOK, # type: ignore[arg-type] ) -def _check(resolver, params): +def _check(resolver: object, params: dict[str, object]) -> dict[str, object]: return handle_check_proposal( - params, resolver=resolver, source_ip=_SRC, identity_token=_TOK, + params, resolver=resolver, source_ip=_SRC, identity_token=_TOK, # type: ignore[arg-type] ) @@ -203,7 +210,7 @@ class TestRpcInternalErrorOnRpcFailure(unittest.TestCase): surfaces as ERR_INTERNAL — the daemon fails closed rather than leaking the cause to the agent.""" - _ARGS = { + _ARGS: dict[str, object] = { "name": _sv.TOOL_EGRESS_ALLOW, "arguments": { "routes_yaml": "routes:\n - host: example.com\n",