create_device.py

import ipaddress
import re
from uuid import uuid4

from orchestrator.db.models import ProductTable, SubscriptionTable
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, choice_list
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status
from orchestrator.workflows.steps import store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form

from gso.products.product_blocks import device as device_pb
from gso.products.product_types import device
from gso.products.product_types.device import DeviceInactive, \
    DeviceProvisioning
from gso.products.product_types.site import Site
from gso.services import provisioning_proxy
from gso.services.provisioning_proxy import await_pp_results, \
    confirm_pp_results


def site_selector() -> list:
    site_subscriptions = {}
    for site_id, site_description in (
            SubscriptionTable.query
                             .join(ProductTable)
                             .filter(
                                ProductTable.product_type == 'Site',
                                SubscriptionTable.status == 'active')
                             .with_entities(SubscriptionTable.subscription_id,
                                            SubscriptionTable.description)
                             .all()
    ):
        site_subscriptions[str(site_id)] = site_description

    # noinspection PyTypeChecker
    return choice_list(
        Choice('site_selection',
               zip(site_subscriptions.keys(),
                   site_subscriptions.items())),  # type:ignore
        min_items=1,
        max_items=1,
    )


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

        device_site: site_selector()
        hostname: str
        ts_address: ipaddress.IPv4Address
        ts_port: int
        device_vendor: device_pb.DeviceVendor
        device_role: device_pb.DeviceRole

    user_input = yield CreateDeviceForm

    return user_input.dict()


@step('Create subscription')
def create_subscription(product: UUIDstr) -> State:
    subscription = DeviceInactive.from_product_id(product, uuid4())

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

def iso_from_ipv4(ipv4_address):
    padded_octets = [f'{x:>03}' for x in ipv4_address.split('.')]
    joined_octets = ''.join(padded_octets)
    re_split = '.'.join(re.findall('....', joined_octets))
    result = '.'.join(['49.51e5.0001', re_split, '00'])
    return (result)

@step('Get information from IPAM')
def get_info_from_ipam(subscription: DeviceInactive) -> State:
    # lo = ipam.new_device_lo_address()
    # subscription.device.lo_ipv4_address = lo.v4
    # subscription.device.lo_ipv6_address = lo.v6
    # TODO: get info about how these should be generated
    subscription.device.device_lo_ipv4_address = \
        ipaddress.ip_address('10.10.10.10')
    subscription.device.device_lo_ipv6_address = \
        ipaddress.ip_address('fc00:798:10::10')
    #subscription.device.device_lo_iso_address \
    #    = "49.51e5.0001.0620.4009.6047.00"
    subscription.device.device_lo_iso_address \
       = iso_from_ipv4(str(subscription.device.device_lo_ipv4_address)) 
    subscription.device.device_si_ipv4_network = "192.168.0.0/31"
    subscription.device.device_ias_lt_ipv4_network = "192.168.1.0/31"
    subscription.device.device_ias_lt_ipv6_network = "fc00:798:1::150/126"
    return {"subscription": subscription}


@step('Initialize subscription')
def initialize_subscription(
        subscription: device.DeviceInactive,
        hostname: str,
        ts_address: ipaddress.IPv4Address,
        ts_port: str,
        device_vendor: device_pb.DeviceVendor,
        device_site: str,
        device_role: device_pb.DeviceRole
) -> State:
    subscription.device.device_ts_address = str(ts_address)
    subscription.device.device_ts_port = str(ts_port)
    subscription.device.device_vendor = device_vendor
    subscription.device.device_site \
        = Site.from_subscription(device_site[0]).site
    fqdn = str(hostname + '.' +
               subscription.device.device_site.site_name.lower() + '.' +
               subscription.device.device_site.site_country_code.lower() +
               '.geant.net')
    subscription.device.device_fqdn = fqdn
    subscription.device.device_role = device_role
    subscription.description = f'Device {fqdn} type \
                                ({subscription.device_type})'
    subscription = device.DeviceProvisioning.from_other_lifecycle(
        subscription, SubscriptionLifecycle.PROVISIONING
    )

    return {'subscription': subscription, 'fqdn': fqdn}


@step('Provision device [DRY RUN]')
def provision_device_dry(subscription: DeviceProvisioning,
                         process_id: UUIDstr) -> State:
    provisioning_proxy.provision_device(subscription, process_id)

    return {'subscription': subscription}


@step('Provision device [FOR REAL]')
def provision_device_real(subscription: DeviceProvisioning,
                          process_id: UUIDstr) -> State:
    provisioning_proxy.provision_device(subscription, process_id, False)

    return {'subscription': subscription}


@workflow(
    'Create device',
    initial_input_form=wrap_create_initial_input_form(
        initial_input_form_generator),
    target=Target.CREATE,
)
def create_device():
    return (
            init
            >> create_subscription
            >> store_process_subscription(Target.CREATE)
            >> get_info_from_ipam
            >> initialize_subscription
            >> provision_device_dry
            >> await_pp_results
            >> confirm_pp_results
            >> provision_device_real
            >> await_pp_results
            >> confirm_pp_results
            >> set_status(SubscriptionLifecycle.ACTIVE)
            >> resync
            >> done
    )