terminate_iptrunk.py

# noinspection PyProtectedMember
import ipaddress

from orchestrator.forms import FormPage
from orchestrator.forms.validators import Label
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, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form

from gso.products.product_types.iptrunk import Iptrunk
from gso.services import ipam, provisioning_proxy
from gso.services.ipam import V4ServiceNetwork, V6ServiceNetwork
from gso.services.provisioning_proxy import pp_interaction


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

    class TerminateForm(FormPage):
        are_you_sure: Label = f"Are you sure you want to remove {subscription.description}?"  # type: ignore

    return TerminateForm  # type: ignore


@step("Set iptrunk ISIS metric to 9000")
def modify_iptrunk_subscription(subscription: Iptrunk) -> State:
    subscription.iptrunk.iptrunk_isis_metric = 9000

    return {"subscription": subscription}


@step("Drain traffic from trunk")
def drain_traffic_from_ip_trunk(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.provision_ip_trunk(subscription, process_id, "isis_interface", False)
    return {
        "subscription": subscription,
        "label_text": "This is setting the ISIS metric of the trunk to 9000"
        "trunk. "
        "Press refresh to get the results\n"
        "When traffic is drained, confirm to continue",
    }


@step("Deprovision IP trunk [DRY RUN]")
def deprovision_ip_trunk_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.deprovision_ip_trunk(subscription, process_id, True)

    return {
        "subscription": subscription,
        "label_text": "This is a dry run for the termination of an IP " "trunk. " "Press refresh to get the results",
    }


@step("Deprovision IP trunk [FOR REAL]")
def deprovision_ip_trunk_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
    provisioning_proxy.deprovision_ip_trunk(subscription, process_id, False)

    return {
        "subscription": subscription,
        "label_text": "This is a termination of an IP trunk. " "Press refresh to get the results",
    }


@step("Deprovision IPv4 networks")
def deprovision_ip_trunk_ipv4(subscription: Iptrunk) -> dict[str, V4ServiceNetwork | V6ServiceNetwork]:
    service_network = ipam.delete_service_network(
        network=ipaddress.ip_network(subscription.iptrunk.iptrunk_ipv4_network),
        service_type="TRUNK",
    )
    return {"service_network": service_network}


@step("Deprovision IPv6 networks")
def deprovision_ip_trunk_ipv6(subscription: Iptrunk) -> dict[str, V4ServiceNetwork | V6ServiceNetwork]:
    service_network = ipam.delete_service_network(
        network=ipaddress.ip_network(subscription.iptrunk.iptrunk_ipv6_network),
        service_type="TRUNK",
    )
    return {"service_network": service_network}


@workflow(
    "Terminate IPtrunk",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.TERMINATE,
)
def terminate_iptrunk() -> StepList:
    return (
        init
        >> store_process_subscription(Target.TERMINATE)
        >> unsync
        >> modify_iptrunk_subscription
        >> pp_interaction(drain_traffic_from_ip_trunk, 3)
        >> pp_interaction(deprovision_ip_trunk_dry, 3)
        >> deprovision_ip_trunk_ipv4
        >> deprovision_ip_trunk_ipv6
        >> set_status(SubscriptionLifecycle.TERMINATED)
        >> resync
        >> done
    )