diff --git a/test/services/test_netbox.py b/test/services/test_netbox.py index b8145e293f416994e45d6937c5e001080e8cdfe2..34d4e2f1667cb69980e9ac840955d685982e5a46 100644 --- a/test/services/test_netbox.py +++ b/test/services/test_netbox.py @@ -1,8 +1,7 @@ """ Unit tests for testing the netbox client """ -import os -import json + import uuid from os import PathLike from unittest.mock import patch @@ -11,20 +10,11 @@ from pynetbox.core.response import Record from gso.services.netbox_client import NetBoxClient from gso.products.product_blocks.site import SiteTier +from gso.utils.exceptions import WorkflowStateError BASE_URL = "https://127.0.0.1:8000" -def _load_fixture(filename): - current_directory = os.path.dirname(os.path.abspath(__file__)) - # Build the full path to the fixture file - fixture_path = os.path.join(current_directory, 'fixtures', filename) - - # Load and return the JSON data - with open(fixture_path, 'r') as f: - return json.load(f) - - @pytest.fixture(scope="module") def device(): values = {"id": 1, "name": "test123"} @@ -61,6 +51,17 @@ def 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) + + @patch("gso.services.netbox_client.pynetbox.api") def test_create_device(mock_api, device, @@ -109,3 +110,42 @@ def test_get_available_lags(mock_api, mock_from_subscription, data_config_filena # 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): + """ + 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