diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py index 0bd1face7dc1fcbdf3f2ef31b7e1e2248698c967..ce629d1449d730dc296bb5928425b715873530f1 100644 --- a/gso/services/netbox_client.py +++ b/gso/services/netbox_client.py @@ -3,8 +3,7 @@ from uuid import UUID import pydantic import pynetbox -from infoblox_client.objects import Interface -from pynetbox.models.dcim import Devices, DeviceTypes, Interfaces +from pynetbox.models.dcim import Devices, DeviceTypes, InterfaceConnection, Interfaces from gso.products.product_types.router import Router from gso.settings import load_oss_params @@ -169,7 +168,7 @@ class NetboxClient: return device def delete_device(self, device_name: str) -> None: - """Delete device by name""" + """Delete device by name.""" self.netbox.dcim.devices.get(name=device_name).delete() return @@ -308,3 +307,13 @@ class NetboxClient: return self.netbox.dcim.interfaces.filter( device=device.name, enabled=False, mark_connected=False, speed=speed_bps ) + + def get_interface_by_name_and_device(self, router_id: UUID, interface_name: str) -> InterfaceConnection: + """Return the interface object by name and device from netbox, or ``None`` if not found.""" + + router = Router.from_subscription(router_id).router.router_fqdn + device = self.get_device_by_name(router) + try: + return self.netbox.dcim.interfaces.get(device=device.name, name=interface_name) + except pynetbox.RequestError: + raise NotFoundError(f"Interface: {interface_name} on device: {device.name} not found.") diff --git a/test/services/test_netbox.py b/test/services/test_netbox.py index ca7ea3931b8ff622878fd447acac5189e992ac20..15508fe77466fcc100ae26ffe7ceae33931db6ff 100644 --- a/test/services/test_netbox.py +++ b/test/services/test_netbox.py @@ -3,11 +3,11 @@ 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 @@ -59,6 +59,7 @@ def interface(): "type": "1000BaseT", "enabled": False, "mark_connected": False, + "lag": None, } return Record(values, None, None) diff --git a/test/workflows/iptrunk/test_terminate_iptrunk.py b/test/workflows/iptrunk/test_terminate_iptrunk.py index 9ddf0240f88c737cefe654927c66fd547e135ff5..f90022cb0c7b64f1e3384c0008a01426ea588f6e 100644 --- a/test/workflows/iptrunk/test_terminate_iptrunk.py +++ b/test/workflows/iptrunk/test_terminate_iptrunk.py @@ -3,6 +3,7 @@ from unittest.mock import patch import pytest from gso.products import Iptrunk +from test.services.conftest import MockedNetboxClient from test.workflows import ( assert_complete, assert_suspended, @@ -13,37 +14,16 @@ 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_interface_by_name_and_by_device_id(self): - return self.BaseMockObject(id=1, name="test") - - def delete_interface(self): - return None - - def free_interface(self): - return self.BaseMockObject(id=1, name="test") - - @pytest.mark.workflow @patch("gso.workflows.iptrunk.terminate_iptrunk.provisioning_proxy.provision_ip_trunk") @patch("gso.workflows.iptrunk.terminate_iptrunk.provisioning_proxy.deprovision_ip_trunk") @patch("gso.workflows.iptrunk.terminate_iptrunk.infoblox.delete_network") @patch("gso.services.netbox_client.NetboxClient.get_device_by_name") -@patch("gso.services.netbox_client.NetboxClient.get_interface_by_name_and_by_device_id") @patch("gso.services.netbox_client.NetboxClient.delete_interface") @patch("gso.services.netbox_client.NetboxClient.free_interface") def test_successful_iptrunk_termination( mocked_free_interface, mocked_delete_interface, - mocked_get_interface_by_name_and_by_device_id, mocked_get_device_by_name, mock_infoblox_delete_network, mock_deprovision_ip_trunk, @@ -55,7 +35,6 @@ def test_successful_iptrunk_termination( product_id = iptrunk_subscription_factory() mocked_netbox = MockedNetboxClient() mocked_get_device_by_name.return_value = mocked_netbox.get_device_by_name() - mocked_get_interface_by_name_and_by_device_id.return_value = mocked_netbox.get_interface_by_name_and_by_device_id() mocked_delete_interface.return_value = mocked_netbox.delete_interface() mocked_free_interface.return_value = mocked_netbox.free_interface()