helpers.py

import re
from ipaddress import IPv4Address
from uuid import UUID

from orchestrator.forms.validators import Choice
from orchestrator.types import UUIDstr

from gso.products.product_blocks.router import RouterVendor
from gso.products.product_types.router import Router
from gso.services.crm import all_customers
from gso.services.netbox_client import NetboxClient


def customer_selector() -> Choice:
    customers = {}
    for customer in all_customers():
        customers[customer["id"]] = customer["name"]

    return Choice("Select a customer", zip(customers.keys(), customers.items()))  # type: ignore[arg-type]


def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
    """Return a list of available interfaces for a given router and speed.

    For Nokia routers, return a list of available interfaces.
    For Juniper routers, return a string.
    """
    if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
        return None
    interfaces = {
        interface["name"]: f"{interface['name']} - {interface['module']['display']} - {interface['description']}"
        for interface in NetboxClient().get_available_interfaces(router_id, speed)
    }
    return Choice("ae member", zip(interfaces.keys(), interfaces.items()))  # type: ignore[arg-type]


def available_lags_choices(router_id: UUID) -> Choice | None:
    """Return a list of available lags for a given router.

    For Nokia routers, return a list of available lags.
    For Juniper routers, return a string.
    """

    if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
        return None
    side_a_ae_iface_list = NetboxClient().get_available_lags(router_id)
    return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list))  # type: ignore[arg-type]


def get_router_vendor(router_id: UUID) -> str:
    """Retrieve the vendor of a router.

    Args:
    ----
    router_id (UUID): The {term}`UUID` of the router.

    Returns:
    -------
    str: The vendor of the router.
    """
    return Router.from_subscription(router_id).router.router_vendor


def iso_from_ipv4(ipv4_address: IPv4Address) -> str:
    """Calculate an :term:`ISO` address, based on an IPv4 address.

    :param IPv4Address ipv4_address: The address that's to be converted
    :returns: An :term:`ISO`-formatted address.
    """
    padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")]
    joined_octets = "".join(padded_octets)
    re_split = ".".join(re.findall("....", joined_octets))
    return ".".join(["49.51e5.0001", re_split, "00"])


def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr | None:
    """Verify if a device exists in Netbox.

    Args:
    ----
    subscription_id (UUID): The {term}`UUID` of the router subscription.

    Returns:
    -------
    UUID: The {term}`UUID` of the router subscription or raises an error.
    """
    router = Router.from_subscription(subscription_id).router
    if router.router_vendor == RouterVendor.NOKIA:
        device = NetboxClient().get_device_by_name(router.router_fqdn)
        if not device:
            raise ValueError("The selected router does not exist in Netbox.")
    return subscription_id