diff --git a/gso/api/v1/imports.py b/gso/api/v1/imports.py
index de08184fb0d3a6cfbc29ff6228fc5b921593f6b8..044dce8352c9b82f06f8c4c9a98fc896c927cf02 100644
--- a/gso/api/v1/imports.py
+++ b/gso/api/v1/imports.py
@@ -15,7 +15,7 @@ from gso.products.product_blocks.router import RouterRole
from gso.services import subscriptions
from gso.services.partners import PartnerNotFoundError, get_partner_by_name
from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
@@ -42,8 +42,8 @@ class RouterImportModel(BaseModel):
ts_port: int
router_vendor: Vendor
router_role: RouterRole
- router_lo_ipv4_address: IPV4AddressType
- router_lo_ipv6_address: IPV6AddressType
+ router_lo_ipv4_address: IPv4AddressType
+ router_lo_ipv6_address: IPv6AddressType
router_lo_iso_address: str
@@ -129,7 +129,7 @@ class SuperPopSwitchImportModel(BaseModel):
super_pop_switch_site: str
hostname: str
super_pop_switch_ts_port: PortNumber
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType
class OfficeRouterImportModel(BaseModel):
@@ -139,8 +139,8 @@ class OfficeRouterImportModel(BaseModel):
office_router_site: str
office_router_fqdn: str
office_router_ts_port: PortNumber
- office_router_lo_ipv4_address: IPV4AddressType
- office_router_lo_ipv6_address: IPV6AddressType
+ office_router_lo_ipv4_address: IPv4AddressType
+ office_router_lo_ipv6_address: IPv6AddressType
def _start_process(process_name: str, data: dict) -> UUID:
diff --git a/gso/monkeypatches.py b/gso/monkeypatches.py
index b5c73b558b8eed81297c36f02909e5f316ceff5f..2e94f50bdd27288e4ce7d829036ffbc8f022ef20 100644
--- a/gso/monkeypatches.py
+++ b/gso/monkeypatches.py
@@ -4,12 +4,8 @@ This adjustment is typically done to extend or modify the functionality of the o
oauth2_lib package to meet specific requirements of the gso application.
"""
-from datetime import datetime
-from ipaddress import IPv4Address, IPv6Address
-
import oauth2_lib.fastapi
import oauth2_lib.settings
-from pydantic import BaseModel
from gso.auth.oidc_policy_helper import HTTPX_SSL_CONTEXT, OIDCUser, OIDCUserModel, opa_decision
from gso.auth.settings import oauth2lib_settings
diff --git a/gso/products/product_blocks/iptrunk.py b/gso/products/product_blocks/iptrunk.py
index de68b83ba5feb8dd33c06dfa791ab07a6ff3ccb8..2d12b9c55924844338ce68d2af0a4a975e7b7c89 100644
--- a/gso/products/product_blocks/iptrunk.py
+++ b/gso/products/product_blocks/iptrunk.py
@@ -1,7 +1,7 @@
"""IP trunk product block that has all parameters of a subscription throughout its lifecycle."""
import ipaddress
-from typing import Annotated, Sequence, TypeVar, Generic
+from typing import Annotated
from annotated_types import Len
from orchestrator.domain.base import ProductBlockModel, T
diff --git a/gso/products/product_blocks/lan_switch_interconnect.py b/gso/products/product_blocks/lan_switch_interconnect.py
index 55a7b97a3ca788dbe3e61f5ddd9c8a5344e5976f..a9b1c77366662e24499f936f9762d515b882b934 100644
--- a/gso/products/product_blocks/lan_switch_interconnect.py
+++ b/gso/products/product_blocks/lan_switch_interconnect.py
@@ -63,7 +63,7 @@ class LanSwitchInterconnectRouterSideBlockProvisioning(
node: RouterBlockProvisioning
ae_iface: str | None = None
- ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+ ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning] # type: ignore[assignment]
class LanSwitchInterconnectRouterSideBlock(
@@ -73,7 +73,7 @@ class LanSwitchInterconnectRouterSideBlock(
node: RouterBlock
ae_iface: str
- ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+ ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock] # type: ignore[assignment]
class LanSwitchInterconnectSwitchSideBlockInactive(
@@ -95,7 +95,7 @@ class LanSwitchInterconnectSwitchSideBlockProvisioning(
node: SwitchBlockProvisioning
ae_iface: str | None = None
- ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+ ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning] # type: ignore[assignment]
class LanSwitchInterconnectSwitchSideBlock(
@@ -105,7 +105,7 @@ class LanSwitchInterconnectSwitchSideBlock(
node: SwitchBlock
ae_iface: str
- ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+ ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock] # type: ignore[assignment]
class LanSwitchInterconnectBlockInactive(
diff --git a/gso/products/product_blocks/office_router.py b/gso/products/product_blocks/office_router.py
index d5bcf1a3b4644a869fb1383a6ad1fd46dad8a5b3..65eab0256a073c699f3ea2ef84d96e3352096722 100644
--- a/gso/products/product_blocks/office_router.py
+++ b/gso/products/product_blocks/office_router.py
@@ -8,7 +8,7 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
class OfficeRouterBlockInactive(
@@ -20,8 +20,8 @@ class OfficeRouterBlockInactive(
office_router_fqdn: str | None = None
office_router_ts_port: PortNumber | None = None
- office_router_lo_ipv4_address: IPV4AddressType | None = None
- office_router_lo_ipv6_address: IPV6AddressType | None = None
+ office_router_lo_ipv4_address: IPv4AddressType | None = None
+ office_router_lo_ipv6_address: IPv6AddressType | None = None
office_router_site: SiteBlockInactive | None
vendor: Vendor | None = None
@@ -31,8 +31,8 @@ class OfficeRouterBlockProvisioning(OfficeRouterBlockInactive, lifecycle=[Subscr
office_router_fqdn: str | None = None
office_router_ts_port: PortNumber | None = None
- office_router_lo_ipv4_address: IPV4AddressType | None = None
- office_router_lo_ipv6_address: IPV6AddressType | None = None
+ office_router_lo_ipv4_address: IPv4AddressType | None = None
+ office_router_lo_ipv6_address: IPv6AddressType | None = None
office_router_site: SiteBlockProvisioning | None
vendor: Vendor | None = None
@@ -45,9 +45,9 @@ class OfficeRouterBlock(OfficeRouterBlockProvisioning, lifecycle=[SubscriptionLi
#: The port of the terminal server that this office router is connected to. Used to offer out of band access.
office_router_ts_port: PortNumber
#: The IPv4 loopback address of the office router.
- office_router_lo_ipv4_address: IPV4AddressType
+ office_router_lo_ipv4_address: IPv4AddressType
#: The IPv6 loopback address of the office router.
- office_router_lo_ipv6_address: IPV6AddressType
+ office_router_lo_ipv6_address: IPv6AddressType
#: The :class:`Site` that this office router resides in. Both physically and computationally.
office_router_site: SiteBlock
#: The vendor of an office router. Defaults to Juniper.
diff --git a/gso/products/product_blocks/pop_vlan.py b/gso/products/product_blocks/pop_vlan.py
index e556ad1e9903772f7fb58a38a4047286595a1fb8..9942a97991ed4ea0e72f3ea2ad5f95b5fc580e83 100644
--- a/gso/products/product_blocks/pop_vlan.py
+++ b/gso/products/product_blocks/pop_vlan.py
@@ -1,7 +1,7 @@
"""Pop VLAN product block that has all parameters of a subscription throughout its lifecycle."""
from ipaddress import IPv4Network, IPv6Network
-from typing import TypeVar, Annotated
+from typing import Annotated, TypeVar
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
@@ -92,7 +92,7 @@ class PopVlanBlock(PopVlanBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
#: The LAN Switch Interconnect that this Pop VLAN is connected to.
lan_switch_interconnect: LanSwitchInterconnectBlock
#: The ports of the Pop VLAN.
- ports: PortList[PopVlanPortBlock]
+ ports: PortList[PopVlanPortBlock] # type: ignore[assignment]
#: The level of the layer preference for the Pop VLAN (L2 or L3).
layer_preference: LayerPreference
#: IPv4 network for the Pop VLAN if layer preference is L3.
diff --git a/gso/products/product_blocks/router.py b/gso/products/product_blocks/router.py
index 85e940a015ceae354efb269569dc3fb8b1539abe..17deeccb1ac8a5ee9bcfaa14fa25f27360881e7c 100644
--- a/gso/products/product_blocks/router.py
+++ b/gso/products/product_blocks/router.py
@@ -8,7 +8,7 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
class RouterRole(strEnum):
@@ -29,8 +29,8 @@ class RouterBlockInactive(
router_fqdn: str | None = None
router_ts_port: PortNumber | None = None
router_access_via_ts: bool | None = None
- router_lo_ipv4_address: IPV4AddressType | None = None
- router_lo_ipv6_address: IPV6AddressType | None = None
+ router_lo_ipv4_address: IPv4AddressType | None = None
+ router_lo_ipv6_address: IPv6AddressType | None = None
router_lo_iso_address: str | None = None
router_role: RouterRole | None = None
router_site: SiteBlockInactive | None
@@ -43,8 +43,8 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
router_fqdn: str
router_ts_port: PortNumber
router_access_via_ts: bool
- router_lo_ipv4_address: IPV4AddressType
- router_lo_ipv6_address: IPV6AddressType
+ router_lo_ipv4_address: IPv4AddressType
+ router_lo_ipv6_address: IPv6AddressType
router_lo_iso_address: str
router_role: RouterRole
router_site: SiteBlockProvisioning
@@ -61,9 +61,9 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
#: Whether this router should be accessed through the terminal server, or through its loopback address.
router_access_via_ts: bool
#: The IPv4 loopback address of the router.
- router_lo_ipv4_address: IPV4AddressType
+ router_lo_ipv4_address: IPv4AddressType
#: The IPv6 loopback address of the router.
- router_lo_ipv6_address: IPV6AddressType
+ router_lo_ipv6_address: IPv6AddressType
#: The :term:`ISO` :term:`NET` of the router, used for :term:`ISIS` support.
router_lo_iso_address: str
#: The role of the router, which can be any of the values defined in :class:`RouterRole`.
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index 1f108a410fe3a39ae8c81eb653ce8ce39db842d3..51a7e168979ba2e6d4df241c221b9f111b5e7a98 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -1,7 +1,7 @@
"""The product block that describes a site subscription."""
import re
-from typing import Annotated, TypeAlias
+from typing import Annotated
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
diff --git a/gso/products/product_blocks/super_pop_switch.py b/gso/products/product_blocks/super_pop_switch.py
index 1b4adafa3854b68277a6036969889d939548f030..3335b28cf90ee9d55abe59be528f404d44d905b8 100644
--- a/gso/products/product_blocks/super_pop_switch.py
+++ b/gso/products/product_blocks/super_pop_switch.py
@@ -8,7 +8,7 @@ from gso.products.product_blocks.site import (
SiteBlockInactive,
SiteBlockProvisioning,
)
-from gso.utils.shared_enums import IPV4AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
class SuperPopSwitchBlockInactive(
@@ -20,7 +20,7 @@ class SuperPopSwitchBlockInactive(
super_pop_switch_fqdn: str | None = None
super_pop_switch_ts_port: PortNumber | None = None
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType | None = None
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType | None = None
super_pop_switch_site: SiteBlockInactive | None
vendor: Vendor | None = None
@@ -30,7 +30,7 @@ class SuperPopSwitchBlockProvisioning(SuperPopSwitchBlockInactive, lifecycle=[Su
super_pop_switch_fqdn: str | None = None
super_pop_switch_ts_port: PortNumber | None = None
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType | None = None
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType | None = None
super_pop_switch_site: SiteBlockProvisioning | None
vendor: Vendor | None = None
@@ -43,7 +43,7 @@ class SuperPopSwitchBlock(SuperPopSwitchBlockProvisioning, lifecycle=[Subscripti
#: The port of the terminal server that this Super PoP switch is connected to. Used to offer out of band access.
super_pop_switch_ts_port: PortNumber
#: The IPv4 management address of the Super PoP switch.
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType
#: The :class:`Site` that this Super PoP switch resides in. Both physically and computationally.
super_pop_switch_site: SiteBlock
#: The vendor of a Super PoP switch. Defaults to Juniper.
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index 1645773fe7063c682a5428a13fa58ed8950ef755..140b4c851b29b649a05243e9f12c44e729debc03 100644
--- a/gso/services/infoblox.py
+++ b/gso/services/infoblox.py
@@ -10,7 +10,7 @@ from infoblox_client.exceptions import (
)
from gso.settings import IPAMParams, load_oss_params
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType
logger = getLogger(__name__)
NULL_MAC = "00:00:00:00:00:00"
@@ -234,8 +234,8 @@ def allocate_host(
def create_host_by_ip(
hostname: str,
- ipv4_address: IPV4AddressType,
- ipv6_address: IPV6AddressType,
+ ipv4_address: IPv4AddressType,
+ ipv6_address: IPv6AddressType,
service_type: str,
comment: str,
) -> None:
@@ -269,7 +269,7 @@ def create_host_by_ip(
new_host.update()
-def find_host_by_ip(ip_addr: IPV4AddressType | ipaddress.IPv6Address) -> objects.HostRecord | None:
+def find_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> objects.HostRecord | None:
"""Find a host record in Infoblox by its associated IP address.
:param ip_addr: The IP address of a host that is searched for.
@@ -315,7 +315,7 @@ def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
)
-def delete_host_by_ip(ip_addr: IPV4AddressType | ipaddress.IPv6Address) -> None:
+def delete_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> None:
"""Delete a host from Infoblox.
Delete a host record in Infoblox, by providing the IP address that is associated with the record. Raises a
diff --git a/gso/services/lso_client.py b/gso/services/lso_client.py
index f707cba2e4d251cf1ee7d7cc017c6cc534505edb..92898940edfdb7a2981156a1fd5e0f5c5a44f786 100644
--- a/gso/services/lso_client.py
+++ b/gso/services/lso_client.py
@@ -130,8 +130,8 @@ def _show_results(state: State) -> FormGenerator:
if "lso_result_extra_label" in state:
extra_label: Label = state["lso_result_extra_label"]
- run_status: ReadOnlyField(state["callback_result"]["status"], default_type=str) # type: ignore[valid-type]
- run_results: ReadOnlyField(json.dumps(state["callback_result"], indent=4), default_type=LongText) # type: ignore[valid-type]
+ run_status: ReadOnlyField(state["callback_result"]["status"], default_type=str) # type: ignore[valid-type]
+ run_results: ReadOnlyField(json.dumps(state["callback_result"], indent=4), default_type=LongText) # type: ignore[valid-type]
yield ConfirmRunPage
[state.pop(key, None) for key in ["run_results", "lso_result_title", "lso_result_extra_label"]]
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index a0c71bb413dd430d30cf5171827847bd1f60fd1a..1daa2d81a1f0c66debfca6bc4fc35fefa33192bb 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -16,7 +16,7 @@ from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordi
from gso.products.product_types.router import Router
from gso.services.netbox_client import NetboxClient
from gso.services.subscriptions import get_active_subscriptions_by_field_and_value
-from gso.utils.shared_enums import IPV4AddressType, Vendor
+from gso.utils.shared_enums import IPv4AddressType, Vendor
class LAGMember(BaseModel):
@@ -105,7 +105,7 @@ def get_router_vendor(router_id: UUID) -> Vendor:
return Router.from_subscription(router_id).router.vendor
-def iso_from_ipv4(ipv4_address: IPV4AddressType) -> str:
+def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
"""Calculate an :term:`ISO` address, based on an IPv4 address.
:param IPv4Address ipv4_address: The address that's to be converted
diff --git a/gso/utils/shared_enums.py b/gso/utils/shared_enums.py
index 86865ba70b9612be6595304d93e23a1006dee941..5ca0657b5e8509ac0604014329e7ddececd66788 100644
--- a/gso/utils/shared_enums.py
+++ b/gso/utils/shared_enums.py
@@ -27,11 +27,11 @@ PortNumber = Annotated[
]
-IPV4AddressType = Annotated[
+IPv4AddressType = Annotated[
ipaddress.IPv4Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
]
-IPV6AddressType = Annotated[
+IPv6AddressType = Annotated[
ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
]
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index 1e89ec894bf9b3029484db458b6ea19e592c6c55..8fe29e3cc372a439befaa176c09f420b5fc7254d 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -10,19 +10,17 @@ ALL_ALIVE_STATES: list[str] = [
SubscriptionLifecycle.ACTIVE,
]
-WF_USABLE_MAP.update(
- {
- "redeploy_base_config": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
- "update_ibgp_mesh": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
- "activate_router": [SubscriptionLifecycle.PROVISIONING],
- "deploy_twamp": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
- "modify_trunk_interface": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
- "activate_iptrunk": [SubscriptionLifecycle.PROVISIONING],
- "terminate_site": ALL_ALIVE_STATES,
- "terminate_router": ALL_ALIVE_STATES,
- "terminate_iptrunk": ALL_ALIVE_STATES,
- }
-)
+WF_USABLE_MAP.update({
+ "redeploy_base_config": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+ "update_ibgp_mesh": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+ "activate_router": [SubscriptionLifecycle.PROVISIONING],
+ "deploy_twamp": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+ "modify_trunk_interface": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+ "activate_iptrunk": [SubscriptionLifecycle.PROVISIONING],
+ "terminate_site": ALL_ALIVE_STATES,
+ "terminate_router": ALL_ALIVE_STATES,
+ "terminate_iptrunk": ALL_ALIVE_STATES,
+})
LazyWorkflowInstance("gso.workflows.iptrunk.activate_iptrunk", "activate_iptrunk")
LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 71f5453bb196c7d79ad67ea1ca0fe5fc97b656f4..99e9f24879ea22c2261ac0fe369859269994710e 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -1,7 +1,7 @@
"""A creation workflow that deploys a new IP trunk service."""
import json
-from typing import Annotated, TypeAlias
+from typing import Annotated
from uuid import uuid4
from annotated_types import Len
@@ -57,7 +57,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
model_config = ConfigDict(title=product_name)
tt_number: str
- partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
+ partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
geant_s_sid: str | None
iptrunk_description: str
iptrunk_type: IptrunkType
@@ -72,9 +72,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class VerifyMinimumLinksForm(FormPage):
info_label: Label = (
- f"This is the calculated minimum-links for this LAG: " f"{initial_user_input.iptrunk_number_of_members - 1}" # type: ignore[assignment]
+ f"This is the calculated minimum-links for this LAG: " f"{initial_user_input.iptrunk_number_of_members - 1}"
)
- info_label2: Label = "Please confirm or modify." # type: ignore[assignment]
+ info_label2: Label = "Please confirm or modify."
yield VerifyMinimumLinksForm
router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
@@ -92,7 +92,14 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
router_a = user_input_router_side_a.side_a_node_id.name
router_a_fqdn = Router.from_subscription(router_a).router.router_fqdn
- JuniperAeMembers = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_number_of_members, max_length=initial_user_input.iptrunk_number_of_members)]
+ juniper_ae_members = Annotated[
+ list[LAGMember],
+ AfterValidator(validate_unique_list),
+ Len(
+ min_length=initial_user_input.iptrunk_number_of_members,
+ max_length=initial_user_input.iptrunk_number_of_members,
+ ),
+ ]
if get_router_vendor(router_a) == Vendor.NOKIA:
@@ -102,16 +109,23 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
initial_user_input.iptrunk_speed,
)
- ae_members_side_a = Annotated[list[NokiaLAGMemberA], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_number_of_members, max_length=initial_user_input.iptrunk_number_of_members)]
+ ae_members_side_a_type = Annotated[
+ list[NokiaLAGMemberA],
+ AfterValidator(validate_unique_list),
+ Len(
+ min_length=initial_user_input.iptrunk_number_of_members,
+ max_length=initial_user_input.iptrunk_number_of_members,
+ ),
+ ]
else:
- ae_members_side_a_type = JuniperAeMembers # type: ignore[assignment]
+ ae_members_side_a_type = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideAForm(FormPage):
model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk.({router_a_fqdn})")
side_a_ae_iface: available_lags_choices(router_a) or str # type: ignore[valid-type]
side_a_ae_geant_a_sid: str | None
- side_a_ae_members: ae_members_side_a # type: ignore[valid-type]
+ side_a_ae_members: ae_members_side_a_type
@field_validator("side_a_ae_members")
def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
@@ -154,14 +168,14 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
),
]
else:
- ae_members_side_b = JuniperAeMembers
+ ae_members_side_b = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideBForm(FormPage):
model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk.({router_b_fqdn})")
side_b_ae_iface: available_lags_choices(router_b) or str # type: ignore[valid-type]
side_b_ae_geant_a_sid: str | None
- side_b_ae_members: ae_members_side_b # type: ignore[valid-type]
+ side_b_ae_members: ae_members_side_b
@field_validator("side_b_ae_members")
def validate_side_b_ae_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
@@ -323,7 +337,7 @@ def check_ip_trunk_connectivity(
execute_playbook(
playbook_name="iptrunks_checks.yaml",
callback_route=callback_route,
- inventory=subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn,
+ inventory=subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, # type: ignore[arg-type]
extra_vars=extra_vars,
)
@@ -397,7 +411,7 @@ def check_ip_trunk_isis(
execute_playbook(
playbook_name="iptrunks_checks.yaml",
callback_route=callback_route,
- inventory=subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn,
+ inventory=subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, # type: ignore[arg-type]
extra_vars=extra_vars,
)
@@ -428,9 +442,9 @@ def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == Vendor.NOKIA:
# Create :term:`LAG` interfaces
lag_interface: Interfaces = nbclient.create_interface(
- iface_name=trunk_side.iptrunk_side_ae_iface,
+ iface_name=trunk_side.iptrunk_side_ae_iface, # type: ignore[arg-type]
interface_type="lag",
- device_name=trunk_side.iptrunk_side_node.router_fqdn,
+ device_name=trunk_side.iptrunk_side_node.router_fqdn, # type: ignore[arg-type]
description=str(subscription.subscription_id),
enabled=True,
)
@@ -439,13 +453,13 @@ def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
# Reserve interfaces
for interface in trunk_side.iptrunk_side_ae_members:
nbclient.attach_interface_to_lag(
- device_name=trunk_side.iptrunk_side_node.router_fqdn,
+ device_name=trunk_side.iptrunk_side_node.router_fqdn, # type: ignore[arg-type]
lag_name=lag_interface.name,
- iface_name=interface.interface_name,
+ iface_name=interface.interface_name, # type: ignore[arg-type]
description=str(subscription.subscription_id),
)
nbclient.reserve_interface(
- device_name=trunk_side.iptrunk_side_node.router_fqdn,
+ device_name=trunk_side.iptrunk_side_node.router_fqdn, # type: ignore[arg-type]
iface_name=interface.interface_name, # type: ignore[arg-type]
)
return {
@@ -485,7 +499,7 @@ def prompt_start_new_checklist(subscription: IptrunkProvisioning) -> FormGenerat
model_config = ConfigDict(title="Start new checklist")
info_label_1: Label = (
- f"Visit {oss_params.SHAREPOINT.checklist_site_url} and start a new Sharepoint checklist for an IPtrunk "
+ f"Visit {oss_params.SHAREPOINT.checklist_site_url} and start a new Sharepoint checklist for an IPtrunk "
f"from {subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn} to "
f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}."
)
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 0a30e01079383504d659324e1e11ccfb793b1594..16dc3f74a1eec22146481872a7627ad0fede7d32 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -125,7 +125,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
),
]
else:
- ae_members = Annotated[
+ ae_members = Annotated[ # type: ignore[assignment, misc]
list[LAGMember],
AfterValidator(validate_unique_list),
Len(
@@ -152,7 +152,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
model_config = ConfigDict(title=form_title)
new_lag_interface: side_a_ae_iface # type: ignore[valid-type]
- existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=list[LAGMember]) # type: ignore[valid-type]
+ existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=list[LAGMember]) # type: ignore[valid-type]
new_lag_member_interfaces: ae_members
@field_validator("new_lag_interface")
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 3395774c577957fcbf1bdfa5a08a9de6d23233ee..234a9dfc22f5f8ca56e70022eba234517ef6a861 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -1,7 +1,7 @@
"""A modification workflow that updates the :term:`LAG` interfaces that are part of an existing IP trunk."""
import json
-from typing import Annotated, TypeVar, Type, Any
+from typing import Annotated, TypeVar
from uuid import UUID, uuid4
from annotated_types import Len
@@ -33,12 +33,14 @@ from gso.utils.helpers import (
validate_iptrunk_unique_interface,
validate_tt_number,
)
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, Vendor
-T = TypeVar('T', bound=LAGMember)
+T = TypeVar("T", bound=LAGMember)
-def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_index: int) -> type[list[Any]]:
+def initialize_ae_members(
+ subscription: Iptrunk, initial_user_input: dict, side_index: int
+) -> Annotated[list[LAGMember], ""]:
"""Initialize the list of AE members."""
router = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_node
router_vendor = get_router_vendor(router.owner_subscription_id)
@@ -62,11 +64,17 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
)
)
- ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members)]
- else:
+ return Annotated[
+ list[NokiaLAGMember],
+ AfterValidator(validate_unique_list),
+ Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
+ ] # type: ignore[return-value]
- ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members)]
- return ae_members # type: ignore[return-value]
+ return Annotated[
+ list[LAGMember],
+ AfterValidator(validate_unique_list),
+ Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
+ ] # type: ignore[return-value]
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -84,9 +92,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
iptrunk_speed: PhysicalPortCapacity = subscription.iptrunk.iptrunk_speed
iptrunk_number_of_members: int = subscription.iptrunk.iptrunk_minimum_links + 1
- iptrunk_isis_metric: ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric, default_type=int)
- iptrunk_ipv4_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv4_network), default_type=IPV4AddressType)
- iptrunk_ipv6_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv6_network), default_type=IPV6AddressType)
+ iptrunk_isis_metric: ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric, default_type=int) # type: ignore[valid-type]
+ iptrunk_ipv4_network: ReadOnlyField( # type: ignore[valid-type]
+ str(subscription.iptrunk.iptrunk_ipv4_network), default_type=IPv4AddressType
+ )
+ iptrunk_ipv6_network: ReadOnlyField( # type: ignore[valid-type]
+ str(subscription.iptrunk.iptrunk_ipv6_network), default_type=IPv6AddressType
+ )
@field_validator("tt_number")
def validate_tt_number(cls, tt_number: str) -> str:
@@ -96,9 +108,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class VerifyMinimumLinksForm(FormPage):
info_label: Label = (
- f"This is the calculated minimum-links for this LAG: " f"{initial_user_input.iptrunk_number_of_members - 1}" # type: ignore[assignment]
+ f"This is the calculated minimum-links for this LAG: " f"{initial_user_input.iptrunk_number_of_members - 1}"
)
- info_label2: Label = "Please confirm or modify." # type: ignore[assignment]
+ info_label2: Label = "Please confirm or modify."
yield VerifyMinimumLinksForm
ae_members_side_a = initialize_ae_members(subscription, initial_user_input.dict(), 0)
@@ -106,8 +118,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class ModifyIptrunkSideAForm(FormPage):
model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")
- side_a_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str)
- side_a_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface, default_type=str)
+ side_a_node: ReadOnlyField( # type: ignore[valid-type]
+ subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str
+ )
+ side_a_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface, default_type=str) # type: ignore[valid-type]
side_a_ae_geant_a_sid: str | None = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid
side_a_ae_members: ae_members_side_a = ( # type: ignore[valid-type]
subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members
@@ -130,8 +144,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class ModifyIptrunkSideBForm(FormPage):
model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")
- side_b_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str)
- side_b_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface, default_type=str)
+ side_b_node: ReadOnlyField( # type: ignore[valid-type]
+ subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str
+ )
+ side_b_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface, default_type=str) # type: ignore[valid-type]
side_b_ae_geant_a_sid: str | None = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_geant_a_sid
side_b_ae_members: ae_members_side_b = ( # type: ignore[valid-type]
subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members
@@ -218,11 +234,11 @@ def modify_iptrunk_subscription(
@step("[DRY RUN] Provision IP trunk interface")
def provision_ip_trunk_iface_dry(
- subscription: Iptrunk,
- process_id: UUIDstr,
- callback_route: str,
- tt_number: str,
- removed_ae_members: list[str],
+ subscription: Iptrunk,
+ process_id: UUIDstr,
+ callback_route: str,
+ tt_number: str,
+ removed_ae_members: list[str],
) -> State:
"""Perform a dry run of deploying the updated IP trunk."""
extra_vars = {
@@ -232,14 +248,14 @@ def provision_ip_trunk_iface_dry(
"config_object": "trunk_interface",
"removed_ae_members": removed_ae_members,
"commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
- f"{subscription.iptrunk.geant_s_sid}",
+ f"{subscription.iptrunk.geant_s_sid}",
}
execute_playbook(
playbook_name="iptrunks.yaml",
callback_route=callback_route,
inventory=f"{subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}\n"
- f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}\n",
+ f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}\n",
extra_vars=extra_vars,
)
@@ -248,11 +264,11 @@ def provision_ip_trunk_iface_dry(
@step("[FOR REAL] Provision IP trunk interface")
def provision_ip_trunk_iface_real(
- subscription: Iptrunk,
- process_id: UUIDstr,
- callback_route: str,
- tt_number: str,
- removed_ae_members: list[str],
+ subscription: Iptrunk,
+ process_id: UUIDstr,
+ callback_route: str,
+ tt_number: str,
+ removed_ae_members: list[str],
) -> State:
"""Provision the new IP trunk with updated interfaces."""
extra_vars = {
@@ -262,14 +278,14 @@ def provision_ip_trunk_iface_real(
"config_object": "trunk_interface",
"removed_ae_members": removed_ae_members,
"commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
- f"{subscription.iptrunk.geant_s_sid}",
+ f"{subscription.iptrunk.geant_s_sid}",
}
execute_playbook(
playbook_name="iptrunks.yaml",
callback_route=callback_route,
inventory=f"{subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}\n"
- f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}\n",
+ f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}\n",
extra_vars=extra_vars,
)
@@ -277,10 +293,10 @@ def provision_ip_trunk_iface_real(
def _netbox_update_interfaces(
- subscription_id: UUID,
- side_block: IptrunkSideBlock,
- removed_ae_members: list[dict],
- previous_ae_members: list[dict],
+ subscription_id: UUID,
+ side_block: IptrunkSideBlock,
+ removed_ae_members: list[dict],
+ previous_ae_members: list[dict],
) -> None:
nbclient = NetboxClient()
@@ -310,7 +326,7 @@ def _netbox_update_interfaces(
@step("Netbox: Reserve side A interfaces")
def netbox_update_interfaces_side_a(
- subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
+ subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
) -> None:
"""Update Netbox such that it contains the new interfaces on side A."""
_netbox_update_interfaces(
@@ -323,7 +339,7 @@ def netbox_update_interfaces_side_a(
@step("Netbox: Reserve side B interfaces")
def netbox_update_interfaces_side_b(
- subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
+ subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
) -> None:
"""Update Netbox such that it contains the new interfaces on side B."""
_netbox_update_interfaces(
@@ -388,25 +404,25 @@ def modify_trunk_interface() -> StepList:
lambda state: get_router_vendor(
state["subscription"]["iptrunk"]["iptrunk_sides"][0]["iptrunk_side_node"]["owner_subscription_id"]
)
- == Vendor.NOKIA
+ == Vendor.NOKIA
)
side_b_is_nokia = conditional(
lambda state: get_router_vendor(
state["subscription"]["iptrunk"]["iptrunk_sides"][1]["iptrunk_side_node"]["owner_subscription_id"]
)
- == Vendor.NOKIA
+ == Vendor.NOKIA
)
return (
- init
- >> store_process_subscription(Target.MODIFY)
- >> unsync
- >> modify_iptrunk_subscription
- >> side_a_is_nokia(netbox_update_interfaces_side_a)
- >> side_b_is_nokia(netbox_update_interfaces_side_b)
- >> lso_interaction(provision_ip_trunk_iface_dry)
- >> lso_interaction(provision_ip_trunk_iface_real)
- >> side_a_is_nokia(allocate_interfaces_in_netbox_side_a)
- >> side_b_is_nokia(allocate_interfaces_in_netbox_side_b)
- >> resync
- >> done
+ init
+ >> store_process_subscription(Target.MODIFY)
+ >> unsync
+ >> modify_iptrunk_subscription
+ >> side_a_is_nokia(netbox_update_interfaces_side_a)
+ >> side_b_is_nokia(netbox_update_interfaces_side_b)
+ >> lso_interaction(provision_ip_trunk_iface_dry)
+ >> lso_interaction(provision_ip_trunk_iface_real)
+ >> side_a_is_nokia(allocate_interfaces_in_netbox_side_a)
+ >> side_b_is_nokia(allocate_interfaces_in_netbox_side_b)
+ >> resync
+ >> done
)
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index 6407dbddb137568141cba287bb118a20269aca56..1ae61b80cd486d46f0f0fbea139868b01dc96d15 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -35,14 +35,14 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class TerminateForm(FormPage):
if iptrunk.status == SubscriptionLifecycle.INITIAL:
info_label_2: Label = (
- "This will immediately mark the subscription as terminated, preventing any other workflows from " # type:ignore[assignment]
+ "This will immediately mark the subscription as terminated, preventing any other workflows from "
"interacting with this product subscription."
)
- info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING." # type:ignore[assignment]
+ info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
tt_number: str
termination_label: Label = (
- "Please confirm whether configuration should get removed from the A and B sides of the trunk." # type: ignore[assignment]
+ "Please confirm whether configuration should get removed from the A and B sides of the trunk."
)
remove_configuration: bool = True
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 56df7e9328960114fbe5844a549434f6e365a015..dd9b2ee9f4b877361e09dc56a8742e0efff69a96 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -42,7 +42,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
model_config = ConfigDict(title=product_name)
tt_number: str
- partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
+ partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
vendor: Vendor
router_site: _site_selector() # type: ignore[valid-type]
hostname: str
@@ -155,7 +155,7 @@ def prompt_reboot_router(subscription: RouterInactive) -> FormGenerator:
if subscription.router.router_site and subscription.router.router_site.site_ts_address:
info_label_1: Label = (
- f"Base config has been deployed. Please log in via the console using https://"
+ f"Base config has been deployed. Please log in via the console using https://"
f"{subscription.router.router_site.site_ts_address}."
)
else:
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 0d46f9abbd7a32e2f141894a09751659d24d52fa..781e75813843db8dd4330fa746cc8a5c45015206 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -34,13 +34,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class TerminateForm(FormPage):
if router.status == SubscriptionLifecycle.INITIAL:
info_label_2: Label = (
- "This will immediately mark the subscription as terminated, preventing any other workflows from " # type:ignore[assignment]
+ "This will immediately mark the subscription as terminated, preventing any other workflows from "
"interacting with this product subscription."
)
- info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING." # type:ignore[assignment]
+ info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
tt_number: str
- termination_label: Label = "Please confirm whether configuration should get removed from the router." # type: ignore[assignment]
+ termination_label: Label = "Please confirm whether configuration should get removed from the router."
remove_configuration: bool = True
user_input = yield TerminateForm
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 6c799e2056634a0ee11e73fa155716ed0b6e3ce8..0aa61cb65a09b64c5d4821b92c4fd85c92a2ef0a 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -1,6 +1,6 @@
"""Update iBGP mesh workflow. Adds a new P router to the mesh of PE routers in the network."""
-from typing import Any, Self
+from typing import Any
from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
@@ -34,7 +34,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
tt_number: str
@model_validator(mode="before")
- def router_has_a_trunk(self) -> Self:
+ def router_has_a_trunk(cls, data: Any) -> Any:
terminating_trunks = get_trunks_that_terminate_on_router(
subscription_id, SubscriptionLifecycle.PROVISIONING
) + get_trunks_that_terminate_on_router(subscription_id, SubscriptionLifecycle.ACTIVE)
@@ -42,7 +42,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
msg = "Selected router does not terminate any available IP trunks."
raise ValueError(msg)
- return self
+ return data
user_input = yield AddBGPSessionForm
diff --git a/gso/workflows/shared/cancel_subscription.py b/gso/workflows/shared/cancel_subscription.py
deleted file mode 100644
index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000
diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py
index 40286044bb73d8c4f3f6bd69b2fef0f0eaad92d7..93ac6f5156344d9c47db476197dbe66b252d972a 100644
--- a/gso/workflows/site/modify_site.py
+++ b/gso/workflows/site/modify_site.py
@@ -68,13 +68,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@step("Modify subscription")
def modify_site_subscription(
- subscription: Site,
- site_city: str,
- site_latitude: LatitudeCoordinate,
- site_longitude: LongitudeCoordinate,
- site_bgp_community_id: int,
- site_internal_id: int,
- site_ts_address: str,
+ subscription: Site,
+ site_city: str,
+ site_latitude: LatitudeCoordinate,
+ site_longitude: LongitudeCoordinate,
+ site_bgp_community_id: int,
+ site_internal_id: int,
+ site_ts_address: str,
) -> State:
"""Update the subscription model in the service database."""
subscription.site.site_city = site_city
@@ -100,11 +100,11 @@ def modify_site() -> StepList:
* Update the subscription model in the service database
"""
return (
- init
- >> store_process_subscription(Target.MODIFY)
- >> unsync
- >> modify_site_subscription
- >> set_status(SubscriptionLifecycle.ACTIVE)
- >> resync
- >> done
+ init
+ >> store_process_subscription(Target.MODIFY)
+ >> unsync
+ >> modify_site_subscription
+ >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> resync
+ >> done
)
diff --git a/gso/workflows/site/terminate_site.py b/gso/workflows/site/terminate_site.py
index 96e807b47ed9c72b101ad3c6303e9b1fbd1405bc..02be6f7c76f4f01ada85ddaf91e552575eb3c18a 100644
--- a/gso/workflows/site/terminate_site.py
+++ b/gso/workflows/site/terminate_site.py
@@ -23,12 +23,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class TerminateForm(FormPage):
if site.status == SubscriptionLifecycle.INITIAL:
info_label_2: Label = (
- "This will immediately mark the subscription as terminated, preventing any other workflows from " # type:ignore[assignment]
+ "This will immediately mark the subscription as terminated, preventing any other workflows from "
"interacting with this product subscription."
)
- info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING." # type:ignore[assignment]
+ info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
- termination_label: Label = "Are you sure you want to delete this site?" # type: ignore[assignment]
+ termination_label: Label = "Are you sure you want to delete this site?"
user_input = yield TerminateForm
return user_input.dict()
diff --git a/gso/workflows/tasks/import_office_router.py b/gso/workflows/tasks/import_office_router.py
index 6784d1c23cf81f17f727a114cbc98bdcc1b387f3..51a2a7accc1ad6e0522a563baf300e4088f16c90 100644
--- a/gso/workflows/tasks/import_office_router.py
+++ b/gso/workflows/tasks/import_office_router.py
@@ -14,7 +14,7 @@ from gso.products.product_types.office_router import OfficeRouterInactive
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_site_by_name
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
@step("Create subscription")
@@ -40,8 +40,8 @@ def initial_input_form_generator() -> FormGenerator:
office_router_site: str
office_router_fqdn: str
office_router_ts_port: PortNumber
- office_router_lo_ipv4_address: IPV4AddressType
- office_router_lo_ipv6_address: IPV6AddressType
+ office_router_lo_ipv4_address: IPv4AddressType
+ office_router_lo_ipv6_address: IPv6AddressType
user_input = yield ImportOfficeRouter
@@ -54,8 +54,8 @@ def initialize_subscription(
office_router_fqdn: str,
office_router_ts_port: PortNumber,
office_router_site: str,
- office_router_lo_ipv4_address: IPV4AddressType | None = None,
- office_router_lo_ipv6_address: IPV6AddressType | None = None,
+ office_router_lo_ipv4_address: IPv4AddressType | None = None,
+ office_router_lo_ipv6_address: IPv6AddressType | None = None,
) -> State:
"""Initialise the office router subscription using input data."""
subscription.office_router.office_router_ts_port = office_router_ts_port
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index 656536925151bfaee8d199d15ac5ffa4759ebd52..6386009896ab1fcead67a8764d1cdbb83fa63b81 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -17,7 +17,7 @@ from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_site_by_name
from gso.utils.helpers import generate_fqdn
-from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
@step("Create subscription")
@@ -45,8 +45,8 @@ def initial_input_form_generator() -> FormGenerator:
ts_port: int
router_vendor: Vendor
router_role: RouterRole
- router_lo_ipv4_address: IPV4AddressType
- router_lo_ipv6_address: IPV6AddressType
+ router_lo_ipv4_address: IPv4AddressType
+ router_lo_ipv6_address: IPv6AddressType
router_lo_iso_address: str
user_input = yield ImportRouter
@@ -62,8 +62,8 @@ def initialize_subscription(
router_site: str,
router_role: router_pb.RouterRole,
router_vendor: Vendor,
- router_lo_ipv4_address: IPV4AddressType | None = None,
- router_lo_ipv6_address: IPV6AddressType | None = None,
+ router_lo_ipv4_address: IPv4AddressType | None = None,
+ router_lo_ipv6_address: IPv6AddressType | None = None,
router_lo_iso_address: str | None = None,
) -> State:
"""Initialise the router subscription using input data."""
diff --git a/gso/workflows/tasks/import_super_pop_switch.py b/gso/workflows/tasks/import_super_pop_switch.py
index cd7fdc24379b110561dbdd97c9c8760af93e4010..ec717c050d0f2a84b146b5163b739e00b3d1224d 100644
--- a/gso/workflows/tasks/import_super_pop_switch.py
+++ b/gso/workflows/tasks/import_super_pop_switch.py
@@ -15,7 +15,7 @@ from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_site_by_name
from gso.utils.helpers import generate_fqdn
-from gso.utils.shared_enums import IPV4AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
@step("Create subscription")
@@ -41,7 +41,7 @@ def initial_input_form_generator() -> FormGenerator:
super_pop_switch_site: str
hostname: str
super_pop_switch_ts_port: PortNumber
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType
user_input = yield ImportSuperPopSwitch
@@ -54,7 +54,7 @@ def initialize_subscription(
hostname: str,
super_pop_switch_ts_port: PortNumber,
super_pop_switch_site: str,
- super_pop_switch_mgmt_ipv4_address: IPV4AddressType | None = None,
+ super_pop_switch_mgmt_ipv4_address: IPv4AddressType | None = None,
) -> State:
"""Initialise the Super PoP switch subscription using input data."""
subscription.super_pop_switch.super_pop_switch_ts_port = super_pop_switch_ts_port
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index c16e0527fb5db87ffd5ff1d41ee27a54919325f7..84039f326f35de5ef08591fb07d686b03bc8d35a 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -55,7 +55,7 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
else:
router_side_b = nokia_router_subscription_factory()
side_b_members = [
- dict(interface_name=f"Interface{interface}", interface_description=faker.sentence())
+ {"interface_name": f"Interface{interface}", "interface_description": faker.sentence()}
for interface in range(2)
]
@@ -73,10 +73,10 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
"side_a_ae_iface": "lag-1",
"side_a_ae_geant_a_sid": None,
"side_a_ae_members": [
- dict(
- interface_name=f"Interface{interface}",
- interface_description=faker.sentence(),
- )
+ {
+ "interface_name": f"Interface{interface}",
+ "interface_description": faker.sentence(),
+ }
for interface in range(2)
],
}