create_router.py

"""A creation workflow for adding a new router to the network."""

from typing import Any

# noinspection PyProtectedMember
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from pydantic import validator

from gso.products.product_blocks.router import (
    PortNumber,
    RouterRole,
    RouterVendor,
    generate_fqdn,
)
from gso.products.product_types.router import RouterInactive, RouterProvisioning
from gso.products.product_types.site import Site
from gso.services import infoblox, subscriptions
from gso.services.crm import customer_selector
from gso.services.netbox_client import NetboxClient
from gso.services.provisioning_proxy import pp_interaction
from gso.utils.helpers import iso_from_ipv4
from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real


def _site_selector() -> Choice:
    site_subscriptions = {}
    for site in subscriptions.get_active_site_subscriptions(includes=["subscription_id", "description"]):
        site_subscriptions[str(site["subscription_id"])] = site["description"]

    # noinspection PyTypeChecker
    return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True))  # type: ignore[arg-type]


def initial_input_form_generator(product_name: str) -> FormGenerator:
    """Gather information about the new router from the operator."""

    class CreateRouterForm(FormPage):
        class Config:
            title = product_name

        tt_number: str
        customer: customer_selector()  # type: ignore[valid-type]
        vendor: RouterVendor
        router_site: _site_selector()  # type: ignore[valid-type]
        hostname: str
        ts_port: PortNumber
        router_role: RouterRole

        @validator("hostname", allow_reuse=True)
        def hostname_must_be_available(cls, hostname: str, **kwargs: dict[str, Any]) -> str:
            router_site = kwargs["values"].get("router_site")
            if not router_site:
                msg = "Please select a site before setting the hostname."
                raise ValueError(msg)

            selected_site = Site.from_subscription(router_site).site
            input_fqdn = generate_fqdn(hostname, selected_site.site_name, selected_site.site_country_code)
            if not infoblox.hostname_available(f"lo0.{input_fqdn}"):
                msg = f'FQDN "{input_fqdn}" is not available.'
                raise ValueError(msg)

            return hostname

    user_input = yield CreateRouterForm

    return user_input.dict()


@step("Create subscription")
def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
    """Create a new subscription object."""
    subscription = RouterInactive.from_product_id(product, customer)

    return {
        "subscription": subscription,
        "subscription_id": subscription.subscription_id,
    }


@step("Initialize subscription")
def initialize_subscription(
    subscription: RouterInactive,
    hostname: str,
    ts_port: PortNumber,
    router_site: str,
    router_role: RouterRole,
    vendor: RouterVendor,
) -> State:
    """Initialise the subscription object in the service database."""
    subscription.router.router_ts_port = ts_port
    subscription.router.router_site = Site.from_subscription(router_site).site
    fqdn = generate_fqdn(
        hostname,
        subscription.router.router_site.site_name,
        subscription.router.router_site.site_country_code,
    )
    subscription.router.router_fqdn = fqdn
    subscription.router.router_role = router_role
    subscription.router.router_access_via_ts = True
    subscription.router.vendor = vendor
    subscription.description = f"Router {fqdn}"

    subscription = RouterProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)

    return {"subscription": subscription}


@step("Allocate loopback interfaces in IPAM")
def ipam_allocate_loopback(subscription: RouterProvisioning) -> State:
    """Allocate :term:`IPAM` resources for the loopback interface."""
    fqdn = subscription.router.router_fqdn
    loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))

    subscription.router.router_lo_ipv4_address = loopback_v4
    subscription.router.router_lo_ipv6_address = loopback_v6
    subscription.router.router_lo_iso_address = iso_from_ipv4(subscription.router.router_lo_ipv4_address)

    return {"subscription": subscription}


@step("Create NetBox Device")
def create_netbox_device(subscription: RouterProvisioning) -> State:
    """Create a new device in Netbox.

    HACK: use a conditional instead for execution of this step
    """
    if subscription.router.vendor == RouterVendor.NOKIA:
        NetboxClient().create_device(
            subscription.router.router_fqdn,
            str(subscription.router.router_site.site_tier),  # type: ignore[union-attr]
        )
        return {"subscription": subscription}
    return {"subscription": subscription}


@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
    """Validate the :term:`IPAM` resources for the loopback interface."""
    host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
    if not host_record or str(subscription.subscription_id) not in host_record.comment:
        return {"ipam_warning": "Loopback record is incorrectly configured in IPAM, please investigate this manually!"}

    return {"subscription": subscription}


@workflow(
    "Create router",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_router() -> StepList:
    """Create a new router in the service database.

    * Create and initialise the subscription object in the service database
    * Allocate :term:`IPAM` resources for the loopback interface
    * Deploy configuration on the new router, first as a dry run
    * Validate :term:`IPAM` resources
    * Create a new device in Netbox
    """
    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> ipam_allocate_loopback
        >> pp_interaction(deploy_base_config_dry)
        >> pp_interaction(deploy_base_config_real)
        >> verify_ipam_loopback
        >> create_netbox_device
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )