tegwick
c63aa398d6
T08: .claude/rules/worker-protocol.md (imported by CLAUDE.md, pointed to from AGENTS.md) and tools/contract-tests/excerpt.py: name the rows, hand every worker one excerpt, one owner per step, check with the excerpt's tests. A test requires every table row to have details and a contract test (found O4). T06: bounded-executor/window_watch.py flags or closes the write role when no run Job is active or recent; stub tests; in the image. Built from the excerpt for O5, E1, C1, E8, E9, R2. 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
110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
"""window_watch.py on stubs (NK-WP-0058-T06, contract row O5): rule, both modes, and errors."""
|
|
import datetime as dt
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
import unittest
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE.parent))
|
|
spec = importlib.util.spec_from_file_location('window_watch', HERE.parent / 'window_watch.py')
|
|
W = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(W)
|
|
OPS = W.OPS
|
|
NOW = dt.datetime(2026, 11, 6, 3, 0, tzinfo=dt.timezone.utc)
|
|
|
|
|
|
def job(minutes_ago, active=0):
|
|
created = (NOW - dt.timedelta(minutes=minutes_ago)).isoformat().replace('+00:00', 'Z')
|
|
return {'metadata': {'creationTimestamp': created}, 'status': {'active': active} if active else {}}
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self, present=True, delete_works=True):
|
|
self.present, self.delete_works, self.calls = present, delete_works, []
|
|
|
|
def _call(self, method, path, body=None, auth=True):
|
|
self.calls.append((method, path))
|
|
assert path == W.ROLE_PATH, path # nothing but the write role is ever touched
|
|
if method == 'GET':
|
|
if not self.present:
|
|
raise OPS.OperationError('openbao GET 404')
|
|
return {'data': {}}
|
|
if method == 'DELETE':
|
|
self.present = not self.delete_works
|
|
return {}
|
|
raise AssertionError(method)
|
|
|
|
|
|
class FakeKube:
|
|
def __init__(self, jobs):
|
|
self.jobs, self.paths = jobs, []
|
|
|
|
def get(self, path):
|
|
self.paths.append(path)
|
|
return {'items': self.jobs}
|
|
|
|
|
|
class Rule(unittest.TestCase):
|
|
def test_decide(self):
|
|
cases = [
|
|
(False, [], 'ok'), # no role: nothing open
|
|
(True, [], 'stale'), # role, never a run: left open
|
|
(True, [job(3)], 'ok'), # role, run just created: a window in use
|
|
(True, [job(30, active=1)], 'ok'), # role, run still active
|
|
(True, [job(30)], 'stale'), # role, last run finished 30 min ago: left open
|
|
(True, [job(30), job(2)], 'ok'), # the newest run counts
|
|
]
|
|
for present, jobs, verdict in cases:
|
|
with self.subTest(present=present, jobs=len(jobs)):
|
|
self.assertEqual(W.decide(present, jobs, NOW)[0], verdict)
|
|
|
|
def test_run_jobs_selector_excludes_other_components(self):
|
|
kube = FakeKube([])
|
|
W.run_jobs(kube)
|
|
self.assertIn('labelSelector=app.kubernetes.io%2Fname%3Dbounded-executor%2C%21app.kubernetes.io%2Fcomponent',
|
|
kube.paths[0])
|
|
self.assertTrue(kube.paths[0].startswith('/apis/batch/v1/namespaces/sso/jobs?'))
|
|
|
|
|
|
class Modes(unittest.TestCase):
|
|
def test_ok_touches_nothing(self):
|
|
bao = FakeBao(present=False)
|
|
result, code = W.watch(bao, FakeKube([]), 'close', NOW)
|
|
self.assertEqual((result['status'], code), ('ok', 0))
|
|
self.assertEqual([m for m, _ in bao.calls], ['GET'])
|
|
|
|
def test_flag_reads_only_and_fails_the_job(self):
|
|
bao = FakeBao()
|
|
result, code = W.watch(bao, FakeKube([job(30)]), 'flag', NOW)
|
|
self.assertEqual((result['status'], code), ('flagged_stale', 3))
|
|
self.assertEqual([m for m, _ in bao.calls], ['GET'])
|
|
self.assertEqual(result['newest_age_s'], 1800)
|
|
|
|
def test_close_deletes_and_proves_absent(self):
|
|
bao = FakeBao()
|
|
result, code = W.watch(bao, FakeKube([]), 'close', NOW)
|
|
self.assertEqual((result['status'], code), ('closed_stale', 0))
|
|
self.assertEqual([m for m, _ in bao.calls], ['GET', 'DELETE', 'GET'])
|
|
self.assertFalse(bao.present)
|
|
|
|
def test_close_that_does_not_take_is_an_error(self):
|
|
result, code = W.watch(FakeBao(delete_works=False), FakeKube([]), 'close', NOW)
|
|
self.assertEqual((result['status'], result['reason'], code), ('error', 'write_role_still_present', 5))
|
|
|
|
def test_a_window_in_use_is_never_closed(self):
|
|
bao = FakeBao()
|
|
result, code = W.watch(bao, FakeKube([job(1, active=1)]), 'close', NOW)
|
|
self.assertEqual((result['status'], code), ('ok', 0))
|
|
self.assertTrue(bao.present)
|
|
|
|
def test_other_openbao_errors_are_not_read_as_absent(self):
|
|
class Denied(FakeBao):
|
|
def _call(self, method, path, body=None, auth=True):
|
|
raise OPS.OperationError('openbao GET 403')
|
|
with self.assertRaises(OPS.OperationError):
|
|
W.watch(Denied(), FakeKube([]), 'close', NOW)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|