From 28b5bcdd3844b7002de303cb8c1533df43fee87a Mon Sep 17 00:00:00 2001
From: Hakan Calim <hakan.calim@fau.de>
Date: Wed, 29 Nov 2023 19:05:11 +0100
Subject: [PATCH] NAT-329: added more tests to check different scenarios with
juniper nodes
---
.../iptrunk/test_modify_trunk_interface.py | 64 ++++++++++++++-----
1 file changed, 48 insertions(+), 16 deletions(-)
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 074219de..3424a5cc 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -4,6 +4,7 @@ import pytest
from gso.products import Iptrunk
from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
+from test.conftest import UseJuniperSide
from test.workflows import (
assert_complete,
assert_pp_interaction_success,
@@ -14,9 +15,41 @@ from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
@pytest.fixture()
-def input_form_iptrunk_data(faker, iptrunk_subscription_factory):
-
- product_id = iptrunk_subscription_factory()
+def input_form_iptrunk_data(
+ request,
+ faker,
+ iptrunk_subscription_factory,
+ juniper_router_subscription_factory,
+ nokia_router_subscription_factory,
+ iptrunk_side_subscription_factory,
+ ):
+ use_juniper= getattr(request, "param", UseJuniperSide.NONE)
+ if use_juniper == UseJuniperSide.SIDE_A:
+ side_node = juniper_router_subscription_factory()
+ 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.generate_junniper_members_list()
+ new_side_b_ae_members = faker.generate_nokia_members_list()
+ elif use_juniper == UseJuniperSide.SIDE_B:
+ side_node = juniper_router_subscription_factory()
+ 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.generate_nokia_members_list()
+ new_side_b_ae_members = faker.generate_junniper_members_list()
+ elif use_juniper == UseJuniperSide.SIDE_BOTH:
+ side_node_1 = juniper_router_subscription_factory()
+ side_node_2 = juniper_router_subscription_factory()
+ 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.generate_junniper_members_list()
+ new_side_b_ae_members = faker.generate_junniper_members_list()
+ else:
+ side_a_node = iptrunk_side_subscription_factory()
+ side_b_node = iptrunk_side_subscription_factory()
+ new_side_a_ae_members = faker.generate_nokia_members_list()
+ new_side_b_ae_members = faker.generate_nokia_members_list()
+
+ product_id = iptrunk_subscription_factory(iptrunk_sides=[side_a_node, side_b_node])
new_sid = faker.geant_sid()
new_description = faker.sentence()
@@ -25,14 +58,8 @@ def input_form_iptrunk_data(faker, iptrunk_subscription_factory):
new_link_count = 2
new_side_a_sid = faker.geant_sid()
- new_side_a_ae_members = [
- {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5)
- ]
new_side_b_sid = faker.geant_sid()
- new_side_b_ae_members = [
- {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5)
- ]
return [
{"subscription_id": product_id},
@@ -55,6 +82,10 @@ def input_form_iptrunk_data(faker, iptrunk_subscription_factory):
]
+@pytest.mark.parametrize("input_form_iptrunk_data", [UseJuniperSide.NONE,
+ UseJuniperSide.SIDE_A,
+ UseJuniperSide.SIDE_B,
+ UseJuniperSide.SIDE_BOTH], indirect=True)
@pytest.mark.workflow()
@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk")
@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
@@ -99,17 +130,18 @@ def test_iptrunk_modify_trunk_interface_success(
assert subscription.status == "active"
assert mock_provision_ip_trunk.call_count == 2
# Assert all Netbox calls have been made
- assert mocked_reserve_interface.call_count == 10 # 5 interfaces per side
- assert mocked_attach_interface_to_lag.call_count == 10 # 5 interfaces per side
- assert mocked_free_interface.call_count == 4 # 2 interfaces per side(The old ones)
- assert mocked_detach_interfaces_from_lag.call_count == 2 # 1 time per side
-
- # Assert all subscription properties have been updated correctly
new_sid = input_form_iptrunk_data[1]["geant_s_sid"]
new_side_a_sid = input_form_iptrunk_data[2]["side_a_ae_geant_a_sid"]
new_side_a_ae_members = input_form_iptrunk_data[2]["side_a_ae_members"]
new_side_b_sid = input_form_iptrunk_data[3]["side_b_ae_geant_a_sid"]
new_side_b_ae_members = input_form_iptrunk_data[3]["side_b_ae_members"]
+ num_ifaces = len(new_side_a_sid ) + len(new_side_b_sid ) # The length are randomly generated
+ assert mocked_reserve_interface.call_count == num_ifaces
+ assert mocked_attach_interface_to_lag.call_count == 10 # 5 interfaces per side
+ assert mocked_free_interface.call_count == 4 # 2 interfaces per side(The old ones)
+ assert mocked_detach_interfaces_from_lag.call_count == 2 # 1 time per side
+
+ # Assert all subscription properties have been updated correctly
assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}"
assert subscription.iptrunk.geant_s_sid == input_form_iptrunk_data[1]["geant_s_sid"]
assert subscription.iptrunk.iptrunk_description == input_form_iptrunk_data[1]["iptrunk_description"]
@@ -120,7 +152,7 @@ def test_iptrunk_modify_trunk_interface_success(
def _find_interface_by_name(interfaces: list[dict[str, str]], 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)
--
GitLab