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 provisioning_proxy
from gso.services.provisioning_proxy import await_pp_results, \
    confirm_pp_results
from gso.services import ipam, _ipam

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:
    # 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
    lo0_alias = re.sub(".geant.net", "", subscription.device.device_fqdn)
    lo0_name = f"lo0.{lo0_alias}" 
    #print(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_ipv4_address = \
    #     ipam.new_service_host()
    #     ipaddress.ip_address('10.10.10.20')
    # subscription.device.device_lo_ipv6_address = \
        # ipaddress.ip_address('fc00:798:10::20')
    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()}.{subscription.device.device_site.site_country_code.lower()}.geant.net"
    #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} ' \
                               f'({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)
            >> 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
    )