Formatting and cleanup to pass TOX

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):

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
     )

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():