diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py index f632766b3e538bd2daad385be7b23aabfbe03d91..3c27e95e2e9229fb3ff6f7651295acb52767b0f7 100644 --- a/gso/services/netbox_client.py +++ b/gso/services/netbox_client.py @@ -208,8 +208,13 @@ class NetboxClient: ) iface.lag = lag.id + # Set description if provided + if description: + iface.description = description + # Update physical interface - return self.netbox.dcim.interfaces.update([iface])[0] + iface.save() + return iface def detach_interface_from_lag(self, device_name: str, lag_name: str, iface_name: str) -> Interfaces: """Remove a given physical interface from a lag. @@ -237,7 +242,8 @@ class NetboxClient: iface.lag = None # Update physical interface - return self.netbox.dcim.interfaces.update([iface])[0] + iface.save() + return iface def reserve_interface(self, device_name: str, iface_name: str) -> Interfaces: """Reserve an interface by enabling it.""" diff --git a/test/services/test_netbox.py b/test/services/test_netbox.py index 430f5db169ce6121bdadcbbd36901c545baf9eef..815c4527239b18f5ea4bfb6c7d5d9e43c84f525e 100644 --- a/test/services/test_netbox.py +++ b/test/services/test_netbox.py @@ -1,8 +1,13 @@ -from unittest.mock import patch +"""Unit tests for testing the netbox client.""" + +import uuid +from os import PathLike +from unittest.mock import Mock, patch import pytest from pynetbox.core.response import Record +from gso.products.product_blocks.site import SiteTier from gso.services.netbox_client import NetboxClient from gso.utils.exceptions import WorkflowStateError @@ -58,6 +63,237 @@ def interface(): return Record(values, None, None) +@pytest.fixture(scope="module") +def lag(): + values = { + "id": 1, + "name": "lag-1", + "type": "lag", + } + return Record(values, None, None) + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_create_device( + mock_api, device, device_type, device_role, site, device_bay, card_type, data_config_filename: PathLike +): + device_name = "mx1.lab.geant.net" + device.name = device_name + site_tier = SiteTier.TIER1 + + # Define mock calls + mock_api.return_value.dcim.device_types.get.return_value = device_type + mock_api.return_value.dcim.device_roles.get.return_value = device_role + mock_api.return_value.dcim.sites.get.return_value = site + mock_api.return_value.dcim.devices.create.return_value = device + mock_api.return_value.dcim.module_bays.filter.return_value = [device_bay] + mock_api.return_value.dcim.module_types.get.return_value = card_type + mock_api.return_value.dcim.module_types.create.return_value = card_type + + new_device = NetboxClient().create_device(device_name, site_tier) + assert new_device is not None + assert new_device.name == device_name + + +@patch("gso.services.netbox_client.Router.from_subscription") +@patch("gso.services.netbox_client.pynetbox.api") +def test_get_available_lags(mock_api, mock_from_subscription, data_config_filename: PathLike): + router_id = uuid.uuid4() + feasible_lags = [f"LAG-{i}" for i in range(1, 11)] + + # Mock the pynetbox API instance + mock_netbox = mock_api.return_value + mock_filter = mock_netbox.dcim.interfaces.filter + mock_filter.return_value = [{"name": f"LAG-{i}", "type": "lag"} for i in range(1, 4)] + + # Mock the Router.from_subscription method + mock_subscription = mock_from_subscription.return_value + mock_router = mock_subscription.router + mock_router.router_fqdn = "test_router" + + netbox_client = NetboxClient() + result = netbox_client.get_available_lags(router_id) + + # Check the result of the function + assert result == [lag for lag in feasible_lags if lag not in [f"LAG-{i}" for i in range(1, 4)]] + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_create_interface(mock_api, device, interface, data_config_filename: PathLike): + # Moch netbox calls + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.create.return_value = interface + + # Create new interface + new_interface = NetboxClient().create_interface(interface.name, interface.type, interface.speed, device.name) + + # Check result + assert new_interface is not None + assert new_interface.name == interface.name + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_reserve_interface_exception(mock_api, device, interface, data_config_filename: PathLike): + """Test for checking if interface is reserved. + + If the interface is already reserved + the method should throw an exception + """ + # Change the interface to reserved + interface.enabled = True + + # expected exception message + exception_message = f"The interface: {interface.name} on device: {device.name} is already reserved." + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.get.return_value = interface + + # Check exception + with pytest.raises(WorkflowStateError) as test_exception: + NetboxClient().reserve_interface(device.name, interface.name) + assert str(test_exception.value) == exception_message + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_reserve_interface(mock_api, device, interface, data_config_filename: PathLike): + """Test a normal reservation of a interface.""" + # Set interface to not reserved + interface.enabled = False + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.get.return_value = interface + + # mock save method + mock_save = Mock() + mock_save.save.return_value = interface + interface.save = mock_save + + # Check reservation of interface + updated_interface = NetboxClient().reserve_interface(device.name, interface.name) + + assert updated_interface is not None + assert updated_interface.enabled is True + mock_save.assert_called_once() + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_allocate_interface_exception(mock_api, device, interface, data_config_filename: PathLike): + """Test to check exception during allocation. + + If the interface is already allocated + the method should throw an exception + """ + # Change the interface to reserved + interface.enabled = True + + # Change interface to allocated + interface.mark_connected = True + + # expected exception message + exception_message = f"The interface: {interface.name} on device: {device.name} is already allocated." + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.get.return_value = interface + + # Check exception + with pytest.raises(WorkflowStateError) as test_exception: + NetboxClient().allocate_interface(device.name, interface.name) + assert str(test_exception.value) == exception_message + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_allocation_interface(mock_api, device, interface, data_config_filename: PathLike): + """Test a normal allocation of a interface.""" + # Set interface to not allocated + interface.mark_connected = False + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.get.return_value = interface + + # mock save method + mock_save = Mock() + mock_save.save.return_value = interface + interface.save = mock_save + + # Check allocation of interface + updated_interface = NetboxClient().allocate_interface(device.name, interface.name) + + assert updated_interface is not None + assert updated_interface.mark_connected is True + mock_save.assert_called_once() + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_delete_device(mock_api, device, data_config_filename: PathLike): + """Test a delete of a device.""" + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + + # mock delete method + mock_delete = Mock() + device.delete = mock_delete + + # Check delete of interface + NetboxClient().delete_device(device.name) + + mock_delete.assert_called_once() + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_get_interfaces_by_device(mock_api, device, interface, data_config_filename: PathLike): + """Test if a interface is returned for a device.""" + # Setup interface speed + speed = 1000 + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.filter.return_value = [interface, interface] + + # Call get interfaces by device + interfaces = NetboxClient().get_interfaces_by_device(device.name, speed) + + assert interfaces is not None + assert len(interfaces) == 2 + + +@patch("gso.services.netbox_client.pynetbox.api") +def test_attach_interface_to_lag(mock_api, device, interface, lag, data_config_filename: PathLike): + """Test if a interface is attached correctly to a lag interface.""" + + # Define site effect function + def get_side_effect(**kwargs): + if kwargs.get("device_id") == 1 and kwargs.get("name") == "lag-1": + return lag + return interface + + # Define a description + description = "test123" + + # At first the interface is not attached to a lag + interface.lag = None + + # Mock netbox api + mock_api.return_value.dcim.devices.get.return_value = device + mock_api.return_value.dcim.interfaces.get.side_effect = get_side_effect + + # mock save method + mock_save = Mock() + mock_save.save.return_value = interface + interface.save = mock_save + + # Check if interface attached to lag + lag_interface = NetboxClient().attach_interface_to_lag(device.name, lag.name, interface.name, description) + + assert lag_interface is not None + assert lag_interface.lag == lag.id + assert lag_interface.description == description + mock_save.assert_called_once() + + @patch("gso.services.netbox_client.pynetbox.api") def test_free_interface(mock_api, device, interface): device_name = "mx1.lab.geant.net"