Allow for creation of Edge Ports on a Juniper router

gso/workflows/edge_port/create_edge_port.py

@@ -31,8 +31,9 @@ from gso.utils.helpers import (
     partner_choice,
     validate_edge_port_number_of_members_based_on_lacp,
 )
+from gso.utils.shared_enums import Vendor
 from gso.utils.types.geant_ids import IMPORTED_GA_ID
-from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
+from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, PhysicalPortCapacity
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.workflow_steps import start_moodi, stop_moodi
 from gso.workflows.shared import create_summary_form
@@ -50,7 +51,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         service_type: EdgePortType
         enable_lacp: bool = False
         speed: PhysicalPortCapacity
-        encapsulation: EncapsulationType = EncapsulationType.DOT1Q
+        encapsulation: EncapsulationType | str = EncapsulationType.DOT1Q  # TODO: remove type hint workaround
         number_of_members: int
         minimum_links: int
         mac_address: str | None = None
@@ -81,13 +82,16 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     initial_user_input = yield CreateEdgePortForm
 
-    class EdgePortLAGMember(LAGMember):
+    class NokiaEdgePortLAGMember(LAGMember):
         interface_name: available_interfaces_choices(  # type: ignore[valid-type]
             initial_user_input.node, initial_user_input.speed
         )
 
+    selected_router = Router.from_subscription(initial_user_input.node)
+    lag_member = JuniperLAGMember if selected_router.router.vendor == Vendor.JUNIPER else NokiaEdgePortLAGMember
+
     lag_ae_members = Annotated[
-        list[EdgePortLAGMember],
+        list[lag_member],  # type: ignore[valid-type]
         AfterValidator(validate_unique_list),
         Len(
             min_length=initial_user_input.number_of_members,
@@ -98,7 +102,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     class SelectInterfaceForm(FormPage):
         model_config = ConfigDict(title="Select Interfaces")
 
-        name: available_service_lags_choices(initial_user_input.node)  # type: ignore[valid-type]
+        name: available_service_lags_choices(initial_user_input.node) or str  # type: ignore[valid-type]
         description: str | None = None
         ae_members: lag_ae_members
 

test/conftest.py

@@ -134,7 +134,7 @@ class FakerProvider(BaseProvider):
         return self.generator.numerify("ae@#")
 
     def nokia_lag_interface_name(self) -> str:
-        return self.generator.numerify("lag-@#")
+        return self.generator.numerify("lag-3#")
 
     def link_members_juniper(self) -> LAGMemberList[LAGMember]:
         iface_amount = self.generator.random_int(min=2, max=5)

test/workflows/edge_port/test_create_edge_port.py

@@ -47,51 +47,57 @@ def _netbox_client_mock():
 
 @pytest.fixture()
 def input_form_wizard_data(request, router_subscription_factory, partner_factory, faker):
-    create_edge_port_step = {
-        "tt_number": faker.tt_number(),
-        "node": str(router_subscription_factory(vendor=Vendor.NOKIA).subscription_id),
-        "partner": partner_factory(name="GAAR", email=faker.email())["partner_id"],
-        "service_type": EdgePortType.PUBLIC,
-        "enable_lacp": True,
-        "speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
-        "encapsulation": EncapsulationType.DOT1Q,
-        "number_of_members": 2,
-        "minimum_links": 1,
-        "custom_service_name": faker.sentence(),
-        "generate_ga_id": False,
-        "ga_id": "GA-12345",
-    }
-    create_edge_port_interface_step = {
-        "name": "lag-21",
-        "description": faker.sentence(),
-        "ae_members": [
-            {
-                "interface_name": f"Interface{interface}",
-                "interface_description": faker.sentence(),
-            }
-            for interface in range(2)
-        ],
-    }
-    summary_view_step = {}
-
-    return [
-        create_edge_port_step,
-        create_edge_port_interface_step,
-        summary_view_step,
-    ]
+    def input_form_factory(vendor: Vendor = Vendor.NOKIA):
+        partner = partner_factory(name="GAAR", email=faker.email())
+        create_edge_port_step = {
+            "tt_number": faker.tt_number(),
+            "node": str(router_subscription_factory(vendor=vendor, partner=partner).subscription_id),
+            "partner": partner["partner_id"],
+            "service_type": EdgePortType.PUBLIC,
+            "enable_lacp": True,
+            "speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
+            "encapsulation": EncapsulationType.DOT1Q,
+            "number_of_members": 2,
+            "minimum_links": 1,
+            "custom_service_name": faker.sentence(),
+            "generate_ga_id": False,
+            "ga_id": "GA-12345",
+        }
+        create_edge_port_interface_step = {
+            "name": faker.nokia_lag_interface_name(),
+            "description": faker.sentence(),
+            "ae_members": [
+                {
+                    "interface_name": faker.network_interface() if vendor == Vendor.JUNIPER else f"Interface{i}",
+                    "interface_description": faker.sentence(),
+                }
+                for i in range(2)
+            ],
+        }
+        summary_view_step = {}
+
+        return [
+            create_edge_port_step,
+            create_edge_port_interface_step,
+            summary_view_step,
+        ]
+
+    return input_form_factory
 
 
 @pytest.mark.workflow()
+@pytest.mark.parametrize("router_vendor", [*Vendor.values()])
 @patch("gso.services.lso_client._send_request")
 def test_successful_edge_port_creation(
     mock_execute_playbook,
+    router_vendor,
     input_form_wizard_data,
     faker,
     _netbox_client_mock,  # noqa: PT019
     test_client,
 ):
     product_id = get_product_id_by_name(ProductName.EDGE_PORT)
-    initial_data = [{"product": product_id}, *input_form_wizard_data]
+    initial_data = [{"product": product_id}, *input_form_wizard_data(vendor=router_vendor)]
     result, process_stat, step_log = run_workflow("create_edge_port", initial_data)
 
     for _ in range(3):
@@ -106,9 +112,12 @@ def test_successful_edge_port_creation(
     subscription = EdgePort.from_subscription(subscription_id)
 
     assert subscription.status == "active"
-    router_fqdn = Router.from_subscription(input_form_wizard_data[0]["node"]).router.router_fqdn
+    router_fqdn = Router.from_subscription(initial_data[1]["node"]).router.router_fqdn
     assert subscription.edge_port.ga_id == "GA-12345"
-    assert subscription.description == f"Edge Port lag-21 on {router_fqdn}, GAAR, {subscription.edge_port.ga_id}"
+    assert (
+        subscription.description
+        == f"Edge Port {initial_data[2]["name"]} on {router_fqdn}, GAAR, {subscription.edge_port.ga_id}"
+    )
     assert len(subscription.edge_port.edge_port_ae_members) == 2
     assert mock_execute_playbook.call_count == 4
 
@@ -123,7 +132,7 @@ def test_successful_edge_port_creation_with_auto_ga_id_creation(
     test_client,
 ):
     product_id = get_product_id_by_name(ProductName.EDGE_PORT)
-    initial_data = [{"product": product_id}, *input_form_wizard_data]
+    initial_data = [{"product": product_id}, *input_form_wizard_data()]
     initial_data[1]["generate_ga_id"] = True
     initial_data[1]["ga_id"] = None
     result, process_stat, step_log = run_workflow("create_edge_port", initial_data)
@@ -152,8 +161,8 @@ def test_edge_port_creation_with_invalid_input(
 ):
     product_id = get_product_id_by_name(ProductName.EDGE_PORT)
     # If the number of members is greater than 1 then LACP must be enabled.
-    input_form_wizard_data[0]["enable_lacp"] = False
-    initial_data = [{"product": product_id}, *input_form_wizard_data]
+    initial_data = [{"product": product_id}, *input_form_wizard_data()]
+    initial_data[1]["enable_lacp"] = False
 
     with pytest.raises(FormValidationError) as error:
         run_workflow("create_edge_port", initial_data)
@@ -175,7 +184,7 @@ def test_edge_port_creation_with_existing_ga_id(
 ):
     subscription = edge_port_subscription_factory(ga_id="GA-12345")
     product_id = get_product_id_by_name(ProductName.EDGE_PORT)
-    initial_data = [{"product": product_id}, *input_form_wizard_data]
+    initial_data = [{"product": product_id}, *input_form_wizard_data()]
     with pytest.raises(FormValidationError) as error:
         run_workflow("create_edge_port", initial_data)