net-kingdom/bounded-executor/tests/test_window_watch.py

"""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)


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()