From 7f05150a46a0838b6a816381cf7fc5fbe919a7c3 Mon Sep 17 00:00:00 2001 From: Christopher Andre Marcel INSELIN <christopher-andre-marcel.inselin@ext.ec.europa.eu> Date: Fri, 13 Oct 2023 17:16:41 +0200 Subject: [PATCH] :rotating_light: Fixed linter errors --- .../workplace/test_create_device_netbox.py | 205 ++++++++++++------ 1 file changed, 140 insertions(+), 65 deletions(-) diff --git a/tests/e2e/workplace/test_create_device_netbox.py b/tests/e2e/workplace/test_create_device_netbox.py index bd4db6d..adabc49 100644 --- a/tests/e2e/workplace/test_create_device_netbox.py +++ b/tests/e2e/workplace/test_create_device_netbox.py @@ -10,13 +10,17 @@ 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 ... """ + """Test case for device creation, config context update and interface creation """ + + # definition of the variables @classmethod def setUpClass(cls) -> None: """Will create a manufacturer, site, device role and device type""" + + list_of_ids: dict = {} + requests.post( url=f"http://{HOST}:{PORT}/api/dcim/manufacturers/", headers=HEADERS, @@ -42,14 +46,14 @@ class TestNetboxDeviceCreate(unittest.TestCase): #Will create a device type manufacturer = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/manufacturers/", headers=HEADERS, timeout=5) - manufactuer_id = json.loads(manufacturer.content)["results"][0]["id"] + list_of_ids["manufacturer_id"] = json.loads(manufacturer.content)["results"][0]["id"] requests.post( url=f"http://{HOST}:{PORT}/api/dcim/device-types/", headers=HEADERS, json={ "model": "WS-C2960X", "slug":"ws-c2960x", - "manufacturer": manufactuer_id + "manufacturer": list_of_ids["manufacturer_id"] }, timeout=5 ) @@ -65,10 +69,10 @@ class TestNetboxDeviceCreate(unittest.TestCase): timeout=5 ) site = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/sites/", headers=HEADERS, timeout=5) - site_id = json.loads(site.content)["results"][0]["id"] + list_of_ids["site_id"] = json.loads(site.content)["results"][0]["id"] device_type = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-types/", headers=HEADERS, timeout=5) - device_type_id = json.loads(device_type.content)["results"][0]["id"] + list_of_ids["device_type_id"] = json.loads(device_type.content)["results"][0]["id"] # Will create a dummy config context @@ -77,8 +81,8 @@ class TestNetboxDeviceCreate(unittest.TestCase): headers=HEADERS, json={ "name": "test_config_context", - "sites": [site_id], - "device_types": [device_type_id], + "sites": [list_of_ids["site_id"]], + "device_types": [list_of_ids["device_type_id"]], "data": { "radius_servers": [ "1.1.1.1", @@ -89,7 +93,7 @@ class TestNetboxDeviceCreate(unittest.TestCase): timeout=5 ) - # will create a custom field + # will create a custom field for access vlan requests.post( url=f"http://{HOST}:{PORT}/api/extras/custom-fields/", @@ -104,17 +108,33 @@ class TestNetboxDeviceCreate(unittest.TestCase): timeout=5 ) + # Will create a custom field for allowed vlans on a trunk interface + + 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 + ) + + @classmethod def tearDownClass(cls) -> None: """ Deleting all data from NETBOX once the all the tests finish running """ + list_of_ids: dict = {} + #deleting the device device = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/devices/", headers=HEADERS, timeout=5) - id_device = json.loads(device.content)["results"][0]["id"] + list_of_ids['id_device'] = json.loads(device.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/dcim/devices/{id_device}/", + url= f"http://{HOST}:{PORT}/api/dcim/devices/{list_of_ids['id_device']}/", headers=HEADERS, timeout=5 ) @@ -122,27 +142,27 @@ class TestNetboxDeviceCreate(unittest.TestCase): #deleting the device type device_type = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-types/", headers=HEADERS, timeout=5) - device_type_id = json.loads(device_type.content)["results"][0]["id"] + list_of_ids['device_type_id'] = json.loads(device_type.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/dcim/device-types/{device_type_id}/", + url= f"http://{HOST}:{PORT}/api/dcim/device-types/{list_of_ids['device_type_id']}/", headers=HEADERS, timeout=5 ) #deleting the manufacturer manufacturer = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/manufacturers/", headers=HEADERS, timeout=5) - manufacturer_id = json.loads(manufacturer.content)["results"][0]["id"] + list_of_ids['manufacturer_id'] = json.loads(manufacturer.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/dcim/manufacturers/{manufacturer_id}/", + url= f"http://{HOST}:{PORT}/api/dcim/manufacturers/{list_of_ids['manufacturer_id']}/", headers=HEADERS, timeout=5 ) # deleting the device role device_role = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-roles/", headers=HEADERS, timeout=5) - device_role_id = json.loads(device_role.content)["results"][0]["id"] + list_of_ids['device_role_id'] = json.loads(device_role.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/dcim/device-roles/{device_role_id}/", + url= f"http://{HOST}:{PORT}/api/dcim/device-roles/{list_of_ids['device_role_id']}/", headers=HEADERS, timeout=5 ) @@ -150,9 +170,9 @@ class TestNetboxDeviceCreate(unittest.TestCase): # deleting the site site = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/sites/", headers=HEADERS, timeout=5) - site_id = json.loads(site.content)["results"][0]["id"] + list_of_ids['site_id'] = json.loads(site.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/dcim/sites/{site_id}/", + url= f"http://{HOST}:{PORT}/api/dcim/sites/{list_of_ids['site_id']}/", headers=HEADERS, timeout=5 ) @@ -164,55 +184,68 @@ class TestNetboxDeviceCreate(unittest.TestCase): headers=HEADERS, timeout=5 ) - config_context_id = json.loads(config_context.content)["results"][0]["id"] + list_of_ids['config_context_id'] = json.loads(config_context.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/extras/config-contexts/{config_context_id}/", + url= f"http://{HOST}:{PORT}/api/extras/config-contexts/{list_of_ids['config_context_id']}/", headers=HEADERS, timeout=5 ) - # deleting the custom_field + # deleting the access_vlan custom_field - custom_field_vlan = requests.get( + custom_field = requests.get( url=f"http://{HOST}:{PORT}/api/extras/custom-fields/", headers=HEADERS, timeout=5 ) - custom_field_vlan_id = json.loads(custom_field_vlan.content)["results"][0]["id"] + + list_of_ids['custom_field_vlan_id'] = json.loads(custom_field.content)["results"][0]["id"] requests.delete( - url= f"http://{HOST}:{PORT}/api/extras/custom-fields/{custom_field_vlan_id}/", + url= f"http://{HOST}:{PORT}/api/extras/custom-fields/{list_of_ids['custom_field_vlan_id']}/", headers=HEADERS, timeout=5 ) + list_of_ids['allowed_vlans_trunk_id'] = json.loads(custom_field.content)["results"][1]["id"] + requests.delete( + url= f"http://{HOST}:{PORT}/api/extras/custom-fields/{list_of_ids['allowed_vlans_trunk_id']}/", + headers=HEADERS, + timeout=5 + ) + + # deleting the allowed_vlans_trunk custom_field + def tearDown(self) -> None: pass def test_device_creation(self) -> None: """" getting the necessary IDs """ + + list_of_ids: dict = {} + device_type = requests.get( url=f"http://{HOST}:{PORT}/api/dcim/device-types/", headers=HEADERS, timeout=5 ) - device_type_id = json.loads(device_type.content)["results"][0]["id"] + list_of_ids['device_type_id'] = json.loads(device_type.content)["results"][0]["id"] device_role = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-roles/", headers=HEADERS, timeout=5) - device_role_id = json.loads(device_role.content)["results"][0]["id"] + list_of_ids['device_role_id'] = json.loads(device_role.content)["results"][0]["id"] site = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/sites/", headers=HEADERS, timeout=5) - site_id = json.loads(site.content)["results"][0]["id"] + list_of_ids['site_id'] = json.loads(site.content)["results"][0]["id"] # Creating the device - response = response = requests.post( + response = requests.post( url=f"http://{HOST}:{PORT}/api/dcim/devices/", headers=HEADERS, json={ "name": "test", - "device_type": device_type_id, - "role": device_role_id, - "site": site_id, + "device_type": list_of_ids['device_type_id'], + "role": list_of_ids['device_role_id'], + "site": list_of_ids['site_id'], "status": "offline" }, timeout=5 @@ -225,71 +258,54 @@ class TestNetboxDeviceCreate(unittest.TestCase): self.assertEqual(json.loads(response.content)["site"]["name"],"ISPRA") self.assertEqual(json.loads(response.content)["status"]["value"],"offline") - def test_put_config_context(self) -> None: - """ Will test the update of config context into the device """ - - # First we need the config context ID + # Creating the config context config_context = requests.get( url=f"http://{HOST}:{PORT}/api/extras/config-contexts/", headers=HEADERS, timeout=5 ) - config_context_id = json.loads(config_context.content)["results"][0]["id"] + list_of_ids['config_context_id'] = json.loads(config_context.content)["results"][0]["id"] config_context_data = requests.get( - url=f"http://{HOST}:{PORT}/api/extras/config-contexts/{config_context_id}/", + url=f"http://{HOST}:{PORT}/api/extras/config-contexts/{list_of_ids['config_context_id']}/", headers=HEADERS, timeout=5 ) data = json.loads(config_context_data.content)["data"] - print(data) - - device_type = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-types/", headers=HEADERS, timeout=5) - device_type_id = json.loads(device_type.content)["results"][0]["id"] - device_role = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/device-roles/", headers=HEADERS, timeout=5) - device_role_id = json.loads(device_role.content)["results"][0]["id"] - - site = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/sites/", headers=HEADERS, timeout=5) - site_id = json.loads(site.content)["results"][0]["id"] + # Getting the device_id device = requests.get(url=f"http://{HOST}:{PORT}/api/dcim/devices/", headers=HEADERS, timeout=5) - device_id = json.loads(device.content)["results"][0]["id"] + list_of_ids['device_id'] = json.loads(device.content)["results"][0]["id"] + + # 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/{device_id}/", + url=f"http://{HOST}:{PORT}/api/dcim/devices/{list_of_ids['device_id']}/", headers=HEADERS, json={ - "device_type": device_type_id, - "role": device_role_id, - "site": site_id, + "device_type": list_of_ids['device_type_id'], + "role": list_of_ids['device_role_id'], + "site": list_of_ids['site_id'], "local_context_data": data }, timeout=5 ) - self.assertEqual(upload_config_context.status_code,200) + # 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") - def test_interface_access_creation(self) -> None: - - """ Test that interface has been created """ - - device = requests.get( - url=f"http://{HOST}:{PORT}/api/dcim/devices/", - headers=HEADERS, - timeout=5 - ) - device_id = json.loads(device.content)["results"][0]["id"] + # Creating a interface with the good vlan response = requests.post( url=f"http://{HOST}:{PORT}/api/dcim/interfaces/", headers=HEADERS, json={ - "device": device_id, + "device": list_of_ids['device_id'], "vdcs": [], "name": "GigabitEthernet2/4", "type": "1000base-t", @@ -303,8 +319,10 @@ class TestNetboxDeviceCreate(unittest.TestCase): timeout=5 ) + # Interface related tests + self.assertEqual(response.status_code,201) - self.assertEqual(json.loads(response.content)["device"]["id"], device_id) + self.assertEqual(json.loads(response.content)["device"]["id"], 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") @@ -313,5 +331,62 @@ class TestNetboxDeviceCreate(unittest.TestCase): 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": 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": 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"], 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() -- GitLab