fix all mypy complains

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/pop_vlan.py

@@ -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/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, TypeAlias, Type
 from uuid import uuid4
 
 from annotated_types import Len
@@ -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]
@@ -104,14 +104,14 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
         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)]
     else:
-        ae_members_side_a_type = JuniperAeMembers # type: ignore[assignment]
+        ae_members_side_a_type = JuniperAeMembers
 
     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
 
         @field_validator("side_a_ae_members")
         def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
@@ -154,14 +154,14 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             ),
         ]
     else:
-        ae_members_side_b = JuniperAeMembers
+        ae_members_side_b = JuniperAeMembers  # 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 +323,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 +397,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 +428,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 +439,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 {

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(

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -38,7 +38,7 @@ from gso.utils.shared_enums import IPV4AddressType, IPV6AddressType, Vendor
 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 +62,10 @@ 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]
+
+    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]
 
-        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]
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -84,9 +83,9 @@ 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(str(subscription.iptrunk.iptrunk_ipv4_network), default_type=IPV4AddressType)  # type: ignore[valid-type]
+        iptrunk_ipv6_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv6_network), default_type=IPV6AddressType)  # type: ignore[valid-type]
 
         @field_validator("tt_number")
         def validate_tt_number(cls, tt_number: str) -> str:
@@ -96,9 +95,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 +105,8 @@ 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(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str)  # type: ignore[valid-type]
+        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 +129,8 @@ 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(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str)  # type: ignore[valid-type]
+        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

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

gso/workflows/router/update_ibgp_mesh.py

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

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