create_device.py

import ipaddress
import re
from uuid import uuid4

from orchestrator.db.models import ProductTable, SubscriptionTable
# noinspection PyProtectedMember
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 _ipam
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: DeviceProvisioning) -> State:
    lo0_alias = re.sub('.geant.net', '', subscription.device.device_fqdn)
    lo0_name = f'lo0.{lo0_alias}'
    lo0_addr = _ipam.allocate_service_host(hostname=lo0_name,
                                           service_type='LO',
                                           cname_aliases=[lo0_alias])
    subscription.device.device_lo_ipv4_address = lo0_addr.v4
    subscription.device.device_lo_ipv6_address = lo0_addr.v6
    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 = f'{hostname}.{subscription.device.device_site.site_name.lower()}.' \
           f'{subscription.device.device_site.site_country_code.lower()}' \
           f'.geant.net'
    subscription.device.device_fqdn = fqdn
    subscription.device.device_role = device_role
    subscription.description = f'Device {fqdn} ' \
                               f'({subscription.device_type})'

    subscription = device.DeviceProvisioning.from_other_lifecycle(
        subscription, SubscriptionLifecycle.PROVISIONING
    )

    return {'subscription': subscription}


@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,
            'label_text': f'This is a dry run for the deployment of a new '
                          f'{subscription.device_type}. Deployment is being '
                          f'taken care of by the provisioning proxy, please '
                          f'wait for the results to come back before '
                          f'continuing.'
            }


@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,
            'label_text': f'This is a live deployment of a new '
                          f'{subscription.device_type}. Deployment is being '
                          f'taken care of by the provisioning proxy, please '
                          f'wait for the results to come back before '
                          f'continuing.'
            }


@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)
            >> initialize_subscription
            >> get_info_from_ipam
            >> provision_device_dry
            >> await_pp_results
            >> confirm_pp_results
            >> provision_device_real
            >> await_pp_results
            >> confirm_pp_results
            >> set_status(SubscriptionLifecycle.ACTIVE)
            >> resync
            >> done
    )