create_router.py

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

from typing import Any

from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import StepList, conditional, done, init, inputstep, 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 pydantic_forms.core import ReadOnlyField

from gso.products.product_blocks.router import RouterRole
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.lso_client import lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.services.partners import get_partner_by_name
from gso.settings import load_oss_params
from gso.utils.helpers import generate_fqdn, iso_from_ipv4
from gso.utils.shared_enums import PortNumber, Vendor
from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real, run_checks_after_base_config


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
        partner: str = ReadOnlyField("GEANT")
        vendor: Vendor
        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, partner: str) -> State:
    """Create a new subscription object."""
    subscription = RouterInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])

    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: Vendor,
) -> 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}"

    return {"subscription": subscription}


@step("Allocate loopback interfaces in IPAM")
def ipam_allocate_loopback(subscription: RouterInactive) -> State:
    """Allocate :term:`IPAM` resources for the loopback interface."""
    fqdn = subscription.router.router_fqdn
    if not fqdn:
        msg = f"Router fqdn for subscription id {subscription.subscription_id} is missing!"
        raise ValueError(msg)
    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: RouterInactive) -> State:
    """Create a new NOKIA device in Netbox."""
    fqdn = subscription.router.router_fqdn
    site_tier = subscription.router.router_site.site_tier if subscription.router.router_site else None
    if not fqdn or not site_tier:
        msg = f"FQDN and/or Site tier missing in router subscription {subscription.subscription_id}!"
        raise ValueError(msg)

    NetboxClient().create_device(fqdn, site_tier)

    return {"subscription": subscription}


@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: RouterInactive) -> 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}


@inputstep("Prompt to reboot", assignee=Assignee.SYSTEM)
def prompt_reboot_router(subscription: RouterInactive) -> FormGenerator:
    """Wait for confirmation from an operator that the router has been rebooted."""

    class RebootPrompt(FormPage):
        class Config:
            title = "Please reboot before continuing"

        if subscription.router.router_site and subscription.router.router_site.site_ts_address:
            info_label_1: Label = (
                f"Base config has been deployed. Please log in via the console using https://"  # type: ignore[assignment]
                f"{subscription.router.router_site.site_ts_address}."
            )
        else:
            info_label_1 = "Base config has been deployed. Please log in via the console."  # type: ignore[assignment]

        info_label_2: Label = "Reboot the router, and once it is up again, press submit to continue the workflow."  # type: ignore[assignment]

    yield RebootPrompt

    return {}


@inputstep("Prompt to test the console", assignee=Assignee.SYSTEM)
def prompt_console_login() -> FormGenerator:
    """Wait for confirmation from an operator that the router can be logged into."""

    class ConsolePrompt(FormPage):
        class Config:
            title = "Verify local authentication"

        info_label_1: Label = (
            "Verify that you are able to log in to the router via the console using the admin account."  # type: ignore[assignment]
        )
        info_label_2: Label = "Once this is done, press submit to continue the workflow."  # type: ignore[assignment]

    yield ConsolePrompt

    return {}


@inputstep("Prompt IMS insertion", assignee=Assignee.SYSTEM)
def prompt_insert_in_ims() -> FormGenerator:
    """Wait for confirmation from an operator that the router has been inserted in IMS."""

    class IMSPrompt(FormPage):
        class Config:
            title = "Update IMS mediation server"

        info_label_1: Label = "Insert the router into IMS."  # type: ignore[assignment]
        info_label_2: Label = "Once this is done, press submit to continue the workflow."  # type: ignore[assignment]

    yield IMSPrompt

    return {}


@inputstep("Prompt RADIUS insertion", assignee=Assignee.SYSTEM)
def prompt_insert_in_radius(subscription: RouterInactive) -> FormGenerator:
    """Wait for confirmation from an operator that the router has been inserted in RADIUS."""

    class RadiusPrompt(FormPage):
        class Config:
            title = "Update RADIUS clients"

        info_label_1: Label = (
            f"Please go to https://kratos.geant.org/add_radius_client and add the {subscription.router.router_fqdn}"  # type: ignore[assignment]
            f" - {subscription.router.router_lo_ipv4_address} to radius authentication"
        )
        info_label_2: Label = "This will be functionally checked later during verification work."  # type: ignore[assignment]

    yield RadiusPrompt

    return {}


@inputstep("Prompt for new Sharepoint checklist", assignee=Assignee.SYSTEM)
def prompt_start_new_checklist(subscription: RouterProvisioning) -> FormGenerator:
    """Prompt the operator to start a new checklist in Sharepoint for approving this new router."""
    oss_params = load_oss_params()

    class SharepointPrompt(FormPage):
        class Config:
            title = "Start new checklist"

        info_label_1: Label = (
            f"Visit {oss_params.SHAREPOINT.checklist_site_url} and start a new Sharepoint checklist for "
            f"{subscription.router.router_fqdn}."  # type: ignore[assignment]
        )
        info_label_2: Label = "Once this is done, click proceed to finish the workflow."  # type: ignore[assignment]

    yield SharepointPrompt

    return {}


@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
    """
    router_is_nokia = conditional(lambda state: state["vendor"] == Vendor.NOKIA)

    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> ipam_allocate_loopback
        >> lso_interaction(deploy_base_config_dry)
        >> lso_interaction(deploy_base_config_real)
        >> verify_ipam_loopback
        >> prompt_reboot_router
        >> prompt_console_login
        >> prompt_insert_in_ims
        >> prompt_insert_in_radius
        >> router_is_nokia(create_netbox_device)
        >> lso_interaction(run_checks_after_base_config)
        >> set_status(SubscriptionLifecycle.PROVISIONING)
        >> prompt_start_new_checklist
        >> resync
        >> done
    )