Feature/nat 286 create unit tests for netbox client

test/services/test_netbox.py

+"""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
+
+BASE_URL = "https://127.0.0.1:8000"
+
+
+@pytest.fixture(scope="module")
+def device():
+    values = {"id": 1, "name": "test123"}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def device_type():
+    values = {"id": 1, "name": "test123"}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def device_role():
+    values = {"id": 1, "name": "test123"}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def site():
+    values = {"id": 1, "name": "test123"}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def device_bay():
+    values = {"id": 1, "name": "bay_test", "position": 1}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def card_type():
+    values = {"id": 1, "name": "test_card_type"}
+    return Record(values, None, None)
+
+
+@pytest.fixture(scope="module")
+def interface():
+    values = {
+        "id": 1,
+        "name": "et-0/0/1",
+        "speed": 1000,
+        "type": "1000BaseT",
+        "enabled": False,
+        "mark_connected": False,
+    }
+    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"
+
+    # 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()