diff --git a/gso/api/v1/imports.py b/gso/api/v1/imports.py
index 6f1cf4964c5f7d79fb31eb3d1170dd98d04e5a4e..51f58c5d62705969f054a2d680aab9e806ff2171 100644
--- a/gso/api/v1/imports.py
+++ b/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)])
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 32d891cac2d66e18a7bed6d235ce448ced4928a7..ff37395b92b3c3d7a8c9b5c61e392afa7b7bbe31 100644
--- a/gso/utils/helpers.py
+++ b/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
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index cd56d9daa3ccc542e0fa45f02d8a2248c3250ac7..f022181b5060632807b5c67aa9548aba9876505e 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/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,
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 3b2c1b4d160dcfa3a51815c6b1b898f7e38e4dfb..4b918d00326114ed1da4f1b87a2a055f406bfe52 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/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__)
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 9e402bbe01270563847124c949b06386a5602ea6..284602504eb7af3d7d7563728fbe401a1ce0f424 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/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:
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index b2310d89c9d1851f168e9c2521fc0052c0c3b74c..c0a0da62cd549962164e6d9cc6afb787a7f63f79 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/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:
diff --git a/gso/workflows/iptrunk/utils.py b/gso/workflows/iptrunk/utils.py
deleted file mode 100644
index 57c1171529194f9499ec279e412525239cf04d99..0000000000000000000000000000000000000000
--- a/gso/workflows/iptrunk/utils.py
+++ /dev/null
@@ -1,28 +0,0 @@
-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",
- }
diff --git a/gso/workflows/tasks/import_iptrunk.py b/gso/workflows/tasks/import_iptrunk.py
index fe58ea11041b5573ae427795b18039e04b5d683c..84c12aa6667962eb1a40184384fad34d0a85ba04 100644
--- a/gso/workflows/tasks/import_iptrunk.py
+++ b/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]:
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 4f40c640d8b0eb2842cacb3b045144db96b926cc..a8f273f77cb87602fcef887599250f0c0eea153c 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/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)
],
}