diff --git a/gso/workflows/edge_port/create_edge_port.py b/gso/workflows/edge_port/create_edge_port.py
index d221c969fa3065ee7bf78de9585662e67b9b3307..e25ad13491ed04791a9498be8689f030b3f02cbe 100644
--- a/gso/workflows/edge_port/create_edge_port.py
+++ b/gso/workflows/edge_port/create_edge_port.py
@@ -9,7 +9,7 @@ from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import SubscriptionLifecycle
from orchestrator.utils.errors import ProcessFailureError
-from orchestrator.workflow import StepList, begin, done
+from orchestrator.workflow import StepList, begin, conditional, done
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from pydantic import AfterValidator, ConfigDict, model_validator
@@ -292,16 +292,18 @@ def create_edge_port() -> StepList:
* Deploy configuration on the new edge port, first as a dry run
* allocate LAG and LAG members in the Netbox.
"""
+ router_is_nokia = conditional(lambda state: state["subscription"]["edge_port"]["node"]["vendor"] == Vendor.NOKIA)
+
return (
begin
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
>> start_moodi()
- >> reserve_interfaces_in_netbox
+ >> router_is_nokia(reserve_interfaces_in_netbox)
>> lso_interaction(create_edge_port_dry)
>> lso_interaction(create_edge_port_real)
- >> allocate_interfaces_in_netbox
+ >> router_is_nokia(allocate_interfaces_in_netbox)
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> stop_moodi()
diff --git a/gso/workflows/edge_port/modify_edge_port.py b/gso/workflows/edge_port/modify_edge_port.py
index 8a82d146761d3d850781226b2d9b3bcd609718ea..3cb953ab1eb019d850931ce969b0a823476a5d4d 100644
--- a/gso/workflows/edge_port/modify_edge_port.py
+++ b/gso/workflows/edge_port/modify_edge_port.py
@@ -25,7 +25,8 @@ from gso.utils.helpers import (
available_interfaces_choices_including_current_members,
validate_edge_port_number_of_members_based_on_lacp,
)
-from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, PhysicalPortCapacity
from gso.utils.types.tt_number import TTNumber
from gso.utils.types.unique_field import validate_field_is_unique
@@ -62,7 +63,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
user_input = yield ModifyEdgePortForm
- class EdgePortLAGMember(LAGMember):
+ class NokiaEdgePortLAGMember(LAGMember):
interface_name: ( # type: ignore[valid-type]
available_interfaces_choices_including_current_members(
subscription.edge_port.node.owner_subscription_id,
@@ -75,8 +76,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
)
+ lag_member = JuniperLAGMember if subscription.edge_port.node.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=user_input.number_of_members,
@@ -86,7 +89,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
current_lag_ae_members = (
[
- EdgePortLAGMember(
+ lag_member(
interface_name=iface.interface_name,
interface_description=iface.interface_description,
)
@@ -272,16 +275,18 @@ def modify_edge_port() -> StepList:
* Modify configuration on the new edge port, first as a dry run
* Change LAG and LAG members in the Netbox.
"""
+ router_is_nokia = conditional(lambda state: state["subscription"]["edge_port"]["node"]["vendor"] == Vendor.NOKIA)
capacity_has_changed = conditional(lambda state: state["capacity_has_changed"])
+
return (
begin
>> store_process_subscription(Target.MODIFY)
>> unsync
>> modify_edge_port_subscription
- >> capacity_has_changed(update_interfaces_in_netbox)
+ >> capacity_has_changed(router_is_nokia(update_interfaces_in_netbox))
>> capacity_has_changed(lso_interaction(update_edge_port_dry))
>> capacity_has_changed(lso_interaction(update_edge_port_real))
- >> capacity_has_changed(allocate_interfaces_in_netbox)
+ >> capacity_has_changed(router_is_nokia(allocate_interfaces_in_netbox))
>> resync
>> done
)
diff --git a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
index 4ebd46d9accd4992e29ead9a5afd78ab232b1ce1..7721e977ef0f54326d1fb2bd5d0f29c6afdfd009 100644
--- a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
@@ -101,7 +101,7 @@ def _initial_input_form(product_name: str) -> FormGenerator:
class InterconnectSwitchSideForm(FormPage):
model_config = ConfigDict(title="Please enter interface names and descriptions for the switch side.")
- switch_side_iface: JuniperPhyInterface
+ switch_side_iface: str
switch_side_ae_members: switch_side_ae_member_list
switch_side_input = yield InterconnectSwitchSideForm
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index ff5126f342cb1d80d501c68134453bf6858f7df0..725885cb4f9f552e1cd0222b94365d6b6cf5434e 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -72,12 +72,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict):
},
"nodeA": {
"name": side_a_node or router_side_a.router.router_fqdn,
- "ae_name": side_a_ae_name or faker.network_interface(),
+ "ae_name": side_a_ae_name or faker.nokia_lag_interface_name(),
"port_ga_id": faker.imported_ga_id(),
"members": side_a_members
or [
{
- "interface_name": faker.network_interface(),
+ "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
}
for _ in range(5)
@@ -87,12 +87,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict):
},
"nodeB": {
"name": side_b_node or router_side_b.router.router_fqdn,
- "ae_name": side_b_ae_name or faker.network_interface(),
+ "ae_name": side_b_ae_name or faker.nokia_lag_interface_name(),
"port_ga_id": faker.imported_ga_id(),
"members": side_b_members
or [
{
- "interface_name": faker.network_interface(),
+ "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
}
for _ in range(5)
@@ -235,7 +235,7 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
"node": str(router_subscription_factory().subscription_id),
"ae_iface": faker.nokia_lag_interface_name(),
"ae_members": [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
for _ in range(2)
],
},
@@ -243,7 +243,10 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
"switch": str(switch_subscription_factory().subscription_id),
"ae_iface": faker.juniper_ae_interface_name(),
"ae_members": [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
for _ in range(2)
],
},
@@ -273,13 +276,10 @@ def edge_port_data(temp_file, faker, router_subscription_factory, partner_factor
"ignore_if_down": False,
"ae_members": [
{
- "interface_name": faker.network_interface(),
+ "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
- },
- {
- "interface_name": faker.network_interface(),
- "interface_description": faker.sentence(),
- },
+ }
+ for _ in range(2)
],
"description": faker.sentence(),
}
@@ -584,7 +584,10 @@ def test_import_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, mock_
@patch("gso.cli.imports.time.sleep")
@patch("gso.cli.imports.start_process")
def test_import_iptrunk_non_unique_members_side_a_and_b(mock_start_process, mock_sleep, iptrunk_data, faker, capfd):
- duplicate_interface = {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
+ duplicate_interface = {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
side_a_members = [duplicate_interface for _ in range(5)]
side_b_members = [duplicate_interface for _ in range(5)]
broken_data = iptrunk_data(side_a_members=side_a_members, side_b_members=side_b_members)
@@ -609,10 +612,12 @@ def test_import_iptrunk_non_unique_members_side_a_and_b(mock_start_process, mock
@patch("gso.cli.imports.start_process")
def test_import_iptrunk_side_a_member_count_mismatch(mock_start_process, mock_sleep, iptrunk_data, faker, capfd):
side_a_members = [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(5)
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
+ for _ in range(5)
]
side_b_members = [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(6)
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
+ for _ in range(6)
]
broken_data = iptrunk_data(side_a_members=side_a_members, side_b_members=side_b_members)
import_iptrunks(broken_data["path"])
diff --git a/test/conftest.py b/test/conftest.py
index 076a485e2b5b868f7be7498762f6531885bec920..44e2901b6623a1105ca2d5b8f9c0a133fe162e5f 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -4,6 +4,7 @@ import ipaddress
import logging
import os
from pathlib import Path
+from uuid import UUID
import orchestrator
import pytest
@@ -36,7 +37,6 @@ from urllib3_mock import Responses
import gso.services.mailer
from gso.services.partners import PartnerSchema, create_partner
from gso.services.subscriptions import is_resource_type_value_unique
-from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import * # noqa: F403
logger = logging.getLogger("faker.factory")
@@ -67,6 +67,10 @@ def pytest_configure(config):
class FakerProvider(BaseProvider):
+ def uuid(self) -> UUID:
+ """Return a faker UUID4 string wrapped in a UUID object to avoid IDE type hint warning."""
+ return UUID(self.generator.uuid4())
+
def ipv4_network(self, *, min_subnet=1, max_subnet=32) -> ipaddress.IPv4Network:
subnet = str(self.generator.random_int(min=min_subnet, max=max_subnet))
ipv4 = self.generator.ipv4()
@@ -125,31 +129,19 @@ class FakerProvider(BaseProvider):
def ipv6_netmask(self) -> int:
return self.generator.random_int(min=1, max=128)
- def network_interface(self) -> str:
+ def nokia_physical_interface_name(self) -> str:
+ return self.generator.numerify("1/x1/%/c@%/%")
+
+ def juniper_physical_interface_name(self) -> str:
interface = self.generator.random_choices(elements=("ge", "et", "xe"))[0]
- number = self.generator.numerify("-%/%%/%%")
+ number = self.generator.numerify("-%/@%/@%")
return f"{interface}{number}"
- def juniper_ae_interface_name(self) -> str:
- return self.generator.numerify("ae@#")
-
def nokia_lag_interface_name(self) -> str:
- return self.generator.numerify("lag-3#")
-
- def link_members_juniper(self) -> LAGMemberList[LAGMember]:
- iface_amount = self.generator.random_int(min=2, max=5)
- interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]
- return [
- LAGMember(interface_name=interface_name, interface_description=self.generator.sentence())
- for interface_name in interface_names
- ]
+ return self.generator.numerify("lag-@%")
- def link_members_nokia(self) -> LAGMemberList[LAGMember]:
- iface_amount = self.generator.random_int(min=2, max=5)
- return [
- LAGMember(interface_name=f"Interface{i}", interface_description=self.generator.sentence())
- for i in range(iface_amount)
- ]
+ def juniper_ae_interface_name(self) -> str:
+ return self.generator.numerify("ae@%")
def vlan_id(self) -> int:
return self.generator.random_int(min=1, max=4095)
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index 7ec1e0d9357e1e6174116c409f4e400a4ad78b95..57f8481c380c83f355074662ed8d0124d3e6573b 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/test/fixtures/edge_port_fixtures.py
@@ -56,7 +56,7 @@ def edge_port_subscription_factory(faker, geant_partner, router_subscription_fac
edge_port_subscription.edge_port.edge_port_description = description or faker.text(max_nb_chars=30)
edge_port_subscription.edge_port.ga_id = ga_id or faker.ga_id()
edge_port_subscription.edge_port.node = node
- edge_port_subscription.edge_port.edge_port_name = name or f"lag-{faker.pyint(21, 50)}"
+ edge_port_subscription.edge_port.edge_port_name = name or faker.nokia_lag_interface_name()
edge_port_subscription.edge_port.edge_port_description = edge_port_description or faker.sentence()
edge_port_subscription.edge_port.enable_lacp = enable_lacp
edge_port_subscription.edge_port.encapsulation = encapsulation
@@ -68,13 +68,17 @@ def edge_port_subscription_factory(faker, geant_partner, router_subscription_fac
edge_port_subscription.edge_port.custom_service_name = custom_service_name or faker.text(max_nb_chars=30)
edge_port_subscription.edge_port.edge_port_ae_members = edge_port_ae_members or [
EdgePortAEMemberBlock.new(
- faker.uuid4(),
- interface_name="Interface2",
+ faker.uuid(),
+ interface_name=faker.nokia_physical_interface_name()
+ if node.vendor == Vendor.NOKIA
+ else faker.juniper_physical_interface_name(),
interface_description=faker.sentence(),
),
EdgePortAEMemberBlock.new(
- faker.uuid4(),
- interface_name="Interface3",
+ faker.uuid(),
+ interface_name=faker.nokia_physical_interface_name()
+ if node.vendor == Vendor.NOKIA
+ else faker.juniper_physical_interface_name(),
interface_description=faker.sentence(),
),
]
diff --git a/test/fixtures/iptrunk_fixtures.py b/test/fixtures/iptrunk_fixtures.py
index 89a61feb25425aaf9ed1a94823423c25d326f411..94f409d47461529cf8d2d858f7e321145daa6d7d 100644
--- a/test/fixtures/iptrunk_fixtures.py
+++ b/test/fixtures/iptrunk_fixtures.py
@@ -27,22 +27,22 @@ def iptrunk_side_subscription_factory(router_subscription_factory, faker):
side_node_access_via_ts: bool | None = False,
) -> IptrunkSideBlock:
return IptrunkSideBlock.new(
- faker.uuid4(),
+ faker.uuid(),
iptrunk_side_node=iptrunk_side_node.router
if iptrunk_side_node
else router_subscription_factory(vendor=Vendor.NOKIA, router_access_via_ts=side_node_access_via_ts).router,
- iptrunk_side_ae_iface=iptrunk_side_ae_iface or faker.pystr(),
+ iptrunk_side_ae_iface=iptrunk_side_ae_iface or faker.nokia_lag_interface_name(),
ga_id=ga_id or faker.ga_id(),
iptrunk_side_ae_members=iptrunk_side_ae_members
or [
IptrunkInterfaceBlock.new(
- faker.uuid4(),
- interface_name=faker.network_interface(),
+ faker.uuid(),
+ interface_name=faker.nokia_physical_interface_name(),
interface_description=faker.sentence(),
),
IptrunkInterfaceBlock.new(
- faker.uuid4(),
- interface_name=faker.network_interface(),
+ faker.uuid(),
+ interface_name=faker.nokia_physical_interface_name(),
interface_description=faker.sentence(),
),
],
diff --git a/test/fixtures/lan_switch_interconnect_fixtures.py b/test/fixtures/lan_switch_interconnect_fixtures.py
index 3879441ee8390cc6d2ee0b7af8a6b11ae8195e5e..50cea443cf5c70cf1593193137b2b8cd4b8496eb 100644
--- a/test/fixtures/lan_switch_interconnect_fixtures.py
+++ b/test/fixtures/lan_switch_interconnect_fixtures.py
@@ -4,7 +4,6 @@ import pytest
from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle
-from pydantic_forms.types import UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.lan_switch_interconnect import (
@@ -17,6 +16,7 @@ from gso.products.product_types.lan_switch_interconnect import (
LanSwitchInterconnectInactive,
)
from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.shared_enums import Vendor
from gso.utils.types.virtual_identifiers import (
DEFAULT_DCN_MANAGEMENT_VLAN_ID,
DEFAULT_SWITCH_MANAGEMENT_VLAN_ID,
@@ -37,7 +37,7 @@ def lan_switch_interconnect_subscription_factory(
minimum_links: int | None = None,
switch_management_vlan_id: VLAN_ID | None = None,
dcn_management_vlan_id: VLAN_ID | None = None,
- router_side_node: UUIDstr | None = None,
+ router_side_node: dict | None = None,
router_side_ae_iface: str | None = None,
router_side_ae_members: list[dict[str, str]] | None = None,
switch_side_switch: SubscriptionModel | None = None,
@@ -55,15 +55,22 @@ def lan_switch_interconnect_subscription_factory(
product_id = get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT)
subscription = LanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])
+ router_side_node = router_side_node or router_subscription_factory().router
router_side_ae_members = router_side_ae_members or [
LanSwitchInterconnectInterfaceBlockInactive.new(
- uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
+ uuid4(),
+ interface_name=(
+ faker.nokia_physical_interface_name()
+ if router_side_node.vendor == Vendor.NOKIA
+ else faker.juniper_physical_interface_name()
+ ),
+ interface_description=faker.sentence(),
)
for _ in range(2)
]
switch_side_ae_members = switch_side_ae_members or [
LanSwitchInterconnectInterfaceBlockInactive.new(
- uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
+ uuid4(), interface_name=faker.juniper_physical_interface_name(), interface_description=faker.sentence()
)
for _ in range(2)
]
@@ -74,14 +81,19 @@ def lan_switch_interconnect_subscription_factory(
subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
uuid4(),
- node=router_side_node or router_subscription_factory().router,
- ae_iface=router_side_ae_iface or faker.network_interface(),
+ node=router_side_node,
+ ae_iface=router_side_ae_iface
+ or (
+ faker.nokia_lag_interface_name()
+ if router_side_node.vendor == Vendor.NOKIA
+ else faker.juniper_ae_interface_name()
+ ),
ae_members=router_side_ae_members,
)
subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
uuid4(),
switch=switch_side_switch.site if switch_side_switch else switch_subscription_factory().switch,
- ae_iface=switch_side_ae_iface or faker.network_interface(),
+ ae_iface=switch_side_ae_iface or faker.juniper_ae_interface_name(),
ae_members=switch_side_ae_members,
)
subscription.lan_switch_interconnect.dcn_management_vlan_id = (
diff --git a/test/services/conftest.py b/test/services/conftest.py
index ee0c59e1ea76e0e12a1ee69365cbe9e71109d0d6..537660f69913ff28747d97d6e852fb3394c53f4f 100644
--- a/test/services/conftest.py
+++ b/test/services/conftest.py
@@ -22,23 +22,24 @@ class MockedNetboxClient:
@staticmethod
def get_available_lags() -> list[str]:
- return [f"lag-{lag}" for lag in range(1, 5)]
+ return [f"lag-{lag}" for lag in range(100)]
@staticmethod
def get_available_services_lags() -> list[str]:
- return [f"lag-{lag}" for lag in range(21, 50)]
+ return [f"lag-{lag}" for lag in range(100)]
@staticmethod
def get_available_interfaces():
- interfaces = []
- for interface in range(5):
- interface_data = {
- "name": f"Interface{interface}",
- "module": {"display": f"Module{interface}"},
- "description": f"Description{interface}",
+ return [
+ {
+ "name": f"1/x1/{i}/c{j}/{k}",
+ "module": {"display": f"Module{i}"},
+ "description": f"Description for interface 1/x1/{i}/c{j}/{k}",
}
- interfaces.append(interface_data)
- return interfaces
+ for i in range(10)
+ for j in range(100)
+ for k in range(10)
+ ]
def create_interface(self):
return self.BaseMockObject(id=1, name="test")
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index ca8d2c034cffce6176f0a127bba7e01795d517b7..f68cd05230fee82143cd5a6abb3491d06d457bf6 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -43,14 +43,14 @@ def generate_tt_numbers(faker, request):
def test_non_nokia_router_returns_none(mock_router, faker):
mock_router.from_subscription.return_value.router.vendor = Vendor.JUNIPER
- result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", [])
+ result = available_interfaces_choices_including_current_members(faker.uuid(), "10G", [])
assert result is None
def test_nokia_router_with_no_interfaces_returns_empty_choice(mock_router, mock_netbox_client, faker):
mock_router.from_subscription.return_value.router.vendor = Vendor.NOKIA
mock_netbox_client().get_available_interfaces.return_value = iter([])
- result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", [])
+ result = available_interfaces_choices_including_current_members(faker.uuid(), "10G", [])
assert len(result) == 0
@@ -71,12 +71,12 @@ def test_nokia_router_with_interfaces_returns_choice(mock_router, mock_netbox_cl
IptrunkInterfaceBlock(
interface_name="interface3",
interface_description="desc3",
- owner_subscription_id=faker.uuid4(),
- subscription_instance_id=faker.uuid4(),
+ owner_subscription_id=faker.uuid(),
+ subscription_instance_id=faker.uuid(),
),
]
- result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", interfaces)
+ result = available_interfaces_choices_including_current_members(faker.uuid(), "10G", interfaces)
assert len(result) == 3
assert hasattr(result, "interface1")
diff --git a/test/workflows/edge_port/test_create_edge_port.py b/test/workflows/edge_port/test_create_edge_port.py
index b73b9fe0af72b099e93e1e98b779346e11c48c61..39ac848fb8e7647bc7b64896ceebc6895d672efb 100644
--- a/test/workflows/edge_port/test_create_edge_port.py
+++ b/test/workflows/edge_port/test_create_edge_port.py
@@ -68,10 +68,12 @@ def input_form_wizard_data(request, router_subscription_factory, partner_factory
"description": faker.sentence(),
"ae_members": [
{
- "interface_name": faker.network_interface() if vendor == Vendor.JUNIPER else f"Interface{i}",
+ "interface_name": faker.juniper_physical_interface_name()
+ if vendor == Vendor.JUNIPER
+ else faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
}
- for i in range(2)
+ for _ in range(2)
],
}
summary_view_step = {}
diff --git a/test/workflows/edge_port/test_create_imported_edge_port.py b/test/workflows/edge_port/test_create_imported_edge_port.py
index 4563c8e503312ae098defd7e17283afdc1645850..d13614eb1e0b5328e55c506875ac507e290ebf84 100644
--- a/test/workflows/edge_port/test_create_imported_edge_port.py
+++ b/test/workflows/edge_port/test_create_imported_edge_port.py
@@ -15,7 +15,7 @@ def imported_edge_port_creation_input_form_data(router_subscription_factory, par
"service_type": EdgePortType.CUSTOMER,
"speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
"encapsulation": EncapsulationType.DOT1Q,
- "name": "lag34",
+ "name": faker.nokia_lag_interface_name(),
"minimum_links": 2,
"ga_id": faker.imported_ga_id(),
"mac_address": faker.mac_address(),
@@ -24,11 +24,11 @@ def imported_edge_port_creation_input_form_data(router_subscription_factory, par
"ignore_if_down": False,
"ae_members": [
{
- "interface_name": faker.network_interface(),
+ "interface_name": faker.juniper_physical_interface_name(),
"interface_description": faker.sentence(),
},
{
- "interface_name": faker.network_interface(),
+ "interface_name": faker.juniper_physical_interface_name(),
"interface_description": faker.sentence(),
},
],
diff --git a/test/workflows/edge_port/test_migrate_edge_port.py b/test/workflows/edge_port/test_migrate_edge_port.py
index 85ab31cfc826bacd30cde8523583d46a35f56341..d81536c6c9d6c848e82f85ebd11dc6677e928880 100644
--- a/test/workflows/edge_port/test_migrate_edge_port.py
+++ b/test/workflows/edge_port/test_migrate_edge_port.py
@@ -54,14 +54,14 @@ def input_form_wizard_data(request, router_subscription_factory, partner, faker)
"node": str(router_subscription_factory(vendor=Vendor.NOKIA).subscription_id),
}
create_edge_port_interface_step = {
- "name": "lag-21",
+ "name": faker.nokia_lag_interface_name(),
"description": faker.sentence(),
"ae_members": [
{
- "interface_name": f"Interface{interface}",
+ "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
}
- for interface in range(2)
+ for _ in range(2)
],
}
summary_view_step = {}
@@ -124,5 +124,8 @@ def test_successful_edge_port_migration(
assert subscription.status == "active"
router_fqdn = Router.from_subscription(input_form_wizard_data[0]["node"]).router.router_fqdn
assert subscription.edge_port.ga_id is not None
- assert subscription.description == f"Edge Port lag-21 on {router_fqdn}, GAAR, {subscription.edge_port.ga_id}"
+ assert (
+ subscription.description
+ == f"Edge Port {input_form_wizard_data[1]["name"]} on {router_fqdn}, GAAR, {subscription.edge_port.ga_id}"
+ )
assert len(subscription.edge_port.edge_port_ae_members) == 2
diff --git a/test/workflows/edge_port/test_modify_edge_port.py b/test/workflows/edge_port/test_modify_edge_port.py
index 987344969177779c2b321d2a69716ce11a1c563a..aa915148c95dba7390bffbb9138f0333f1cf2294 100644
--- a/test/workflows/edge_port/test_modify_edge_port.py
+++ b/test/workflows/edge_port/test_modify_edge_port.py
@@ -4,6 +4,7 @@ import pytest
from pydantic_forms.exceptions import FormValidationError
from gso.products.product_types.edge_port import EdgePort
+from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import (
assert_complete,
@@ -15,40 +16,43 @@ from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
@pytest.fixture()
-def input_form_wizard_data(request, faker, edge_port_subscription_factory, partner_factory):
- subscription_id = str(edge_port_subscription_factory().subscription_id)
-
- return [
- {"subscription_id": subscription_id},
- {
- "tt_number": faker.tt_number(),
- "ga_id": faker.ga_id(),
- "member_speed": PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND,
- "number_of_members": 1,
- "custom_service_name": faker.sentence(),
- },
- {
- "description": faker.sentence(),
- "ae_members": [
- {
- "interface_name": "Interface1",
- "interface_description": faker.sentence(),
- }
- ],
- },
- ]
+def input_form_wizard_data(
+ request, faker, router_subscription_factory, edge_port_subscription_factory, partner_factory
+):
+ def _input_form_data_generator(vendor: Vendor) -> list[dict]:
+ router = router_subscription_factory(vendor=vendor).router
+ subscription_id = str(edge_port_subscription_factory(node=router).subscription_id)
+
+ return [
+ {"subscription_id": subscription_id},
+ {
+ "tt_number": faker.tt_number(),
+ "ga_id": faker.ga_id(),
+ "member_speed": PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND,
+ "number_of_members": 1,
+ "custom_service_name": faker.sentence(),
+ },
+ {
+ "description": faker.sentence(),
+ "ae_members": [
+ {
+ "interface_name": faker.nokia_physical_interface_name()
+ if vendor == Vendor.NOKIA
+ else faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ ],
+ },
+ ]
+
+ return _input_form_data_generator
@pytest.mark.workflow()
@pytest.mark.parametrize("invalid_ga_id", ["GS-11111", "GA-1234", "GA_12345", "GA-100000"])
-def test_modify_edge_port_with_invalid_ga_id(
- input_form_wizard_data, faker, invalid_ga_id, iptrunk_side_subscription_factory, iptrunk_subscription_factory
-):
- input_data = input_form_wizard_data
+def test_modify_edge_port_with_invalid_ga_id(input_form_wizard_data, faker, invalid_ga_id):
+ input_data = input_form_wizard_data(vendor=Vendor.NOKIA)
input_data[1]["ga_id"] = invalid_ga_id
- iptrunk_subscription_factory(
- iptrunk_sides=[iptrunk_side_subscription_factory(ga_id="GA-11111"), iptrunk_side_subscription_factory()]
- )
# Run workflow
with pytest.raises(FormValidationError):
@@ -56,6 +60,7 @@ def test_modify_edge_port_with_invalid_ga_id(
@pytest.mark.workflow()
+@pytest.mark.parametrize("router_vendor", Vendor.values())
@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag")
@patch("gso.services.netbox_client.NetboxClient.reserve_interface")
@@ -72,20 +77,23 @@ def test_modify_edge_port_with_changing_capacity(
mocked_attach_interface_to_lag,
mocked_get_available_interfaces,
input_form_wizard_data,
+ router_vendor: Vendor,
faker,
):
# Set up mock return values
- mocked_netbox = MockedNetboxClient()
- mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
- mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
- mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
- mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
- mocked_free_interface.return_value = mocked_netbox.free_interface()
- mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
- mocked_get_interface_by_name_and_device.side_effect = mocked_netbox.get_interface_by_name_and_device
+ if router_vendor == Vendor.NOKIA:
+ mocked_netbox = MockedNetboxClient()
+ mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
+ mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
+ mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
+ mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
+ mocked_free_interface.return_value = mocked_netbox.free_interface()
+ mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
+ mocked_get_interface_by_name_and_device.side_effect = mocked_netbox.get_interface_by_name_and_device
+ input_form_data = input_form_wizard_data(vendor=router_vendor)
# Run workflow
- result, process_stat, step_log = run_workflow("modify_edge_port", input_form_wizard_data)
+ result, process_stat, step_log = run_workflow("modify_edge_port", input_form_data)
for _ in range(2):
result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
@@ -99,36 +107,44 @@ def test_modify_edge_port_with_changing_capacity(
assert subscription.status == "active"
# The number of members have been changed from 2 to 1
- assert mocked_reserve_interface.call_count == 1
- assert mocked_attach_interface_to_lag.call_count == 1
- assert mocked_free_interface.call_count == 2
- assert mocked_detach_interfaces_from_lag.call_count == 1
- assert subscription.edge_port.ga_id == input_form_wizard_data[1]["ga_id"]
+ assert mocked_reserve_interface.call_count == (1 if router_vendor == Vendor.NOKIA else 0)
+ assert mocked_attach_interface_to_lag.call_count == (1 if router_vendor == Vendor.NOKIA else 0)
+ assert mocked_free_interface.call_count == (2 if router_vendor == Vendor.NOKIA else 0)
+ assert mocked_detach_interfaces_from_lag.call_count == (1 if router_vendor == Vendor.NOKIA else 0)
+ assert subscription.edge_port.ga_id == input_form_data[1]["ga_id"]
assert len(subscription.edge_port.edge_port_ae_members) == 1
@pytest.fixture()
-def input_form_wizard_without_changing_capacity(request, faker, edge_port_subscription_factory, partner_factory):
- subscription_id = str(edge_port_subscription_factory().subscription_id)
- subscription = EdgePort.from_subscription(subscription_id)
-
- return [
- {"subscription_id": subscription_id},
- {"tt_number": faker.tt_number(), "ga_id": faker.ga_id()},
- {
- "description": faker.sentence(),
- "ae_members": [
- {
- "interface_name": interface.interface_name,
- "interface_description": interface.interface_description,
- }
- for interface in subscription.edge_port.edge_port_ae_members
- ],
- },
- ]
+def input_form_wizard_without_changing_capacity(
+ request, faker, router_subscription_factory, edge_port_subscription_factory, partner_factory
+):
+ def _input_form_wizard(router_vendor: Vendor) -> list[dict]:
+ router = router_subscription_factory(vendor=router_vendor).router
+ subscription_id = str(edge_port_subscription_factory(node=router).subscription_id)
+ subscription = EdgePort.from_subscription(subscription_id)
+
+ form_data: list[dict] = [
+ {"subscription_id": subscription_id},
+ {"tt_number": faker.tt_number(), "ga_id": faker.ga_id()},
+ {
+ "description": faker.sentence(),
+ "ae_members": [
+ {
+ "interface_name": interface.interface_name,
+ "interface_description": interface.interface_description,
+ }
+ for interface in subscription.edge_port.edge_port_ae_members
+ ],
+ },
+ ]
+ return form_data
+
+ return _input_form_wizard
@pytest.mark.workflow()
+@pytest.mark.parametrize("router_vendor", Vendor.values())
@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag")
@patch("gso.services.netbox_client.NetboxClient.reserve_interface")
@@ -145,20 +161,23 @@ def test_modify_edge_port_without_changing_capacity(
mocked_attach_interface_to_lag,
mocked_get_available_interfaces,
input_form_wizard_without_changing_capacity,
+ router_vendor: Vendor,
faker,
):
# Set up mock return values
- mocked_netbox = MockedNetboxClient()
- mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
- mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
- mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
- mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
- mocked_free_interface.return_value = mocked_netbox.free_interface()
- mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
- mocked_get_interface_by_name_and_device.side_effect = mocked_netbox.get_interface_by_name_and_device
+ if router_vendor == Vendor.NOKIA:
+ mocked_netbox = MockedNetboxClient()
+ mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
+ mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
+ mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
+ mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
+ mocked_free_interface.return_value = mocked_netbox.free_interface()
+ mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
+ mocked_get_interface_by_name_and_device.side_effect = mocked_netbox.get_interface_by_name_and_device
+ input_form = input_form_wizard_without_changing_capacity(router_vendor)
# Run workflow
- result, _, _ = run_workflow("modify_edge_port", input_form_wizard_without_changing_capacity)
+ result, _, _ = run_workflow("modify_edge_port", input_form)
assert_complete(result)
state = extract_state(result)
@@ -173,6 +192,6 @@ def test_modify_edge_port_without_changing_capacity(
assert mocked_free_interface.call_count == 0
assert mocked_detach_interfaces_from_lag.call_count == 0
- assert subscription.edge_port.ga_id == input_form_wizard_without_changing_capacity[1]["ga_id"]
+ assert subscription.edge_port.ga_id == input_form[1]["ga_id"]
assert len(subscription.edge_port.edge_port_ae_members) == 2
- assert subscription.edge_port.edge_port_description == input_form_wizard_without_changing_capacity[2]["description"]
+ assert subscription.edge_port.edge_port_description == input_form[2]["description"]
diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py
index bd38256cd2b1f4b0df777e80228cbc7b19ac9435..7f83775bb0d8ba1309bc008a8440f90b541e0f53 100644
--- a/test/workflows/iptrunk/test_create_imported_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py
@@ -24,16 +24,18 @@ def workflow_input_data(faker, router_subscription_factory):
"iptrunk_isis_metric": 10000,
"iptrunk_description_suffix": faker.word(),
"side_a_node_id": str(router_subscription_factory().subscription_id),
- "side_a_ae_iface": faker.network_interface(),
+ "side_a_ae_iface": faker.nokia_lag_interface_name(),
"side_a_ga_id": faker.imported_ga_id(),
"side_a_ae_members": [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(3)
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
+ for _ in range(3)
],
"side_b_node_id": str(router_subscription_factory().subscription_id),
- "side_b_ae_iface": faker.network_interface(),
+ "side_b_ae_iface": faker.nokia_lag_interface_name(),
"side_b_ga_id": faker.imported_ga_id(),
"side_b_ae_members": [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(3)
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
+ for _ in range(3)
],
"iptrunk_ipv4_network": faker.ipv4_network(max_subnet=31),
"iptrunk_ipv6_network": faker.ipv6_network(max_subnet=126),
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 95522cfe16af44c6b135d3cb509b9b33f29e0e07..6996e20a03cf1c69ab0afa7e05e3f3fdd80d60ed 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -53,12 +53,18 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
# Set side b router to Juniper
if vendor == Vendor.JUNIPER:
router_side_b = str(router_subscription_factory(vendor=Vendor.JUNIPER).subscription_id)
- side_b_members = faker.link_members_juniper()[0:2]
+ side_b_members = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
else:
router_side_b = str(router_subscription_factory().subscription_id)
side_b_members = [
- {"interface_name": f"Interface{interface}", "interface_description": faker.sentence()}
- for interface in range(2)
+ {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
+ for _ in range(2)
]
create_ip_trunk_step = {
@@ -72,18 +78,18 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
create_ip_trunk_confirm_step = {"iptrunk_minimum_links": 1}
create_ip_trunk_side_a_router_name = {"side_a_node_id": router_side_a}
create_ip_trunk_side_a_step = {
- "side_a_ae_iface": "lag-1",
+ "side_a_ae_iface": faker.nokia_lag_interface_name(),
"side_a_ae_members": [
{
- "interface_name": f"Interface{interface}",
+ "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(),
}
- for interface in range(2)
+ for _ in range(2)
],
}
create_ip_trunk_side_b_router_name = {"side_b_node_id": router_side_b}
create_ip_trunk_side_b_step = {
- "side_b_ae_iface": "lag-4",
+ "side_b_ae_iface": faker.nokia_lag_interface_name(),
"side_b_ae_members": side_b_members,
}
summary_view_step = {}
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
index d2bd637b248bdda453d6a1f68685562a4dcacf7c..179788307a3b963858aedc80f7e7d27b2473b89d 100644
--- a/test/workflows/iptrunk/test_migrate_iptrunk.py
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -27,10 +27,22 @@ def migrate_form_input(
iptrunk_subscription_factory,
router_subscription_factory,
iptrunk_side_subscription_factory,
-):
+) -> list[dict]:
use_juniper = getattr(request, "param", UseJuniperSide.NONE)
- new_side_ae_members_nokia = faker.link_members_nokia()[0:2]
- new_side_ae_members_juniper = faker.link_members_juniper()[0:2]
+ new_side_ae_members_nokia = [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
+ new_side_ae_members_juniper = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
if use_juniper == UseJuniperSide.SIDE_A:
# Nokia -> Juniper
@@ -38,7 +50,7 @@ def migrate_form_input(
new_router = str(router_subscription_factory(vendor=Vendor.JUNIPER, router_access_via_ts=False).subscription_id)
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id)
new_side_ae_members = new_side_ae_members_juniper
- lag_name = "ae1"
+ lag_name = faker.juniper_ae_interface_name()
elif use_juniper == UseJuniperSide.SIDE_B:
# Juniper -> Nokia
old_side_a_node = router_subscription_factory(vendor=Vendor.JUNIPER)
@@ -69,7 +81,7 @@ def migrate_form_input(
new_router = str(router_subscription_factory().subscription_id)
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
new_side_ae_members = new_side_ae_members_nokia
- lag_name = "lag-1"
+ lag_name = faker.nokia_lag_interface_name()
return [
{"subscription_id": str(old_subscription.subscription_id)},
@@ -94,7 +106,8 @@ def interface_lists_are_equal(list1, list2):
for item1 in list1:
if not any(
- item1.interface_name == item2.interface_name and item1.interface_description == item2.interface_description
+ item1["interface_name"] == item2.interface_name
+ and item1["interface_description"] == item2.interface_description
for item2 in list2
):
return False
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 4b0f2577208fc5b001dbc2fa90865596445a7be1..4e72fbd72893222cd5f6fc516c12dfc08e7b27ca 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -6,7 +6,7 @@ from pydantic_forms.exceptions import FormValidationError
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.iptrunk import Iptrunk
from gso.utils.shared_enums import Vendor
-from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
+from gso.utils.types.interfaces import PhysicalPortCapacity
from test.conftest import UseJuniperSide
from test.workflows import (
assert_complete,
@@ -30,26 +30,74 @@ def input_form_iptrunk_data(
side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node)
side_b_node = iptrunk_side_subscription_factory()
- new_side_a_ae_members = faker.link_members_juniper()[0:2]
- new_side_b_ae_members = faker.link_members_nokia()[0:2]
+ new_side_a_ae_members = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
+ new_side_b_ae_members = [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
elif use_juniper == UseJuniperSide.SIDE_B:
side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory()
side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node)
- new_side_a_ae_members = faker.link_members_nokia()[0:2]
- new_side_b_ae_members = faker.link_members_juniper()[0:2]
+ new_side_a_ae_members = [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
+ new_side_b_ae_members = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
elif use_juniper == UseJuniperSide.SIDE_BOTH:
side_node_1 = router_subscription_factory(vendor=Vendor.JUNIPER)
side_node_2 = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_1)
side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_2)
- new_side_a_ae_members = faker.link_members_juniper()[0:2]
- new_side_b_ae_members = faker.link_members_juniper()[0:2]
+ new_side_a_ae_members = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
+ new_side_b_ae_members = [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
else:
side_a_node = iptrunk_side_subscription_factory()
side_b_node = iptrunk_side_subscription_factory()
- new_side_a_ae_members = faker.link_members_nokia()[0:2]
- new_side_b_ae_members = faker.link_members_nokia()[0:2]
+ new_side_a_ae_members = [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
+ new_side_b_ae_members = [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ]
product_id = str(iptrunk_subscription_factory(iptrunk_sides=[side_a_node, side_b_node]).subscription_id)
@@ -105,7 +153,7 @@ def test_iptrunk_modify_trunk_interface_success(
mocked_reserve_interface,
mocked_attach_interface_to_lag,
mocked_get_available_interfaces,
- input_form_iptrunk_data,
+ input_form_iptrunk_data: list[dict],
faker,
):
# Set up mock return values
@@ -174,9 +222,9 @@ def test_iptrunk_modify_trunk_interface_success(
assert subscription.iptrunk.iptrunk_sides[0].ga_id == new_side_a_gid
assert subscription.iptrunk.iptrunk_description_suffix == input_form_iptrunk_data[1]["iptrunk_description_suffix"]
- def _find_interface_by_name(interfaces: LAGMemberList, name: str):
+ def _find_interface_by_name(interfaces: list[dict], name: str):
for interface in interfaces:
- if interface.interface_name == name:
+ if interface["interface_name"] == name:
return interface
msg = f"Interface {name} not found!"
raise IndexError(msg)
@@ -184,7 +232,7 @@ def test_iptrunk_modify_trunk_interface_success(
for member in subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members:
assert (
member.interface_description
- == _find_interface_by_name(new_side_a_ae_members, member.interface_name).interface_description
+ == _find_interface_by_name(new_side_a_ae_members, member.interface_name)["interface_description"]
)
assert subscription.iptrunk.iptrunk_sides[1].ga_id == new_side_b_gid
@@ -192,7 +240,7 @@ def test_iptrunk_modify_trunk_interface_success(
for member in subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members:
assert (
member.interface_description
- == _find_interface_by_name(new_side_b_ae_members, member.interface_name).interface_description
+ == _find_interface_by_name(new_side_b_ae_members, member.interface_name)["interface_description"]
)
diff --git a/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
index a04f7e42674b61152429d88c631feb82257f22d2..bce3d1fa01b7bbd3a431e0ed160c2dcf8c60c0de 100644
--- a/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
@@ -19,12 +19,22 @@ def workflow_input_data(faker, router_subscription_factory, switch_subscription_
"router_side": {
"node": str(router_subscription_factory().subscription_id),
"ae_iface": faker.nokia_lag_interface_name(),
- "ae_members": faker.link_members_nokia(),
+ "ae_members": [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ ],
},
"switch_side": {
"switch": str(switch_subscription_factory().subscription_id),
"ae_iface": faker.juniper_ae_interface_name(),
- "ae_members": faker.link_members_juniper(),
+ "ae_members": [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ ],
},
}
diff --git a/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
index f7a8fab1cd21ea7ce5904aac458927b2a16decc6..f5e423c16a7b8428627518d42f3e1d76c2e4e1f1 100644
--- a/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
@@ -47,12 +47,24 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact
"minimum_link_count": 2,
},
{
- "router_side_iface": "lag-1",
- "router_side_ae_members": faker.link_members_nokia()[:2],
+ "router_side_iface": faker.nokia_lag_interface_name(),
+ "router_side_ae_members": [
+ {
+ "interface_name": faker.nokia_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ],
},
{
- "switch_side_iface": faker.network_interface(),
- "switch_side_ae_members": faker.link_members_juniper()[:2],
+ "switch_side_iface": faker.juniper_ae_interface_name(),
+ "switch_side_ae_members": [
+ {
+ "interface_name": faker.juniper_physical_interface_name(),
+ "interface_description": faker.sentence(),
+ }
+ for _ in range(2)
+ ],
},
{},
]