Skip to content

Instantly share code, notes, and snippets.

@commonism
Created August 9, 2023 13:35
Show Gist options
  • Save commonism/18bd0a13dee523c47f5453f3f8b587cf to your computer and use it in GitHub Desktop.
Save commonism/18bd0a13dee523c47f5453f3f8b587cf to your computer and use it in GitHub Desktop.
netbox_api_test.py
import asyncio
import logging
from typing import Dict
from pathlib import Path
import aiopenapi3
import httpx
import pytest
import pytest_asyncio
import aiopenapi3.plugin
@pytest.fixture(scope="session")
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
class SetServer(aiopenapi3.plugin.Document):
def parsed(self, ctx: "Document.Context") -> "Document.Context":
ctx.document["servers"][0]["url"] = "/"
return ctx
class FixSchemas(aiopenapi3.plugin.Init):
def schemas(self, ctx: "Init.Context") -> "Init.Context":
from aiopenapi3 import v30
ctx.schemas: Dict[str, v30.Schema]
# dcim_device_roles_create response
ctx.schemas["DeviceRole"].required = list(filter(lambda x: x not in ["device_count", "virtualmachine_count"], ctx.schemas["DeviceRole"].required))
# dcim_sites_create response
ctx.schemas["Site"].required = list(filter(lambda x: x not in ["circuit_count","device_count", "prefix_count","rack_count","virtualmachine_count","vlan_count"], ctx.schemas["Site"].required))
# dcim_devices_create response
ctx.schemas["Device"].required = list(filter(
lambda x: x not in ["parent_device", "primary_ip"], ctx.schemas["Device"].required))
# dcim_devices_create response
ctx.schemas["DeviceWithConfigContext"].required = list(filter(
lambda x: x not in ["parent_device", "primary_ip"], ctx.schemas["DeviceWithConfigContext"].required))
for i in ["parent_device", "primary_ip"]:
ctx.schemas["DeviceWithConfigContext"].properties[i].nullable = True
# ipam_vrfs_create response
ctx.schemas["VRF"].required = list(filter(
lambda x: x not in ["ipaddress_count", "prefix_count"], ctx.schemas["VRF"].required))
# tenancy_tenants_create response
ctx.schemas["Tenant"].required = list(filter(
lambda x: x not in ["circuit_count", "device_count","ipaddress_count","prefix_count","rack_count","site_count","virtualmachine_count","vlan_count","vrf_count","cluster_count"], ctx.schemas["Tenant"].required))
# dcim_device_types_create
ctx.schemas["DeviceType"].required = list(filter(
lambda x: x not in ["device_count"], ctx.schemas["DeviceType"].required))
# dcim_interfaces_list
ctx.schemas["Interface"].required = list(filter(
lambda x: x not in ["link_peers_type","connected_endpoints","connected_endpoints_type","connected_endpoints_reachable"], ctx.schemas["Interface"].required))
# dcim_manufacturers_create
ctx.schemas["Manufacturer"].required = list(filter(
lambda x: x not in ["devicetype_count","inventoryitem_count","platform_count"], ctx.schemas["Manufacturer"].required))
return ctx
class FixPaths(aiopenapi3.plugin.Init):
@staticmethod
def operationbyId(ctx, name):
for path,item in ctx.paths.items():
for i in ["get","post","patch","head"]:
if (operation := getattr(item, i, None)) == None:
continue
if operation.operationId == name:
return path,i,operation
raise KeyError(name)
@staticmethod
def parameterbyname(operation, name):
for parameter in operation.parameters:
if parameter.name == name:
return parameter
raise KeyError(name)
def paths(self, ctx: "Init.Context") -> "Init.Context":
for i in []:
_,_,op = self.operationbyId(ctx, i)
p = self.parameterbyname(op, "slug")
p.schema_ = aiopenapi3.v30.Schema(type="string")
from aiopenapi3.debug import httpx_debug_event_hooks_async
@pytest_asyncio.fixture(scope="session")
async def client(event_loop):
url = f"http://127.0.0.1:8000/api/schema/"
def sf(*args, **kwargs) -> httpx.AsyncClient:
event_hooks = httpx_debug_event_hooks_async()
return httpx.AsyncClient(*args, timeout=30, event_hooks=event_hooks, **kwargs)
plugins = [SetServer(), FixSchemas(), FixPaths()]
path = Path("/tmp/nb.pickle")
try:
api = aiopenapi3.OpenAPI.cache_load(path, plugins=plugins, session_factory=sf)
except FileNotFoundError:
api = await aiopenapi3.OpenAPI.load_async(url, session_factory=sf, plugins = plugins)
api.cache_store(path)
api.authenticate(tokenAuth="Token 0a2c1764de239b00de09f8b89758bf778ed916f4")
return api
@pytest_asyncio.fixture(scope="session")
async def DeviceRole(client):
"""dcim_device_roles_create"""
t = client._.dcim_device_roles_create.data.get_type()
data = t(name="Server", slug="server")
try:
r = await client._.dcim_device_roles_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
r = await client._.dcim_device_roles_list(parameters={"slug":[data.slug]})
assert r.count == 1
r = r.results[0]
else:
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def DeviceType(client, Manufacturer):
"""dcim_device_types_create"""
t = client._.dcim_device_types_create.data.get_type()
data = t(manufacturer=Manufacturer.id, model="PowerEdge T320", slug="t320")
try:
r = await client._.dcim_device_types_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
print(e.response.json())
r = await client._.dcim_device_types_list(parameters={"slug":[data.slug]})
assert r.count == 1
r = r.results[0]
else:
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def Manufacturer(client):
"""dcim_manufacturers_create"""
t = client._.dcim_manufacturers_create.data.get_type()
data = t(name="Dell", slug="dell")
try:
r = await client._.dcim_manufacturers_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
print(e.response.json())
r = await client._.dcim_manufacturers_list(parameters={"slug":[data.slug]})
assert r.count == 1
r = r.results[0]
else:
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def Site(client):
"""dcim_sites_create"""
t = client._.dcim_sites_create.data.get_type()
data = t(name="Home", slug="home", status="active")
try:
r = await client._.dcim_sites_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
r = await client._.dcim_sites_list(parameters={"slug":[data.slug]})
assert r.count == 1
r = r.results[0]
else:
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def Device(client, DeviceRole, DeviceType, Site, Tenant):
"""dcim_devices_create"""
t = client._.dcim_devices_create.data.get_type()
data = t(device_type=DeviceType.id, device_role=DeviceRole.id, site=Site.id, status="active", name="Server2", tenant=Tenant.id)
try:
r = await client._.dcim_devices_list(parameters={"name":[data.name]})
r = r.results[0]
except IndexError:
try:
r = await client._.dcim_devices_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
print(e.response.json())
logging.exception(e)
raise e
except aiopenapi3.errors.ResponseSchemaError as e:
import json
print(json.dumps(e.response.json()))
raise
return r
@pytest_asyncio.fixture(scope="session")
async def Interface(client, Device):
"""dcim_interfaces_create"""
t = client._.dcim_interfaces_create.data.get_type()
data = t(device=Device.id, name="eth0", type="1000base-t", vdcs=[])
try:
r = await client._.dcim_interfaces_list(parameters={"name":[data.name], "device_id":[data.device]})
r = r.results[0]
except IndexError:
try:
r = await client._.dcim_interfaces_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
print(e.response.json())
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def VRF(client):
"""ipam_vrfs_create"""
t = client._.ipam_vrfs_create.data.get_type()
data = t(name="vrf1")
try:
r = await client._.ipam_vrfs_list(parameters={"name": [data.name]})
r = r.results[0]
except IndexError:
try:
r = await client._.ipam_vrfs_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
print(e.response.json())
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def Tenant(client):
"""tenancy_tenants_create"""
t = client._.ipam_vrfs_create.data.get_type()
data = t(name="TheTenant", slug="thetenant")
try:
r = await client._.tenancy_tenants_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
if e.response.status_code == 400:
r = await client._.tenancy_tenants_list(parameters={"slug":[data.slug]})
assert r.count == 1
r = r.results[0]
else:
logging.exception(e)
raise e
return r
@pytest_asyncio.fixture(scope="session")
async def IPAddress(client, VRF, Tenant, Interface):
"""ipam_ip_addresses_create"""
t = client._.ipam_ip_addresses_create.data.get_type()
data = t(address="10.0.0.1/24", vrf=VRF.id, tenant=Tenant.id, status="active", role="", dns_name="Server2")
try:
r = await client._.ipam_ip_addresses_list(parameters={"address": [data.address], "interface_id":[Interface.id]})
r = r.results[0]
except IndexError:
try:
r = await client._.ipam_ip_addresses_create(data=data.model_dump(mode="json", exclude_unset=True))
except aiopenapi3.errors.HTTPStatusError as e:
logging.exception(e)
raise e
return r
@pytest.mark.asyncio
async def test_client(client):
assert client
@pytest.mark.asyncio
async def test_create(client, Device, Interface, IPAddress):
assert Device
if Device.primary_ip4 is None:
t = client._.ipam_ip_addresses_update.data.get_type()
data = t(address=IPAddress.address, assigned_object_type = "dcim.interface", assigned_object_id = Interface.id)
try:
r = await client._.ipam_ip_addresses_update(parameters={"id":IPAddress.id}, data=data.model_dump(mode="json", exclude_unset=True))
print(r)
except aiopenapi3.errors.HTTPStatusError as e:
print(e.response.json())
raise
t = client._.dcim_devices_update.data.get_type()
data = t(device_type=Device.device_type.id, device_role=Device.device_role.id, site=Device.site.id, primary_ip4=IPAddress.id)
try:
r = await client._.dcim_devices_update(parameters={"id":Device.id}, data=data.model_dump(mode="json", exclude_unset=True))
print(r)
except aiopenapi3.errors.HTTPStatusError as e:
print(e.response.json())
raise
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment