Fixed interface list drop down.

gso/api/v1/imports.py

@@ -15,7 +15,7 @@ from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_blocks.site import SiteTier
 from gso.services import subscriptions
 from gso.services.crm import CustomerNotFoundError, get_customer_by_name
-from gso.workflows.iptrunk.utils import LAGMember
+from gso.utils.helpers import LAGMember
 
 router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
 

gso/utils/helpers.py

@@ -2,14 +2,40 @@ import re
 from ipaddress import IPv4Address
 from uuid import UUID
 
-from orchestrator.forms.validators import Choice
-from orchestrator.types import UUIDstr
+from orchestrator import step
+from orchestrator.types import State, UUIDstr
+from pydantic import BaseModel
+from pydantic_forms.validators import Choice
 
 from gso.products.product_blocks.router import RouterVendor
+from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
+from gso.services import provisioning_proxy
 from gso.services.netbox_client import NetboxClient
 
 
+class LAGMember(BaseModel):
+    #  TODO: validate interface name
+    interface_name: str
+    interface_description: str
+
+    def __hash__(self) -> int:
+        return hash((self.interface_name, self.interface_description))
+
+
+@step("[COMMIT] Set ISIS metric to 90000")
+def set_isis_to_90000(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State:
+    old_isis_metric = subscription.iptrunk.iptrunk_isis_metric
+    subscription.iptrunk.iptrunk_isis_metric = 90000
+    provisioning_proxy.provision_ip_trunk(subscription, process_id, tt_number, "isis_interface", False)
+
+    return {
+        "subscription": subscription,
+        "old_isis_metric": old_isis_metric,
+        "label_text": "ISIS is being set to 90K by the provisioning proxy, please wait for the results",
+    }
+
+
 def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
     """Return a list of available interfaces for a given router and speed.
 
@@ -81,3 +107,20 @@ def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr | None:
         if not device:
             raise ValueError("The selected router does not exist in Netbox.")
     return subscription_id
+
+
+def validate_iptrunk_unique_interface(interfaces: list[LAGMember]) -> list[LAGMember]:
+    """Verify if the interfaces are unique.
+
+    Args:
+    ----
+    interfaces (list[LAGMember]): The list of interfaces.
+
+    Returns:
+    -------
+    list[LAGMember]: The list of interfaces or raises an error.
+    """
+    interface_names = [member.interface_name for member in interfaces]
+    if len(interface_names) != len(set(interface_names)):
+        raise ValueError("Interfaces must be unique.")
+    return interfaces

gso/workflows/iptrunk/create_iptrunk.py

@@ -16,15 +16,16 @@ from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvision
 from gso.products.product_types.router import Router
 from gso.services import infoblox, provisioning_proxy, subscriptions
 from gso.services.crm import customer_selector
-from gso.services.netbox_client import NetboxClient, NotFoundError
+from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
 from gso.utils.helpers import (
+    LAGMember,
     available_interfaces_choices,
     available_lags_choices,
     get_router_vendor,
+    validate_iptrunk_unique_interface,
     validate_router_in_netbox,
 )
-from gso.workflows.iptrunk.utils import LAGMember
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -66,27 +67,22 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     user_input_router_side_a = yield SelectRouterSideA
     router_a = user_input_router_side_a.side_a_node_id.name
 
-    if get_router_vendor(router_a) == RouterVendor.NOKIA:
-        available_interfaces = available_interfaces_choices(router_a, initial_user_input.iptrunk_speed)
-        if available_interfaces is None:
-            raise NotFoundError(f"Router {router_a} could not be found in Netbox.")
+    class JuniperAeMembers(UniqueConstrainedList[LAGMember]):
+        min_items = initial_user_input.iptrunk_minimum_links
 
-        class NokiaLAGMember(LAGMember):
-            interface_name: Choice = available_interfaces  # type: ignore[assignment]
+    if get_router_vendor(router_a) == RouterVendor.NOKIA:
 
-            def __hash__(self) -> int:
-                return hash((self.interface_name, self.interface_description))
+        class NokiaLAGMemberA(LAGMember):
+            interface_name: available_interfaces_choices(  # type: ignore[valid-type]
+                router_a, initial_user_input.iptrunk_speed
+            )
 
-        class NokiaAeMembersA(UniqueConstrainedList[NokiaLAGMember]):
+        class NokiaAeMembersA(UniqueConstrainedList[NokiaLAGMemberA]):
             min_items = initial_user_input.iptrunk_minimum_links
 
         ae_members_side_a = NokiaAeMembersA
     else:
-
-        class JuniperAeMembersA(UniqueConstrainedList[LAGMember]):
-            min_items = initial_user_input.iptrunk_minimum_links
-
-        ae_members_side_a = JuniperAeMembersA  # type: ignore[assignment]
+        ae_members_side_a = JuniperAeMembers  # type: ignore[assignment]
 
     class CreateIptrunkSideAForm(FormPage):
         class Config:
@@ -96,6 +92,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_a_ae_geant_a_sid: str
         side_a_ae_members: ae_members_side_a  # type: ignore[valid-type]
 
+        @validator("side_a_ae_members", allow_reuse=True)
+        def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
+            return validate_iptrunk_unique_interface(side_a_ae_members)
+
     user_input_side_a = yield CreateIptrunkSideAForm
     # Remove the selected router for side A, to prevent any loops
     routers.pop(str(router_a))
@@ -115,28 +115,20 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     router_b = user_input_router_side_b.side_b_node_id.name
 
     if get_router_vendor(router_b) == RouterVendor.NOKIA:
-        available_interfaces = available_interfaces_choices(router_b, initial_user_input.iptrunk_speed)
-        if available_interfaces is None:
-            raise NotFoundError(f"Router {router_b} could not be found in Netbox.")
 
-        class NokiaLAGMember(LAGMember):  # type: ignore[no-redef]
-            interface_name: Choice = available_interfaces  # type: ignore[assignment]
-
-            def __hash__(self) -> int:
-                return hash((self.interface_name, self.interface_description))
+        class NokiaLAGMemberB(LAGMember):
+            interface_name: available_interfaces_choices(  # type: ignore[valid-type]
+                router_b, initial_user_input.iptrunk_speed
+            )
 
         class NokiaAeMembersB(UniqueConstrainedList):
             min_items = len(user_input_side_a.side_a_ae_members)
             max_items = len(user_input_side_a.side_a_ae_members)
-            item_type = NokiaLAGMember
+            item_type = NokiaLAGMemberB
 
         ae_members_side_b = NokiaAeMembersB
     else:
-
-        class JuniperAeMembersB(UniqueConstrainedList[LAGMember]):
-            min_items = len(user_input_side_a.side_a_ae_members)
-
-        ae_members_side_b = JuniperAeMembersB  # type: ignore[assignment]
+        ae_members_side_b = JuniperAeMembers  # type: ignore[assignment]
 
     class CreateIptrunkSideBForm(FormPage):
         class Config:
@@ -146,6 +138,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_b_ae_geant_a_sid: str
         side_b_ae_members: ae_members_side_b  # type: ignore[valid-type]
 
+        @validator("side_b_ae_members", allow_reuse=True)
+        def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
+            return validate_iptrunk_unique_interface(side_b_ae_members)
+
     user_input_side_b = yield CreateIptrunkSideBForm
 
     return (
@@ -307,12 +303,12 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
                 nbclient.attach_interface_to_lag(
                     device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
                     lag_name=lag_interface.name,
-                    iface_name=interface,
+                    iface_name=interface.interface_name,
                     description=str(subscription.subscription_id),
                 )
                 nbclient.reserve_interface(
                     device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
-                    iface_name=interface,
+                    iface_name=interface.interface_name,
                 )
     return {
         "subscription": subscription,
@@ -328,7 +324,7 @@ def allocate_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
             for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
                 NetboxClient().allocate_interface(
                     device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
-                    iface_name=interface,
+                    iface_name=interface.interface_name,
                 )
     return {
         "subscription": subscription,

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -18,7 +18,7 @@ from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
 from gso.services import provisioning_proxy
 from gso.services.provisioning_proxy import pp_interaction
-from gso.workflows.iptrunk.utils import set_isis_to_90000
+from gso.utils.helpers import set_isis_to_90000
 
 logger = getLogger(__name__)
 

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -13,7 +13,7 @@ from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkTy
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import provisioning_proxy
 from gso.services.provisioning_proxy import pp_interaction
-from gso.workflows.iptrunk.utils import LAGMember
+from gso.utils.helpers import LAGMember
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:

gso/workflows/iptrunk/terminate_iptrunk.py

@@ -11,7 +11,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import infoblox, provisioning_proxy
 from gso.services.provisioning_proxy import pp_interaction
-from gso.workflows.iptrunk.utils import set_isis_to_90000
+from gso.utils.helpers import set_isis_to_90000
 
 
 def initial_input_form_generator() -> FormGenerator:

gso/workflows/iptrunk/utils.py

-from orchestrator import step
-from orchestrator.types import State, UUIDstr
-from pydantic import BaseModel
-
-from gso.products.product_types.iptrunk import Iptrunk
-from gso.services import provisioning_proxy
-
-
-class LAGMember(BaseModel):
-    #  TODO: validate interface name
-    interface_name: str
-    interface_description: str
-
-    def __hash__(self) -> int:
-        return hash((self.interface_name, self.interface_description))
-
-
-@step("[COMMIT] Set ISIS metric to 90000")
-def set_isis_to_90000(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State:
-    old_isis_metric = subscription.iptrunk.iptrunk_isis_metric
-    subscription.iptrunk.iptrunk_isis_metric = 90000
-    provisioning_proxy.provision_ip_trunk(subscription, process_id, tt_number, "isis_interface", False)
-
-    return {
-        "subscription": subscription,
-        "old_isis_metric": old_isis_metric,
-        "label_text": "ISIS is being set to 90K by the provisioning proxy, please wait for the results",
-    }

gso/workflows/tasks/import_iptrunk.py

@@ -13,8 +13,8 @@ from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.services import subscriptions
 from gso.services.crm import get_customer_by_name
+from gso.utils.helpers import LAGMember
 from gso.workflows.iptrunk.create_iptrunk import initialize_subscription
-from gso.workflows.iptrunk.utils import LAGMember
 
 
 def _generate_routers() -> dict[str, str]:

test/workflows/iptrunk/test_create_iptrunk.py

@@ -7,7 +7,7 @@ from gso.products import Iptrunk, ProductType
 from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.services.crm import customer_selector, get_customer_by_name
 from gso.services.subscriptions import get_product_id_by_name
-from gso.workflows.iptrunk.utils import LAGMember
+from gso.utils.helpers import LAGMember
 from test.workflows import (
     assert_aborted,
     assert_complete,
@@ -33,7 +33,7 @@ class MockedNetboxClient:
 
     def get_available_interfaces(self):
         interfaces = []
-        for interface in range(1, 5):
+        for interface in range(5):
             interface_data = {
                 "name": f"Interface{interface}",
                 "module": {"display": f"Module{interface}"},
@@ -97,8 +97,8 @@ def input_form_wizard_data(router_subscription_factory, faker):
         "side_a_ae_iface": "LAG1",
         "side_a_ae_geant_a_sid": faker.geant_sid(),
         "side_a_ae_members": [
-            LAGMember(interface_name=faker.network_interface(), interface_description=faker.sentence())
-            for _ in range(5)
+            LAGMember(interface_name=f"Interface{interface}", interface_description=faker.sentence())
+            for interface in range(5)
         ],
     }
     create_ip_trunk_side_b_router_name = {"side_b_node_id": router_side_b}
@@ -106,8 +106,8 @@ def input_form_wizard_data(router_subscription_factory, faker):
         "side_b_ae_iface": "LAG4",
         "side_b_ae_geant_a_sid": faker.geant_sid(),
         "side_b_ae_members": [
-            LAGMember(interface_name=faker.network_interface(), interface_description=faker.sentence())
-            for _ in range(5)
+            LAGMember(interface_name=f"Interface{interface}", interface_description=faker.sentence())
+            for interface in range(5)
         ],
     }