test_create_device_netbox.py

"""Test case for Netbox Device Creation"""

import unittest
import os
import json
import requests

HOST = os.getenv("HOST", default="localhost")
PORT = os.getenv("PORT", default="8080")
API_TOKEN = os.getenv("API_TOKEN", default="only4testingpurpose")
HEADERS = {"Authorization": f"Token {API_TOKEN}"}

class TestNetboxDeviceCreate(unittest.TestCase):
    """Test case for device creation, config context update and interface creation """

    list_of_ids: dict = {}

    # definition of the variables

    def setUp(self) -> None:
        """Will create a manufacturer, site, device role and device type"""

        manufacturer = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/manufacturers/",
            headers=HEADERS,
            json={
                "name": "Cisco",
                "slug": "cisco"
            },
            timeout=5
        )

        self.list_of_ids['manufacturer_id'] = json.loads(manufacturer.content)["id"]

        # Will create a device role

        device_role = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/device-roles/",
            headers=HEADERS,
            json={
                "name": "Switch",
                "slug":"switch"
            },
            timeout=5
        )

        self.list_of_ids['device_role_id'] = json.loads(device_role.content)["id"]

        #Will create a device type

        device_type = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/device-types/",
            headers=HEADERS,
            json={
                "model": "WS-C2960X",
                "slug":"ws-c2960x",
                "manufacturer": self.list_of_ids["manufacturer_id"]
            },
            timeout=5
        )

        self.list_of_ids['device_type_id'] = json.loads(device_type.content)["id"]

        # will create a site

        site = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/sites/",
            headers=HEADERS,
            json={
                "name": "ISPRA",
                "slug":"ispra"
            },
            timeout=5
        )

        self.list_of_ids["site_id"] = json.loads(site.content)["id"]


        # Will create a dummy config context

        config_context = requests.post(
            url=f"http://{HOST}:{PORT}/api/extras/config-contexts/",
            headers=HEADERS,
            json={
                "name": "test_config_context",
                "sites": [self.list_of_ids["site_id"]],
                "device_types": [self.list_of_ids["device_type_id"]],
                "data": {
                    "radius_servers": [
                        "1.1.1.1",
                        "2.2.2.2"
                    ]
                }
            },
            timeout=5
        )


        self.list_of_ids['config_context_id'] = json.loads(config_context.content)["id"]

        # will create a custom field for access vlan

        access_vlan = requests.post(
            url=f"http://{HOST}:{PORT}/api/extras/custom-fields/",
            headers=HEADERS,
            json={
                "content_types": ["dcim.interface"],
                "name": "access_vlan",
                "type": "integer",
                "validation_minimum": 1,
                "validation_maximum": 4096
            },
            timeout=5
        )

        self.list_of_ids['custom_field_vlan_id'] = json.loads(access_vlan.content)["id"]

        # Will create a custom field for allowed vlans on a trunk interface

        allowed_vlans_trunk = requests.post(
            url=f"http://{HOST}:{PORT}/api/extras/custom-fields/",
            headers=HEADERS,
            json={
                "content_types": ["dcim.interface"],
                "name": "allowed_vlans_trunk",
                "type": "text"
            },
            timeout=5
        )

        self.list_of_ids['allowed_vlans_trunk_id'] = json.loads(allowed_vlans_trunk.content)["id"]



    def tearDown(self) -> None:
        """ Deleting all data from NETBOX once the all the tests finish running """

        #deleting the device

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/dcim/devices/{self.list_of_ids['device_id']}/",
            headers=HEADERS,
            timeout=5
        )

        #deleting the device type

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/dcim/device-types/{self.list_of_ids['device_type_id']}/",
            headers=HEADERS,
            timeout=5
        )

        #deleting the manufacturer

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/dcim/manufacturers/{self.list_of_ids['manufacturer_id']}/",
            headers=HEADERS,
            timeout=5
        )

        # deleting the device role

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/dcim/device-roles/{self.list_of_ids['device_role_id']}/",
            headers=HEADERS,
            timeout=5
        )

        # deleting the site

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/dcim/sites/{self.list_of_ids['site_id']}/",
            headers=HEADERS,
            timeout=5
        )

        # Deleting the config context


        requests.delete(
            url= f"http://{HOST}:{PORT}/api/extras/config-contexts/{self.list_of_ids['config_context_id']}/",
            headers=HEADERS,
            timeout=5
        )

        # deleting the access_vlan custom_field


        requests.delete(
            url= f"http://{HOST}:{PORT}/api/extras/custom-fields/{self.list_of_ids['custom_field_vlan_id']}/",
            headers=HEADERS,
            timeout=5
        )

        requests.delete(
            url= f"http://{HOST}:{PORT}/api/extras/custom-fields/{self.list_of_ids['allowed_vlans_trunk_id']}/",
            headers=HEADERS,
            timeout=5
        )

        # deleting the allowed_vlans_trunk custom_field


    def test_device_creation(self) -> None:
        """" getting the necessary IDs """


        # Creating the device

        response = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/devices/",
            headers=HEADERS,
            json={
                "name": "test",
                "device_type": self.list_of_ids['device_type_id'],
                "role": self.list_of_ids['device_role_id'],
                "site": self.list_of_ids['site_id'],
                "status": "offline"
            },
            timeout=5
        )

        self.list_of_ids['device_id'] = json.loads(response.content)["id"]

        self.assertEqual(response.status_code,201)
        self.assertEqual(json.loads(response.content)["name"],"test")
        self.assertEqual(json.loads(response.content)["device_type"]["display"],"WS-C2960X")
        self.assertEqual(json.loads(response.content)["role"]["name"],"Switch")
        self.assertEqual(json.loads(response.content)["site"]["name"],"ISPRA")
        self.assertEqual(json.loads(response.content)["status"]["value"],"offline")

        # getting the config context data

        config_context_data = requests.get(
            url=f"http://{HOST}:{PORT}/api/extras/config-contexts/{self.list_of_ids['config_context_id']}/",
            headers=HEADERS,
            timeout=5
        )
        data = json.loads(config_context_data.content)["data"]

        # Putting the data from the config context into the local context data of the device

        upload_config_context = requests.put(
            url=f"http://{HOST}:{PORT}/api/dcim/devices/{self.list_of_ids['device_id']}/",
            headers=HEADERS,
            json={
                "device_type": self.list_of_ids['device_type_id'],
                "role": self.list_of_ids['device_role_id'],
                "site": self.list_of_ids['site_id'],
                "local_context_data": data
            },
            timeout=5
        )

        # Config context related tests

        self.assertEqual(upload_config_context.status_code,200)
        self.assertEqual(json.loads(upload_config_context.content)["local_context_data"]["radius_servers"][0],"1.1.1.1")
        self.assertEqual(json.loads(upload_config_context.content)["local_context_data"]["radius_servers"][1],"2.2.2.2")

        # Creating a interface with the good vlan

        response = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/interfaces/",
            headers=HEADERS,
            json={
                "device": self.list_of_ids['device_id'],
                "vdcs": [],
                "name": "GigabitEthernet2/4",
                "type": "1000base-t",
                "mode": "access",
                "description": "test",
                "duplex": "auto",
                "custom_fields": {
                    "access_vlan": 10
                }
            },
            timeout=5
        )

        # Interface related tests

        self.assertEqual(response.status_code,201)
        self.assertEqual(json.loads(response.content)["device"]["id"], self.list_of_ids['device_id'])
        self.assertEqual(json.loads(response.content)["name"], "GigabitEthernet2/4")
        self.assertEqual(json.loads(response.content)["type"]["value"], "1000base-t")
        self.assertEqual(json.loads(response.content)["mode"]["value"], "access")
        self.assertEqual(json.loads(response.content)["description"], "test")
        self.assertEqual(json.loads(response.content)["speed"], None)
        self.assertEqual(json.loads(response.content)["duplex"]["value"], "auto")
        self.assertEqual(json.loads(response.content)["custom_fields"]["access_vlan"], 10)

        # Creating an interface with an out of range vlan

        bad_request = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/interfaces/",
            headers=HEADERS,
            json={
                "device": self.list_of_ids['device_id'],
                "vdcs": [],
                "name": "GigabitEthernet2/5",
                "type": "1000base-t",
                "mode": "access",
                "description": "test",
                "duplex": "auto",
                "custom_fields": {
                    "access_vlan": 4097
                }
            },
            timeout=5
        )

        # Status code Should be a 400 - bad request
        self.assertEqual(bad_request.status_code, 400)

        # Test the creation of a trunk interface

        trunk_interface = requests.post(
            url=f"http://{HOST}:{PORT}/api/dcim/interfaces/",
            headers=HEADERS,
            json={
                "device": self.list_of_ids['device_id'],
                "vdcs": [],
                "name": "TenGigabitEthernet1/1",
                "type": "10gbase-x-sfpp",
                "mode": "tagged",
                "description": "test",
                "duplex": "auto",
                "custom_fields": {
                    "allowed_vlans_trunk": "10,50,100-1000,2000-2599"
                }
            },
            timeout=5
        )

        # Trunk interface related tests

        self.assertEqual(trunk_interface.status_code,201)
        self.assertEqual(json.loads(trunk_interface.content)["device"]["id"], self.list_of_ids['device_id'])
        self.assertEqual(json.loads(trunk_interface.content)["name"], "TenGigabitEthernet1/1")
        self.assertEqual(json.loads(trunk_interface.content)["type"]["value"], "10gbase-x-sfpp")
        self.assertEqual(json.loads(trunk_interface.content)["mode"]["value"], "tagged")
        self.assertEqual(json.loads(trunk_interface.content)["description"], "test")
        self.assertEqual(json.loads(trunk_interface.content)["speed"], None)
        self.assertEqual(json.loads(trunk_interface.content)["duplex"]["value"], "auto")
        self.assertEqual(
            json.loads(trunk_interface.content)["custom_fields"]["allowed_vlans_trunk"], "10,50,100-1000,2000-2599"
        )

if __name__ == "__main__":
    unittest.main()