computer-playbook/cli/administration/inventory/mesh/model.py

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class MeshSpec:
    """One declared mesh.

    Args:
        name: identifier of the mesh, used as the interface name.
        subnet: the mesh address range, as ``a.b.c.0/24``.
        listen_port: UDP port the hub, or every member of a full mesh,
            listens on.
        routed_range: the range a spoke routes through the hub.
    """

    name: str
    subnet: str
    listen_port: int
    routed_range: str = ""


@dataclass(frozen=True)
class Topology:
    """Who an inventory puts into each mesh.

    Args:
        hub: the host that routes for every mesh, or None for full meshes.
        spokes: mesh name to the hosts that joined it, never the hub.
    """

    hub: str | None
    spokes: dict[str, tuple[str, ...]] = field(default_factory=dict)


@dataclass(frozen=True)
class MeshMember:
    """One host's place in a resolved mesh.

    Args:
        host: inventory hostname.
        address: the member's address inside the mesh subnet.
        endpoint: the underlay address its peers dial.
        private_key: the member's own secret half, or None when the member
            already holds one.
        public_key: the half its peers receive.
        is_hub: whether this member routes for the others.
    """

    host: str
    address: str
    private_key: str | None
    public_key: str
    is_hub: bool
    endpoint: str = ""


@dataclass(frozen=True)
class Mesh:
    """A spec resolved against an inventory.

    Args:
        spec: the declaration this mesh was resolved from.
        members: every host in the mesh, hub first.
    """

    spec: MeshSpec
    members: tuple[MeshMember, ...] = field(default_factory=tuple)

    @property
    def hub(self) -> MeshMember:
        for member in self.members:
            if member.is_hub:
                return member
        raise ValueError(f"mesh '{self.spec.name}' resolved without a hub")

    @property
    def spokes(self) -> tuple[MeshMember, ...]:
        return tuple(m for m in self.members if not m.is_hub)

    @property
    def is_full(self) -> bool:
        return not any(member.is_hub for member in self.members)

    def peers_of(self, host: str) -> tuple[MeshMember, ...]:
        member = self.member(host)
        if member is None:
            return ()
        if member.is_hub:
            return self.spokes
        if self.is_full:
            return tuple(m for m in self.members if m.host != host)
        return (self.hub,)

    def member(self, host: str) -> MeshMember | None:
        for candidate in self.members:
            if candidate.host == host:
                return candidate
        return None