From 261dcba2135947a0355972017f4b14eb716888bf Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Wed, 25 Oct 2023 14:59:16 +0200
Subject: [PATCH] Updated modification IP trunk workflow's unit tests.
---
gso/workflows/iptrunk/migrate_iptrunk.py | 2 +-
.../iptrunk/modify_trunk_interface.py | 10 ++---
test/services/conftest.py | 40 +++++++++++++++++++
test/workflows/iptrunk/test_create_iptrunk.py | 37 +----------------
.../iptrunk/test_modify_trunk_interface.py | 33 ++++++++++++++-
5 files changed, 78 insertions(+), 44 deletions(-)
create mode 100644 test/services/conftest.py
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index e31d0011..b34d7d2c 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -69,7 +69,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
]
routers = {}
- for router_id, router_description in get_active_router_subscriptions(fields=["subscription_id", "description"]):
+ for router_id, router_description in get_active_router_subscriptions(includes=["subscription_id", "description"]):
if router_id not in current_routers:
current_router_site = Router.from_subscription(router_id).router.router_site.subscription
old_side_site = Router.from_subscription(migrate_form_input.replace_side).router.router_site
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index a2448438..26d9c9cc 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -258,11 +258,11 @@ def allocate_interfaces_in_netbox(subscription: Iptrunk, previous_ae_members: di
device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
iface_name=interface.interface_name,
)
- # detach the old interfaces from lag
- nbclient.detach_interfaces_from_lag(
- device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
- lag_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface,
- )
+ # detach the old interfaces from lag
+ nbclient.detach_interfaces_from_lag(
+ device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+ lag_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface,
+ )
return {"subscription": subscription}
diff --git a/test/services/conftest.py b/test/services/conftest.py
new file mode 100644
index 00000000..45019637
--- /dev/null
+++ b/test/services/conftest.py
@@ -0,0 +1,40 @@
+class MockedNetboxClient:
+ class BaseMockObject:
+ def __init__(self, **kwargs):
+ for key, value in kwargs.items():
+ setattr(self, key, value)
+
+ def get_device_by_name(self):
+ return self.BaseMockObject(id=1, name="test")
+
+ def get_available_lags(self) -> list[str]:
+ return [f"LAG{lag}" for lag in range(1, 5)]
+
+ def get_available_interfaces(self):
+ interfaces = []
+ for interface in range(5):
+ interface_data = {
+ "name": f"Interface{interface}",
+ "module": {"display": f"Module{interface}"},
+ "description": f"Description{interface}",
+ }
+ interfaces.append(interface_data)
+ return interfaces
+
+ def create_interface(self):
+ return self.BaseMockObject(id=1, name="test")
+
+ def attach_interface_to_lag(self):
+ return self.BaseMockObject(id=1, name="test")
+
+ def reserve_interface(self):
+ return self.BaseMockObject(id=1, name="test")
+
+ def allocate_interface(self):
+ return {"id": 1, "name": "test"}
+
+ def free_interface(self):
+ return self.BaseMockObject(id=1, name="test")
+
+ def detach_interfaces_from_lag(self):
+ return None
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index a8f273f7..77ca88fd 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -8,6 +8,7 @@ from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
from gso.services.crm import customer_selector, get_customer_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.helpers import LAGMember
+from test.services.conftest import MockedNetboxClient
from test.workflows import (
assert_aborted,
assert_complete,
@@ -19,42 +20,6 @@ from test.workflows import (
)
-class MockedNetboxClient:
- class BaseMockObject:
- def __init__(self, **kwargs):
- for key, value in kwargs.items():
- setattr(self, key, value)
-
- def get_device_by_name(self):
- return self.BaseMockObject(id=1, name="test")
-
- def get_available_lags(self) -> list[str]:
- return [f"LAG{lag}" for lag in range(1, 5)]
-
- def get_available_interfaces(self):
- interfaces = []
- for interface in range(5):
- interface_data = {
- "name": f"Interface{interface}",
- "module": {"display": f"Module{interface}"},
- "description": f"Description{interface}",
- }
- interfaces.append(interface_data)
- return interfaces
-
- def create_interface(self):
- return self.BaseMockObject(id=1, name="test")
-
- def attach_interface_to_lag(self):
- return self.BaseMockObject(id=1, name="test")
-
- def reserve_interface(self):
- return self.BaseMockObject(id=1, name="test")
-
- def allocate_interface(self):
- return {"id": 1, "name": "test"}
-
-
@pytest.fixture
def netbox_client_mock():
# Mock NetboxClient methods
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 8f39c4d7..5bda8463 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -12,16 +12,40 @@ from test.workflows import (
run_workflow,
user_accept_and_assert_suspended,
)
+from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
@pytest.mark.workflow
@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk")
+@patch("gso.services.netbox_client.NetboxClient.get_device_by_name")
+@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")
+@patch("gso.services.netbox_client.NetboxClient.allocate_interface")
+@patch("gso.services.netbox_client.NetboxClient.free_interface")
+@patch("gso.services.netbox_client.NetboxClient.detach_interfaces_from_lag")
def test_iptrunk_modify_trunk_interface_success(
+ mocked_detach_interfaces_from_lag,
+ mocked_free_interface,
+ mocked_allocate_interface,
+ mocked_reserve_interface,
+ mocked_attach_interface_to_lag,
+ mocked_get_available_interfaces,
+ mocked_get_device_by_name,
mock_provision_ip_trunk,
iptrunk_subscription_factory,
faker,
):
# Set up mock return values
+ mocked_netbox = MockedNetboxClient()
+ mocked_get_device_by_name.return_value = mocked_netbox.get_device_by_name()
+ 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()
+
product_id = iptrunk_subscription_factory()
new_sid = faker.geant_sid()
new_description = faker.sentence()
@@ -31,12 +55,12 @@ def test_iptrunk_modify_trunk_interface_success(
new_side_a_sid = faker.geant_sid()
new_side_a_ae_members = [
- {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(5)
+ {"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": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(5)
+ {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5)
]
# Run workflow
@@ -86,6 +110,11 @@ def test_iptrunk_modify_trunk_interface_success(
assert "active" == subscription.status
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
assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}"
--
GitLab