test: satisfy pyright strict annotations on the new supervise RPC tests
tracker-policy-pr / check-pr (pull_request) Successful in 12s
test / integration-docker (pull_request) Successful in 19s
test / unit (pull_request) Failing after 50s
lint / lint (push) Successful in 2m51s
test / integration-firecracker (pull_request) Successful in 3m34s
test / coverage (pull_request) Has been skipped
test / publish-infra (pull_request) Has been skipped

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 <noreply@anthropic.com>
This commit is contained in:
2026-07-24 02:20:56 +00:00
parent f0d2a4b855
commit 8a82861459
4 changed files with 33 additions and 20 deletions
@@ -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({