helpers.py

import re
from ipaddress import IPv4Address
from uuid import UUID

from orchestrator import step
from orchestrator.types import State, UUIDstr
from pydantic import BaseModel
from pydantic_forms.validators import Choice

from gso.products.product_blocks.router import RouterVendor
from gso.products.product_types.iptrunk import Iptrunk
from gso.products.product_types.router import Router
from gso.services import provisioning_proxy
from gso.services.netbox_client import NetboxClient


class LAGMember(BaseModel):
    #  TODO: validate interface name
    interface_name: str
    interface_description: str

    def __hash__(self) -> int:
        return hash((self.interface_name, self.interface_description))


@step("[COMMIT] Set ISIS metric to 90000")
def set_isis_to_90000(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State:
    old_isis_metric = subscription.iptrunk.iptrunk_isis_metric
    subscription.iptrunk.iptrunk_isis_metric = 90000
    provisioning_proxy.provision_ip_trunk(subscription, process_id, tt_number, "isis_interface", False)

    return {
        "subscription": subscription,
        "old_isis_metric": old_isis_metric,
        "label_text": "ISIS is being set to 90K by the provisioning proxy, please wait for the results",
    }


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


def validate_iptrunk_unique_interface(interfaces: list[LAGMember]) -> list[LAGMember]:
    """Verify if the interfaces are unique.

    Args:
    ----
    interfaces (list[LAGMember]): The list of interfaces.

    Returns:
    -------
    list[LAGMember]: The list of interfaces or raises an error.
    """
    interface_names = [member.interface_name for member in interfaces]
    if len(interface_names) != len(set(interface_names)):
        raise ValueError("Interfaces must be unique.")
    return interfaces