Skip to content
Snippets Groups Projects
Verified Commit 0a30ba67 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Rework the faker providers for generating Juniper and Nokia interface names

parent 9f5ac783
Branches
Tags
1 merge request!415Juniper Edge Port modification
Pipeline #93863 passed
Showing
with 211 additions and 112 deletions
...@@ -79,7 +79,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: ...@@ -79,7 +79,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
lag_member = JuniperLAGMember if subscription.edge_port.node.vendor == Vendor.JUNIPER else NokiaEdgePortLAGMember lag_member = JuniperLAGMember if subscription.edge_port.node.vendor == Vendor.JUNIPER else NokiaEdgePortLAGMember
lag_ae_members = Annotated[ lag_ae_members = Annotated[
list[lag_member], list[lag_member], # type: ignore[valid-type]
AfterValidator(validate_unique_list), AfterValidator(validate_unique_list),
Len( Len(
min_length=user_input.number_of_members, min_length=user_input.number_of_members,
......
...@@ -101,7 +101,7 @@ def _initial_input_form(product_name: str) -> FormGenerator: ...@@ -101,7 +101,7 @@ def _initial_input_form(product_name: str) -> FormGenerator:
class InterconnectSwitchSideForm(FormPage): class InterconnectSwitchSideForm(FormPage):
model_config = ConfigDict(title="Please enter interface names and descriptions for the switch side.") 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_ae_members: switch_side_ae_member_list
switch_side_input = yield InterconnectSwitchSideForm switch_side_input = yield InterconnectSwitchSideForm
......
...@@ -72,12 +72,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict): ...@@ -72,12 +72,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict):
}, },
"nodeA": { "nodeA": {
"name": side_a_node or router_side_a.router.router_fqdn, "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(), "port_ga_id": faker.imported_ga_id(),
"members": side_a_members "members": side_a_members
or [ or [
{ {
"interface_name": faker.network_interface(), "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(), "interface_description": faker.sentence(),
} }
for _ in range(5) for _ in range(5)
...@@ -87,12 +87,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict): ...@@ -87,12 +87,12 @@ def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict):
}, },
"nodeB": { "nodeB": {
"name": side_b_node or router_side_b.router.router_fqdn, "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(), "port_ga_id": faker.imported_ga_id(),
"members": side_b_members "members": side_b_members
or [ or [
{ {
"interface_name": faker.network_interface(), "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(), "interface_description": faker.sentence(),
} }
for _ in range(5) for _ in range(5)
...@@ -235,7 +235,7 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory, ...@@ -235,7 +235,7 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
"node": str(router_subscription_factory().subscription_id), "node": str(router_subscription_factory().subscription_id),
"ae_iface": faker.nokia_lag_interface_name(), "ae_iface": faker.nokia_lag_interface_name(),
"ae_members": [ "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) for _ in range(2)
], ],
}, },
...@@ -243,7 +243,10 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory, ...@@ -243,7 +243,10 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
"switch": str(switch_subscription_factory().subscription_id), "switch": str(switch_subscription_factory().subscription_id),
"ae_iface": faker.juniper_ae_interface_name(), "ae_iface": faker.juniper_ae_interface_name(),
"ae_members": [ "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) for _ in range(2)
], ],
}, },
...@@ -273,13 +276,10 @@ def edge_port_data(temp_file, faker, router_subscription_factory, partner_factor ...@@ -273,13 +276,10 @@ def edge_port_data(temp_file, faker, router_subscription_factory, partner_factor
"ignore_if_down": False, "ignore_if_down": False,
"ae_members": [ "ae_members": [
{ {
"interface_name": faker.network_interface(), "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(), "interface_description": faker.sentence(),
}, }
{ for _ in range(2)
"interface_name": faker.network_interface(),
"interface_description": faker.sentence(),
},
], ],
"description": faker.sentence(), "description": faker.sentence(),
} }
...@@ -584,7 +584,10 @@ def test_import_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, mock_ ...@@ -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.time.sleep")
@patch("gso.cli.imports.start_process") @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): 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_a_members = [duplicate_interface for _ in range(5)]
side_b_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) 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 ...@@ -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") @patch("gso.cli.imports.start_process")
def test_import_iptrunk_side_a_member_count_mismatch(mock_start_process, mock_sleep, iptrunk_data, faker, capfd): def test_import_iptrunk_side_a_member_count_mismatch(mock_start_process, mock_sleep, iptrunk_data, faker, capfd):
side_a_members = [ 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 = [ 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) broken_data = iptrunk_data(side_a_members=side_a_members, side_b_members=side_b_members)
import_iptrunks(broken_data["path"]) import_iptrunks(broken_data["path"])
...@@ -701,7 +706,7 @@ def test_import_l3_core_service_with_invalid_partner(mock_start_process, mock_sl ...@@ -701,7 +706,7 @@ def test_import_l3_core_service_with_invalid_partner(mock_start_process, mock_sl
def test_import_l3_core_service_with_invalid_edge_port( def test_import_l3_core_service_with_invalid_edge_port(
mock_start_process, mock_sleep, faker, l3_core_service_data, edge_port_subscription_factory, capfd mock_start_process, mock_sleep, faker, l3_core_service_data, edge_port_subscription_factory, capfd
): ):
fake_uuid = faker.uuid() fake_uuid = faker.uuid4()
broken_data = l3_core_service_data( broken_data = l3_core_service_data(
service_binding_ports=[ service_binding_ports=[
{ {
......
...@@ -37,7 +37,6 @@ from urllib3_mock import Responses ...@@ -37,7 +37,6 @@ from urllib3_mock import Responses
import gso.services.mailer import gso.services.mailer
from gso.services.partners import PartnerSchema, create_partner from gso.services.partners import PartnerSchema, create_partner
from gso.services.subscriptions import is_resource_type_value_unique from gso.services.subscriptions import is_resource_type_value_unique
from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import * # noqa: F403 from test.fixtures import * # noqa: F403
logger = logging.getLogger("faker.factory") logger = logging.getLogger("faker.factory")
...@@ -130,31 +129,19 @@ class FakerProvider(BaseProvider): ...@@ -130,31 +129,19 @@ class FakerProvider(BaseProvider):
def ipv6_netmask(self) -> int: def ipv6_netmask(self) -> int:
return self.generator.random_int(min=1, max=128) 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] interface = self.generator.random_choices(elements=("ge", "et", "xe"))[0]
number = self.generator.numerify("-%/%%/%%") number = self.generator.numerify("-%/@%/@%")
return f"{interface}{number}" return f"{interface}{number}"
def juniper_ae_interface_name(self) -> str:
return self.generator.numerify("ae@#")
def nokia_lag_interface_name(self) -> str: def nokia_lag_interface_name(self) -> str:
return self.generator.numerify("lag-3#") return self.generator.numerify("lag-@%")
def link_members_juniper(self) -> LAGMemberList[LAGMember]: def juniper_ae_interface_name(self) -> str:
iface_amount = self.generator.random_int(min=2, max=5) return self.generator.numerify("ae@%")
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
]
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 vlan_id(self) -> int: def vlan_id(self) -> int:
return self.generator.random_int(min=1, max=4095) return self.generator.random_int(min=1, max=4095)
......
from uuid import UUID
import pytest import pytest
from orchestrator.db import db from orchestrator.db import db
from orchestrator.domain import SubscriptionModel from orchestrator.domain import SubscriptionModel
...@@ -58,7 +56,7 @@ def edge_port_subscription_factory(faker, geant_partner, router_subscription_fac ...@@ -58,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.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.ga_id = ga_id or faker.ga_id()
edge_port_subscription.edge_port.node = node 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.edge_port_description = edge_port_description or faker.sentence()
edge_port_subscription.edge_port.enable_lacp = enable_lacp edge_port_subscription.edge_port.enable_lacp = enable_lacp
edge_port_subscription.edge_port.encapsulation = encapsulation edge_port_subscription.edge_port.encapsulation = encapsulation
...@@ -71,16 +69,16 @@ def edge_port_subscription_factory(faker, geant_partner, router_subscription_fac ...@@ -71,16 +69,16 @@ def edge_port_subscription_factory(faker, geant_partner, router_subscription_fac
edge_port_subscription.edge_port.edge_port_ae_members = edge_port_ae_members or [ edge_port_subscription.edge_port.edge_port_ae_members = edge_port_ae_members or [
EdgePortAEMemberBlock.new( EdgePortAEMemberBlock.new(
faker.uuid(), faker.uuid(),
interface_name="Interface2" interface_name=faker.nokia_physical_interface_name()
if node.vendor == Vendor.NOKIA if node.vendor == Vendor.NOKIA
else faker.link_members_juniper()[0].interface_name, else faker.juniper_physical_interface_name(),
interface_description=faker.sentence(), interface_description=faker.sentence(),
), ),
EdgePortAEMemberBlock.new( EdgePortAEMemberBlock.new(
faker.uuid(), faker.uuid(),
interface_name="Interface3" interface_name=faker.nokia_physical_interface_name()
if node.vendor == Vendor.NOKIA if node.vendor == Vendor.NOKIA
else faker.link_members_juniper()[1].interface_name, else faker.juniper_physical_interface_name(),
interface_description=faker.sentence(), interface_description=faker.sentence(),
), ),
] ]
......
...@@ -31,18 +31,18 @@ def iptrunk_side_subscription_factory(router_subscription_factory, faker): ...@@ -31,18 +31,18 @@ def iptrunk_side_subscription_factory(router_subscription_factory, faker):
iptrunk_side_node=iptrunk_side_node.router iptrunk_side_node=iptrunk_side_node.router
if iptrunk_side_node if iptrunk_side_node
else router_subscription_factory(vendor=Vendor.NOKIA, router_access_via_ts=side_node_access_via_ts).router, 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(), ga_id=ga_id or faker.ga_id(),
iptrunk_side_ae_members=iptrunk_side_ae_members iptrunk_side_ae_members=iptrunk_side_ae_members
or [ or [
IptrunkInterfaceBlock.new( IptrunkInterfaceBlock.new(
faker.uuid(), faker.uuid(),
interface_name=faker.network_interface(), interface_name=faker.nokia_physical_interface_name(),
interface_description=faker.sentence(), interface_description=faker.sentence(),
), ),
IptrunkInterfaceBlock.new( IptrunkInterfaceBlock.new(
faker.uuid(), faker.uuid(),
interface_name=faker.network_interface(), interface_name=faker.nokia_physical_interface_name(),
interface_description=faker.sentence(), interface_description=faker.sentence(),
), ),
], ],
......
...@@ -4,7 +4,6 @@ import pytest ...@@ -4,7 +4,6 @@ import pytest
from orchestrator.db import db from orchestrator.db import db
from orchestrator.domain import SubscriptionModel from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle from orchestrator.types import SubscriptionLifecycle
from pydantic_forms.types import UUIDstr
from gso.products import ProductName from gso.products import ProductName
from gso.products.product_blocks.lan_switch_interconnect import ( from gso.products.product_blocks.lan_switch_interconnect import (
...@@ -17,6 +16,7 @@ from gso.products.product_types.lan_switch_interconnect import ( ...@@ -17,6 +16,7 @@ from gso.products.product_types.lan_switch_interconnect import (
LanSwitchInterconnectInactive, LanSwitchInterconnectInactive,
) )
from gso.services.subscriptions import get_product_id_by_name from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import Vendor
from gso.utils.types.virtual_identifiers import ( from gso.utils.types.virtual_identifiers import (
DEFAULT_DCN_MANAGEMENT_VLAN_ID, DEFAULT_DCN_MANAGEMENT_VLAN_ID,
DEFAULT_SWITCH_MANAGEMENT_VLAN_ID, DEFAULT_SWITCH_MANAGEMENT_VLAN_ID,
...@@ -37,7 +37,7 @@ def lan_switch_interconnect_subscription_factory( ...@@ -37,7 +37,7 @@ def lan_switch_interconnect_subscription_factory(
minimum_links: int | None = None, minimum_links: int | None = None,
switch_management_vlan_id: VLAN_ID | None = None, switch_management_vlan_id: VLAN_ID | None = None,
dcn_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_iface: str | None = None,
router_side_ae_members: list[dict[str, str]] | None = None, router_side_ae_members: list[dict[str, str]] | None = None,
switch_side_switch: SubscriptionModel | None = None, switch_side_switch: SubscriptionModel | None = None,
...@@ -55,15 +55,22 @@ def lan_switch_interconnect_subscription_factory( ...@@ -55,15 +55,22 @@ def lan_switch_interconnect_subscription_factory(
product_id = get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT) product_id = get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT)
subscription = LanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"]) 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 [ router_side_ae_members = router_side_ae_members or [
LanSwitchInterconnectInterfaceBlockInactive.new( 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) for _ in range(2)
] ]
switch_side_ae_members = switch_side_ae_members or [ switch_side_ae_members = switch_side_ae_members or [
LanSwitchInterconnectInterfaceBlockInactive.new( 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) for _ in range(2)
] ]
...@@ -74,14 +81,19 @@ def lan_switch_interconnect_subscription_factory( ...@@ -74,14 +81,19 @@ def lan_switch_interconnect_subscription_factory(
subscription.lan_switch_interconnect.minimum_links = minimum_links or 1 subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new( subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
uuid4(), uuid4(),
node=router_side_node or router_subscription_factory().router, node=router_side_node,
ae_iface=router_side_ae_iface or faker.network_interface(), 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, ae_members=router_side_ae_members,
) )
subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new( subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
uuid4(), uuid4(),
switch=switch_side_switch.site if switch_side_switch else switch_subscription_factory().switch, 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, ae_members=switch_side_ae_members,
) )
subscription.lan_switch_interconnect.dcn_management_vlan_id = ( subscription.lan_switch_interconnect.dcn_management_vlan_id = (
......
...@@ -22,23 +22,24 @@ class MockedNetboxClient: ...@@ -22,23 +22,24 @@ class MockedNetboxClient:
@staticmethod @staticmethod
def get_available_lags() -> list[str]: 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 @staticmethod
def get_available_services_lags() -> list[str]: 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 @staticmethod
def get_available_interfaces(): def get_available_interfaces():
interfaces = [] return [
for interface in range(5): {
interface_data = { "name": f"1/x1/{i}/c{j}/{k}",
"name": f"Interface{interface}", "module": {"display": f"Module{i}"},
"module": {"display": f"Module{interface}"}, "description": f"Description for interface 1/x1/{i}/c{j}/{k}",
"description": f"Description{interface}",
} }
interfaces.append(interface_data) for i in range(10)
return interfaces for j in range(100)
for k in range(10)
]
def create_interface(self): def create_interface(self):
return self.BaseMockObject(id=1, name="test") return self.BaseMockObject(id=1, name="test")
......
...@@ -68,10 +68,12 @@ def input_form_wizard_data(request, router_subscription_factory, partner_factory ...@@ -68,10 +68,12 @@ def input_form_wizard_data(request, router_subscription_factory, partner_factory
"description": faker.sentence(), "description": faker.sentence(),
"ae_members": [ "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(), "interface_description": faker.sentence(),
} }
for i in range(2) for _ in range(2)
], ],
} }
summary_view_step = {} summary_view_step = {}
......
...@@ -15,7 +15,7 @@ def imported_edge_port_creation_input_form_data(router_subscription_factory, par ...@@ -15,7 +15,7 @@ def imported_edge_port_creation_input_form_data(router_subscription_factory, par
"service_type": EdgePortType.CUSTOMER, "service_type": EdgePortType.CUSTOMER,
"speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND, "speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
"encapsulation": EncapsulationType.DOT1Q, "encapsulation": EncapsulationType.DOT1Q,
"name": "lag34", "name": faker.nokia_lag_interface_name(),
"minimum_links": 2, "minimum_links": 2,
"ga_id": faker.imported_ga_id(), "ga_id": faker.imported_ga_id(),
"mac_address": faker.mac_address(), "mac_address": faker.mac_address(),
......
...@@ -54,14 +54,14 @@ def input_form_wizard_data(request, router_subscription_factory, partner, faker) ...@@ -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), "node": str(router_subscription_factory(vendor=Vendor.NOKIA).subscription_id),
} }
create_edge_port_interface_step = { create_edge_port_interface_step = {
"name": "lag-21", "name": faker.nokia_lag_interface_name(),
"description": faker.sentence(), "description": faker.sentence(),
"ae_members": [ "ae_members": [
{ {
"interface_name": f"Interface{interface}", "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(), "interface_description": faker.sentence(),
} }
for interface in range(2) for _ in range(2)
], ],
} }
summary_view_step = {} summary_view_step = {}
...@@ -124,5 +124,8 @@ def test_successful_edge_port_migration( ...@@ -124,5 +124,8 @@ def test_successful_edge_port_migration(
assert subscription.status == "active" assert subscription.status == "active"
router_fqdn = Router.from_subscription(input_form_wizard_data[0]["node"]).router.router_fqdn 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.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 assert len(subscription.edge_port.edge_port_ae_members) == 2
...@@ -36,9 +36,9 @@ def input_form_wizard_data( ...@@ -36,9 +36,9 @@ def input_form_wizard_data(
"description": faker.sentence(), "description": faker.sentence(),
"ae_members": [ "ae_members": [
{ {
"interface_name": "Interface1" "interface_name": faker.nokia_physical_interface_name()
if vendor == Vendor.NOKIA if vendor == Vendor.NOKIA
else faker.link_members_juniper()[0].interface_name, else faker.juniper_physical_interface_name(),
"interface_description": faker.sentence(), "interface_description": faker.sentence(),
} }
], ],
......
...@@ -24,16 +24,18 @@ def workflow_input_data(faker, router_subscription_factory): ...@@ -24,16 +24,18 @@ def workflow_input_data(faker, router_subscription_factory):
"iptrunk_isis_metric": 10000, "iptrunk_isis_metric": 10000,
"iptrunk_description_suffix": faker.word(), "iptrunk_description_suffix": faker.word(),
"side_a_node_id": str(router_subscription_factory().subscription_id), "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_ga_id": faker.imported_ga_id(),
"side_a_ae_members": [ "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_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_ga_id": faker.imported_ga_id(),
"side_b_ae_members": [ "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_ipv4_network": faker.ipv4_network(max_subnet=31),
"iptrunk_ipv6_network": faker.ipv6_network(max_subnet=126), "iptrunk_ipv6_network": faker.ipv6_network(max_subnet=126),
......
...@@ -53,12 +53,18 @@ def input_form_wizard_data(request, router_subscription_factory, faker): ...@@ -53,12 +53,18 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
# Set side b router to Juniper # Set side b router to Juniper
if vendor == Vendor.JUNIPER: if vendor == Vendor.JUNIPER:
router_side_b = str(router_subscription_factory(vendor=Vendor.JUNIPER).subscription_id) 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: else:
router_side_b = str(router_subscription_factory().subscription_id) router_side_b = str(router_subscription_factory().subscription_id)
side_b_members = [ side_b_members = [
{"interface_name": f"Interface{interface}", "interface_description": faker.sentence()} {"interface_name": faker.nokia_physical_interface_name(), "interface_description": faker.sentence()}
for interface in range(2) for _ in range(2)
] ]
create_ip_trunk_step = { create_ip_trunk_step = {
...@@ -72,18 +78,18 @@ def input_form_wizard_data(request, router_subscription_factory, faker): ...@@ -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_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_router_name = {"side_a_node_id": router_side_a}
create_ip_trunk_side_a_step = { 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": [ "side_a_ae_members": [
{ {
"interface_name": f"Interface{interface}", "interface_name": faker.nokia_physical_interface_name(),
"interface_description": faker.sentence(), "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_router_name = {"side_b_node_id": router_side_b}
create_ip_trunk_side_b_step = { 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, "side_b_ae_members": side_b_members,
} }
summary_view_step = {} summary_view_step = {}
......
...@@ -27,10 +27,22 @@ def migrate_form_input( ...@@ -27,10 +27,22 @@ def migrate_form_input(
iptrunk_subscription_factory, iptrunk_subscription_factory,
router_subscription_factory, router_subscription_factory,
iptrunk_side_subscription_factory, iptrunk_side_subscription_factory,
): ) -> list[dict]:
use_juniper = getattr(request, "param", UseJuniperSide.NONE) use_juniper = getattr(request, "param", UseJuniperSide.NONE)
new_side_ae_members_nokia = faker.link_members_nokia()[0:2] new_side_ae_members_nokia = [
new_side_ae_members_juniper = faker.link_members_juniper()[0:2] {
"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: if use_juniper == UseJuniperSide.SIDE_A:
# Nokia -> Juniper # Nokia -> Juniper
...@@ -38,7 +50,7 @@ def migrate_form_input( ...@@ -38,7 +50,7 @@ def migrate_form_input(
new_router = str(router_subscription_factory(vendor=Vendor.JUNIPER, router_access_via_ts=False).subscription_id) 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) replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id)
new_side_ae_members = new_side_ae_members_juniper 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: elif use_juniper == UseJuniperSide.SIDE_B:
# Juniper -> Nokia # Juniper -> Nokia
old_side_a_node = router_subscription_factory(vendor=Vendor.JUNIPER) old_side_a_node = router_subscription_factory(vendor=Vendor.JUNIPER)
...@@ -69,7 +81,7 @@ def migrate_form_input( ...@@ -69,7 +81,7 @@ def migrate_form_input(
new_router = str(router_subscription_factory().subscription_id) new_router = str(router_subscription_factory().subscription_id)
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.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 new_side_ae_members = new_side_ae_members_nokia
lag_name = "lag-1" lag_name = faker.nokia_lag_interface_name()
return [ return [
{"subscription_id": str(old_subscription.subscription_id)}, {"subscription_id": str(old_subscription.subscription_id)},
...@@ -94,7 +106,8 @@ def interface_lists_are_equal(list1, list2): ...@@ -94,7 +106,8 @@ def interface_lists_are_equal(list1, list2):
for item1 in list1: for item1 in list1:
if not any( 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 for item2 in list2
): ):
return False return False
......
...@@ -6,7 +6,7 @@ from pydantic_forms.exceptions import FormValidationError ...@@ -6,7 +6,7 @@ from pydantic_forms.exceptions import FormValidationError
from gso.products.product_blocks.iptrunk import IptrunkType from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.iptrunk import Iptrunk from gso.products.product_types.iptrunk import Iptrunk
from gso.utils.shared_enums import Vendor 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.conftest import UseJuniperSide
from test.workflows import ( from test.workflows import (
assert_complete, assert_complete,
...@@ -30,26 +30,74 @@ def input_form_iptrunk_data( ...@@ -30,26 +30,74 @@ def input_form_iptrunk_data(
side_node = router_subscription_factory(vendor=Vendor.JUNIPER) side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node) side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node)
side_b_node = iptrunk_side_subscription_factory() side_b_node = iptrunk_side_subscription_factory()
new_side_a_ae_members = faker.link_members_juniper()[0:2] new_side_a_ae_members = [
new_side_b_ae_members = faker.link_members_nokia()[0:2] {
"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: elif use_juniper == UseJuniperSide.SIDE_B:
side_node = router_subscription_factory(vendor=Vendor.JUNIPER) side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory() side_a_node = iptrunk_side_subscription_factory()
side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node) 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_a_ae_members = [
new_side_b_ae_members = faker.link_members_juniper()[0:2] {
"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: elif use_juniper == UseJuniperSide.SIDE_BOTH:
side_node_1 = router_subscription_factory(vendor=Vendor.JUNIPER) side_node_1 = router_subscription_factory(vendor=Vendor.JUNIPER)
side_node_2 = 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_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) 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_a_ae_members = [
new_side_b_ae_members = faker.link_members_juniper()[0:2] {
"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: else:
side_a_node = iptrunk_side_subscription_factory() side_a_node = iptrunk_side_subscription_factory()
side_b_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_a_ae_members = [
new_side_b_ae_members = faker.link_members_nokia()[0:2] {
"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) 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( ...@@ -105,7 +153,7 @@ def test_iptrunk_modify_trunk_interface_success(
mocked_reserve_interface, mocked_reserve_interface,
mocked_attach_interface_to_lag, mocked_attach_interface_to_lag,
mocked_get_available_interfaces, mocked_get_available_interfaces,
input_form_iptrunk_data, input_form_iptrunk_data: list[dict],
faker, faker,
): ):
# Set up mock return values # Set up mock return values
...@@ -174,9 +222,9 @@ def test_iptrunk_modify_trunk_interface_success( ...@@ -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_sides[0].ga_id == new_side_a_gid
assert subscription.iptrunk.iptrunk_description_suffix == input_form_iptrunk_data[1]["iptrunk_description_suffix"] 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: for interface in interfaces:
if interface.interface_name == name: if interface["interface_name"] == name:
return interface return interface
msg = f"Interface {name} not found!" msg = f"Interface {name} not found!"
raise IndexError(msg) raise IndexError(msg)
...@@ -184,7 +232,7 @@ def test_iptrunk_modify_trunk_interface_success( ...@@ -184,7 +232,7 @@ def test_iptrunk_modify_trunk_interface_success(
for member in subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members: for member in subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members:
assert ( assert (
member.interface_description 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 assert subscription.iptrunk.iptrunk_sides[1].ga_id == new_side_b_gid
...@@ -192,7 +240,7 @@ def test_iptrunk_modify_trunk_interface_success( ...@@ -192,7 +240,7 @@ def test_iptrunk_modify_trunk_interface_success(
for member in subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members: for member in subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members:
assert ( assert (
member.interface_description 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"]
) )
......
...@@ -19,12 +19,22 @@ def workflow_input_data(faker, router_subscription_factory, switch_subscription_ ...@@ -19,12 +19,22 @@ def workflow_input_data(faker, router_subscription_factory, switch_subscription_
"router_side": { "router_side": {
"node": str(router_subscription_factory().subscription_id), "node": str(router_subscription_factory().subscription_id),
"ae_iface": faker.nokia_lag_interface_name(), "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_side": {
"switch": str(switch_subscription_factory().subscription_id), "switch": str(switch_subscription_factory().subscription_id),
"ae_iface": faker.juniper_ae_interface_name(), "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(),
}
],
}, },
} }
......
...@@ -47,12 +47,24 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact ...@@ -47,12 +47,24 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact
"minimum_link_count": 2, "minimum_link_count": 2,
}, },
{ {
"router_side_iface": "lag-1", "router_side_iface": faker.nokia_lag_interface_name(),
"router_side_ae_members": faker.link_members_nokia()[:2], "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_iface": faker.juniper_ae_interface_name(),
"switch_side_ae_members": faker.link_members_juniper()[:2], "switch_side_ae_members": [
{
"interface_name": faker.juniper_physical_interface_name(),
"interface_description": faker.sentence(),
}
for _ in range(2)
],
}, },
{}, {},
] ]
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment