netbox_api_test.py

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