window_watch: wait up to 30 s for OpenBao and the API server, and name an unreachable endpoint
All checks were successful
Bounded-execution contracts / contracts (push) Successful in 1m42s
Build and Publish bounded-executor / build-and-push (push) Successful in 5s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Identity provider journey acceptance / provider (push) Successful in 53s

The first two watch runs after the deploy (2026-11-06 09:20Z, 09:30Z) exited 5 with a bare URLError:
k3s programs a new pod's NetworkPolicy a few seconds late (as for the executor, 2026-10-31). The live
policies admit the watch pod to both endpoints. NK-WP-0058-T06.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 943911@bnt-lap001
Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0
This commit is contained in:
tegwick 2026-11-06 10:33:30 +01:00
parent c53de12f21
commit b4773235bc
2 changed files with 37 additions and 1 deletions

View file

@ -106,5 +106,29 @@ class Modes(unittest.TestCase):
W.watch(Denied(), FakeKube([]), 'close', NOW)
class Network(unittest.TestCase):
"""2026-11-06: the first watch runs failed with a bare URLError before the pod's NetworkPolicy was live."""
def test_waits_for_both_endpoints_with_the_api_ca(self):
calls = []
W.wait_for_network('http://openbao.openbao.svc:8200', wait=lambda addr, **kw: calls.append((addr, kw)))
self.assertEqual([c[0] for c in calls], ['http://openbao.openbao.svc:8200', W.KUBE_API])
self.assertEqual(calls[1][1]['ca_file'], W.KUBE_CA)
self.assertTrue(all(c[1]['timeout'] == 30 for c in calls))
def test_an_unreachable_endpoint_is_named_in_the_line(self):
import contextlib, io, json as js
original = W.wait_for_network
def unreachable(addr, wait=None):
raise OPS.Unreachable('openbao not reachable')
W.wait_for_network = unreachable
self.addCleanup(setattr, W, 'wait_for_network', original)
with contextlib.redirect_stdout(io.StringIO()) as out:
code = W.main(['--mode', 'flag'])
self.assertEqual(code, 5)
self.assertEqual(js.loads(out.getvalue())['reason'], 'openbao not reachable')
if __name__ == '__main__':
unittest.main()