fix all mypy complains

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:

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

gso/products/product_blocks/iptrunk.py

 """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

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(

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.

gso/products/product_blocks/pop_vlan.py

 """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.

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`.

gso/products/product_blocks/site.py

 """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

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.

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

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"]]

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

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")
 ]
 

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

gso/workflows/iptrunk/create_iptrunk.py

 """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}."
         )

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

gso/workflows/iptrunk/modify_trunk_interface.py

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

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
 

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:

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