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

"""Kubernetes adapters of the pilot operations against a fake API server (NK-WP-0056-T12). No cluster."""

from __future__ import annotations

import datetime as dt
import importlib.util
import json
import pathlib
import sys
import tempfile
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HERE = pathlib.Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("operations", HERE.parent / "operations.py")
OPS = importlib.util.module_from_spec(spec)
sys.modules["operations"] = OPS
spec.loader.exec_module(OPS)

ES = "/apis/external-secrets.io/v1/namespaces/sso/externalsecrets/authelia-secrets"
DEP = "/apis/apps/v1/namespaces/sso/deployments/authelia"
T0 = dt.datetime(2026, 11, 3, 9, 30, tzinfo=dt.UTC)


class Api:
    def __init__(self):
        self.patches = []
        self.auth = []
        self.es_status = {"conditions": [{"type": "Ready", "status": "True"}], "refreshTime": "2026-11-03T09:29:00Z"}
        self.dep = {"metadata": {"generation": 5}, "spec": {"replicas": 1},
                    "status": {"observedGeneration": 5, "replicas": 1, "updatedReplicas": 1, "availableReplicas": 1}}
        self.gets_until_ready = 0
        self.forbidden = False


def serve(api: Api):
    class Handler(BaseHTTPRequestHandler):
        def _send(self, code, body):
            raw = json.dumps(body).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            self.end_headers()
            self.wfile.write(raw)

        def do_GET(self):
            api.auth.append(self.headers.get("Authorization"))
            if api.forbidden:
                return self._send(403, {"kind": "Status"})
            if self.path == ES:
                return self._send(200, {"status": api.es_status})
            if self.path == DEP:
                if api.gets_until_ready > 0:
                    api.gets_until_ready -= 1
                    return self._send(200, {**api.dep, "status": {**api.dep["status"], "updatedReplicas": 0,
                                                                  "unavailableReplicas": 1}})
                return self._send(200, api.dep)
            self._send(404, {})

        def do_PATCH(self):
            api.patches.append((self.path, self.headers.get("Content-Type"),
                                json.loads(self.rfile.read(int(self.headers["Content-Length"])))))
            if api.forbidden:
                return self._send(403, {"kind": "Status"})
            if self.path == DEP:
                api.dep["metadata"]["generation"] += 1
                api.dep["status"]["observedGeneration"] = api.dep["metadata"]["generation"]
            self._send(200, {})

        def log_message(self, *a):
            return

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()
    return server


class KubeTests(unittest.TestCase):
    def setUp(self):
        self.api = Api()
        self.server = serve(self.api)
        self.addCleanup(self.server.shutdown)
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        token = pathlib.Path(tmp.name) / "token"
        token.write_text("synthetic-sa-token\n")
        self.kube = OPS.Kube(api=f"http://127.0.0.1:{self.server.server_port}", token_file=str(token), ca_file=None)

    def test_force_sync_patches_only_the_annotation(self):
        delivery = OPS.KubeDelivery(self.kube, "sso", "authelia-secrets", clock=lambda: T0)
        self.assertEqual(delivery.force_sync(), T0)
        ((path, ctype, body),) = self.api.patches
        self.assertEqual((path, ctype), (ES, "application/merge-patch+json"))
        self.assertEqual(body, {"metadata": {"annotations": {"force-sync": str(int(T0.timestamp()))}}})
        self.assertEqual(self.api.auth, [])  # no read before the patch

    def test_synced_needs_ready_and_a_refresh_after_the_sync(self):
        delivery = OPS.KubeDelivery(self.kube, "sso", "authelia-secrets")
        self.assertFalse(delivery.synced(T0))  # refreshed before T0
        self.api.es_status["refreshTime"] = "2026-11-03T09:30:05Z"
        self.assertTrue(delivery.synced(T0))
        self.api.es_status["conditions"][0]["status"] = "False"
        self.assertFalse(delivery.synced(T0))

    def test_restart_patches_restarted_at_and_waits_for_the_rollout(self):
        self.api.gets_until_ready = 2
        restart = OPS.KubeRestart(self.kube, "sso", "authelia", timeout=10, poll=0.01, clock=lambda: T0)
        restart.restart()
        ((path, _, body),) = self.api.patches
        self.assertEqual(path, DEP)
        self.assertEqual(body["spec"]["template"]["metadata"]["annotations"]["kubectl.kubernetes.io/restartedAt"],
                         "2026-11-03T09:30:00Z")
        self.assertEqual(self.api.gets_until_ready, 0)

    def test_rollout_that_never_becomes_ready_is_a_definite_failure(self):
        self.api.gets_until_ready = 10 ** 6
        ticks = iter(range(10 ** 6))
        restart = OPS.KubeRestart(self.kube, "sso", "authelia", timeout=5, poll=0, monotonic=lambda: next(ticks),
                                  sleep=lambda s: None)
        with self.assertRaises(OPS.RolloutFailed) as caught:
            restart.restart()
        self.assertTrue(caught.exception.definite_failure)

    def test_forbidden_is_an_operation_error_and_sends_the_bearer(self):
        self.api.forbidden = True
        with self.assertRaises(OPS.OperationError):
            OPS.KubeDelivery(self.kube, "sso", "authelia-secrets").synced(T0)
        self.assertEqual(self.api.auth, ["Bearer synthetic-sa-token"])


if __name__ == "__main__":
    unittest.main()