How to Mock Kubernetes Services with a ClusterIP Registry in Python

import json
from dataclasses import dataclass, asdict
from typing import Dict, Optional


@dataclass
class Service:
    name: str
    namespace: str
    cluster_ip: str
    selector: Dict[str, str]
    port: int
    target_port: Optional[int] = None


class ClusterIPServiceRegistry:
    _ip_counter = 0

    def __init__(self) -> None:
        self._services: Dict[str, Service] = {}

    def _next_cluster_ip(self) -> str:
        self._ip_counter += 1
        return f"10.96.{self._ip_counter // 256}.{self._ip_counter % 256}"

    def create_service(self, name: str, namespace: str, selector: Dict[str, str], port: int, target_port: Optional[int] = None) -> Service:
        key = f"{namespace}/{name}"
        if key in self._services:
            raise ValueError(f"Service {key} already exists")

        service = Service(
            name=name,
            namespace=namespace,
            cluster_ip=self._next_cluster_ip(),
            selector=selector,
            port=port,
            target_port=target_port or port,
        )
        self._services[key] = service
        return service

    def get_service(self, name: str, namespace: str = "default") -> Optional[Service]:
        return self._services.get(f"{namespace}/{name}")

    def list_services(self) -> list[Service]:
        return list(self._services.values())

    def to_json(self) -> str:
        return json.dumps([asdict(s) for s in self.list_services()], indent=2)


if __name__ == "__main__":
    registry = ClusterIPServiceRegistry()

    web_service = registry.create_service(
        name="web",
        namespace="default",
        selector={"app": "web"},
        port=80,
        target_port=8080,
    )
    api_service = registry.create_service(
        name="api",
        namespace="prod",
        selector={"app": "api"},
        port=443,
    )

    print(registry.to_json())

    found = registry.get_service("web")
    print(f"\nFound: {found.cluster_ip} -> {found.target_port}")