create_router.py

import ipaddress
import re

# 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 gso import repository
from gso.products.product_blocks import router as router_pb
from gso.products.product_types import router
from gso.products.product_types.router import RouterInactive, RouterProvisioning
from gso.products.product_types.site import Site
from gso.services import ipam, provisioning_proxy
from gso.services.provisioning_proxy import pp_interaction
from gso.workflows.utils import customer_selector


def site_selector() -> Choice:
    site_subscriptions = {}
    for site_id, site_description in repository.all_active_site_subscriptions(
        fields=["subscription_id", "description"]
    ):
        site_subscriptions[str(site_id)] = site_description

    # noinspection PyTypeChecker
    return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items()))  # type: ignore


def initial_input_form_generator(product_name: str) -> FormGenerator:
    class CreateRouterForm(FormPage):
        class Config:
            title = product_name

        customer: customer_selector()  # type: ignore
        router_site: site_selector()  # type: ignore
        hostname: str
        ts_port: int
        router_vendor: router_pb.RouterVendor
        router_role: router_pb.RouterRole
        is_ias_connected: bool

    user_input = yield CreateRouterForm

    return user_input.dict()


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

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


def iso_from_ipv4(ipv4_address: ipaddress.IPv4Address) -> str:
    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"])


@step("Get information from IPAM")
def get_info_from_ipam(subscription: RouterProvisioning, is_ias_connected: bool) -> State:
    lo0_alias = re.sub(".geant.net", "", subscription.router.router_fqdn)
    lo0_name = f"lo0.{lo0_alias}"
    lo0_addr = ipam.allocate_host(hostname=lo0_name, service_type="LO", cname_aliases=[lo0_alias])
    subscription.router.router_lo_ipv4_address = lo0_addr.v4
    subscription.router.router_lo_ipv6_address = lo0_addr.v6
    subscription.router.router_lo_iso_address = iso_from_ipv4(subscription.router.router_lo_ipv4_address)
    subscription.router.router_is_ias_connected = is_ias_connected

    if is_ias_connected is True:
        subscription.router.router_si_ipv4_network = ipam.allocate_ipv4_network(
            service_type="SI", comment=f"SI for {lo0_name}"
        ).v4
        subscription.router.router_ias_lt_ipv4_network = ipam.allocate_ipv4_network(
            service_type="LT_IAS", comment=f"LT for {lo0_name}"
        ).v4
        subscription.router.router_ias_lt_ipv6_network = ipam.allocate_ipv6_network(
            service_type="LT_IAS", comment=f"LT for {lo0_name}"
        ).v6

    return {"subscription": subscription}


@step("Initialize subscription")
def initialize_subscription(
    subscription: router.RouterInactive,
    hostname: str,
    ts_port: int,
    router_vendor: router_pb.RouterVendor,
    router_site: str,
    router_role: router_pb.RouterRole,
) -> State:
    subscription.router.router_ts_port = ts_port
    subscription.router.router_vendor = router_vendor
    subscription.router.router_site = Site.from_subscription(router_site).site
    fqdn = (
        f"{hostname}.{subscription.router.router_site.site_name.lower()}."
        f"{subscription.router.router_site.site_country_code.lower()}"
        ".geant.net"
    )
    subscription.router.router_fqdn = fqdn
    subscription.router.router_role = router_role
    subscription.router.router_access_via_ts = True
    subscription.description = f"Router {fqdn}"

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

    return {"subscription": subscription}


@step("Provision router [DRY RUN]")
def provision_router_dry(subscription: RouterProvisioning, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_router(subscription, process_id)

    return {
        "subscription": subscription,
        "label_text": (
            "Dry run for the deployment of base config on a new router. Deployment is done by the"
            " provisioning proxy, please wait for the results to come back before continuing."
        ),
    }


@step("Provision router [FOR REAL]")
def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_router(subscription, process_id, False)

    return {
        "subscription": subscription,
        "label_text": (
            "Deployment of base config for a new router. Deployment is being taken care of by the"
            " provisioning proxy, please wait for the results to come back before continuing."
        ),
    }


@workflow(
    "Create router",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_router() -> StepList:
    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> get_info_from_ipam
        >> pp_interaction(provision_router_dry, 3)
        >> pp_interaction(provision_router_real, 3)
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )