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/pop_vlan.py b/gso/products/product_blocks/pop_vlan.py
index e556ad1e9903772f7fb58a38a4047286595a1fb8..34f90b6f5f2dc0ef8d46f593d61f55c7d3c7f450 100644
--- a/gso/products/product_blocks/pop_vlan.py
+++ b/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.
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 71f5453bb196c7d79ad67ea1ca0fe5fc97b656f4..9fcbff6bf9e835031dbeca9ac4852120f54e83ff 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, 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 {
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 0a30e01079383504d659324e1e11ccfb793b1594..c631a1740eed806ab92953381ab875d07c4e7c9d 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(
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 3395774c577957fcbf1bdfa5a08a9de6d23233ee..719c43ca565b0b7eef97966dd4b41abc5dbb39ef 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/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
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index 6407dbddb137568141cba287bb118a20269aca56..e1d4a82639bff61b56250f85f96a64e418d53b27 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/terminate_router.py b/gso/workflows/router/terminate_router.py
index 0d46f9abbd7a32e2f141894a09751659d24d52fa..a779c0fe572b885ab5feda4779a63ed2895ba4d2 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..ad4420bcbefe0ba42f6a77d4ee03a5f7d9f6c8d3 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/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
diff --git a/gso/workflows/site/terminate_site.py b/gso/workflows/site/terminate_site.py
index 96e807b47ed9c72b101ad3c6303e9b1fbd1405bc..4beb6125f0528ca83cca3cb8c30a23cbe1dff818 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()