From 0956a7ea6387853cd93c37185ea3e3fd41585b40 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Fri, 18 Apr 2025 14:38:21 +0200
Subject: [PATCH] Allow for creation of Edge Ports on a Juniper router
---
gso/workflows/edge_port/create_edge_port.py | 14 +--
test/conftest.py | 2 +-
.../edge_port/test_create_edge_port.py | 87 ++++++++++---------
3 files changed, 58 insertions(+), 45 deletions(-)
diff --git a/gso/workflows/edge_port/create_edge_port.py b/gso/workflows/edge_port/create_edge_port.py
index 2455ee3de..d221c969f 100644
--- a/gso/workflows/edge_port/create_edge_port.py
+++ b/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
diff --git a/test/conftest.py b/test/conftest.py
index 926cd183b..2d2c1a784 100644
--- a/test/conftest.py
+++ b/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)
diff --git a/test/workflows/edge_port/test_create_edge_port.py b/test/workflows/edge_port/test_create_edge_port.py
index 941995bcd..b907a4138 100644
--- a/test/workflows/edge_port/test_create_edge_port.py
+++ b/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)
--
GitLab