diff --git a/data_uploader/exceptions/__init__.py b/data_uploader/exceptions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data_uploader/exceptions/ip_address_collision_error.py b/data_uploader/exceptions/ip_address_collision_error.py new file mode 100644 index 0000000..c4420ac --- /dev/null +++ b/data_uploader/exceptions/ip_address_collision_error.py @@ -0,0 +1,2 @@ +class IPAddressCollisionError(ValueError): + pass diff --git a/data_uploader/exceptions/mac_address_collision_error.py b/data_uploader/exceptions/mac_address_collision_error.py new file mode 100644 index 0000000..007813b --- /dev/null +++ b/data_uploader/exceptions/mac_address_collision_error.py @@ -0,0 +1,2 @@ +class MacAddressCollisionError(ValueError): + pass diff --git a/data_uploader/exceptions/missing_mandatory_param_error.py b/data_uploader/exceptions/missing_mandatory_param_error.py new file mode 100644 index 0000000..0cd9208 --- /dev/null +++ b/data_uploader/exceptions/missing_mandatory_param_error.py @@ -0,0 +1,2 @@ +class MissingMandatoryParamError(ValueError): + pass diff --git a/data_uploader/netbox_api/__init__.py b/data_uploader/netbox_api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data_uploader/netbox_api/netbox_base.py b/data_uploader/netbox_api/netbox_base.py new file mode 100644 index 0000000..4f51ddd --- /dev/null +++ b/data_uploader/netbox_api/netbox_base.py @@ -0,0 +1,9 @@ +from data_uploader.netbox_api.netbox_connection import NetboxApi + + +class NetboxBase: + def __init__(self, url, token, cert): + self.url = url + self.token = token + self.cert = cert + self._netbox_api = NetboxApi.api_object(url, token, cert) diff --git a/data_uploader/netbox_api/netbox_connection.py b/data_uploader/netbox_api/netbox_connection.py new file mode 100644 index 0000000..c1d510c --- /dev/null +++ b/data_uploader/netbox_api/netbox_connection.py @@ -0,0 +1,30 @@ +from data_uploader.exceptions.missing_mandatory_param_error import MissingMandatoryParamError +import requests +from pynetbox import api + + +class NetboxApi: + """ + Wraps a netbox connection as a context manager + + """ + def __init__(self): + pass + + @staticmethod + def api_object(netbox_url: str, token: str, cert=None): + if not netbox_url: + raise MissingMandatoryParamError( + "NetBox URL is required but not provided." + ) + if not token: + raise MissingMandatoryParamError( + "A token is required but not provided." + ) + session = requests.Session() + session.verify = cert if cert else False + + connection = api(url=netbox_url, token=token) + connection.http_session = session + + return connection diff --git a/data_uploader/netbox_api/netbox_dcim.py b/data_uploader/netbox_api/netbox_dcim.py new file mode 100644 index 0000000..3bd8b5b --- /dev/null +++ b/data_uploader/netbox_api/netbox_dcim.py @@ -0,0 +1,130 @@ +from typing import Union +from data_uploader.exceptions.mac_address_collision_error import MacAddressCollisionError +from data_uploader.netbox_api.netbox_base import NetboxBase +from pynetbox.core.response import Record + + +class NetboxDcim(NetboxBase): + """ + Class for retrieving or creating DCIM objects in Netbox + """ + def __init__(self, url, token, cert): + super().__init__(url, token, cert) + + def get_device(self, hostname: str) -> Union[Record, None]: + """ + Search for the name of a device (server) in Netbox by its name + :param hostname: name of the server to search for in Netbox + :return: dict of record in netbox or None if not found + """ + self._netbox_api.dcim.devices.get(name=hostname) + return self._netbox_api.dcim.devices.get(name=hostname) + + def get_device_types(self, model: str) -> Union[Record, None]: + """ + Get device model type from netbox + :param model: model type to search for in netbox + :return: dict of record in netbox or None if not found + """ + + return self._netbox_api.dcim.device_types.get(slug=model) + + def get_interface(self, interface_name: str, hostname: str) -> Union[Record, None]: + """ + Get an interface from Netbox + :param interface_name: Name of interface to search for + :param hostname: Name of device the interface is attached to + """ + return self._netbox_api.dcim.interfaces.get(name=interface_name, device=hostname) + + def find_mac_addr(self, mac_address: str) -> bool: + """ + Checks whether there is an interface in Netbox that already uses a specific mac address + :param mac_address: mac address to search Netbox for + :returns: Boolean indicating whether an interface with that address exists (True) or not (False) + """ + check = self._netbox_api.interface.get(mac_address=mac_address) + return True if check else False + + def create_device(self, hostname: str, site: str, location: str, tenant: str, manufacturer: str, + rack: str, rack_position: int, device_type: str, serial_no: str) -> Record: + """ + Create a new device in NetBox. + + :param hostname: Name of device + :param site: Building and Room the device is located in + :param location: Row the device is in + :param tenant: Tenant for the device + :param manufacturer: Device manufacturer + :param rack: Rack the device is in + :param rack_position: Rack position (Optional) + :param device_type: Device type + :param serial_no: serial number (optional) + + :return: Netbox Record object + """ + netbox = self._netbox_api + netbox_device = netbox.dcim.devices.create( + name=hostname, + site=netbox.dcim.devices.get(name=site).id, + location=netbox.dcim.locations.get(name=location).id, + tenant=netbox.tenancy.tenants.get(name=tenant).id, + manufacturer=netbox.dcim.manufacturers.get(name=manufacturer).id, # optional field + rack=netbox.dcim.racks.get(name=rack).id, + postion=rack_position, # optional field - omit this field if the device is unracked + device_type=netbox.dcim.device_types.get(slug=device_type).id, + serial=serial_no, # optional field + #tags=[{"name": "Tag 1"}, {"name": "Tag 2"}] # optional field + ) + return netbox_device + + def create_device_type(self, model: str, manufacturer: str, slug: str, unit_height: int) -> Record: + """ + Create a new device type in NetBox + :param model: Name of device type + :param manufacturer: name of manufacturer + :param slug: slug of model name + :param unit_height: height of devices of this type + + :return: Netbox record of new device type + """ + netbox = self._netbox_api + new_device_type = netbox.dcim.device_types.create( + manufacturer=netbox.dcim.manufacturers.get(name=manufacturer).id, + model=model, + slug=slug, + u_height=unit_height, + # custom_fields={'cf_1': 'Custom data 1'} # optional field + ) + return new_device_type + + def create_interface(self, interface_name: str, hostname: str, interface_type: str, description: str, + mac_address: str) -> Record: + """ + Create an interface for a device already in NetBox + :param interface_name: name of interface + :param hostname: name of device the interface is attached to + :param interface_type: interface type + :param description: description + :param mac_address: mac_address + :return: netbox_interface: Record object with details of newly created netbox interface + """ + # verify whether the mac address already exists in Netbox on a specific interface + mac_addr_match = NetboxDcim.find_mac_addr(self, mac_address) + + if mac_addr_match: + raise MacAddressCollisionError( + "MAC Address already exists in Netbox" + ) + + netbox = self._netbox_api + + netbox_interface = netbox.dcim.interfaces.create( + name=interface_name, + device=netbox.dcim.devices.get(name=hostname).id, + type=interface_type, + description=description, + mac_address=mac_address, + # tags = [{'name': 'Tag 1'}] + ) + return netbox_interface diff --git a/data_uploader/netbox_api/netbox_ipam.py b/data_uploader/netbox_api/netbox_ipam.py new file mode 100644 index 0000000..e93a493 --- /dev/null +++ b/data_uploader/netbox_api/netbox_ipam.py @@ -0,0 +1,52 @@ +from data_uploader.netbox_api.netbox_dcim import NetboxDcim +from data_uploader.netbox_api.netbox_connection import NetboxApi +import pynetbox + +class NetboxIpam: + def __init__(self, url, token, cert): + self.url = url + self.token = token + self. cert = cert + self._netbox_api = NetboxApi.api_object(url, token, cert) + + def search_mac_address(self, mac_addr: str): + """ + Search Netbox for a mac address assigned to any interface + This will call a method from NetBox Dcim to do this + :return: Record object with interface the mac address is associated with or None if not found + """ + netbox = self._netbox_api + interface = netbox.dcim.interfaces.get(mac_address=mac_addr) + + return interface + + def get_ip_address(self, ip_addr: str): + """ + Search netbox for an IP Address + :param ip_addr: IP address to search for in Netbox + :return: Record object with IP Address or None if not found + """ + netbox = self._netbox_api + return netbox.ipam.ip_addresses.get(address=ip_addr) + + def create_ip_address(self, hostname: str, interface:str, ip_address: str, tenant: str): + """ + Create a new IP address in Netbox + :param hostname: Name of device in Netbox (must be present in Netbox) + :param interface: Interface attached to device (must be present in Netbox) + :param ip_address: IP address to create + :param tenant: Tenant IP address belongs to (must be present in Netbox) + """ + + interface = NetboxDcim.get_interface(name=interface, device=hostname) + netbox_ip = netbox.ipam.ip_addresses.create( + address="ip-address", + tenant=netbox.tenancy.tenants.get(name='tenant-name').id, + tags=[{'name': 'Tag 1'}], + ) + # assign IP Address to device's network interface + netbox_ip.assigned_object = interface + netbox_ip.assigned_object_id = interface.id + netbox_ip.assigned_object_type = 'dcim.interface' + # save changes to IP record in Netbox + netbox_ip.save() diff --git a/data_uploader/parsers/__init__.py b/data_uploader/parsers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data_uploader/requirements.txt b/data_uploader/requirements.txt new file mode 100644 index 0000000..0c905f6 --- /dev/null +++ b/data_uploader/requirements.txt @@ -0,0 +1,6 @@ +pynetbox +pytest +argparse +logger +requests +nose \ No newline at end of file diff --git a/data_uploader/system_data_class.py b/data_uploader/system_data_class.py new file mode 100644 index 0000000..635914a --- /dev/null +++ b/data_uploader/system_data_class.py @@ -0,0 +1,14 @@ +from dataclasses import dataclass + +@dataclass +class SystemDataClass: + """ + Class for keeping track of server properties + """ + manufacturer: str + model: str + primaryMAC: str + hostname: str + serviceTag: str + ipmiIPAddress: str + diff --git a/data_uploader/test/__init__.py b/data_uploader/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data_uploader/test/test_netbox_connection.py b/data_uploader/test/test_netbox_connection.py new file mode 100644 index 0000000..7b76c5a --- /dev/null +++ b/data_uploader/test/test_netbox_connection.py @@ -0,0 +1,55 @@ +import unittest +from unittest.mock import NonCallableMock, patch +from nose.tools import raises +from data_uploader.exceptions.missing_mandatory_param_error import MissingMandatoryParamError +from data_uploader.netbox_api.netbox_connection import NetboxApi + + +class NetboxApiTests(unittest.TestCase): + @staticmethod + @patch('data_uploader.netbox_api.netbox_connection.api') + def test_api_object_create(mock_api): + """ + Tests that we do get an API object an + """ + mock_url = "example.com" + mock_token = NonCallableMock() + mock_cert = NonCallableMock() + + api = NetboxApi.api_object(mock_url, mock_token, mock_cert) + assert mock_api.return_value == api + + @staticmethod + @patch('data_uploader.netbox_api.netbox_connection.api') + def test_api_no_cert(mock_api): + """ + Test that an api object is created even when we don't have a path for a certificate + """ + mock_url = "example.com" + mock_token = NonCallableMock() + mock_cert = None + + api = NetboxApi.api_object(mock_url, mock_token, mock_cert) + assert mock_api.return_value == api + + @raises(MissingMandatoryParamError) + def test_api_throws_for_no_url(self): + """ + Tests a None type will throw error if used as url + """ + missing_url = None + mock_token = NonCallableMock() + mock_cert = NonCallableMock() + api = NetboxApi.api_object(missing_url, mock_token, mock_cert) + api.assertRaises(MissingMandatoryParamError) + + @raises(MissingMandatoryParamError) + def test_api_throws_for_no_token(self): + """ + Tests a None type will throw if used as token + """ + mock_url = NonCallableMock() + missing_token = None + mock_cert = NonCallableMock() + NetboxApi.api_object(mock_url, missing_token, mock_cert) + self.assertRaises(MissingMandatoryParamError) diff --git a/data_uploader/test/test_netbox_dcim.py b/data_uploader/test/test_netbox_dcim.py new file mode 100644 index 0000000..beb18a7 --- /dev/null +++ b/data_uploader/test/test_netbox_dcim.py @@ -0,0 +1,146 @@ +import unittest +from unittest.mock import NonCallableMock, MagicMock, patch +from data_uploader.netbox_api.netbox_dcim import NetboxDcim +from nose.tools import assert_true, assert_is_not_none, raises +from data_uploader.exceptions.mac_address_collision_error import MacAddressCollisionError +from pynetbox.core.response import Record + +# WIP +class NetboxDcimTests(unittest.TestCase): + + """ + Runs tests to ensure we are interacting with the NetBox + API for dcim objects in the expected way + """ + + def test_get_device_response(self): + """ + Test that we can interact with netbox and get a response + """ + self._netbox_api = MagicMock() + mock_device_name = NonCallableMock() + response = NetboxDcim.get_device(self._netbox_api, hostname=mock_device_name) + assert_true(response.ok) + assert_is_not_none(response) + + def test_get_device_none_found(self): + """ + Test that get_device does return None if a device is not in Netbox + """ + pass + + def test_get_device_type(self): + """ + Test that we can interact with NetBox to get a device type and + get a response back + :return: + """ + self._netbox_api = MagicMock() + mock_model = NonCallableMock() + response = NetboxDcim.get_device_types(model=mock_model) + assert_true(response.ok) + assert_is_not_none(response) + # TO DO assert a value is actually returned by get device type + + def test_get_device_type_none_found(self): + """ + Test None is returned if device type not in Netbox + """ + pass + + def test_get_interface(self): + """ + Test get_interface method + """ + self._netbox_api = MagicMock() + mock_interface = NonCallableMock() + mock_hostname = NonCallableMock() + + response = NetboxDcim.get_interface(self, interface_name=mock_interface, hostname=mock_hostname) + assert_true(response.ok) + assert_is_not_none(response) + + def test_find_mac_addr(self): + """ + Test checking for a mac address in Netbox + """ + self._netbox_api = MagicMock() + mock_mac_addr = NonCallableMock() + + response = NetboxDcim.find_mac_addr(self, mac_address=mock_mac_addr) + assert_true(response) + + def test_no_mac_addr_found(self): + """ + Test False is returned when a mac address is missing in Netbox + """ + # WIP + pass + + def test_create_device(self): + """ + Test creating a device + """ + # WIP + self._netbox_api = MagicMock() + mock_hostname = NonCallableMock() + mock_site = NonCallableMock() + mock_location = NonCallableMock() + mock_tenant = NonCallableMock() + mock_manufacturer = NonCallableMock() + mock_rack = NonCallableMock() + mock_rack_position = NonCallableMock() + mock_device_type = NonCallableMock() + mock_serial_no = NonCallableMock() + + device = NetboxDcim.create_device(self, hostname=mock_hostname, site=mock_site, location=mock_location, tenant=mock_tenant, manufacturer=mock_manufacturer, + rack=mock_rack, rack_position=mock_rack_position, device_type=mock_device_type, serial_no=mock_serial_no) + + pass + + def test_create_device_type(self): + """ + Test creating a device type + """ + # WIP + self._netbox_api = MagicMock() + mock_model = NonCallableMock() + mock_manufacturer = NonCallableMock() + mock_slug = NonCallableMock() + mock_unit_height = NonCallableMock() + + device_type = NetboxDcim.create_device_type(self, model=mock_model, manufacturer=mock_manufacturer, slug=mock_slug, unit_height=mock_unit_height) + pass + + # currently failing test as raising MAC address error + def test_create_interface(self): + """ + Test interface creation + """ + # WIP + self._netbox_api = MagicMock() + mock_interface = NonCallableMock() + mock_hostname = NonCallableMock() + mock_interface_type = NonCallableMock() + mock_description = NonCallableMock() + mock_mac_addr = NonCallableMock() + mock_mac_addr_match = NetboxDcim.find_mac_addr(self._netbox_api, mac_address=mock_mac_addr) + + interface = NetboxDcim.create_interface(self._netbox_api, interface_name=mock_interface, hostname=mock_hostname, interface_type=mock_interface_type, + description=mock_description, mac_address=mock_mac_addr) + + # assert value is returned + + #@patch(NetboxDcim.find_mac_addr) + #@raises(MacAddressCollisionError) + def test_mac_address_collision(self): + """ + Test if interface creation is attempted + and mac address already exists then an error + is raised + """ + # WIP + self._netbox_api = MagicMock() + mac_addr_match = False + + pass \ No newline at end of file diff --git a/data_uploader/test/test_netbox_ipam.py b/data_uploader/test/test_netbox_ipam.py new file mode 100644 index 0000000..5b494d7 --- /dev/null +++ b/data_uploader/test/test_netbox_ipam.py @@ -0,0 +1,11 @@ +# WIP +import unittest +from unittest.mock import NonCallableMock, MagicMock + +class NetboxIpamTests(unittest.TestCase): + """ + Run tests to ensure that we are interacting with the Netbox API + for ipam objects in the expected way + """ + def test(self): + pass \ No newline at end of file