integrate the iptrunk model rework and netbox functionality

gso/utils/helpers.py

@@ -7,18 +7,9 @@ from orchestrator.types import UUIDstr
 
 from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.router import Router
-from gso.services.crm import all_customers
 from gso.services.netbox_client import NetboxClient
 
 
-def customer_selector() -> Choice:
-    customers = {}
-    for customer in all_customers():
-        customers[customer["id"]] = customer["name"]
-
-    return Choice("Select a customer", zip(customers.keys(), customers.items()))  # type: ignore[arg-type]
-
-
 def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
     """Return a list of available interfaces for a given router and speed.
 

gso/workflows/iptrunk/create_iptrunk.py

 from uuid import uuid4
 
 from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, ChoiceList, UniqueConstrainedList
+from orchestrator.forms.validators import Choice, UniqueConstrainedList
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
 from orchestrator.workflow import StepList, done, init, step, workflow
@@ -15,17 +15,15 @@ from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router
 from gso.services import infoblox, provisioning_proxy, subscriptions
-from gso.services.netbox_client import NetboxClient
 from gso.services.crm import customer_selector
+from gso.services.netbox_client import NetboxClient, NotFoundError
 from gso.services.provisioning_proxy import pp_interaction
-from gso.workflows.utils import (
+from gso.utils.helpers import (
     available_interfaces_choices,
     available_lags_choices,
-    customer_selector,
     get_router_vendor,
     validate_router_in_netbox,
 )
-from gso.utils.types.phy_port import PhyPortCapacity
 from gso.workflows.iptrunk.utils import LAGMember
 
 
@@ -53,84 +51,100 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     initial_user_input = yield CreateIptrunkForm
 
-    router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items()))  # type: ignore
+    router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items()))  # type: ignore[arg-type]
 
     class SelectRouterSideA(FormPage):
         class Config:
             title = "Select a router for side A of the trunk."
 
-        iptrunk_sideA_node_id: router_enum_a  # type: ignore[valid-type]
+        side_a_node_id: router_enum_a  # type: ignore[valid-type]
 
-        @validator("iptrunk_sideA_node_id", allow_reuse=True)
-        def validate_device_exists_in_netbox(cls, iptrunk_sideA_node_id: UUIDstr) -> str | None:
-            return validate_router_in_netbox(iptrunk_sideA_node_id)
+        @validator("side_a_node_id", allow_reuse=True)
+        def validate_device_exists_in_netbox(cls, side_a_node_id: UUIDstr) -> str | None:
+            return validate_router_in_netbox(side_a_node_id)
 
     user_input_router_side_a = yield SelectRouterSideA
-    router_a = user_input_router_side_a.iptrunk_sideA_node_id.name
-    side_a_ae_iface = available_lags_choices(router_a) or str
+    router_a = user_input_router_side_a.side_a_node_id.name
 
-    class AeMembersListA(ChoiceList):
-        min_items = initial_user_input.iptrunk_minimum_links
-        item_type = available_interfaces_choices(router_a, initial_user_input.iptrunk_speed)  # type: ignore
-        unique_items = True
+    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
-        unique_items = True
+        class NokiaLAGMember(LAGMember):
+            interface_name: Choice = available_interfaces  # type: ignore[assignment]
 
-    ae_members_side_a = AeMembersListA if get_router_vendor(router_a) == RouterVendor.NOKIA else JuniperAeMembers
+            def __hash__(self) -> int:
+                return hash((self.interface_name, self.interface_description))
 
-    class AeMembersDescriptionListA(UniqueConstrainedList[str]):
-        min_items = initial_user_input.iptrunk_minimum_links
+        class NokiaAeMembersA(UniqueConstrainedList[NokiaLAGMember]):
+            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]
 
     class CreateIptrunkSideAForm(FormPage):
         class Config:
             title = "Provide subscription details for side A of the trunk."
 
-        side_a_ae_iface: side_a_ae_iface  # type: ignore[valid-type]
+        side_a_ae_iface: available_lags_choices(router_a) or str  # type: ignore[valid-type]
         side_a_ae_geant_a_sid: str
         side_a_ae_members: ae_members_side_a  # type: ignore[valid-type]
-        side_a_ae_members_descriptions: AeMembersDescriptionListA
 
     user_input_side_a = yield CreateIptrunkSideAForm
     # Remove the selected router for side A, to prevent any loops
-    routers.pop(str(user_input_side_a.side_a_node_id.name))
+    routers.pop(str(router_a))
     router_enum_b = Choice("Select a router", zip(routers.keys(), routers.items()))  # type: ignore[arg-type]
 
     class SelectRouterSideB(FormPage):
         class Config:
             title = "Select a router for side B of the trunk."
 
-        iptrunk_sideB_node_id: router_enum_b  # type: ignore[valid-type]
+        side_b_node_id: router_enum_b  # type: ignore[valid-type]
 
-        @validator("iptrunk_sideB_node_id", allow_reuse=True)
-        def validate_device_exists_in_netbox(cls, iptrunk_sideB_node_id: UUIDstr) -> str | None:
-            return validate_router_in_netbox(iptrunk_sideB_node_id)
+        @validator("side_b_node_id", allow_reuse=True)
+        def validate_device_exists_in_netbox(cls, side_b_node_id: UUIDstr) -> str | None:
+            return validate_router_in_netbox(side_b_node_id)
 
     user_input_router_side_b = yield SelectRouterSideB
-    router_b = user_input_router_side_b.iptrunk_sideB_node_id.name
-    side_b_ae_iface = available_lags_choices(router_b) or str
+    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 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
 
-    class AeMembersListB(ChoiceList):
-        min_items = len(user_input_side_a.side_a_ae_members)
-        max_items = len(user_input_side_a.side_a_ae_members)
-        item_type = available_interfaces_choices(router_b, initial_user_input.iptrunk_speed)  # type: ignore
-        unique_items = True
+        ae_members_side_b = NokiaAeMembersB
+    else:
 
-    ae_members_side_b = AeMembersListB if get_router_vendor(router_b) == RouterVendor.NOKIA else JuniperAeMembers
+        class JuniperAeMembersB(UniqueConstrainedList[LAGMember]):
+            min_items = len(user_input_side_a.side_a_ae_members)
 
-    class AeMembersDescriptionListB(UniqueConstrainedList[LAGMember]):
-        min_items = len(user_input_side_a.side_a_ae_members)
-        max_items = len(user_input_side_a.side_a_ae_members)
+        ae_members_side_b = JuniperAeMembersB  # type: ignore[assignment]
 
     class CreateIptrunkSideBForm(FormPage):
         class Config:
             title = "Provide subscription details for side B of the trunk."
 
-        side_b_ae_iface: side_b_ae_iface  # type: ignore[valid-type]
+        side_b_ae_iface: available_lags_choices(router_b) or str  # type: ignore[valid-type]
         side_b_ae_geant_a_sid: str
         side_b_ae_members: ae_members_side_b  # type: ignore[valid-type]
-        side_b_ae_members_descriptions: AeMembersDescriptionListB
 
     user_input_side_b = yield CreateIptrunkSideBForm
 

gso/workflows/router/create_router.py

@@ -159,7 +159,8 @@ def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr,
 def create_netbox_device(subscription: RouterProvisioning) -> State:
     if subscription.router.router_vendor == RouterVendor.NOKIA:
         NetboxClient().create_device(
-            subscription.router.router_fqdn, subscription.router.router_site.site_tier  # type: ignore[arg-type, union-attr]
+            subscription.router.router_fqdn,
+            str(subscription.router.router_site.site_tier),  # type: ignore[union-attr]
         )
         return {"subscription": subscription, "label_text": "Creating NetBox device"}
     return {"subscription": subscription, "label_text": "Skipping NetBox device creation for Juniper router."}

test/workflows/iptrunk/test_create_iptrunk.py

@@ -7,6 +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 test.workflows import (
     assert_aborted,
     assert_complete,
@@ -83,31 +84,30 @@ def input_form_wizard_data(router_subscription_factory, faker):
     router_side_b = router_subscription_factory()
 
     create_ip_trunk_step = {
-        "tt_number": faker.pystr(),
+        "tt_number": faker.tt_number(),
         "customer": getattr(customer_selector(), get_customer_by_name("GÉANT")["id"]),
-        "geant_s_sid": faker.pystr(),
+        "geant_s_sid": faker.geant_sid(),
         "iptrunk_type": IptrunkType.DARK_FIBER,
         "iptrunk_description": faker.sentence(),
         "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
         "iptrunk_minimum_links": 2,
     }
-    create_ip_trunk_side_a_router_name = {"iptrunk_sideA_node_id": router_side_a}
+    create_ip_trunk_side_a_router_name = {"side_a_node_id": router_side_a}
     create_ip_trunk_side_a_step = {
-        "side_a_node_id": router_side_a,
-        "side_a_ae_iface": faker.pystr(),
-        "side_a_ae_geant_a_sid": faker.pystr(),
+        "side_a_ae_iface": "LAG1",
+        "side_a_ae_geant_a_sid": faker.geant_sid(),
         "side_a_ae_members": [
-            {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(5)
+            LAGMember(interface_name=faker.network_interface(), interface_description=faker.sentence())
+            for _ in range(5)
         ],
     }
-
-    create_ip_trunk_side_b_router_name = {"iptrunk_sideB_node_id": router_side_b}
+    create_ip_trunk_side_b_router_name = {"side_b_node_id": router_side_b}
     create_ip_trunk_side_b_step = {
-        "side_b_node_id": router_side_b,
-        "side_b_ae_iface": faker.pystr(),
-        "side_b_ae_geant_a_sid": faker.pystr(),
+        "side_b_ae_iface": "LAG4",
+        "side_b_ae_geant_a_sid": faker.geant_sid(),
         "side_b_ae_members": [
-            {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(5)
+            LAGMember(interface_name=faker.network_interface(), interface_description=faker.sentence())
+            for _ in range(5)
         ],
     }
 

test/workflows/router/test_create_router.py

@@ -29,7 +29,7 @@ def router_creation_input_form_data(site_subscription_factory, faker):
         "router_site": router_site,
         "hostname": faker.pystr(),
         "ts_port": faker.pyint(),
-        "router_vendor": faker.random_choices(elements=(RouterVendor.NOKIA, RouterVendor.JUNIPER), length=1)[0],
+        "router_vendor": RouterVendor.NOKIA,
         "router_role": faker.random_choices(elements=(RouterRole.P, RouterRole.PE, RouterRole.AMT), length=1)[0],
         "is_ias_connected": True,
     }
@@ -37,7 +37,7 @@ def router_creation_input_form_data(site_subscription_factory, faker):
 
 @pytest.mark.workflow
 @patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
-@patch("gso.workflows.router.create_router.NetBoxClient.create_device")
+@patch("gso.workflows.router.create_router.NetboxClient.create_device")
 @patch("gso.workflows.router.create_router.infoblox.hostname_available")
 @patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
 @patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
@@ -134,7 +134,7 @@ def test_create_router_success(
 
 @pytest.mark.workflow
 @patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
-@patch("gso.workflows.router.create_router.NetBoxClient.create_device")
+@patch("gso.workflows.router.create_router.NetboxClient.create_device")
 @patch("gso.workflows.router.create_router.infoblox.hostname_available")
 @patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
 @patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")

test/workflows/router/test_terminate_router.py

@@ -12,7 +12,7 @@ def router_termination_input_form_data(site_subscription_factory, faker):
 
 
 @pytest.mark.workflow
-@patch("gso.workflows.router.terminate_router.NetBoxClient.delete_device")
+@patch("gso.workflows.router.terminate_router.NetboxClient.delete_device")
 @patch("gso.workflows.router.terminate_router.infoblox.delete_host_by_ip")
 @patch("gso.workflows.router.terminate_router.infoblox.delete_network")
 def test_terminate_router_success(