Updated modification IP trunk workflow's unit tests.

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

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}
 

test/services/conftest.py

+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

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

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}"