test: satisfy pyright strict annotations on the new supervise RPC tests
test / integration-docker (pull_request) Successful in 35s
test / unit (pull_request) Failing after 43s
tracker-policy-pr / check-pr (pull_request) Successful in 13s
lint / lint (push) Successful in 59s
test / integration-firecracker (pull_request) Successful in 3m33s
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 2c496dc3d0
commit 2acaa082e4
4 changed files with 33 additions and 20 deletions
+17 -10
View File
@@ -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",