modify_generic.py

import ipaddress

from orchestrator.forms import FormPage, ReadOnlyField
from orchestrator.forms.validators import UniqueConstrainedList
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form

from gso.products.product_blocks import PhyPortCapacity
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.iptrunk import Iptrunk
from gso.services import provisioning_proxy
from gso.services.provisioning_proxy import pp_interaction


def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    subscription = Iptrunk.from_subscription(subscription_id)

    class ModifyIptrunkForm(FormPage):
        geant_s_sid: str = subscription.iptrunk.geant_s_sid
        iptrunk_description: str = subscription.iptrunk.iptrunk_description
        iptrunk_type: IptrunkType = subscription.iptrunk.iptrunk_type
        iptrunk_speed: PhyPortCapacity = subscription.iptrunk.iptrunk_speed  # type: ignore
        iptrunk_minimum_links: int = subscription.iptrunk.iptrunk_minimum_links
        iptrunk_isis_metric: int = ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric)
        iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv4_network)
        iptrunk_ipv6_network: ipaddress.IPv6Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv6_network)

    initial_user_input = yield ModifyIptrunkForm

    class AeMembersListA(UniqueConstrainedList[str]):
        min_items = initial_user_input.iptrunk_minimum_links

    class ModifyIptrunkSideAForm(FormPage):
        class Config:
            title = "Provide subscription details for side A of the trunk."

        iptrunk_sideA_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideA_node.device_fqdn)
        iptrunk_sideA_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideA_ae_iface)
        iptrunk_sideA_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid
        iptrunk_sideA_ae_members: AeMembersListA = subscription.iptrunk.iptrunk_sideA_ae_members  # type: ignore
        iptrunk_sideA_ae_members_descriptions: AeMembersListA = (
            subscription.iptrunk.iptrunk_sideA_ae_members_description  # type: ignore
        )

    user_input_side_a = yield ModifyIptrunkSideAForm

    class AeMembersListB(UniqueConstrainedList[str]):
        min_items = len(user_input_side_a.iptrunk_sideA_ae_members)
        max_items = len(user_input_side_a.iptrunk_sideA_ae_members)

    class ModifyIptrunkSideBForm(FormPage):
        class Config:
            title = "Provide subscription details for side B of the trunk."

        iptrunk_sideB_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideB_node.device_fqdn)
        iptrunk_sideB_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideB_ae_iface)
        iptrunk_sideB_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid
        iptrunk_sideB_ae_members: AeMembersListB = subscription.iptrunk.iptrunk_sideB_ae_members  # type: ignore
        iptrunk_sideB_ae_members_descriptions: AeMembersListB = (
            subscription.iptrunk.iptrunk_sideB_ae_members_description  # type: ignore
        )

    user_input_side_b = yield ModifyIptrunkSideBForm

    return initial_user_input.dict() | user_input_side_a.dict() | user_input_side_b.dict()


@step("Update subscription")
def modify_iptrunk_subscription(
    subscription: Iptrunk,
    geant_s_sid: str,
    iptrunk_type: IptrunkType,
    iptrunk_description: str,
    iptrunk_speed: str,
    iptrunk_minimum_links: int,
    iptrunk_sideA_ae_geant_a_sid: str,
    iptrunk_sideA_ae_members: list[str],
    iptrunk_sideA_ae_members_descriptions: list[str],
    iptrunk_sideB_ae_geant_a_sid: str,
    iptrunk_sideB_ae_members: list[str],
    iptrunk_sideB_ae_members_descriptions: list[str],
) -> State:
    subscription.iptrunk.geant_s_sid = geant_s_sid
    subscription.iptrunk.iptrunk_description = iptrunk_description
    subscription.iptrunk.iptrunk_type = iptrunk_type
    subscription.iptrunk.iptrunk_speed = iptrunk_speed
    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links

    subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid = iptrunk_sideA_ae_geant_a_sid
    subscription.iptrunk.iptrunk_sideA_ae_members = iptrunk_sideA_ae_members
    subscription.iptrunk.iptrunk_sideA_ae_members_description = iptrunk_sideA_ae_members_descriptions

    subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid = iptrunk_sideB_ae_geant_a_sid
    subscription.iptrunk.iptrunk_sideB_ae_members = iptrunk_sideB_ae_members
    subscription.iptrunk.iptrunk_sideB_ae_members_description = iptrunk_sideB_ae_members_descriptions

    subscription.description = f"IP trunk, geant_s_sid:{geant_s_sid}"

    return {"subscription": subscription}


@step("Provision IP trunk interface [DRY RUN]")
def provision_ip_trunk_iface_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_ip_trunk(subscription, process_id, "trunk_interface")

    return {
        "subscription": subscription,
        "label_text": "[DRY RUN] Provisioning trunk interface, please refresh to get the results of the playbook.",
    }


@step("Provision IP trunk interface [FOR REAL]")
def provision_ip_trunk_iface_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_ip_trunk(subscription, process_id, "trunk_interface", False)

    return {
        "subscription": subscription,
        "label_text": "Provisioning trunk interface, please refresh to get the results of the playbook.",
    }


@step("Provision IP trunk LLDP interface [DRY RUN]")
def provision_ip_trunk_lldp_iface_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_ip_trunk(subscription, process_id, "lldp_interface")

    return {
        "subscription": subscription,
        "label_text": "[DRY RUN] Provisioning LLDP interface, please refresh to get the results of the playbook.",
    }


@step("Provision IP trunk LLDP interface [FOR REAL]")
def provision_ip_trunk_lldp_iface_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_ip_trunk(subscription, process_id, "lldp_interface", False)

    return {
        "subscription": subscription,
        "label_text": "Provisioning LLDP interface, please refresh to get the results of the playbook.",
    }


@workflow(
    "Modify IP trunk",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.MODIFY,
)
def modify_generic() -> StepList:
    return (
        init
        >> store_process_subscription(Target.MODIFY)
        >> unsync
        >> modify_iptrunk_subscription
        >> pp_interaction(provision_ip_trunk_iface_dry, 3)
        >> pp_interaction(provision_ip_trunk_iface_real, 3)
        >> pp_interaction(provision_ip_trunk_lldp_iface_dry, 3)
        >> pp_interaction(provision_ip_trunk_lldp_iface_real, 3)
        >> resync
        >> done
    )