Skip to content
Snippets Groups Projects
Commit f78b1638 authored by Neda Moeini's avatar Neda Moeini
Browse files

Updated tests.

parent ec17726c
Branches
Tags
1 merge request!94Feature/Netbox integration terminate ip trunk
Pipeline #84329 passed
......@@ -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.")
......@@ -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)
......
......@@ -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()
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment