Skip to content
Snippets Groups Projects
Verified Commit 0649d9ee authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

update formatting

parent e5ec2c24
No related branches found
No related tags found
1 merge request!12Add sites, and add integration of IPtrunks and routers with LSO deployment
Hide whitespace changes
Inline Side-by-side
gso/services/provisioning_proxy.py
+ 10
− 6
View file @ 0649d9ee
......@@ -4,6 +4,7 @@ import logging
import requests
from orchestrator import inputstep
from orchestrator.config.assignee import Assignee
# noinspection PyProtectedMember
from orchestrator.forms import FormPage, ReadOnlyField
from orchestrator.forms.validators import Accept, Label, LongText
from orchestrator.types import UUIDstr, State, strEnum
......@@ -92,7 +93,7 @@ def provision_device(
def provision_ip_trunk(subscription: IptrunkProvisioning,
process_id: UUIDstr,
config_object: str,
config_object: str,
dry_run: bool = True):
"""
Function that provisions an IP trunk service using LSO.
......@@ -100,20 +101,21 @@ def provision_ip_trunk(subscription: IptrunkProvisioning,
:param :class:`IptrunkProvisioning` subscription: The subscription object
that is to be provisioned.
:param UUIDstr process_id: The related process ID, used for callback.
:param str config_object: The type of object that is deployed
:param bool dry_run: A boolean indicating whether this should be a dry run
or not, defaults to ``True``.
"""
parameters = {
'subscription': json.loads(json_dumps(subscription)),
'dry_run': dry_run,
'verb': "deploy", # FIXME
'object': config_object # FIXME
'verb': "deploy",
'object': config_object
}
_send_request('ip_trunk', parameters, process_id, CUDOperation.POST)
#def modify_ip_trunk(old_subscription: Iptrunk,
# def modify_ip_trunk(old_subscription: Iptrunk,
# new_subscription: Iptrunk,
# process_id: UUIDstr,
# dry_run: bool = True):
......@@ -137,9 +139,10 @@ def provision_ip_trunk(subscription: IptrunkProvisioning,
#
# _send_request('ip_trunk', parameters, process_id, CUDOperation.PUT)
def deprovision_ip_trunk(subscription: Iptrunk,
process_id: UUIDstr,
dry_run: bool = True):
process_id: UUIDstr,
dry_run: bool = True):
"""
Function that provisions an IP trunk service using LSO.
......@@ -157,6 +160,7 @@ def deprovision_ip_trunk(subscription: Iptrunk,
_send_request('ip_trunk', parameters, process_id, CUDOperation.DELETE)
@inputstep('Await provisioning proxy results', assignee=Assignee('SYSTEM'))
def await_pp_results() -> State:
class ProvisioningResultPage(FormPage):
......
gso/workflows/device/create_device.py
+ 1
− 2
View file @ 0649d9ee
......@@ -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
from orchestrator.targets import Target
......@@ -92,8 +93,6 @@ def get_info_from_ipam(subscription: DeviceInactive) -> State:
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 \
# = "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'
......
gso/workflows/device/terminate_device.py
+ 2
− 2
View file @ 0649d9ee
......@@ -19,8 +19,8 @@ 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}?"
) # type:ignore
f'Are you sure you want to remove {subscription.description}?'
)
return TerminateForm
......
gso/workflows/iptrunk/create_iptrunk.py
+ 34
− 24
View file @ 0649d9ee
......@@ -2,6 +2,7 @@ import ipaddress
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
......@@ -25,16 +26,17 @@ def device_selector(choice_value: str) -> list:
device_subscriptions = {}
for device_id, device_description in (
SubscriptionTable.query.join(ProductTable)
.filter(
.filter(
ProductTable.product_type == 'Device',
SubscriptionTable.status == 'active',
)
.with_entities(SubscriptionTable.subscription_id,
SubscriptionTable.description)
.all()
.with_entities(SubscriptionTable.subscription_id,
SubscriptionTable.description)
.all()
):
device_subscriptions[str(device_id)] = device_description
# noinspection PyTypeChecker
return choice_list(
Choice(choice_value, zip(device_subscriptions.keys(),
device_subscriptions.items())), # type:ignore
......@@ -68,10 +70,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
iptrunk_sideB_ae_members: list[str]
iptrunk_sideB_ae_members_descriptions: list[str]
# TODO: we need additional validation:
# sideA fqdn must be different from sideB fqdn
# the lenght of iptrunk_sideA_ae_members should
# be the same as iptrunk_sideA_ae_members
# interface names must be validated
# * sideA fqdn must be different from sideB fqdn
# * the length of iptrunk_sideA_ae_members should
# be the same as iptrunk_sideB_ae_members
# * interface names must be validated
user_input = yield CreateIptrunkForm
......@@ -152,64 +154,72 @@ def initialize_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")
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'trunk_interface')
return {'subscription': subscription}
@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)
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'trunk_interface', False)
return {'subscription': subscription}
@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")
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'isis_interface')
return {'subscription': subscription}
@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)
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'isis_interface', False)
return {'subscription': subscription}
@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")
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'ldp_interface')
return {'subscription': subscription}
@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)
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'ldp_interface', False)
return {'subscription': subscription}
@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")
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'lldp_interface')
return {'subscription': subscription}
@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)
process_id: UUIDstr) -> State:
provisioning_proxy.provision_ip_trunk(subscription, process_id,
'lldp_interface', False)
return {'subscription': subscription}
......
gso/workflows/iptrunk/terminate_iptrunk.py
+ 21
− 24
View file @ 0649d9ee
# noinspection PyProtectedMember
from orchestrator.forms import FormPage
from orchestrator.types import FormGenerator, State
from orchestrator.forms.validators import Label
from orchestrator.targets import Target
from orchestrator.types import InputForm, SubscriptionLifecycle, UUIDstr
from orchestrator.types import State
from orchestrator.workflow import done, init, step, workflow
from orchestrator.workflows.steps import (
resync,
......@@ -23,50 +24,46 @@ 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}?"
) # type:ignore
f'Are you sure you want to remove {subscription.description}?'
)
return TerminateForm
def _deprovision_in_user_management_system(fqdn: str) -> str:
pass
@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 )
process_id: UUIDstr) -> State:
provisioning_proxy.deprovision_ip_trunk(subscription, process_id)
return {'subscription': subscription}
@step('Deprovision IP trunk [FOR REAL]')
def deprovision_ip_trunk_real(subscription: Iptrunk,
process_id: UUIDstr) -> State:
process_id: UUIDstr) -> State:
provisioning_proxy.deprovision_ip_trunk(subscription, process_id, False)
return {'subscription': subscription}
@workflow(
"Terminate IPtrunk",
'Terminate IPtrunk',
initial_input_form=wrap_modify_initial_input_form(
initial_input_form_generator),
initial_input_form_generator),
target=Target.TERMINATE,
)
def terminate_iptrunk():
return (
init
>> store_process_subscription(Target.TERMINATE)
>> unsync
>> deprovision_ip_trunk_dry
>> await_pp_results
>> confirm_pp_results
>> deprovision_ip_trunk_real
>> await_pp_results
>> confirm_pp_results
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
>> done
init
>> store_process_subscription(Target.TERMINATE)
>> unsync
>> deprovision_ip_trunk_dry
>> await_pp_results
>> confirm_pp_results
>> deprovision_ip_trunk_real
>> await_pp_results
>> confirm_pp_results
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
>> done
)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment