diff --git a/gso/settings.py b/gso/settings.py
index a0be09616096e5916526878f0875a27108eb8fc3..a28b4eae0e033d463e41aff59ddd150222dfff0d 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -4,7 +4,8 @@ GSO settings, ensuring that the required parameters are set correctly.
import ipaddress
import json
import os
-from pydantic import BaseSettings, Field
+# from pydantic import BaseSettings, Field
+from pydantic import BaseSettings
class GeneralParams(BaseSettings):
diff --git a/gso/workflows/device/create_device.py b/gso/workflows/device/create_device.py
index 62d319586de87a52fdb4026ecd5a43ab07c0ac19..02f7eeba00e2c09003fcec8621b8f7f792f1a61a 100644
--- a/gso/workflows/device/create_device.py
+++ b/gso/workflows/device/create_device.py
@@ -3,6 +3,7 @@ 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
@@ -16,34 +17,40 @@ 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.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
+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()
+ 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
+ Choice(
+ "site_selection",
+ zip(site_subscriptions.keys(), site_subscriptions.items()),
+ ), # type:ignore
min_items=1,
max_items=1,
)
@@ -66,123 +73,144 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
return user_input.dict()
-@step('Create subscription')
+@step("Create subscription")
def create_subscription(product: UUIDstr) -> State:
subscription = DeviceInactive.from_product_id(product, uuid4())
return {
- 'subscription': subscription,
- 'subscription_id': subscription.subscription_id,
+ "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'])
+ 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')
+@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])
+ 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 \
- = _ipam.allocate_service_ipv4_network(service_type='SI', comment=f"SI for {lo0_name}").v4
- subscription.device.device_ias_lt_ipv4_network \
- = _ipam.allocate_service_ipv4_network(service_type='LT_IAS', comment=f"LT for {lo0_name}").v4
- subscription.device.device_ias_lt_ipv6_network \
- = _ipam.allocate_service_ipv6_network(service_type='LT_IAS', comment=f"LT for {lo0_name}").v6
- return {'subscription': subscription}
-
-
-@step('Initialize subscription')
+ subscription.device.device_lo_iso_address = iso_from_ipv4(
+ str(subscription.device.device_lo_ipv4_address)
+ )
+ subscription.device.device_si_ipv4_network = (
+ _ipam.allocate_service_ipv4_network(
+ service_type="SI", comment=f"SI for {lo0_name}"
+ ).v4
+ )
+ subscription.device.device_ias_lt_ipv4_network = (
+ _ipam.allocate_service_ipv4_network(
+ service_type="LT_IAS", comment=f"LT for {lo0_name}"
+ ).v4
+ )
+ subscription.device.device_ias_lt_ipv6_network = (
+ _ipam.allocate_service_ipv6_network(
+ service_type="LT_IAS", comment=f"LT for {lo0_name}"
+ ).v6
+ )
+ 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
+ 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_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()}"
+ ".geant.net"
+ )
subscription.device.device_fqdn = fqdn
subscription.device.device_role = device_role
subscription.device.device_access_via_ts = True
- subscription.description = f'Device {fqdn} ' \
- f'({subscription.device_type})'
+ subscription.description = f"Device {fqdn} ({subscription.device_type})"
subscription = device.DeviceProvisioning.from_other_lifecycle(
subscription, SubscriptionLifecycle.PROVISIONING
)
- return {'subscription': subscription}
+ return {"subscription": subscription}
-@step('Provision device [DRY RUN]')
-def provision_device_dry(subscription: DeviceProvisioning,
- process_id: UUIDstr) -> State:
+@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'Dry run for the deployment of base config on a'
- f'new {subscription.device_type}. Deployment is '
- f'done by the provisioning proxy, please '
- f'wait for the results to come back before '
- f'continuing.'
- }
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "Dry run for the deployment of base config on a"
+ f"new {subscription.device_type}. Deployment is "
+ "done by the provisioning proxy, please "
+ "wait for the results to come back before "
+ "continuing."
+ ),
+ }
-@step('Provision device [FOR REAL]')
-def provision_device_real(subscription: DeviceProvisioning,
- process_id: UUIDstr) -> State:
+@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'Deployment of base config for 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.'
- }
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "Deployment of base config for a new "
+ f"{subscription.device_type}. Deployment is being "
+ "taken care of by the provisioning proxy, please "
+ "wait for the results to come back before "
+ "continuing."
+ ),
+ }
@workflow(
- 'Create device',
+ "Create device",
initial_input_form=wrap_create_initial_input_form(
- initial_input_form_generator),
+ 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
+ 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
)
diff --git a/gso/workflows/device/terminate_device.py b/gso/workflows/device/terminate_device.py
index 7659e78d8d5c05582f7964757203825a88c8ec6b..c8b9147372fbb76b8433393fd5fc0066e4fc72b4 100644
--- a/gso/workflows/device/terminate_device.py
+++ b/gso/workflows/device/terminate_device.py
@@ -21,7 +21,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> InputForm:
class TerminateForm(FormPage):
are_you_sure: Label = (
- f'Are you sure you want to remove {subscription.description}?'
+ f"Are you sure you want to remove {subscription.description}?"
)
return TerminateForm
@@ -34,44 +34,63 @@ def _deprovision_in_user_management_system(fqdn: str) -> str:
@step("Deprovision loopback IPs from IPAM/DNS")
def deprovision_loopback_ips(subscription: Device) -> None:
input_host_addresses = ipam.HostAddresses(
- v4=ipaddress.ip_address(subscription.device.device_lo_ipv4_address),
- v6=ipaddress.ip_address(subscription.device.device_lo_ipv6_address)
- )
- fqdn_as_list=subscription.device.device_fqdn.split(".")
- hostname=str(fqdn_as_list[0])+"."+str(fqdn_as_list[1])+"."+str(fqdn_as_list[2])
- lo0_name="lo0."+hostname
- host_addresses=ipam.delete_service_host(
+ v4=ipaddress.ip_address(subscription.device.device_lo_ipv4_address),
+ v6=ipaddress.ip_address(subscription.device.device_lo_ipv6_address),
+ )
+ fqdn_as_list = subscription.device.device_fqdn.split(".")
+ hostname = (
+ str(fqdn_as_list[0])
+ + "."
+ + str(fqdn_as_list[1])
+ + "."
+ + str(fqdn_as_list[2])
+ )
+ lo0_name = "lo0." + hostname
+ host_addresses = ipam.delete_service_host(
hostname=lo0_name,
host_addresses=input_host_addresses,
cname_aliases=[hostname],
- service_type='LO'
- )
- return {'cleanup_for': hostname }
+ service_type="LO",
+ )
+ return {"addresses": host_addresses}
@step("Deprovision SI- interface IPs from IPAM/DNS")
def deprovision_si_ips(subscription: Device) -> None:
- service_network=ipam.delete_service_network(
- network=ipaddress.ip_network(subscription.device.device_si_ipv4_network)
- ,service_type='SI')
- return {'service_network': service_network}
+ service_network = ipam.delete_service_network(
+ network=ipaddress.ip_network(
+ subscription.device.device_si_ipv4_network
+ ),
+ service_type="SI",
+ )
+ return {"service_network": service_network}
-@step("Deprovision LT- (IAS) interface IPs from IPAM/DNS")
+@step("Deprovision LT- interface (IAS) IPs from IPAM/DNS")
def deprovision_lt_ips(subscription: Device) -> None:
- service_network_v4=ipam.delete_service_network(
- network=ipaddress.ip_network(subscription.device.device_ias_lt_ipv4_network),
- service_type='LT_IAS')
- service_network_v6=ipam.delete_service_network(
- network=ipaddress.ip_network(subscription.device.device_ias_lt_ipv6_network),
- service_type='LT_IAS')
- return {'service_network_v4': service_network_v4,
- 'service_network_v6': service_network_v6}
+ service_network_v4 = ipam.delete_service_network(
+ network=ipaddress.ip_network(
+ subscription.device.device_ias_lt_ipv4_network
+ ),
+ service_type="LT_IAS",
+ )
+ service_network_v6 = ipam.delete_service_network(
+ network=ipaddress.ip_network(
+ subscription.device.device_ias_lt_ipv6_network
+ ),
+ service_type="LT_IAS",
+ )
+ return {
+ "service_network_v4": service_network_v4,
+ "service_network_v6": service_network_v6,
+ }
+
@workflow(
"Terminate device",
initial_input_form=wrap_modify_initial_input_form(
- initial_input_form_generator),
+ initial_input_form_generator
+ ),
target=Target.TERMINATE,
)
def terminate_device():
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index e90d42841a92bbeadd49956c25c99fbbcaee6dcf..9c0e1c8c2215d12153830b538c1ea37a99133009 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -1,6 +1,7 @@
from uuid import uuid4
from orchestrator.db.models import ProductTable, SubscriptionTable
+
# noinspection PyProtectedMember
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, UniqueConstrainedList
@@ -15,11 +16,15 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
from gso.products.product_blocks import PhyPortCapacity
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.device import Device
-from gso.products.product_types.iptrunk import IptrunkInactive, \
- IptrunkProvisioning
+from gso.products.product_types.iptrunk import (
+ IptrunkInactive,
+ IptrunkProvisioning,
+)
from gso.services import provisioning_proxy, _ipam
-from gso.services.provisioning_proxy import confirm_pp_results, \
- await_pp_results
+from gso.services.provisioning_proxy import (
+ confirm_pp_results,
+ await_pp_results,
+)
def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -28,14 +33,15 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
devices = {}
for device_id, device_description in (
- SubscriptionTable.query.join(ProductTable)
- .filter(
- ProductTable.product_type == 'Device',
- SubscriptionTable.status == 'active',
- )
- .with_entities(SubscriptionTable.subscription_id,
- SubscriptionTable.description)
- .all()
+ SubscriptionTable.query.join(ProductTable)
+ .filter(
+ ProductTable.product_type == "Device",
+ SubscriptionTable.status == "active",
+ )
+ .with_entities(
+ SubscriptionTable.subscription_id, SubscriptionTable.description
+ )
+ .all()
):
devices[str(device_id)] = device_description
@@ -54,11 +60,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class AeMembersListA(UniqueConstrainedList[str]):
min_items = initial_user_input.iptrunk_minimum_links
- DeviceEnumA = Choice('Device A', zip(devices.keys(), devices.items()))
+ DeviceEnumA = Choice("Device A", zip(devices.keys(), devices.items()))
class CreateIptrunkSideAForm(FormPage):
class Config:
- title = 'Provide subscription details for side A of the trunk.'
+ title = "Provide subscription details for side A of the trunk."
iptrunk_sideA_node_id: DeviceEnumA
iptrunk_sideA_ae_iface: str
@@ -70,7 +76,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
# We remove the selected device for side A, to prevent any loops
devices.pop(str(user_input_side_a.iptrunk_sideA_node_id.name))
- DeviceEnumB = Choice('Device B', zip(devices.keys(), devices.items()))
+ DeviceEnumB = Choice("Device B", zip(devices.keys(), devices.items()))
class AeMembersListB(UniqueConstrainedList[str]):
min_items = len(user_input_side_a.iptrunk_sideA_ae_members)
@@ -78,7 +84,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class CreateIptrunkSideBForm(FormPage):
class Config:
- title = 'Provide subscription details for side B of the trunk.'
+ title = "Provide subscription details for side B of the trunk."
iptrunk_sideB_node_id: DeviceEnumB
iptrunk_sideB_ae_iface: str
@@ -88,49 +94,59 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
user_input_side_b = yield CreateIptrunkSideBForm
- return initial_user_input.dict() | \
- user_input_side_a.dict() | \
- user_input_side_b.dict()
+ return (
+ initial_user_input.dict()
+ | user_input_side_a.dict()
+ | user_input_side_b.dict()
+ )
-@step('Create subscription')
+@step("Create subscription")
def create_subscription(product: UUIDstr) -> State:
subscription = IptrunkInactive.from_product_id(product, uuid4())
return {
- 'subscription': subscription,
- 'subscription_id': subscription.subscription_id,
+ "subscription": subscription,
+ "subscription_id": subscription.subscription_id,
}
-@step('Get information from IPAM')
+@step("Get information from IPAM")
def get_info_from_ipam(subscription: IptrunkProvisioning) -> State:
# TODO: get info about how these should be generated
- subscription.iptrunk.iptrunk_ipv4_network \
- = _ipam.allocate_service_ipv4_network(service_type="TRUNK", comment=subscription.iptrunk.iptrunk_description).v4
- subscription.iptrunk.iptrunk_ipv6_network \
- = _ipam.allocate_service_ipv6_network(service_type="TRUNK", comment=subscription.iptrunk.iptrunk_description).v6
- return {'subscription': subscription}
+ subscription.iptrunk.iptrunk_ipv4_network = (
+ _ipam.allocate_service_ipv4_network(
+ service_type="TRUNK",
+ comment=subscription.iptrunk.iptrunk_description,
+ ).v4
+ )
+ subscription.iptrunk.iptrunk_ipv6_network = (
+ _ipam.allocate_service_ipv6_network(
+ service_type="TRUNK",
+ comment=subscription.iptrunk.iptrunk_description,
+ ).v6
+ )
+ return {"subscription": subscription}
-@step('Initialize subscription')
+@step("Initialize subscription")
def initialize_subscription(
- subscription: IptrunkInactive,
- geant_s_sid: str,
- iptrunk_type: IptrunkType,
- iptrunk_description: str,
- iptrunk_speed: str,
- iptrunk_minimum_links: int,
- iptrunk_sideA_node_id: str,
- iptrunk_sideA_ae_iface: str,
- iptrunk_sideA_ae_geant_a_sid: str,
- iptrunk_sideA_ae_members: list[str],
- iptrunk_sideA_ae_members_descriptions: list[str],
- iptrunk_sideB_node_id: str,
- iptrunk_sideB_ae_iface: str,
- iptrunk_sideB_ae_geant_a_sid: str,
- iptrunk_sideB_ae_members: list[str],
- iptrunk_sideB_ae_members_descriptions: list[str]
+ subscription: IptrunkInactive,
+ geant_s_sid: str,
+ iptrunk_type: IptrunkType,
+ iptrunk_description: str,
+ iptrunk_speed: str,
+ iptrunk_minimum_links: int,
+ iptrunk_sideA_node_id: str,
+ iptrunk_sideA_ae_iface: str,
+ iptrunk_sideA_ae_geant_a_sid: str,
+ iptrunk_sideA_ae_members: list[str],
+ iptrunk_sideA_ae_members_descriptions: list[str],
+ iptrunk_sideB_node_id: str,
+ iptrunk_sideB_ae_iface: 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
@@ -139,189 +155,235 @@ def initialize_subscription(
subscription.iptrunk.iptrunk_isis_metric = 9000
subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
- subscription.iptrunk.iptrunk_sideA_node = \
- Device.from_subscription(iptrunk_sideA_node_id).device
+ subscription.iptrunk.iptrunk_sideA_node = Device.from_subscription(
+ iptrunk_sideA_node_id
+ ).device
subscription.iptrunk.iptrunk_sideA_ae_iface = iptrunk_sideA_ae_iface
- subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid \
- = iptrunk_sideA_ae_geant_a_sid
+ 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_node = \
- Device.from_subscription(iptrunk_sideB_node_id).device
- subscription.iptrunk.iptrunk_sideB_ae_iface \
- = iptrunk_sideB_ae_iface
- subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid \
- = iptrunk_sideB_ae_geant_a_sid
+ subscription.iptrunk.iptrunk_sideA_ae_members_description = (
+ iptrunk_sideA_ae_members_descriptions
+ )
+
+ subscription.iptrunk.iptrunk_sideB_node = Device.from_subscription(
+ iptrunk_sideB_node_id
+ ).device
+ subscription.iptrunk.iptrunk_sideB_ae_iface = iptrunk_sideB_ae_iface
+ 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.iptrunk.iptrunk_sideB_ae_members_description = (
+ iptrunk_sideB_ae_members_descriptions
+ )
- subscription.description = f'IP trunk, geant_s_sid:{geant_s_sid}'
+ subscription.description = f"IP trunk, geant_s_sid:{geant_s_sid}"
subscription = IptrunkProvisioning.from_other_lifecycle(
subscription, SubscriptionLifecycle.PROVISIONING
)
- return {'subscription': subscription}
-
-
-@step('Provision IP trunk interface [DRY RUN]')
-def provision_ip_trunk_iface_dry(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'trunk_interface')
-
- return {'subscription': subscription,
- 'label_text': 'This is a dry run for the deployment of a new IP '
- 'trunk. Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk interface [FOR REAL]')
-def provision_ip_trunk_iface_real(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'trunk_interface', False)
-
- return {'subscription': subscription,
- 'label_text': 'This is a live deployment of a new IP trunk. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk ISIS interface [DRY RUN]')
-def provision_ip_trunk_isis_iface_dry(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'isis_interface')
-
- return {'subscription': subscription,
- 'label_text': 'This is a dry run for the deployment of a new IP '
- 'trunk ISIS interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk ISIS interface [FOR REAL]')
-def provision_ip_trunk_isis_iface_real(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'isis_interface', False)
-
- return {'subscription': subscription,
- 'label_text': 'This is a live deployment of a new IP trunk '
- 'ISIS interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk LDP interface [DRY RUN]')
-def provision_ip_trunk_ldp_iface_dry(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'ldp_interface')
-
- return {'subscription': subscription,
- 'label_text': 'This is a dry run for the deployment of a new IP '
- 'trunk LDP interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk LDP interface [FOR REAL]')
-def provision_ip_trunk_ldp_iface_real(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'ldp_interface', False)
-
- return {'subscription': subscription,
- 'label_text': 'This is a live deployment of a new IP trunk '
- 'LDP interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk LLDP interface [DRY RUN]')
-def provision_ip_trunk_lldp_iface_dry(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'lldp_interface')
-
- return {'subscription': subscription,
- 'label_text': 'This is a dry run for the deployment of a new IP '
- 'trunk LLDP interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
-
-
-@step('Provision IP trunk LLDP interface [FOR REAL]')
-def provision_ip_trunk_lldp_iface_real(subscription: IptrunkProvisioning,
- process_id: UUIDstr) -> State:
- provisioning_proxy.provision_ip_trunk(subscription, process_id,
- 'lldp_interface', False)
-
- return {'subscription': subscription,
- 'label_text': 'This is a live deployment of a new IP trunk '
- 'LLDP interface. '
- 'Deployment is being taken care of by the '
- 'provisioning proxy, please wait for the results to '
- 'come back before continuing.'
- }
+ return {"subscription": subscription}
+
+
+@step("Provision IP trunk interface [DRY RUN]")
+def provision_ip_trunk_iface_dry(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "trunk_interface"
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a dry run for the deployment of a new IP "
+ "trunk. Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk interface [FOR REAL]")
+def provision_ip_trunk_iface_real(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "trunk_interface", False
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a live deployment of a new IP trunk. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk ISIS interface [DRY RUN]")
+def provision_ip_trunk_isis_iface_dry(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "isis_interface"
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a dry run for the deployment of a new IP "
+ "trunk ISIS interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk ISIS interface [FOR REAL]")
+def provision_ip_trunk_isis_iface_real(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "isis_interface", False
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a live deployment of a new IP trunk "
+ "ISIS interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk LDP interface [DRY RUN]")
+def provision_ip_trunk_ldp_iface_dry(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "ldp_interface"
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a dry run for the deployment of a new IP "
+ "trunk LDP interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk LDP interface [FOR REAL]")
+def provision_ip_trunk_ldp_iface_real(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "ldp_interface", False
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a live deployment of a new IP trunk "
+ "LDP interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk LLDP interface [DRY RUN]")
+def provision_ip_trunk_lldp_iface_dry(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "lldp_interface"
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a dry run for the deployment of a new IP "
+ "trunk LLDP interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
+
+
+@step("Provision IP trunk LLDP interface [FOR REAL]")
+def provision_ip_trunk_lldp_iface_real(
+ subscription: IptrunkProvisioning, process_id: UUIDstr
+) -> State:
+ provisioning_proxy.provision_ip_trunk(
+ subscription, process_id, "lldp_interface", False
+ )
+
+ return {
+ "subscription": subscription,
+ "label_text": (
+ "This is a live deployment of a new IP trunk "
+ "LLDP interface. "
+ "Deployment is being taken care of by the "
+ "provisioning proxy, please wait for the results to "
+ "come back before continuing."
+ ),
+ }
@workflow(
- 'Create IP trunk',
+ "Create IP trunk",
initial_input_form=wrap_create_initial_input_form(
- initial_input_form_generator),
+ initial_input_form_generator
+ ),
target=Target.CREATE,
)
def create_iptrunk():
return (
- init
- >> create_subscription
- >> store_process_subscription(Target.CREATE)
- >> initialize_subscription
- >> get_info_from_ipam
- >> provision_ip_trunk_iface_dry
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_iface_real
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_isis_iface_dry
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_isis_iface_real
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_ldp_iface_dry
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_ldp_iface_real
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_lldp_iface_dry
- >> await_pp_results
- >> confirm_pp_results
- >> provision_ip_trunk_lldp_iface_real
- >> await_pp_results
- >> confirm_pp_results
- >> set_status(SubscriptionLifecycle.ACTIVE)
- >> resync
- >> done
+ init
+ >> create_subscription
+ >> store_process_subscription(Target.CREATE)
+ >> initialize_subscription
+ >> get_info_from_ipam
+ >> provision_ip_trunk_iface_dry
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_iface_real
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_isis_iface_dry
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_isis_iface_real
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_ldp_iface_dry
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_ldp_iface_real
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_lldp_iface_dry
+ >> await_pp_results
+ >> confirm_pp_results
+ >> provision_ip_trunk_lldp_iface_real
+ >> await_pp_results
+ >> confirm_pp_results
+ >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> resync
+ >> done
)