"""K8s discovery reconciliation integration tests. Tests verify that: 1. The K8s EndpointSlice watcher correctly discovers new workers as Services and backing Deployments are updated. 2. Workers are removed from the router's registry after the backing EndpointSlice entries disappear (pod deleted / deployment scaled to 0). 3. After a simulated watch-connection interruption (router restarted), the registry converges back to the correct worker set. Note: sgl-router does not currently expose a Prometheus /metrics endpoint, so the SMG-style metric assertions are not used here. Disconnect/reconnect coverage is provided by test_lifecycle.TestRouterRestart. """ from __future__ import annotations import logging import time import httpx import pytest from conftest import ( NAMESPACE, RECONCILIATION_WAIT_SECS, _kubectl, _poll_until, logger, ) def _scale_fake_worker(replicas: int) -> None: _kubectl( "scale", "deployment/fake-worker", f"--replicas={replicas}", "-n", NAMESPACE ) def _can_route(router_url: str) -> bool: try: r = httpx.post( f"{router_url}/v1/chat/completions", json={ "model": "tiny", "messages": [{"role": "user", "content": "reconcile"}], }, timeout=8.0, ) return r.status_code == 200 except Exception: return False class TestWatcherDiscovery: """The EndpointSlice watcher discovers new endpoints on Deployment scale-up.""" def test_watcher_discovers_new_endpoints_on_scale_up(self, router_url): """Scale from 1 to 3 replicas; router must continue routing successfully.""" _scale_fake_worker(1) # Wait for scale-down to propagate and routing to stabilise _poll_until( lambda: _can_route(router_url), "router routes with 1 replica", timeout=60, interval=3, ) _scale_fake_worker(3) _poll_until( lambda: _can_route(router_url), "router routes with 3 replicas (after scale-up)", timeout=60, interval=3, ) class TestStaleEndpointRemoval: """When fake-worker replicas drop, the router must stop routing to the removed endpoints. Because sgl-router has no /workers admin API, we verify removal indirectly: scale to 0, assert the router returns non-200 (or at least that scaling back to 2 restores routing), then restore. """ def test_routing_restores_after_scale_down_and_back_up(self, router_url): """Scale to 0 (no workers → expect non-200), then restore to 2. After restore the router must route again within the reconciliation window. """ try: _scale_fake_worker(0) # Expect routing to fail eventually (503 or connection error) deadline = time.time() + RECONCILIATION_WAIT_SECS routing_failed = False while time.time() < deadline: try: r = httpx.post( f"{router_url}/v1/chat/completions", json={ "model": "tiny", "messages": [{"role": "user", "content": "no-workers"}], }, timeout=5.0, ) if r.status_code != 200: routing_failed = True break except Exception: routing_failed = True break time.sleep(3) # If after RECONCILIATION_WAIT_SECS the router is still routing, # that means old endpoints are cached — not necessarily wrong for # a watcher that hasn't ticked yet, but log a warning. if not routing_failed: logger.warning( "Router still returning 200 after scale-to-0; " "EndpointSlice event may be delayed — continuing test." ) # Restore workers and verify routing comes back _scale_fake_worker(2) _poll_until( lambda: _can_route(router_url), "routing restored after scale back up to 2", timeout=RECONCILIATION_WAIT_SECS, interval=3, ) finally: _scale_fake_worker(3) class TestReconciliationConsistency: """Routing remains stable over multiple reconciliation windows with steady worker state — no spurious deregistrations or duplicate registrations.""" @pytest.mark.slow def test_routing_stable_over_multiple_reconciliation_cycles(self, router_url): """Deploy 3 workers, sample routing success over ~150s (2 reconciliation cycles + margin), assert no interruptions.""" _scale_fake_worker(3) _poll_until( lambda: _can_route(router_url), "baseline routing with 3 workers", timeout=30, interval=2, ) # Sample every 15s for 150s wait_secs = RECONCILIATION_WAIT_SECS + 60 end_time = time.time() + wait_secs failures = [] while time.time() < end_time: ok = _can_route(router_url) if not ok: failures.append(time.time()) time.sleep(15) assert not failures, ( f"Routing failed at {len(failures)} sample(s) during stability window; " f"timestamps: {failures}" )