fix black linting errors

gso/products/product_blocks/device.py

@@ -18,9 +18,9 @@ class DeviceRole(strEnum):
     amt = "amt"
 
 
-class DeviceBlockInactive(ProductBlockModel,
-                          lifecycle=[SubscriptionLifecycle.INITIAL],
-                          product_block_name="DeviceBlock"):
+class DeviceBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="DeviceBlock"
+):
     device_fqdn: Optional[str] = None
     device_ts_address: Optional[str] = None
     device_ts_port: Optional[int] = None
@@ -36,8 +36,7 @@ class DeviceBlockInactive(ProductBlockModel,
     device_site: Optional[SiteBlockInactive]
 
 
-class DeviceBlockProvisioning(DeviceBlockInactive,
-                              lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class DeviceBlockProvisioning(DeviceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     device_fqdn: str
     device_ts_address: str
     device_ts_port: int
@@ -53,8 +52,7 @@ class DeviceBlockProvisioning(DeviceBlockInactive,
     device_site: Optional[SiteBlockProvisioning]
 
 
-class DeviceBlock(DeviceBlockProvisioning,
-                  lifecycle=[SubscriptionLifecycle.ACTIVE]):
+class DeviceBlock(DeviceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     device_fqdn: str
     device_ts_address: str
     device_ts_port: int

gso/products/product_blocks/iptrunk.py

@@ -13,9 +13,9 @@ class IptrunkType(strEnum):
     Leased = "Leased"
 
 
-class IptrunkBlockInactive(ProductBlockModel,
-                           lifecycle=[SubscriptionLifecycle.INITIAL],
-                           product_block_name="IptrunkBlock"):
+class IptrunkBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="IptrunkBlock"
+):
     geant_s_sid: Optional[str] = None
     iptrunk_description: Optional[str] = None
     iptrunk_type: Optional[IptrunkType] = None
@@ -29,19 +29,16 @@ class IptrunkBlockInactive(ProductBlockModel,
     iptrunk_sideA_ae_iface: Optional[str] = None
     iptrunk_sideA_ae_geant_a_sid: Optional[str] = None
     iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideA_ae_members_description: list[str] \
-        = Field(default_factory=list)
+    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
     #
     iptrunk_sideB_node: DeviceBlockInactive
     iptrunk_sideB_ae_iface: Optional[str] = None
     iptrunk_sideB_ae_geant_a_sid: Optional[str] = None
     iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] \
-        = Field(default_factory=list)
+    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)
 
 
-class IptrunkBlockProvisioning(IptrunkBlockInactive,
-                               lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     geant_s_sid: Optional[str] = None
     iptrunk_description: Optional[str] = None
     iptrunk_type: Optional[IptrunkType] = None
@@ -55,19 +52,16 @@ class IptrunkBlockProvisioning(IptrunkBlockInactive,
     iptrunk_sideA_ae_iface: Optional[str] = None
     iptrunk_sideA_ae_geant_a_sid: Optional[str] = None
     iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideA_ae_members_description: list[str] = Field(
-            default_factory=list)
+    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
     #
     iptrunk_sideB_node: DeviceBlockProvisioning
     iptrunk_sideB_ae_iface: Optional[str] = None
     iptrunk_sideB_ae_geant_a_sid: Optional[str] = None
     iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] = Field(
-            default_factory=list)
+    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)
 
 
-class IptrunkBlock(IptrunkBlockProvisioning,
-                   lifecycle=[SubscriptionLifecycle.ACTIVE]):
+class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     geant_s_sid: str
     iptrunk_description: str
     iptrunk_type: IptrunkType
@@ -81,12 +75,10 @@ class IptrunkBlock(IptrunkBlockProvisioning,
     iptrunk_sideA_ae_iface: str
     iptrunk_sideA_ae_geant_a_sid: str
     iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideA_ae_members_description: list[str] = Field(
-            default_factory=list)
+    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
     #
     iptrunk_sideB_node: DeviceBlock
     iptrunk_sideB_ae_iface: str
     iptrunk_sideB_ae_geant_a_sid: str
     iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] = Field(
-            default_factory=list)
+    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)

gso/products/product_blocks/site.py

@@ -11,9 +11,7 @@ class SiteTier(strEnum):
     tier4 = 4
 
 
-class SiteBlockInactive(ProductBlockModel,
-                        lifecycle=[SubscriptionLifecycle.INITIAL],
-                        product_block_name="SiteBlock"):
+class SiteBlockInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="SiteBlock"):
     site_name: Optional[str] = None
     site_city: Optional[str] = None
     site_country: Optional[str] = None
@@ -25,8 +23,7 @@ class SiteBlockInactive(ProductBlockModel,
     site_tier: Optional[SiteTier] = None
 
 
-class SiteBlockProvisioning(SiteBlockInactive,
-                            lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     site_name: Optional[str] = None
     site_city: Optional[str] = None
     site_country: Optional[str] = None
@@ -38,8 +35,7 @@ class SiteBlockProvisioning(SiteBlockInactive,
     site_tier: Optional[SiteTier] = None
 
 
-class SiteBlock(SiteBlockProvisioning,
-                lifecycle=[SubscriptionLifecycle.ACTIVE]):
+class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     site_name: str
     site_city: str
     site_country: str

gso/products/product_types/device.py

@@ -14,8 +14,7 @@ class DeviceInactive(SubscriptionModel, is_base=True):
     device: DeviceBlockInactive
 
 
-class DeviceProvisioning(DeviceInactive,
-                         lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class DeviceProvisioning(DeviceInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     device_type: DeviceType
     device: DeviceBlockProvisioning
 

gso/products/product_types/iptrunk.py

@@ -8,8 +8,7 @@ class IptrunkInactive(SubscriptionModel, is_base=True):
     iptrunk: IptrunkBlockInactive
 
 
-class IptrunkProvisioning(IptrunkInactive,
-                          lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class IptrunkProvisioning(IptrunkInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     iptrunk: IptrunkBlockProvisioning
 
 

gso/products/product_types/site.py

@@ -8,8 +8,7 @@ class SiteInactive(SubscriptionModel, is_base=True):
     site: SiteBlockInactive
 
 
-class SiteProvisioning(SiteInactive,
-                       lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+class SiteProvisioning(SiteInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     site: SiteBlockProvisioning
 
 

gso/services/ipam.py

@@ -32,51 +32,43 @@ class HostAddresses(BaseSettings):
     v6: ipaddress.IPv6Address
 
 
-def new_service_networks(service_type='',
-                         comment="",
-                         extattrs={}) -> ServiceNetworks:
+def new_service_networks(service_type="", comment="", extattrs={}) -> ServiceNetworks:
     v4_service_network = _ipam.allocate_service_ipv4_network(
-        service_type=service_type, comment=comment, extattrs=extattrs)
+        service_type=service_type, comment=comment, extattrs=extattrs
+    )
     v6_service_network = _ipam.allocate_service_ipv6_network(
-        service_type=service_type, comment=comment, extattrs=extattrs)
-    return ServiceNetworks(
-        v4=v4_service_network.v4,
-        v6=v6_service_network.v6)
-
-
-def new_service_host(hostname,
-                     service_type='',
-                     service_networks: ServiceNetworks = None,
-                     host_addresses: HostAddresses = None,
-                     cname_aliases=None,
-                     extattrs={}) -> HostAddresses:
+        service_type=service_type, comment=comment, extattrs=extattrs
+    )
+    return ServiceNetworks(v4=v4_service_network.v4, v6=v6_service_network.v6)
+
+
+def new_service_host(
+    hostname,
+    service_type="",
+    service_networks: ServiceNetworks = None,
+    host_addresses: HostAddresses = None,
+    cname_aliases=None,
+    extattrs={},
+) -> HostAddresses:
     return _ipam.allocate_service_host(
         hostname=hostname,
         service_type=service_type,
         service_networks=service_networks,
         host_addresses=host_addresses,
         cname_aliases=cname_aliases,
-        extattrs=extattrs)
+        extattrs=extattrs,
+    )
 
 
 def delete_service_network(
-    network: ipaddress.ip_network = None, service_type=''
+    network: ipaddress.ip_network = None, service_type=""
 ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
-    return _ipam.delete_service_network(
-        ipnetwork=network,
-        service_type=service_type
-    )
+    return _ipam.delete_service_network(ipnetwork=network, service_type=service_type)
 
 
 def delete_service_host(
-    hostname='',
-    host_addresses: HostAddresses = None,
-    cname_aliases=[],
-    service_type=''
+    hostname="", host_addresses: HostAddresses = None, cname_aliases=[], service_type=""
 ) -> HostAddresses:
     return _ipam.delete_service_host(
-        hostname=hostname,
-        host_addresses=host_addresses,
-        cname_aliases=cname_aliases,
-        service_type=service_type
+        hostname=hostname, host_addresses=host_addresses, cname_aliases=cname_aliases, service_type=service_type
     )

gso/services/provisioning_proxy.py

@@ -29,16 +29,16 @@ class CUDOperation(strEnum):
     Enum for different C(R)UD operations that the provisioning proxy supports.
     Read is not applicable, hence these become CUD and not CRUD operations.
     """
+
     #: Creation is done with a POST request
-    POST = 'POST'
+    POST = "POST"
     #: Updating is done with a PUT request
-    PUT = 'PUT'
+    PUT = "PUT"
     #: Removal is done with a DELETE request
-    DELETE = 'DELETE'
+    DELETE = "DELETE"
 
 
-def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr,
-                  operation: CUDOperation):
+def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operation: CUDOperation):
     """
     Internal function for sending a request to LSO. The callback address is
     derived using the process ID provided.
@@ -57,13 +57,11 @@ def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr,
     pp_params = oss.PROVISIONING_PROXY
     assert pp_params
 
-    callback_url = f'{settings.load_oss_params().GENERAL.public_hostname}' \
-                   f'/api/processes/{process_id}/resume'
-    logger.debug('[provisioning proxy] provisioning for process %s',
-                 process_id)
+    callback_url = f"{settings.load_oss_params().GENERAL.public_hostname}" f"/api/processes/{process_id}/resume"
+    logger.debug("[provisioning proxy] provisioning for process %s", process_id)
 
-    parameters.update({'callback': callback_url})
-    url = f'{pp_params.scheme}://{pp_params.api_base}/api/{endpoint}'
+    parameters.update({"callback": callback_url})
+    url = f"{pp_params.scheme}://{pp_params.api_base}/api/{endpoint}"
 
     request = None
 
@@ -79,10 +77,7 @@ def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr,
         raise AssertionError(request.content)
 
 
-def provision_device(
-        subscription: DeviceProvisioning,
-        process_id: UUIDstr,
-        dry_run: bool = True):
+def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_run: bool = True):
     """
     Function that provisions a new device using LSO.
 
@@ -92,18 +87,14 @@ def provision_device(
     :param bool dry_run: A boolean indicating whether this should be a dry run
         or not, defaults to ``True``.
     """
-    parameters = {
-        'dry_run': dry_run,
-        'subscription': json.loads(json_dumps(subscription))
-    }
+    parameters = {"dry_run": dry_run, "subscription": json.loads(json_dumps(subscription))}
 
-    _send_request('device', parameters, process_id, CUDOperation.POST)
+    _send_request("device", parameters, process_id, CUDOperation.POST)
 
 
-def provision_ip_trunk(subscription: IptrunkProvisioning,
-                       process_id: UUIDstr,
-                       config_object: str,
-                       dry_run: bool = True):
+def provision_ip_trunk(
+    subscription: IptrunkProvisioning, process_id: UUIDstr, config_object: str, dry_run: bool = True
+):
     """
     Function that provisions an IP trunk service using LSO.
 
@@ -115,13 +106,13 @@ def provision_ip_trunk(subscription: IptrunkProvisioning,
         or not, defaults to ``True``.
     """
     parameters = {
-        'subscription': json.loads(json_dumps(subscription)),
-        'dry_run': dry_run,
-        'verb': 'deploy',
-        'object': config_object
+        "subscription": json.loads(json_dumps(subscription)),
+        "dry_run": dry_run,
+        "verb": "deploy",
+        "object": config_object,
     }
 
-    _send_request('ip_trunk', parameters, process_id, CUDOperation.POST)
+    _send_request("ip_trunk", parameters, process_id, CUDOperation.POST)
 
 
 # def modify_ip_trunk(old_subscription: Iptrunk,
@@ -149,9 +140,7 @@ 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):
+def deprovision_ip_trunk(subscription: Iptrunk, process_id: UUIDstr, dry_run: bool = True):
     """
     Function that provisions an IP trunk service using LSO.
 
@@ -161,31 +150,25 @@ def deprovision_ip_trunk(subscription: Iptrunk,
     :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': 'terminate'
-    }
+    parameters = {"subscription": json.loads(json_dumps(subscription)), "dry_run": dry_run, "verb": "terminate"}
 
-    _send_request('ip_trunk', parameters, process_id, CUDOperation.DELETE)
+    _send_request("ip_trunk", parameters, process_id, CUDOperation.DELETE)
 
 
-@inputstep('Await provisioning proxy results', assignee=Assignee('SYSTEM'))
-def await_pp_results(subscription: SubscriptionModel,
-                     label_text: str) -> State:
+@inputstep("Await provisioning proxy results", assignee=Assignee("SYSTEM"))
+def await_pp_results(subscription: SubscriptionModel, label_text: str) -> State:
     class ProvisioningResultPage(FormPage):
         class Config:
-            title = f'Deploying {subscription.product.name}...'
+            title = f"Deploying {subscription.product.name}..."
 
         warning_label: Label = label_text
         pp_run_results: dict = None
-        confirm: Accept = Accept('INCOMPLETE')
+        confirm: Accept = Accept("INCOMPLETE")
 
-        @validator('pp_run_results', allow_reuse=True, pre=True, always=True)
+        @validator("pp_run_results", allow_reuse=True, pre=True, always=True)
         def run_results_must_be_given(cls, run_results):
             if run_results is None:
-                raise ValueError('Run results may not be empty. '
-                                 'Wait for the provisioning proxy to finish.')
+                raise ValueError("Run results may not be empty. " "Wait for the provisioning proxy to finish.")
             return run_results
 
     result_page = yield ProvisioningResultPage
@@ -193,18 +176,19 @@ def await_pp_results(subscription: SubscriptionModel,
     return result_page.dict()
 
 
-@inputstep('Confirm provisioning proxy results', assignee=Assignee('SYSTEM'))
+@inputstep("Confirm provisioning proxy results", assignee=Assignee("SYSTEM"))
 def confirm_pp_results(state: State) -> State:
     class ConfirmRunPage(FormPage):
         class Config:
-            title = f"Execution for " \
-                    f"{state['subscription']['product']['name']} " \
-                    f"completed, please confirm the results below."
-
-        run_status: str = ReadOnlyField(state['pp_run_results']['status'])
-        run_results: LongText = ReadOnlyField(
-            f"{state['pp_run_results']['output']}")
-        confirm: Accept = Accept('INCOMPLETE')
+            title = (
+                f"Execution for "
+                f"{state['subscription']['product']['name']} "
+                f"completed, please confirm the results below."
+            )
+
+        run_status: str = ReadOnlyField(state["pp_run_results"]["status"])
+        run_results: LongText = ReadOnlyField(f"{state['pp_run_results']['output']}")
+        confirm: Accept = Accept("INCOMPLETE")
 
     yield ConfirmRunPage
 

gso/services/resource_manager.py

@@ -4,26 +4,22 @@ from gso import settings
 
 
 def import_new_router(router_name, oss_params=settings.OSSParams):
-    r = requests.post(
-        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
-        f'/api/interfaces/initialize-router/{router_name}')
+    r = requests.post(f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/initialize-router/{router_name}")
     r.raise_for_status()
 
 
 def next_lag(router_name, oss_params=settings.OSSParams):
-    r = requests.post(
-        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
-        f'/api/interfaces/next-lag/{router_name}')
+    r = requests.post(f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-lag/{router_name}")
     r.raise_for_status()
     response = r.json()
-    return response['name']
+    return response["name"]
 
 
 def next_physical(router_name, lag_name, oss_params=settings.OSSParams):
     # TODO: speed needed (if first interface)
     r = requests.post(
-        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
-        f'/api/interfaces/next-physical/{router_name}/{lag_name}')
+        f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-physical/{router_name}/{lag_name}"
+    )
     r.raise_for_status()
     response = r.json()
-    return response['name']
+    return response["name"]

gso/settings.py

@@ -13,6 +13,7 @@ class GeneralParams(BaseSettings):
     """
     General parameters for a GSO configuration file.
     """
+
     #: The hostname that GSO is publicly served at, used for building the
     #: callback URL that the provisioning proxy uses.
     public_hostname: str
@@ -22,6 +23,7 @@ class InfoBloxParams(BaseSettings):
     """
     Parameters related to InfoBlox.
     """
+
     scheme: str
     wapi_version: str
     host: str
@@ -33,6 +35,7 @@ class V4NetworkParams(BaseSettings):
     """
     A set of parameters that describe an IPv4 network in InfoBlox.
     """
+
     containers: list[ipaddress.IPv4Network]
     networks: list[ipaddress.IPv4Network]
     mask: int  # TODO: validation on mask?
@@ -42,6 +45,7 @@ class V6NetworkParams(BaseSettings):
     """
     A set of parameters that describe an IPv6 network in InfoBlox.
     """
+
     containers: list[ipaddress.IPv6Network]
     networks: list[ipaddress.IPv6Network]
     mask: int  # TODO: validation on mask?
@@ -52,6 +56,7 @@ class ServiceNetworkParams(BaseSettings):
     Parameters for InfoBlox that describe IPv4 and v6 networks, and the
     corresponding domain name that should be used as a suffix.
     """
+
     V4: V4NetworkParams
     V6: V6NetworkParams
     domain_name: str
@@ -62,6 +67,7 @@ class IPAMParams(BaseSettings):
     """
     A set of parameters related to IPAM.
     """
+
     INFOBLOX: InfoBloxParams
     LO: ServiceNetworkParams
     TRUNK: ServiceNetworkParams
@@ -74,6 +80,7 @@ class ProvisioningProxyParams(BaseSettings):
     """
     Parameters for the provisioning proxy.
     """
+
     scheme: str
     api_base: str
     auth: str  # FIXME: unfinished
@@ -84,6 +91,7 @@ class OSSParams(BaseSettings):
     """
     The set of parameters required for running GSO.
     """
+
     GENERAL: GeneralParams
     IPAM: IPAMParams
     RESOURCE_MANAGER_API_PREFIX: str
@@ -95,9 +103,9 @@ def load_oss_params() -> OSSParams:
     look for OSS_PARAMS_FILENAME in the environment and load the parameters
     from that file.
     """
-    with open(os.environ['OSS_PARAMS_FILENAME'], encoding='utf-8') as file:
+    with open(os.environ["OSS_PARAMS_FILENAME"], encoding="utf-8") as file:
         return OSSParams(**json.loads(file.read()))
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     print(load_oss_params())

gso/workflows/__init__.py

@@ -4,16 +4,11 @@ init class that imports all workflows into GSO.
 from orchestrator.workflows import LazyWorkflowInstance
 
 LazyWorkflowInstance("gso.workflows.device.create_device", "create_device")
-LazyWorkflowInstance(
-    "gso.workflows.device.terminate_device", "terminate_device"
-)
+LazyWorkflowInstance("gso.workflows.device.terminate_device", "terminate_device")
 LazyWorkflowInstance("gso.workflows.device.get_facts", "get_facts")
 LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.modify_generic", "modify_generic")
-LazyWorkflowInstance("gso.workflows.iptrunk.terminate_iptrunk",
-                     "terminate_iptrunk")
-LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric",
-                     "modify_isis_metric")
-LazyWorkflowInstance("gso.workflows.iptrunk.modify_generic",
-                     "modify_generic")
+LazyWorkflowInstance("gso.workflows.iptrunk.terminate_iptrunk", "terminate_iptrunk")
+LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric", "modify_isis_metric")
+LazyWorkflowInstance("gso.workflows.iptrunk.modify_generic", "modify_generic")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")

gso/workflows/device/create_device.py

@@ -29,9 +29,7 @@ def site_selector() -> Choice:
             ProductTable.product_type == "Site",
             SubscriptionTable.status == "active",
         )
-        .with_entities(
-            SubscriptionTable.subscription_id, SubscriptionTable.description
-        )
+        .with_entities(SubscriptionTable.subscription_id, SubscriptionTable.description)
         .all()
     ):
         site_subscriptions[str(site_id)] = site_description
@@ -82,48 +80,36 @@ def iso_from_ipv4(ipv4_address):
 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_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
-    )
+    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
-    ).site
+    subscription.device.device_site = Site.from_subscription(device_site).site
     fqdn = (
         f"{hostname}.{subscription.device.device_site.site_name.lower()}."
         f"{subscription.device.device_site.site_country_code.lower()}"
@@ -134,17 +120,13 @@ def initialize_subscription(
     subscription.device.device_access_via_ts = True
     subscription.description = f"Device {fqdn} ({subscription.device_type})"
 
-    subscription = device.DeviceProvisioning.from_other_lifecycle(
-        subscription, SubscriptionLifecycle.PROVISIONING
-    )
+    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:
+def provision_device_dry(subscription: DeviceProvisioning, process_id: UUIDstr) -> State:
     provisioning_proxy.provision_device(subscription, process_id)
 
     return {
@@ -160,9 +142,7 @@ def provision_device_dry(
 
 
 @step("Provision device [FOR REAL]")
-def provision_device_real(
-        subscription: DeviceProvisioning, process_id: UUIDstr
-) -> State:
+def provision_device_real(subscription: DeviceProvisioning, process_id: UUIDstr) -> State:
     provisioning_proxy.provision_device(subscription, process_id, False)
 
     return {
@@ -179,25 +159,23 @@ def provision_device_real(
 
 @workflow(
     "Create device",
-    initial_input_form=wrap_create_initial_input_form(
-        initial_input_form_generator
-    ),
+    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
+        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/get_facts.py

@@ -17,9 +17,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from gso.products.product_types.device import Device
 
 
-def initial_input_form_generator(
-    subscription_id: UUIDstr, organisation: UUIDstr
-) -> InputForm:
+def initial_input_form_generator(subscription_id: UUIDstr, organisation: UUIDstr) -> InputForm:
     subscription = Device.from_subscription(subscription_id)
 
     class TerminateForm(FormPage):
@@ -47,9 +45,7 @@ def get_facts(subscription_id) -> None:
 
 @workflow(
     "Get Facts from Device",
-    initial_input_form=wrap_modify_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
     target=Target.SYSTEM,
 )
 def get_facts_from_device():

gso/workflows/device/terminate_device.py

@@ -16,9 +16,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> InputForm:
     subscription = Device.from_subscription(subscription_id)
 
     class TerminateForm(FormPage):
-        are_you_sure: Label = (
-            f"Are you sure you want to remove {subscription.description}?"
-        )
+        are_you_sure: Label = f"Are you sure you want to remove {subscription.description}?"
 
     return TerminateForm
 
@@ -34,13 +32,7 @@ def deprovision_loopback_ips(subscription: Device) -> None:
         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])
-    )
+    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,
@@ -54,9 +46,7 @@ def deprovision_loopback_ips(subscription: Device) -> None:
 @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
-        ),
+        network=ipaddress.ip_network(subscription.device.device_si_ipv4_network),
         service_type="SI",
     )
     return {"service_network": service_network}
@@ -65,15 +55,11 @@ def deprovision_si_ips(subscription: Device) -> None:
 @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
-        ),
+        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
-        ),
+        network=ipaddress.ip_network(subscription.device.device_ias_lt_ipv6_network),
         service_type="LT_IAS",
     )
     return {
@@ -84,9 +70,7 @@ def deprovision_lt_ips(subscription: Device) -> None:
 
 @workflow(
     "Terminate device",
-    initial_input_form=wrap_modify_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
     target=Target.TERMINATE,
 )
 def terminate_device():

gso/workflows/iptrunk/create_iptrunk.py

@@ -30,9 +30,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             ProductTable.product_type == "Device",
             SubscriptionTable.status == "active",
         )
-        .with_entities(
-            SubscriptionTable.subscription_id, SubscriptionTable.description
-        )
+        .with_entities(SubscriptionTable.subscription_id, SubscriptionTable.description)
         .all()
     ):
         devices[str(device_id)] = device_description
@@ -52,8 +50,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     class AeMembersListA(UniqueConstrainedList[str]):
         min_items = initial_user_input.iptrunk_minimum_links
 
-    DeviceEnumA = Choice("Select a device", zip(devices.keys(),
-                                                devices.items()))
+    DeviceEnumA = Choice("Select a device", zip(devices.keys(), devices.items()))
 
     class CreateIptrunkSideAForm(FormPage):
         class Config:
@@ -69,8 +66,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("Select a device", zip(devices.keys(),
-                                                devices.items()))
+    DeviceEnumB = Choice("Select a device", zip(devices.keys(), devices.items()))
 
     class AeMembersListB(UniqueConstrainedList[str]):
         min_items = len(user_input_side_a.iptrunk_sideA_ae_members)
@@ -88,11 +84,7 @@ 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")
@@ -108,18 +100,14 @@ def create_subscription(product: UUIDstr) -> State:
 @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
-    )
+    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}
 
 
@@ -149,179 +137,107 @@ 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_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_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_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 = IptrunkProvisioning.from_other_lifecycle(
-        subscription, SubscriptionLifecycle.PROVISIONING
-    )
+    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"
-    )
+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": (
-            "Provision of the Trunk interface [DRY] "
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the Trunk interface [DRY] " "Please refresh to get the results of the playbook"),
     }
 
 
 @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
-    )
+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": (
-            "Provision of the Trunk interface [REAL] "
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the Trunk interface [REAL] " "Please refresh to get the results of the playbook"),
     }
 
 
 @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"
-    )
+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": (
-            "Provision of the ISIS interface [DRY]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the ISIS interface [DRY]" "Please refresh to get the results of the playbook"),
     }
 
 
 @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
-    )
+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": (
-            "Provision of the ISIS interface [REAL]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the ISIS interface [REAL]" "Please refresh to get the results of the playbook"),
     }
 
 
 @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"
-    )
+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": (
-            "Provision of the LDP interface [DRY]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LDP interface [DRY]" "Please refresh to get the results of the playbook"),
     }
 
 
 @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
-    )
+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": (
-            "Provision of the LDP interface [REAL]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LDP interface [REAL]" "Please refresh to get the results of the playbook"),
     }
 
 
 @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"
-    )
+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": (
-            "Provision of the LLDP interface [DRY]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LLDP interface [DRY]" "Please refresh to get the results of the playbook"),
     }
 
 
 @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
-    )
+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": (
-            "Provision of the LLDP interface [REAL]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LLDP interface [REAL]" "Please refresh to get the results of the playbook"),
     }
 
 
 @workflow(
     "Create IP trunk",
-    initial_input_form=wrap_create_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
     target=Target.CREATE,
 )
 def create_iptrunk():

gso/workflows/iptrunk/modify_generic.py

@@ -24,15 +24,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         iptrunk_type: IptrunkType = subscription.iptrunk.iptrunk_type
         iptrunk_speed: PhyPortCapacity = subscription.iptrunk.iptrunk_speed
         iptrunk_minimum_links: int = subscription.iptrunk.iptrunk_minimum_links
-        iptrunk_isis_metric: int = ReadOnlyField(
-            subscription.iptrunk.iptrunk_isis_metric
-        )
-        iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(
-            subscription.iptrunk.iptrunk_ipv4_network
-        )
-        iptrunk_ipv6_network: ipaddress.IPv6Network = ReadOnlyField(
-            subscription.iptrunk.iptrunk_ipv6_network
-        )
+        iptrunk_isis_metric: int = ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric)
+        iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv4_network)
+        iptrunk_ipv6_network: ipaddress.IPv6Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv6_network)
 
     initial_user_input = yield ModifyIptrunkForm
 
@@ -43,18 +37,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         class Config:
             title = "Provide subscription details for side A of the trunk."
 
-        iptrunk_sideA_node: str = ReadOnlyField(
-            subscription.iptrunk.iptrunk_sideA_node.device_fqdn
-        )
-        iptrunk_sideA_ae_iface: str = ReadOnlyField(
-            subscription.iptrunk.iptrunk_sideA_ae_iface
-        )
-        iptrunk_sideA_ae_geant_a_sid: str = (
-            subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid
-        )
-        iptrunk_sideA_ae_members: AeMembersListA = (
-            subscription.iptrunk.iptrunk_sideA_ae_members
-        )
+        iptrunk_sideA_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideA_node.device_fqdn)
+        iptrunk_sideA_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideA_ae_iface)
+        iptrunk_sideA_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid
+        iptrunk_sideA_ae_members: AeMembersListA = subscription.iptrunk.iptrunk_sideA_ae_members
         iptrunk_sideA_ae_members_descriptions: AeMembersListA = (
             subscription.iptrunk.iptrunk_sideA_ae_members_description
         )
@@ -69,29 +55,17 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         class Config:
             title = "Provide subscription details for side B of the trunk."
 
-        iptrunk_sideB_node: str = ReadOnlyField(
-            subscription.iptrunk.iptrunk_sideB_node.device_fqdn
-        )
-        iptrunk_sideB_ae_iface: str = ReadOnlyField(
-            subscription.iptrunk.iptrunk_sideB_ae_iface
-        )
-        iptrunk_sideB_ae_geant_a_sid: str = (
-            subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid
-        )
-        iptrunk_sideB_ae_members: AeMembersListB = (
-            subscription.iptrunk.iptrunk_sideB_ae_members
-        )
+        iptrunk_sideB_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideB_node.device_fqdn)
+        iptrunk_sideB_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sideB_ae_iface)
+        iptrunk_sideB_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid
+        iptrunk_sideB_ae_members: AeMembersListB = subscription.iptrunk.iptrunk_sideB_ae_members
         iptrunk_sideB_ae_members_descriptions: AeMembersListB = (
             subscription.iptrunk.iptrunk_sideB_ae_members_description
         )
 
     user_input_side_b = yield ModifyIptrunkSideBForm
 
-    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("Update subscription")
@@ -115,21 +89,13 @@ def modify_iptrunk_subscription(
     subscription.iptrunk.iptrunk_speed = iptrunk_speed
     subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
 
-    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_sideA_ae_members_description = iptrunk_sideA_ae_members_descriptions
 
-    subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid = (
-        iptrunk_sideB_ae_geant_a_sid
-    )
+    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}"
 
@@ -137,78 +103,48 @@ def modify_iptrunk_subscription(
 
 
 @step("Provision IP trunk interface [DRY RUN]")
-def provision_ip_trunk_iface_dry(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "trunk_interface"
-    )
+def provision_ip_trunk_iface_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "trunk_interface")
 
     return {
         "subscription": subscription,
-        "label_text": (
-            "Provision of the Trunk interface [DRY] "
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the Trunk interface [DRY] " "Please refresh to get the results of the playbook"),
     }
 
 
 @step("Provision IP trunk interface [FOR REAL]")
-def provision_ip_trunk_iface_real(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "trunk_interface", False
-    )
+def provision_ip_trunk_iface_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "trunk_interface", False)
 
     return {
         "subscription": subscription,
-        "label_text": (
-            "Provision of the Trunk interface [REAL] "
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the Trunk interface [REAL] " "Please refresh to get the results of the playbook"),
     }
 
 
 @step("Provision IP trunk LLDP interface [DRY RUN]")
-def provision_ip_trunk_lldp_iface_dry(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "lldp_interface"
-    )
+def provision_ip_trunk_lldp_iface_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "lldp_interface")
 
     return {
         "subscription": subscription,
-        "label_text": (
-            "Provision of the LLDP interface [DRY]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LLDP interface [DRY]" "Please refresh to get the results of the playbook"),
     }
 
 
 @step("Provision IP trunk LLDP interface [FOR REAL]")
-def provision_ip_trunk_lldp_iface_real(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "lldp_interface", False
-    )
+def provision_ip_trunk_lldp_iface_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "lldp_interface", False)
 
     return {
         "subscription": subscription,
-        "label_text": (
-            "Provision of the LLDP interface [REAL]"
-            "Please refresh to get the results of the playbook"
-        ),
+        "label_text": ("Provision of the LLDP interface [REAL]" "Please refresh to get the results of the playbook"),
     }
 
 
 @workflow(
     "Modify IP trunk",
-    initial_input_form=wrap_modify_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
     target=Target.MODIFY,
 )
 def modify_generic():

gso/workflows/iptrunk/modify_isis_metric.py

@@ -22,21 +22,15 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
 
 @step("Update subscription")
-def modify_iptrunk_subscription(
-    subscription: Iptrunk, isis_metric: int
-) -> State:
+def modify_iptrunk_subscription(subscription: Iptrunk, isis_metric: int) -> State:
     subscription.iptrunk.iptrunk_isis_metric = isis_metric
 
     return {"subscription": subscription}
 
 
 @step("Provision IP trunk ISIS interface [DRY RUN]")
-def provision_ip_trunk_isis_iface_dry(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "isis_interface"
-    )
+def provision_ip_trunk_isis_iface_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "isis_interface")
 
     return {
         "subscription": subscription,
@@ -51,12 +45,8 @@ def provision_ip_trunk_isis_iface_dry(
 
 
 @step("Provision IP trunk ISIS interface [FOR REAL]")
-def provision_ip_trunk_isis_iface_real(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "isis_interface", False
-    )
+def provision_ip_trunk_isis_iface_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "isis_interface", False)
 
     return {
         "subscription": subscription,
@@ -72,9 +62,7 @@ def provision_ip_trunk_isis_iface_real(
 
 @workflow(
     "Modify IP trunk",
-    initial_input_form=wrap_modify_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
     target=Target.MODIFY,
 )
 def modify_isis_metric():

gso/workflows/iptrunk/terminate_iptrunk.py

@@ -18,9 +18,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> InputForm:
     subscription = Iptrunk.from_subscription(subscription_id)
 
     class TerminateForm(FormPage):
-        are_you_sure: Label = (
-            f"Are you sure you want to remove {subscription.description}?"
-        )
+        are_you_sure: Label = f"Are you sure you want to remove {subscription.description}?"
 
     return TerminateForm
 
@@ -33,12 +31,8 @@ def modify_iptrunk_subscription(subscription: Iptrunk) -> State:
 
 
 @step("Drain traffic from trunk")
-def drain_traffic_from_ip_trunk(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
-    provisioning_proxy.provision_ip_trunk(
-        subscription, process_id, "isis_interface", False
-    )
+def drain_traffic_from_ip_trunk(subscription: Iptrunk, process_id: UUIDstr) -> State:
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, "isis_interface", False)
     return {
         "subscription": subscription,
         "label_text": "This is setting the ISIS metric of the trunk to 9000"
@@ -49,38 +43,29 @@ def drain_traffic_from_ip_trunk(
 
 
 @step("Deprovision IP trunk [DRY RUN]")
-def deprovision_ip_trunk_dry(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
+def deprovision_ip_trunk_dry(subscription: Iptrunk, process_id: UUIDstr) -> State:
     provisioning_proxy.deprovision_ip_trunk(subscription, process_id, True)
 
     return {
         "subscription": subscription,
-        "label_text": "This is a dry run for the termination of an IP "
-        "trunk. "
-        "Press refresh to get the results",
+        "label_text": "This is a dry run for the termination of an IP " "trunk. " "Press refresh to get the results",
     }
 
 
 @step("Deprovision IP trunk [FOR REAL]")
-def deprovision_ip_trunk_real(
-    subscription: Iptrunk, process_id: UUIDstr
-) -> State:
+def deprovision_ip_trunk_real(subscription: Iptrunk, process_id: UUIDstr) -> State:
     provisioning_proxy.deprovision_ip_trunk(subscription, process_id, False)
 
     return {
         "subscription": subscription,
-        "label_text": "This is a termination of an IP trunk. "
-        "Press refresh to get the results",
+        "label_text": "This is a termination of an IP trunk. " "Press refresh to get the results",
     }
 
 
 @step("Deprovision IPv4 networks")
 def deprovision_ip_trunk_ipv4(subscription: Iptrunk) -> None:
     service_network = ipam.delete_service_network(
-        network=ipaddress.ip_network(
-            subscription.iptrunk.iptrunk_ipv4_network
-        ),
+        network=ipaddress.ip_network(subscription.iptrunk.iptrunk_ipv4_network),
         service_type="TRUNK",
     )
     return {"service_network": service_network}
@@ -89,9 +74,7 @@ def deprovision_ip_trunk_ipv4(subscription: Iptrunk) -> None:
 @step("Deprovision IPv6 networks")
 def deprovision_ip_trunk_ipv6(subscription: Iptrunk) -> None:
     service_network = ipam.delete_service_network(
-        network=ipaddress.ip_network(
-            subscription.iptrunk.iptrunk_ipv6_network
-        ),
+        network=ipaddress.ip_network(subscription.iptrunk.iptrunk_ipv6_network),
         service_type="TRUNK",
     )
     return {"service_network": service_network}
@@ -99,9 +82,7 @@ def deprovision_ip_trunk_ipv6(subscription: Iptrunk) -> None:
 
 @workflow(
     "Terminate IPtrunk",
-    initial_input_form=wrap_modify_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
     target=Target.TERMINATE,
 )
 def terminate_iptrunk():

gso/workflows/site/create_site.py

@@ -66,18 +66,14 @@ def initialize_subscription(
 
     subscription.description = f"Site {site_name}"
 
-    subscription = site.SiteProvisioning.from_other_lifecycle(
-        subscription, SubscriptionLifecycle.PROVISIONING
-    )
+    subscription = site.SiteProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
 
     return {"subscription": subscription}
 
 
 @workflow(
     "Create Site",
-    initial_input_form=wrap_create_initial_input_form(
-        initial_input_form_generator
-    ),
+    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
     target=Target.CREATE,
 )
 def create_site():