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,
        "lag": None,
    }
    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()


@patch("gso.services.netbox_client.pynetbox.api")
def test_free_interface(mock_api, device, interface):
    device_name = "mx1.lab.geant.net"
    interface_name = "et-0/0/1"

    # Define mock calls
    mock_api.return_value.dcim.devices.get.return_value = device
    mock_api.return_value.dcim.interfaces.get.return_value = interface

    # Create a NetboxClient instance
    netbox_client = NetboxClient()

    # Test free_interface method on success
    interface.mark_connected = True
    interface.enabled = True
    cleared_interface = netbox_client.free_interface(device_name, interface_name)
    assert cleared_interface.enabled is False
    assert cleared_interface.mark_connected is False
    assert cleared_interface.description == ""