Netbox Service Tests

Overview

The Netbox Service Tests (test_netbox_service.py) provide comprehensive testing coverage for NORFAB's Netbox integration service. These tests validate the Netbox worker's ability to interact with Netbox instances.

Test Summary

The test suite is organized into 17 test classes:

Test Class	Purpose	Documentation
TestNetboxWorker	Core service discovery and version checks	Details
TestNetboxGrapQL	GraphQL query operations	Details
TestGetInterfaces	Interface retrieval and filtering	Details
TestGetDevices	Device retrieval and filtering	Details
TestGetConnections	Device connection queries	Details
TestGetNornirInventory	Nornir inventory generation from Netbox	Details
TestGetCircuits	Circuit information retrieval	Details
TestGetBgpPeerings	BGP peering data	Details
TestSyncDeviceFacts	Device fact synchronization	Details
TestSyncDeviceInterfaces	Device interface synchronization	Details
TestCreateDeviceInterfaces	Device interface creation	Details
TestSyncDeviceIP	IP address synchronization	Details
TestCreateIP	IP address creation	Details
TestNetboxCache	Caching functionality	Details
TestGetContainerlabInventory	Containerlab inventory generation	Details
TestCreatePrefix	IP prefix creation	Details
TestCreateIPBulk	Bulk IP address creation	Details

Test Classes

TestNetboxWorker

Core Netbox service functionality and health checks.

Purpose: Validate basic service operations, version compatibility, and status monitoring.

Test Get Netbox Inventory

Source code in tests\test_netbox_service.py

def test_get_netbox_inventory(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_inventory",
        workers="any",
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["service", "instances"]
        ), f"{worker} - not all netbox inventory data returned"
        assert all(
            k in res["result"]["instances"] for k in ["dev", "preprod", "prod"]
        ), f"{worker} - not all netbox instances inventory data returned"

Test Get Netbox Version

Source code in tests\test_netbox_service.py

def test_get_netbox_version(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_version",
        workers="any",
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"]
            for k in ["platform", "pynetbox", "python", "requests"]
        ), f"{worker} - not all netbox version data returned"

Test Get Netbox Status

Source code in tests\test_netbox_service.py

def test_get_netbox_status(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_netbox_status",
        workers="any",
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["dev", "preprod", "prod"]
        ), f"{worker} - not all netbox instances inventory data returned"
        for instance, status_data in res["result"].items():
            assert all(
                k in status_data
                for k in [
                    "django-version",
                    "error",
                    "netbox-version",
                    "plugins",
                    "python-version",
                    "rq-workers-running",
                    "status",
                ]
            ), f"{worker}:{instance} - not all netbox instances status data returned"

Test Get Netbox Compatibility

Source code in tests\test_netbox_service.py

def test_get_netbox_compatibility(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_compatibility",
        workers="any",
    )
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["dev", "preprod", "prod"]
        ), f"{worker} - not all netbox instances inventory data returned"
        for instance, compatible in res["result"].items():
            assert compatible == True, f"{worker}:{instance} - not compatible"

TestNetboxGrapQL

GraphQL query operations against Netbox instances.

Purpose: Validate GraphQL query execution, dry-run mode, and error handling.

Test Graphql Query String

Source code in tests\test_netbox_service.py

def test_graphql_query_string(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "graphql",
        workers="any",
        kwargs={"query_string": "query DeviceListQuery { device_list { name } }"},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert "device_list" in res["result"], f"{worker} no device list returned"
        assert isinstance(
            res["result"]["device_list"], list
        ), f"{worker} unexpected device list payload type, was expecting list"
        assert (
            len(res["result"]["device_list"]) > 0
        ), f"{worker} returned no devices in device list"

Test Graphql Query String With Instance

Source code in tests\test_netbox_service.py

def test_graphql_query_string_with_instance(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "graphql",
        workers="any",
        kwargs={
            "query_string": "query DeviceListQuery { device_list { name } }",
            "instance": "prod",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert "device_list" in res["result"], f"{worker} no device list returned"
        assert isinstance(
            res["result"]["device_list"], list
        ), f"{worker} unexpected device list payload type, was expecting list"
        assert (
            len(res["result"]["device_list"]) > 0
        ), f"{worker} returned no devices in device list"

Test Graphql Query String Dry Run

Source code in tests\test_netbox_service.py

def test_graphql_query_string_dry_run(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "graphql",
        workers="any",
        kwargs={
            "query_string": "query DeviceListQuery { device_list { name } }",
            "dry_run": True,
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["headers", "data", "verify", "url"]
        ), f"{worker} - not all dry run data returned"

Test Graphql Query String Error

Source code in tests\test_netbox_service.py

def test_graphql_query_string_error(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "graphql",
        workers="any",
        kwargs={
            "query_string": "query DeviceListQuery { device_list { name } ",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert res[
            "errors"
        ], f"{worker} did not return errors for malformed graphql query"

Test Form Graphql Query Dry Run

Source code in tests\test_netbox_service.py

def test_form_graphql_query_dry_run(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 4:
        ret = nfclient.run_job(
            "netbox",
            "graphql",
            workers="any",
            kwargs={
                "obj": "device_list",
                "fields": ["name", "platform {name}"],
                "filters": {"q": "ceos", "platform": "arista_eos"},
                "dry_run": True,
            },
        )
        pprint.pprint(ret, width=200)
        for worker, res in ret.items():
            assert res["result"]["data"] == (
                '{"query": "query {device_list(filters: {q: \\"ceos\\", platform: \\"arista_eos\\"}) {name platform {name}}}"}'
            ), f"{worker} did not return correct query string"
    elif self.nb_version[0] == 3:
        ret = nfclient.run_job(
            "netbox",
            "graphql",
            workers="any",
            kwargs={
                "obj": "device_list",
                "fields": ["name", "platform {name}"],
                "filters": {"name__ic": "ceos", "platform": "arista_eos"},
                "dry_run": True,
            },
        )
        pprint.pprint(ret, width=200)
        for worker, res in ret.items():
            assert res["result"]["data"] == (
                '{"query": "query {device_list(name__ic: \\"ceos\\", platform: \\"arista_eos\\") {name platform {name}}}"}'
            ), f"{worker} did not return correct query string"

TestGetInterfaces

Interface retrieval and filtering operations.

Purpose: Validate device interface queries with various filters and parameters.

Test Get Interfaces

Source code in tests\test_netbox_service.py

def test_get_interfaces(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)
    clear_nb_cache("get_interfaces*", nfclient)
    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"]},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert all(
                    k in intf_data
                    for k in [
                        "enabled",
                        "description",
                        "mtu",
                        "parent",
                        "mode",
                        "untagged_vlan",
                        "vrf",
                        "tagged_vlans",
                        "tags",
                        "custom_fields",
                        "last_updated",
                        "bridge",
                        "child_interfaces",
                        "bridge_interfaces",
                        "member_interfaces",
                        "wwn",
                        "duplex",
                        "speed",
                    ]
                ), f"{worker}:{device}:{intf_name} not all data returned"
                if self.nb_version >= (4, 2, 0):
                    assert "mac_addresses" in intf_data
                else:
                    assert "mac_address" in intf_data

Test Get Interfaces With Instance

Source code in tests\test_netbox_service.py

def test_get_interfaces_with_instance(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)
    clear_nb_cache("get_interfaces*", nfclient)
    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"], "instance": "prod"},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert all(
                    k in intf_data
                    for k in [
                        "enabled",
                        "description",
                        "mtu",
                        "parent",
                        "mode",
                        "untagged_vlan",
                        "vrf",
                        "tagged_vlans",
                        "tags",
                        "custom_fields",
                        "last_updated",
                        "bridge",
                        "child_interfaces",
                        "bridge_interfaces",
                        "member_interfaces",
                        "wwn",
                        "duplex",
                        "speed",
                    ]
                ), f"{worker}:{device}:{intf_name} not all data returned"
                if self.nb_version >= (4, 2, 0):
                    assert "mac_addresses" in intf_data
                else:
                    assert "mac_address" in intf_data

Test Get Interfaces Dry Run

Source code in tests\test_netbox_service.py

def test_get_interfaces_dry_run(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"], "dry_run": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert res["result"] == {
            "filter_params": {"device": ["ceos1", "fceos4"]}
        }, f"{worker} did not return correct query string"

Test Get Interfaces Add Ip

Source code in tests\test_netbox_service.py

def test_get_interfaces_add_ip(self, nfclient):
    clear_nb_cache("get_interfaces*", nfclient)
    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"], "ip_addresses": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert (
                    "ip_addresses" in intf_data
                ), f"{worker}:{device}:{intf_name} no IP addresses data returned"
                for ip in intf_data["ip_addresses"]:
                    assert all(
                        k in ip
                        for k in [
                            "address",
                            "family",
                        ]
                    ), f"{worker}:{device}:{intf_name} not all IP data returned"

Test Get Interfaces Add Inventory Items

Source code in tests\test_netbox_service.py

def test_get_interfaces_add_inventory_items(self, nfclient):
    clear_nb_cache("get_interfaces*", nfclient)
    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos1", "fceos4"],
            "inventory_items": True,
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert (
                    "inventory_items" in intf_data
                ), f"{worker}:{device}:{intf_name} no inventory items data returned"
                for item in intf_data["inventory_items"]:
                    assert all(
                        k in item
                        for k in [
                            "name",
                            "role",
                            "manufacturer",
                            "custom_fields",
                            "serial",
                        ]
                    ), f"{worker}:{device}:{intf_name} not all inventory item data returned"

Test Get Interfaces With Interface Regex

Source code in tests\test_netbox_service.py

def test_get_interfaces_with_interface_regex(self, nfclient):
    clear_nb_cache("get_interfaces*", nfclient)
    ret = nfclient.run_job(
        "netbox",
        "get_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos1", "fceos4"],
            "interface_regex": "loop.+",
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert (
                    "loopback" in intf_name.lower()
                ), f"{worker}:{device}:{intf_name} interface name does not match regex pattern"

TestGetDevices

Device retrieval and advanced filtering.

Purpose: Validate comprehensive device queries with filtering, selection, and data structure validation.

Test With Devices List

Source code in tests\test_netbox_service.py

def test_with_devices_list(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"]},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            print(list(device_data.keys()))
            assert isinstance(
                device_data, dict
            ), f"{worker}:{device} did not return device data as dictionary"
            assert all(
                k in device_data for k in self.device_data_keys
            ), f"{worker}:{device} not all data returned"
            assert (
                "role" in device_data
            ), f"{worker}:{device} nodevice role info returned"

Test With Filters

Source code in tests\test_netbox_service.py

def test_with_filters(self, nfclient):
    # REST API filter syntax: plain dicts with standard query params
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "filters": [
                {"name": ["ceos1", "fceos4"]},
                {"name__ic": "390"},
            ]
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert (
            "fceos3_390" in res["result"]
        ), f"{worker} returned no results for fceos3_390"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            assert isinstance(
                device_data, dict
            ), f"{worker}:{device} did not return device data as dictionary"
            assert all(
                k in device_data for k in self.device_data_keys
            ), f"{worker}:{device} not all data returned"
            assert (
                "role" in device_data or "devcie_role" in device_data
            ), f"{worker}:{device} nodevice role info returned"

Test With Filters Dry Run

Source code in tests\test_netbox_service.py

def test_with_filters_dry_run(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "filters": [
                {"name": ["ceos1", "fceos4"]},
                {"name__ic": "390"},
            ],
            "dry_run": True,
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert (
            "get_devices_dry_run" in res["result"]
        ), f"{worker} - dry run key missing from result"
        dry_run_data = res["result"]["get_devices_dry_run"]
        assert (
            "filters" in dry_run_data
        ), f"{worker} - 'filters' key missing from dry run result"
        assert isinstance(
            dry_run_data["filters"], list
        ), f"{worker} - dry run filters should be a list"

Test Get Devices Cache

Source code in tests\test_netbox_service.py

@pytest.mark.parametrize("cache", cache_options)
def test_get_devices_cache(self, nfclient, cache):
    # REST API filter syntax works across all Netbox versions
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "devices": ["ceos1", "fceos4"],
            "filters": [
                {"name": ["ceos1", "fceos4"]},
                {"name__ic": "390"},
            ],
            "cache": cache,
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "ceos1" in res["result"], f"{worker} returned no results for ceos1"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            assert isinstance(
                device_data, dict
            ), f"{worker}:{device} did not return device data as dictionary"
            assert all(
                k in device_data for k in self.device_data_keys
            ), f"{worker}:{device} not all data returned"
            assert (
                "role" in device_data or "devcie_role" in device_data
            ), f"{worker}:{device} nodevice role info returned"

TestGetConnections

Device connection and cable management queries.

Purpose: Validate device connection retrieval and relationship mapping.

Test Get Connections

Source code in tests\test_netbox_service.py

def test_get_connections(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        assert (
            "ConsolePort1" in res["result"]["fceos4"]
        ), f"{worker}:fceos4 no console ports data returned"
        assert (
            "ConsoleServerPort1" in res["result"]["fceos5"]
        ), f"{worker}:fceos5 no console server ports data returned"
        assert (
            "eth11.123" in res["result"]["fceos5"]
        ), f"{worker}:fceos5 no virtual ports data returned"
        assert (
            "Port-Channel1" in res["result"]["fceos4"]
        ), f"{worker}:fceos5 no lag ports data returned"
        assert (
            "Port-Channel1.101" in res["result"]["fceos4"]
        ), f"{worker}:fceos5 no lag virtual ports data returned"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                assert all(
                    k in intf_data
                    for k in [
                        "remote_device",
                        "remote_interface",
                        "remote_termination_type",
                        "termination_type",
                    ]
                ), f"{worker}:{device}:{intf_name} not all data returned"
                # verify provider network connection handling
                if device == "fceos4" and intf_name == "eth201":
                    assert (
                        "provider" in intf_data
                    ), f"{worker}:{device}:{intf_name} no provider data"
                    assert intf_data["remote_termination_type"] == "providernetwork"
                    assert intf_data["remote_device"] == None
                    assert intf_data["remote_interface"] == None
                    assert intf_data["remote_interface_label"] == None
                # verify breakout handling
                if device == "fceos5" and intf_name == "eth1":
                    assert (
                        intf_data["breakout"] == True
                    ), f"{worker}:{device}:{intf_name} was expecting breakout connection"
                    assert isinstance(intf_data["remote_interface"], list)
                    assert len(intf_data["remote_interface"]) > 1
                # verify virtual ports handling
                if device == "fceos5" and intf_name == "eth11.123":
                    assert intf_data["remote_device"] == "fceos4"
                    assert intf_data["remote_device_status"] == "active"
                    assert intf_data["remote_interface"] == "eth11.123"
                    assert intf_data["termination_type"] == "virtual"
                # verify lag ports handling
                if device == "fceos4" and intf_name == "Port-Channel1":
                    assert intf_data["remote_device"] == "fceos5"
                    assert intf_data["remote_interface"] == "ae5"
                    assert intf_data["termination_type"] == "lag"
                    assert "remote_interface_label" not in intf_data
                # verify lag virtual interfaces handling
                if device == "fceos4" and intf_name == "Port-Channel1.101":
                    assert intf_data["remote_device"] == "fceos5"
                    assert intf_data["remote_interface"] == "ae5.101"
                    assert intf_data["termination_type"] == "virtual"
                    assert "remote_interface_label" not in intf_data
                if device == "fceos5" and intf_name == "ae6.0":
                    assert intf_data["remote_device"] == "fceos4"
                    assert intf_data["remote_interface"] == "Port-Channel2"
                    assert intf_data["termination_type"] == "virtual"
                    assert intf_data["remote_termination_type"] == "lag"
                if device == "fceos4" and intf_name == "eth103.0":
                    assert intf_data["remote_device"] == "fceos5"
                    assert intf_data["remote_interface"] == "eth103"
                    assert intf_data["termination_type"] == "virtual"
                    assert intf_data["remote_termination_type"] == "interface"
                    assert intf_data["remote_interface_label"] == ""
                if device == "fceos4" and intf_name == "ConsolePort2":
                    assert intf_data["remote_device"] == "fceos5"
                    assert intf_data["remote_interface"] == "ConsoleServerPort2"
                    assert intf_data["termination_type"] == "consoleport"
                    assert (
                        intf_data["remote_termination_type"] == "consoleserverport"
                    )
                    assert intf_data["remote_interface_label"] == ""

Test Get Connections Physical Interface Regex

Source code in tests\test_netbox_service.py

def test_get_connections_physical_interface_regex(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "interface_regex": "eth10.*",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert interfaces, f"{worker}:{device} no connections data returned"
            for intf_name, intf_data in interfaces.items():
                assert intf_name.startswith(
                    "eth10"
                ), f"{worker}:{device}:{intf_name} not matching regex"

Test Get Connections Virtual Interface Regex

Source code in tests\test_netbox_service.py

def test_get_connections_virtual_interface_regex(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "interface_regex": "(Port-Channel1|ae5).101",  # match Port-Channel1.101 and ae5.101
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        assert (
            res["result"]["fceos4"]["Port-Channel1.101"]["remote_interface"]
            == "ae5.101"
        ), "Unexpected interface name"
        assert (
            res["result"]["fceos5"]["ae5.101"]["remote_interface"]
            == "Port-Channel1.101"
        ), "Unexpected interface name"

Test Get Connections Dry Run

Source code in tests\test_netbox_service.py

def test_get_connections_dry_run(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "dry_run": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        dry_run_payload = res["result"]
        assert all(
            k in dry_run_payload for k in ["headers", "data", "verify", "url"]
        ), f"{worker} - not all dry run data returned"

Test Get Connections And Cables

Source code in tests\test_netbox_service.py

def test_get_connections_and_cables(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"]},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, interfaces in res["result"].items():
            assert isinstance(
                interfaces, dict
            ), f"{worker}:{device} did not return interfaces dictionary"
            for intf_name, intf_data in interfaces.items():
                if intf_data["termination_type"] in ["virtual", "lag"]:
                    continue
                assert (
                    "cable" in intf_data
                ), f"{worker}:{device}:{intf_name} no cable data returned"
                assert all(
                    k in intf_data["cable"]
                    for k in [
                        "custom_fields",
                        "label",
                        "peer_device",
                        "peer_interface",
                        "peer_termination_type",
                        "status",
                        "tags",
                        "tenant",
                        "type",
                    ]
                ), f"{worker}:{device}:{intf_name} not all cable data returned"

TestGetNornirInventory

Nornir inventory generation from Netbox data.

Purpose: Validate conversion of Netbox data into Nornir-compatible inventory format.

Test With Devices

Source code in tests\test_netbox_service.py

def test_with_devices(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4", "nonexist"]},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert (
            "ceos1" in res["result"]["hosts"]
        ), f"{worker} returned no results for ceos1"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert all(
                k in data for k in ["data", "hostname", "platform"]
            ), f"{worker}:{device} not all data returned"

Test With Filters

Source code in tests\test_netbox_service.py

def test_with_filters(self, nfclient):

    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={
            "filters": [
                {"name": ["ceos1"]},
                {"name__ic": "fceos"},
            ]
        },
    )
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert (
            "ceos1" in res["result"]["hosts"]
        ), f"{worker} returned no results for ceos1"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert all(
                k in data for k in ["data", "hostname", "platform"]
            ), f"{worker}:{device} not all data returned"

Test Source Platform From Config Context

Source code in tests\test_netbox_service.py

def test_source_platform_from_config_context(self, nfclient):
    # for iosxr1 platform data encoded in config context
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["iosxr1"]},
    )
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert (
            "iosxr1" in res["result"]["hosts"]
        ), f"{worker} returned no results for iosxr1"
        for device, data in res["result"]["hosts"].items():
            assert all(
                k in data for k in ["data", "hostname", "platform"]
            ), f"{worker}:{device} not all data returned"

Test With Devices Nbdata Is True

Source code in tests\test_netbox_service.py

def test_with_devices_nbdata_is_true(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"], "nbdata": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert (
            "ceos1" in res["result"]["hosts"]
        ), f"{worker} returned no results for ceos1"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert all(
                k in data for k in ["data", "hostname", "platform"]
            ), f"{worker}:{device} not all device data returned"
            assert all(
                k in data["data"] for k in self.device_data_keys
            ), f"{worker}:{device} not all nbdata returned"

Test With Devices Add Interfaces

Source code in tests\test_netbox_service.py

def test_with_devices_add_interfaces(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["ceos1", "fceos4"], "interfaces": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert (
            "ceos1" in res["result"]["hosts"]
        ), f"{worker} returned no results for ceos1"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert data["data"][
                "interfaces"
            ], f"{worker}:{device} no interfaces data returned"
            for intf_name, intf_data in data["data"]["interfaces"].items():
                assert all(
                    k in intf_data
                    for k in [
                        "vrf",
                        "mode",
                        "description",
                    ]
                ), f"{worker}:{device}:{intf_name} not all interface data returned"

Test With Devices Add Interfaces With Ip And Inventory

Source code in tests\test_netbox_service.py

def test_with_devices_add_interfaces_with_ip_and_inventory(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={
            "devices": ["ceos1", "fceos4"],
            "interfaces": {"ip_addresses": True, "inventory_items": True},
        },
    )
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert (
            "ceos1" in res["result"]["hosts"]
        ), f"{worker} returned no results for ceos1"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert data["data"][
                "interfaces"
            ], f"{worker}:{device} no interfaces data returned"
            for intf_name, intf_data in data["data"]["interfaces"].items():
                assert (
                    "ip_addresses" in intf_data
                ), f"{worker}:{device}:{intf_name} no ip addresses data returned"
                assert (
                    "inventory_items" in intf_data
                ), f"{worker}:{device}:{intf_name} no invetnory data returned"

Test With Devices Add Connections

Source code in tests\test_netbox_service.py

def test_with_devices_add_connections(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "connections": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert (
            "fceos5" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos5"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert data["data"][
                "connections"
            ], f"{worker}:{device} no connections data returned"
            for intf_name, intf_data in data["data"]["connections"].items():
                assert all(
                    k in intf_data
                    for k in [
                        "remote_interface",
                        "remote_device",
                    ]
                ), f"{worker}:{device}:{intf_name} not all connection data returned"

Test With Devices Add Bgp Peerings

Source code in tests\test_netbox_service.py

def test_with_devices_add_bgp_peerings(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_nornir_inventory",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "bgp_peerings": True},
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert (
            "fceos5" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos5"
        assert (
            "fceos4" in res["result"]["hosts"]
        ), f"{worker} returned no results for fceos4"
        for device, data in res["result"]["hosts"].items():
            assert data["data"][
                "bgp_peerings"
            ], f"{worker}:{device} no bgp_peerings data returned"
            for peering, peering_data in data["data"]["bgp_peerings"].items():
                assert all(
                    k in peering_data
                    for k in [
                        "id",
                        "name",
                    ]
                ), f"{worker}:{device}:{peering} not all peerings data returned"

TestGetCircuits

Circuit information retrieval from Netbox.

Purpose: Validate circuit data queries and filtering.

Test Get Circuits Dry Run

Source code in tests\test_netbox_service.py

def test_get_circuits_dry_run(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 3:
        ret = nfclient.run_job(
            "netbox",
            "get_circuits",
            workers="any",
            kwargs={
                "devices": ["fceos4", "fceos5"],
                "dry_run": True,
            },
        )
        pprint.pprint(ret, width=200)
        for worker, res in ret.items():
            assert res["result"]["data"] == (
                '{"query": "query {circuit_list(site: '
                + '[\\"saltnornir-lab\\"]) {cid tags {name} '
                + "provider {name} commit_rate description status "
                + "type {name} provider_account {name} tenant "
                + "{name} termination_a {id} termination_z {id} "
                + 'custom_fields comments}}"}'
            ), f"{worker} did not return correct query string"

Test Get Circuits

Source code in tests\test_netbox_service.py

def test_get_circuits(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_circuits",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
        },
    )
    pprint.pprint(ret, width=200)
    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            assert device_data, f"{worker}:{device} no circuit data returned"
            for cid, cid_data in device_data.items():
                if cid == "CID3":
                    assert all(
                        k in cid_data
                        for k in [
                            "tags",
                            "provider",
                            "commit_rate",
                            "description",
                            "status",
                            "type",
                            "provider_account",
                            "tenant",
                            "custom_fields",
                            "comments",
                            "provider_account",
                            "provider_network",
                        ]
                    ), f"{worker}:{device}:{cid} not all circuit data returned"
                else:
                    assert all(
                        k in cid_data
                        for k in [
                            "tags",
                            "provider",
                            "commit_rate",
                            "description",
                            "status",
                            "type",
                            "provider_account",
                            "tenant",
                            "custom_fields",
                            "comments",
                            "remote_device",
                            "remote_interface",
                        ]
                    ), f"{worker}:{device}:{cid} not all circuit data returned"

Test Get Circuits By Cid

Source code in tests\test_netbox_service.py

def test_get_circuits_by_cid(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_circuits",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "cid": ["CID1"]},
    )
    pprint.pprint(ret, width=200)
    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            assert device_data, f"{worker}:{device} no circuit data returned"
            for cid, cid_data in device_data.items():
                assert (
                    cid == "CID1"
                ), f"{worker}:{device}:{cid} wrong circuit returned, was expecting 'CID1' only"

Test Get Circuits Cache

Source code in tests\test_netbox_service.py

@pytest.mark.parametrize("cache", cache_options)
def test_get_circuits_cache(self, nfclient, cache):
    print(f"cache: {cache}")
    ret = nfclient.run_job(
        "netbox",
        "get_circuits",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "cache": cache},
    )
    pprint.pprint(ret, width=200)
    for worker, res in ret.items():
        assert "fceos5" in res["result"], f"{worker} returned no results for fceos5"
        assert "fceos4" in res["result"], f"{worker} returned no results for fceos4"
        for device, device_data in res["result"].items():
            assert device_data, f"{worker}:{device} no circuit data returned"
            for cid, cid_data in device_data.items():
                if cid == "CID3":
                    assert all(
                        k in cid_data
                        for k in [
                            "tags",
                            "provider",
                            "commit_rate",
                            "description",
                            "status",
                            "type",
                            "provider_account",
                            "tenant",
                            "custom_fields",
                            "comments",
                            "provider_account",
                            "provider_network",
                        ]
                    ), f"{worker}:{device}:{cid} not all circuit data returned"
                else:
                    assert all(
                        k in cid_data
                        for k in [
                            "tags",
                            "provider",
                            "commit_rate",
                            "description",
                            "status",
                            "type",
                            "provider_account",
                            "tenant",
                            "custom_fields",
                            "comments",
                            "remote_device",
                            "remote_interface",
                        ]
                    ), f"{worker}:{device}:{cid} not all circuit data returned"

Test Get Circuits Cache Content

Source code in tests\test_netbox_service.py

def test_get_circuits_cache_content(self, nfclient):
    circuits_cache = nfclient.run_job(
        "netbox",
        "cache_get",
        workers="all",
        kwargs={"keys": "get_circuits*"},
    )

    pprint.pprint(circuits_cache)

    for worker, res in circuits_cache.items():
        if "get_circuits::CID1" in res["result"]:
            assert (
                res["result"]["get_circuits::CID1"]["fceos4"]["remote_device"]
                == "fceos5"
            )
            assert (
                res["result"]["get_circuits::CID1"]["fceos5"]["remote_device"]
                == "fceos4"
            )
        if "get_circuits::CID2" in res["result"]:
            assert (
                res["result"]["get_circuits::CID2"]["fceos4"]["remote_device"]
                == "fceos5"
            )
            assert (
                res["result"]["get_circuits::CID2"]["fceos5"]["remote_device"]
                == "fceos4"
            )
        if "get_circuits::CID3" in res["result"]:
            assert (
                res["result"]["get_circuits::CID3"]["fceos4"]["provider_network"]
                == "Provider1-Net1"
            )

TestGetBgpPeerings

BGP peering information retrieval.

Purpose: Validate BGP peering data extraction and filtering.

Test Get Bgp Peerings

Test basic BGP peerings retrieval

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings(self, nfclient):
    """Test basic BGP peerings retrieval"""
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"]},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["result"], f"{worker} returned no results"
        assert "fceos4" in res["result"], f"{worker} missing fceos4 in results"
        assert "fceos5" in res["result"], f"{worker} missing fceos5 in results"
        # Check that each device has a dictionary (may be empty if no BGP sessions)
        for device, bgp_sessions in res["result"].items():
            assert isinstance(
                bgp_sessions, dict
            ), f"{worker}:{device} BGP sessions should be a dictionary"
            # If there are BGP sessions, verify the structure
            for session_name, session_data in bgp_sessions.items():
                assert isinstance(
                    session_data, dict
                ), f"{worker}:{device}:{session_name} session data should be a dictionary"

                # Verify required top-level fields
                required_fields = [
                    "id",
                    "name",
                    "description",
                    "device",
                    "local_address",
                    "local_as",
                    "remote_address",
                    "remote_as",
                    "status",
                    "last_updated",
                    "created",
                    "url",
                    "display",
                    "site",
                    "tenant",
                    "tags",
                    "comments",
                    "custom_fields",
                ]
                for field in required_fields:
                    assert (
                        field in session_data
                    ), f"{worker}:{device}:{session_name} missing field '{field}'"

Test Get Bgp Peerings With Instance

Test BGP peerings retrieval with explicit instance

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_with_instance(self, nfclient):
    """Test BGP peerings retrieval with explicit instance"""
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["fceos4", "fceos5"], "instance": "prod"},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["result"], f"{worker} returned no results"
        assert "fceos4" in res["result"], f"{worker} missing fceos4 in results"
        assert "fceos5" in res["result"], f"{worker} missing fceos5 in results"
        # Check that each device has a dictionary (may be empty if no BGP sessions)
        for device, bgp_sessions in res["result"].items():
            assert isinstance(
                bgp_sessions, dict
            ), f"{worker}:{device} BGP sessions should be a dictionary"

Test Get Bgp Peerings Nonexistent Device

Test error handling for non-existent device

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_nonexistent_device(self, nfclient):
    """Test error handling for non-existent device"""
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["nonexistent-device-12345"]},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["result"], f"{worker} returned no results"
        assert (
            "nonexistent-device-12345" in res["result"]
        ), f"{worker} should have entry for nonexistent device"
        # The result for non-existent device should be empty dict
        assert (
            res["result"]["nonexistent-device-12345"] == {}
        ), f"{worker} should return empty dict for nonexistent device"

Test Get Bgp Peerings Empty Devices List

Test with empty devices list

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_empty_devices_list(self, nfclient):
    """Test with empty devices list"""
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": []},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert isinstance(
            res["result"], dict
        ), f"{worker} should return a dictionary"
        assert (
            len(res["result"]) == 0
        ), f"{worker} should return empty dict for empty devices list"

Test Get Bgp Peerings Cache True

Test cache content for BGP peerings

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_cache_true(self, nfclient):
    """Test cache content for BGP peerings"""
    # get cache brief info
    cache_before = nfclient.run_job(
        "netbox",
        "cache_list",
        workers=["netbox-worker-1.1"],
        kwargs={"keys": "get_bgp_peerings*", "details": True},
    )

    # Clear any existing cache
    nfclient.run_job(
        "netbox",
        "cache_clear",
        workers=["netbox-worker-1.1"],
        kwargs={"keys": "get_bgp_peerings*"},
    )

    # cache data
    nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers=["netbox-worker-1.1"],
        kwargs={"devices": ["fceos4", "fceos5"], "cache": True},
    )

    # Now retrieve cache content
    cache_after = nfclient.run_job(
        "netbox",
        "cache_list",
        workers=["netbox-worker-1.1"],
        kwargs={"keys": "get_bgp_peerings*", "details": True},
    )

    print("cache_before:")
    pprint.pprint(cache_before, width=200)

    print("cache_after:")
    pprint.pprint(cache_after, width=200)

    for worker, res in cache_after.items():
        for cache_item_after in res["result"]:
            key = cache_item_after["key"]
            for cache_item_before in cache_before[worker]["result"]:
                if cache_item_before["key"] == key:
                    assert (
                        cache_item_before["creation"]
                        != cache_item_after["creation"]
                    ), f"{worker}:{key} cache not re-created"

Test Get Bgp Peerings Cache Refresh

Test cache content for BGP peerings

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_cache_refresh(self, nfclient):
    """Test cache content for BGP peerings"""
    # get cache brief info
    cache_before = nfclient.run_job(
        "netbox",
        "cache_list",
        workers=["netbox-worker-1.1"],
        kwargs={"keys": "get_bgp_peerings*", "details": True},
    )

    # cache data
    nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers=["netbox-worker-1.1"],
        kwargs={"devices": ["fceos4", "fceos5"], "cache": "refresh"},
    )

    # Now retrieve cache content
    cache_after = nfclient.run_job(
        "netbox",
        "cache_list",
        workers=["netbox-worker-1.1"],
        kwargs={"keys": "get_bgp_peerings*", "details": True},
    )

    print("cache_before:")
    pprint.pprint(cache_before, width=200)

    print("cache_after:")
    pprint.pprint(cache_after, width=200)

    for worker, res in cache_after.items():
        for cache_item_after in res["result"]:
            key = cache_item_after["key"]
            for cache_item_before in cache_before[worker]["result"]:
                if cache_item_before["key"] == key:
                    assert (
                        cache_item_before["creation"]
                        != cache_item_after["creation"]
                    ), f"{worker}:{key} cache not re-created"

Test Get Bgp Peerings Cache Force

Test cache force mode (use cache without checking)

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_cache_force(self, nfclient):
    """Test cache force mode (use cache without checking)"""
    # First, populate cache
    nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["fceos4"], "cache": True},
    )

    # Use cache="force" to retrieve from cache only
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["fceos4"], "cache": "force"},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["result"], f"{worker} returned no results"
        assert "fceos4" in res["result"], f"{worker} missing fceos4 in results"

Test Get Bgp Peerings Cache False

Test with cache disabled

Source code in tests\test_netbox_service.py

def test_get_bgp_peerings_cache_false(self, nfclient):
    """Test with cache disabled"""
    # Clear any existing cache
    nfclient.run_job(
        "netbox",
        "cache_clear",
        workers="all",
        kwargs={"keys": "get_bgp_peerings*"},
    )

    # Fetch with cache=False
    ret = nfclient.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={"devices": ["fceos4"], "cache": False},
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["result"], f"{worker} returned no results"
        assert "fceos4" in res["result"], f"{worker} missing fceos4 in results"

    # Verify nothing was cached
    bgp_cache = nfclient.run_job(
        "netbox",
        "cache_get",
        workers="all",
        kwargs={"keys": "get_bgp_peerings::fceos4"},
    )

    for worker, res in bgp_cache.items():
        # Cache should be empty or the key should not exist
        assert (
            "get_bgp_peerings::fceos4" not in res["result"]
        ), f"{worker} should not have cached data when cache=False"

TestSyncDeviceFacts

Device fact synchronization from external sources to Netbox.

Purpose: Validate pushing collected device facts (from Nornir) back to Netbox.

Test Sync Device Facts Basic Update

Test basic sync with devices list - updates serial numbers

Source code in tests\test_netbox_service.py

def test_sync_device_facts_basic_update(self, nfclient):
    """Test basic sync with devices list - updates serial numbers"""
    # Setup: update serial for spine to force a change
    pynb = get_pynetbox(nfclient)
    nb_device = pynb.dcim.devices.get(name="ceos-spine-1")
    nb_device.serial = "123456"
    nb_device.save()

    # Execute sync job
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1", "ceos-spine-2"],
        },
    )
    pprint.pprint(ret, width=200)

    # Verify results
    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed - {res.get('errors')}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "ceos-spine-2" in res["result"]
        ), f"{worker} returned no results for ceos-spine-2"

        # ceos-spine-1 should have been updated
        assert res["result"]["ceos-spine-1"][
            "sync_device_facts"
        ], "ceos-spine-1 no data returned"
        assert res["result"]["ceos-spine-1"]["sync_device_facts"][
            "serial"
        ], "ceos-spine-1 serial not synced"

        # ceos-spine-2 should be in sync or updated
        assert res["result"]["ceos-spine-2"][
            "sync_device_facts"
        ], "ceos-spine-2 no data returned"

Test Sync Device Facts Already In Sync

Test when device facts are already synchronized

Source code in tests\test_netbox_service.py

def test_sync_device_facts_already_in_sync(self, nfclient):
    """Test when device facts are already synchronized"""
    # First sync to ensure devices are up to date
    nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
        },
    )

    # Second sync should show devices in sync
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert (
            res["result"]["ceos-spine-1"]["sync_device_facts"]
            == "Device facts in sync"
        ), "Expected device to be in sync"

Test Sync Device Facts With Filters

Test sync using Nornir filters instead of device list

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_filters(self, nfclient):
    """Test sync using Nornir filters instead of device list"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "FC": "spine",
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed - {res.get('errors')}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "ceos-spine-2" in res["result"]
        ), f"{worker} returned no results for ceos-spine-2"

        for device, device_data in res["result"].items():
            assert (
                "sync_device_facts" in device_data
            ), f"{worker}:{device} no sync data"

Test Sync Device Facts Dry Run

Test dry run mode - should not modify NetBox

Source code in tests\test_netbox_service.py

def test_sync_device_facts_dry_run(self, nfclient):
    """Test dry run mode - should not modify NetBox"""
    # Setup: change serial to create a difference
    pynb = get_pynetbox(nfclient)
    nb_device = pynb.dcim.devices.get(name="ceos-spine-1")
    original_serial = nb_device.serial
    nb_device.serial = "DRY-RUN-TEST-123"
    nb_device.save()

    # Execute dry run
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1", "ceos-spine-2"],
            "dry_run": True,
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert res["dry_run"] is True, "dry_run flag not set in result"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "ceos-spine-2" in res["result"]
        ), f"{worker} returned no results for ceos-spine-2"

        for device, device_data in res["result"].items():
            assert (
                "sync_device_facts_dry_run" in device_data
            ), f"{worker}:{device} no dry run data"

    # Verify NetBox was not modified
    nb_device = pynb.dcim.devices.get(name="ceos-spine-1")
    assert (
        nb_device.serial == "DRY-RUN-TEST-123"
    ), "NetBox was modified during dry run"

    # Cleanup
    nb_device.serial = original_serial
    nb_device.save()

Test Sync Device Facts With Diff

Test that diff is properly populated when changes are made

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_diff(self, nfclient):
    """Test that diff is properly populated when changes are made"""
    # Setup: force a change
    pynb = get_pynetbox(nfclient)
    nb_device = pynb.dcim.devices.get(name="ceos-spine-1")
    nb_device.serial = "OLD-SERIAL-123"
    nb_device.save()

    # Execute sync
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        if (
            res["result"]["ceos-spine-1"]["sync_device_facts"]
            != "Device facts in sync"
        ):
            assert (
                "ceos-spine-1" in res["diff"]
            ), "diff not populated for changed device"
            assert (
                "serial" in res["diff"]["ceos-spine-1"]
            ), "serial field not in diff"
            assert (
                "-" in res["diff"]["ceos-spine-1"]["serial"]
            ), "old value (-) not in diff"
            assert (
                "+" in res["diff"]["ceos-spine-1"]["serial"]
            ), "new value (+) not in diff"

Test Sync Device Facts With Branch

Test sync with NetBox branching plugin

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_branch(self, nfclient):
    """Test sync with NetBox branching plugin"""
    delete_branch("sync_facts_test_branch", nfclient)

    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1", "ceos-spine-2"],
            "branch": "sync_facts_test_branch",
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed - {res.get('errors')}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "ceos-spine-2" in res["result"]
        ), f"{worker} returned no results for ceos-spine-2"

        for device, device_data in res["result"].items():
            assert "branch" in device_data, f"{worker}:{device} no branch info"
            assert (
                device_data["branch"] == "sync_facts_test_branch"
            ), "Wrong branch name"

    # Cleanup
    delete_branch("sync_facts_test_branch", nfclient)

Test Sync Device Facts With Custom Instance

Test sync with explicit NetBox instance

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_custom_instance(self, nfclient):
    """Test sync with explicit NetBox instance"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
            "instance": "prod",
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert "prod" in res["resources"], "instance not in resources"

Test Sync Device Facts With Batch Size

Test sync with custom batch size

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_batch_size(self, nfclient):
    """Test sync with custom batch size"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1", "ceos-spine-2"],
            "batch_size": 1,  # Process one device at a time
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert "ceos-spine-1" in res["result"], "ceos-spine-1 not processed"
        assert "ceos-spine-2" in res["result"], "ceos-spine-2 not processed"

Test Sync Device Facts With Timeout

Test sync with custom timeout

Source code in tests\test_netbox_service.py

def test_sync_device_facts_with_timeout(self, nfclient):
    """Test sync with custom timeout"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
            "timeout": 120,
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert "ceos-spine-1" in res["result"], "ceos-spine-1 not processed"

Test Sync Device Facts Non Existing Device

Test error handling when device doesn't exist in NetBox

Source code in tests\test_netbox_service.py

def test_sync_device_facts_non_existing_device(self, nfclient):
    """Test error handling when device doesn't exist in NetBox"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["nonexistent-device-12345"],
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        # Should fail or report error
        assert res["errors"], "Should report error for non-existent device"

Test Sync Device Facts Empty Device List

Test sync with empty device list and no filters

Source code in tests\test_netbox_service.py

def test_sync_device_facts_empty_device_list(self, nfclient):
    """Test sync with empty device list and no filters"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": [],
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        # Should complete without error, just process no devices
        assert not res["failed"], f"{worker} failed"
        assert (
            res["result"] == {} or len(res["result"]) == 0
        ), "Should have empty result"

Test Sync Device Facts Single Device

Test sync with a single device

Source code in tests\test_netbox_service.py

def test_sync_device_facts_single_device(self, nfclient):
    """Test sync with a single device"""
    ret = nfclient.run_job(
        "netbox",
        "sync_device_facts",
        workers="any",
        kwargs={
            "datasource": "nornir",
            "devices": ["ceos-spine-1"],
        },
    )
    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert not res["failed"], f"{worker} failed"
        assert "ceos-spine-1" in res["result"], "Device not in result"
        assert len(res["result"]) == 1, "Should only process one device"

TestSyncDeviceInterfaces

Interface status and configuration synchronization.

Purpose: Validate pushing interface data from network devices to Netbox.

Test Sync Device Interfaces

Clean TEST_SYNC interfaces from spines then sync. All TEST_SYNC interfaces must be created from live data; result must carry live-run keys.

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces(self, nfclient):
    """Clean TEST_SYNC interfaces from spines then sync. All TEST_SYNC interfaces
    must be created from live data; result must carry live-run keys."""
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(nfclient, self.SPINE_DEVICES)
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no results for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.LIVE_RUN_KEYS <= device_data.keys()
            ), f"{worker}:{device} missing keys in result, got: {set(device_data)}"
            assert device_data[
                "created"
            ], f"{worker}:{device} no interfaces created after cleanup"

Test Sync Device Interfaces Dry Run

Clean TEST_SYNC from spines then dry_run sync. The plan must list cleaned interfaces under 'create' and expose correct key/type structure.

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces_dry_run(self, nfclient):
    """Clean TEST_SYNC from spines then dry_run sync. The plan must list
    cleaned interfaces under 'create' and expose correct key/type structure."""
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(nfclient, self.SPINE_DEVICES, dry_run=True)
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no results for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.DRY_RUN_KEYS <= device_data.keys()
            ), f"{worker}:{device} dry-run result missing keys, got: {set(device_data)}"
            assert isinstance(
                device_data["create"], list
            ), f"{worker}:{device} create not a list"
            assert isinstance(
                device_data["delete"], list
            ), f"{worker}:{device} delete not a list"
            assert isinstance(
                device_data["update"], dict
            ), f"{worker}:{device} update not a dict"
            assert isinstance(
                device_data["in_sync"], list
            ), f"{worker}:{device} in_sync not a list"
            assert device_data[
                "create"
            ], f"{worker}:{device} create list is empty after cleanup"

Test Sync Device Interfaces Create

Clean TEST_SYNC from spine-2 then verify sync creates Loopback10 (TEST_SYNC_LOOPBACK_IPV4) and Loopback11 (TEST_SYNC_LOOPBACK_IPV6) from live data.

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces_create(self, nfclient):
    """Clean TEST_SYNC from spine-2 then verify sync creates Loopback10
    (TEST_SYNC_LOOPBACK_IPV4) and Loopback11 (TEST_SYNC_LOOPBACK_IPV6) from live data.
    """
    self._cleanup(nfclient, ["ceos-spine-2"])

    ret = self._sync(nfclient, ["ceos-spine-2"])
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        device_data = res["result"]["ceos-spine-2"]
        assert (
            "Loopback10" in device_data["created"]
        ), f"{worker} Loopback10 not in created list"
        assert (
            "Loopback11" in device_data["created"]
        ), f"{worker} Loopback11 not in created list"

    # Validate Loopback10 record in NetBox
    nb_lb10 = self._get_nb_intf(nfclient, "ceos-spine-2", "Loopback10")
    assert nb_lb10 is not None, "Loopback10 not found in NetBox after sync"
    assert (
        nb_lb10.description == "TEST_SYNC_LOOPBACK_IPV4"
    ), f"Loopback10 description mismatch: got {nb_lb10.description!r}"
    assert (
        nb_lb10.type.value == "virtual"
    ), f"Loopback10 type mismatch: got {nb_lb10.type.value!r}"
    # Validate Loopback11 record in NetBox
    nb_lb11 = self._get_nb_intf(nfclient, "ceos-spine-2", "Loopback11")
    assert nb_lb11 is not None, "Loopback11 not found in NetBox after sync"
    assert (
        nb_lb11.description == "TEST_SYNC_LOOPBACK_IPV6"
    ), f"Loopback11 description mismatch: got {nb_lb11.description!r}"
    assert (
        nb_lb11.type.value == "virtual"
    ), f"Loopback11 type mismatch: got {nb_lb11.type.value!r}"

Test Sync Device Interfaces Update Description

Clean TEST_SYNC for spine-2, run sync to create Loopback10 with the correct description, then corrupt it and verify sync restores it. Field-level diff must be present in res['diff'].

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces_update_description(self, nfclient):
    """Clean TEST_SYNC for spine-2, run sync to create Loopback10 with the
    correct description, then corrupt it and verify sync restores it.
    Field-level diff must be present in res['diff']."""
    self._cleanup(nfclient, ["ceos-spine-2"])

    # Create correct NB state from live data
    setup = self._sync(nfclient, ["ceos-spine-2"])
    for worker, res in setup.items():
        assert not res["failed"], f"Setup sync failed for {worker}: {res['errors']}"
        assert (
            "Loopback10" in res["result"]["ceos-spine-2"]["created"]
        ), f"{worker} Loopback10 not created during setup"

    # Corrupt description
    intf_id = self._get_intf_id(nfclient, "ceos-spine-2", "Loopback10")
    self._patch_intf(nfclient, intf_id, {"description": "corrupted-by-test"})

    ret = self._sync(nfclient, ["ceos-spine-2"])
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        device_data = res["result"]["ceos-spine-2"]
        assert (
            "Loopback10" in device_data["updated"]
        ), f"{worker} Loopback10 not in updated list after description corruption"
        assert "ceos-spine-2" in res["diff"], f"{worker} diff not populated"
        assert (
            "Loopback10" in res["diff"]["ceos-spine-2"]["update"]
        ), f"{worker} Loopback10 field diff missing"
        assert (
            "description" in res["diff"]["ceos-spine-2"]["update"]["Loopback10"]
        ), f"{worker} description field missing from diff"

    # Validate description restored in NetBox
    nb_lb10 = self._get_nb_intf(nfclient, "ceos-spine-2", "Loopback10")
    assert nb_lb10 is not None, "Loopback10 not found in NetBox after sync"
    assert (
        nb_lb10.description == "TEST_SYNC_LOOPBACK_IPV4"
    ), f"Loopback10 description not restored in NetBox: got {nb_lb10.description!r}"

Test Sync Device Interfaces Non Existing Device

Sync of a device not in NetBox should report an error.

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces_non_existing_device(self, nfclient):
    """Sync of a device not in NetBox should report an error."""
    ret = self._sync(nfclient, ["nonexistent-device-12345"])
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert (
            len(res["errors"]) > 0
        ), f"{worker} should have errors for nonexistent device"

Test Sync Device Interfaces With Branch

Clean TEST_SYNC from spines, delete the branch, then sync into a new branch. Result must carry live-run keys and at least one interface must be created.

Source code in tests\test_netbox_service.py

def test_sync_device_interfaces_with_branch(self, nfclient):
    """Clean TEST_SYNC from spines, delete the branch, then sync into a new branch.
    Result must carry live-run keys and at least one interface must be created."""
    delete_branch("update_interfaces_branch_1", nfclient)
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(
        nfclient,
        self.SPINE_DEVICES,
        branch="update_interfaces_branch_1",
    )
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no results for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.LIVE_RUN_KEYS <= device_data.keys()
            ), f"{worker}:{device} missing keys in branch-run result"

TestCreateDeviceInterfaces

Device interface creation in Netbox.

Purpose: Validate programmatic interface creation and configuration.

Test Create Device Interfaces Single

Test creating a single interface on a device

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_single(self, nfclient):
    """Test creating a single interface on a device"""
    delete_interfaces(nfclient, "ceos-spine-1", "TestInterface1")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "TestInterface1",
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "TestInterface1" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create TestInterface1"

Test Create Device Interfaces Multiple Devices

Test creating interfaces on multiple devices

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_multiple_devices(self, nfclient):
    """Test creating interfaces on multiple devices"""
    delete_interfaces(nfclient, "ceos-spine-1", "TestInterface2")
    delete_interfaces(nfclient, "ceos-spine-2", "TestInterface2")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1", "ceos-spine-2"],
            "interface_name": "TestInterface2",
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "ceos-spine-2" in res["result"]
        ), f"{worker} returned no results for ceos-spine-2"
        assert (
            "TestInterface2" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create TestInterface2 on ceos-spine-1"
        assert (
            "TestInterface2" in res["result"]["ceos-spine-2"]["created"]
        ), f"{worker} did not create TestInterface2 on ceos-spine-2"

Test Create Device Interfaces With Range Numeric

Test creating interfaces with numeric range expansion

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_with_range_numeric(self, nfclient):
    """Test creating interfaces with numeric range expansion"""
    for i in range(1, 4):
        delete_interfaces(nfclient, "ceos-spine-1", f"Loopback{i}")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "Loopback[1-3]",
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            len(res["result"]["ceos-spine-1"]["created"]) == 3
        ), f"{worker} did not create 3 interfaces"
        assert (
            "Loopback1" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create Loopback1"
        assert (
            "Loopback2" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create Loopback2"
        assert (
            "Loopback3" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create Loopback3"

Test Create Device Interfaces With Range List

Test creating interfaces with comma-separated list expansion

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_with_range_list(self, nfclient):
    """Test creating interfaces with comma-separated list expansion"""
    delete_interfaces(nfclient, "ceos-spine-1", "ge-0/0/0")
    delete_interfaces(nfclient, "ceos-spine-1", "xe-0/0/0")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "[ge,xe]-0/0/0",
            "interface_type": "1000base-t",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            len(res["result"]["ceos-spine-1"]["created"]) == 2
        ), f"{worker} did not create 2 interfaces"
        assert (
            "ge-0/0/0" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create ge-0/0/0"
        assert (
            "xe-0/0/0" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create xe-0/0/0"

Test Create Device Interfaces With Multiple Ranges

Test creating interfaces with multiple range patterns

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_with_multiple_ranges(self, nfclient):
    """Test creating interfaces with multiple range patterns"""
    for prefix in ["ge", "xe"]:
        for i in range(0, 2):
            delete_interfaces(nfclient, "ceos-spine-1", f"{prefix}-0/0/{i}")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "[ge,xe]-0/0/[0-1]",
            "interface_type": "1000base-t",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            len(res["result"]["ceos-spine-1"]["created"]) == 4
        ), f"{worker} did not create 4 interfaces"
        assert "ge-0/0/0" in res["result"]["ceos-spine-1"]["created"]
        assert "ge-0/0/1" in res["result"]["ceos-spine-1"]["created"]
        assert "xe-0/0/0" in res["result"]["ceos-spine-1"]["created"]
        assert "xe-0/0/1" in res["result"]["ceos-spine-1"]["created"]

Test Create Device Interfaces Multiple Names List

Test creating multiple interfaces passed as a list

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_multiple_names_list(self, nfclient):
    """Test creating multiple interfaces passed as a list"""
    delete_interfaces(nfclient, "ceos-spine-1", "TestIntf1")
    delete_interfaces(nfclient, "ceos-spine-1", "TestIntf2")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": ["TestIntf1", "TestIntf2"],
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            len(res["result"]["ceos-spine-1"]["created"]) == 2
        ), f"{worker} did not create 2 interfaces"
        assert "TestIntf1" in res["result"]["ceos-spine-1"]["created"]
        assert "TestIntf2" in res["result"]["ceos-spine-1"]["created"]

Test Create Device Interfaces Skip Existing

Test that existing interfaces are skipped

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_skip_existing(self, nfclient):
    """Test that existing interfaces are skipped"""
    # First create the interface
    ret1 = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "TestExisting",
            "interface_type": "virtual",
        },
    )

    # Try to create it again
    ret2 = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "TestExisting",
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret2)
    for worker, res in ret2.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "TestExisting" in res["result"]["ceos-spine-1"]["skipped"]
        ), f"{worker} did not skip existing TestExisting interface"
        assert (
            len(res["result"]["ceos-spine-1"]["created"]) == 0
        ), f"{worker} should not have created any interfaces"

    # Cleanup
    delete_interfaces(nfclient, "ceos-spine-1", "TestExisting")

Test Create Device Interfaces Dry Run

Test dry run mode

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_dry_run(self, nfclient):
    """Test dry run mode"""
    delete_interfaces(nfclient, "ceos-spine-1", "TestDryRun")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "TestDryRun",
            "interface_type": "virtual",
            "dry_run": True,
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "TestDryRun" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not mark TestDryRun for creation in dry run"

    # Verify interface was not actually created
    resp_get = nfclient.run_job(
        "netbox",
        "rest",
        workers="any",
        kwargs={
            "method": "get",
            "api": "/dcim/interfaces/",
            "params": {"device": "ceos-spine-1", "name": "TestDryRun"},
        },
    )
    worker, interfaces = tuple(resp_get.items())[0]
    assert (
        len(interfaces["result"]["results"]) == 0
    ), "Interface should not exist after dry run"

Test Create Device Interfaces With Branch

Test creating interfaces with a branch

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_with_branch(self, nfclient):
    """Test creating interfaces with a branch"""
    delete_branch("create_interfaces_branch_1", nfclient)
    delete_interfaces(nfclient, "ceos-spine-1", "TestBranch")

    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["ceos-spine-1"],
            "interface_name": "TestBranch",
            "interface_type": "virtual",
            "branch": "create_interfaces_branch_1",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        assert (
            "ceos-spine-1" in res["result"]
        ), f"{worker} returned no results for ceos-spine-1"
        assert (
            "TestBranch" in res["result"]["ceos-spine-1"]["created"]
        ), f"{worker} did not create TestBranch"

    # Cleanup
    delete_interfaces(nfclient, "ceos-spine-1", "TestBranch")
    delete_branch("create_interfaces_branch_1", nfclient)

Test Create Device Interfaces Non Existing Device

Test handling of non-existing device

Source code in tests\test_netbox_service.py

def test_create_device_interfaces_non_existing_device(self, nfclient):
    """Test handling of non-existing device"""
    ret = nfclient.run_job(
        "netbox",
        "create_device_interfaces",
        workers="any",
        kwargs={
            "devices": ["nonexistent-device-12345"],
            "interface_name": "TestInterface",
            "interface_type": "virtual",
        },
    )

    pprint.pprint(ret)
    for worker, res in ret.items():
        assert len(res["errors"]) > 0, f"{worker} should have errors"

TestSyncDeviceIP

IP address and interface IP synchronization.

Purpose: Validate IP address assignment and synchronization to device interfaces.

Test Sync Device Ip

Clean IPs from both spines then sync. Both spine IPs must be created; result must carry the correct RESULT_KEYS per device.

Source code in tests\test_netbox_service.py

def test_sync_device_ip(self, nfclient):
    """Clean IPs from both spines then sync. Both spine IPs must be created;
    result must carry the correct RESULT_KEYS per device."""
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(nfclient, self.SPINE_DEVICES)
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no result for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.RESULT_KEYS <= device_data.keys()
            ), f"{worker}:{device} missing keys in result, got: {set(device_data)}"
            assert device_data[
                "created"
            ], f"{worker}:{device} no IPs created after cleanup"

    # Validate both spine IPs exist in NetBox assigned to the correct interface
    pynb = get_pynetbox(nfclient)
    nb_spine1 = list(
        pynb.ipam.ip_addresses.filter(address=self.SPINE1_IP, device="ceos-spine-1")
    )
    assert nb_spine1, f"{self.SPINE1_IP} not found in NetBox for ceos-spine-1"
    assert (
        nb_spine1[0].assigned_object is not None
    ), f"{self.SPINE1_IP} not assigned to any interface on ceos-spine-1"
    assert (
        nb_spine1[0].assigned_object.name == self.SPINE1_INTF
    ), f"{self.SPINE1_IP} assigned to {nb_spine1[0].assigned_object.name!r}, expected {self.SPINE1_INTF!r}"
    nb_spine2 = list(
        pynb.ipam.ip_addresses.filter(address=self.SPINE2_IP, device="ceos-spine-2")
    )
    assert nb_spine2, f"{self.SPINE2_IP} not found in NetBox for ceos-spine-2"
    assert (
        nb_spine2[0].assigned_object is not None
    ), f"{self.SPINE2_IP} not assigned to any interface on ceos-spine-2"

Test Sync Device Ip Dry Run

Clean IPs from both spines then dry_run. Result keys must be RESULT_KEYS and 'created' must be non-empty (no actual NB writes).

Source code in tests\test_netbox_service.py

def test_sync_device_ip_dry_run(self, nfclient):
    """Clean IPs from both spines then dry_run. Result keys must be RESULT_KEYS
    and 'created' must be non-empty (no actual NB writes)."""
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(nfclient, self.SPINE_DEVICES, dry_run=True)
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no result for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.RESULT_KEYS <= device_data.keys()
            ), f"{worker}:{device} dry-run result missing keys, got: {set(device_data)}"
            assert device_data[
                "created"
            ], f"{worker}:{device} dry-run created list is empty after cleanup"

    # Verify dry-run made no writes — IPs must still be absent from NetBox
    pynb = get_pynetbox(nfclient)
    ips_in_nb = list(
        pynb.ipam.ip_addresses.filter(address=self.SPINE1_IP, device="ceos-spine-1")
    )
    assert (
        not ips_in_nb
    ), f"dry-run wrote IP {self.SPINE1_IP} to NetBox: {[str(i) for i in ips_in_nb]}"

Test Sync Device Ip With Branch

Clean spine IPs, delete the test branch, then sync into a new branch. Result must carry RESULT_KEYS and at least one IP must be created.

Source code in tests\test_netbox_service.py

def test_sync_device_ip_with_branch(self, nfclient):
    """Clean spine IPs, delete the test branch, then sync into a new branch.
    Result must carry RESULT_KEYS and at least one IP must be created."""
    branch = "sync_device_ip_branch_1"
    delete_branch(branch, nfclient)
    self._cleanup(nfclient, self.SPINE_DEVICES)

    ret = self._sync(nfclient, self.SPINE_DEVICES, branch=branch)
    pprint.pprint(ret)
    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} failed - {res}"
        for device in self.SPINE_DEVICES:
            assert (
                device in res["result"]
            ), f"{worker} returned no result for {device}"
        for device, device_data in res["result"].items():
            assert (
                self.RESULT_KEYS <= device_data.keys()
            ), f"{worker}:{device} missing keys in branch-run result"

    # Validate IPs were NOT written to the main NetBox context (branch-only writes)
    pynb = get_pynetbox(nfclient)
    nb_ips = list(
        pynb.ipam.ip_addresses.filter(address=self.SPINE1_IP, device="ceos-spine-1")
    )
    assert not nb_ips, (
        f"Branch sync wrote {self.SPINE1_IP} to main NetBox context (should be branch-only): "
        f"{[str(i) for i in nb_ips]}"
    )

    delete_branch(branch, nfclient)

TestCreateIP

IP address creation in Netbox.

Purpose: Validate IP address allocation and creation.

Test Create Ip By Prefix

Source code in tests\test_netbox_service.py

def test_create_ip_by_prefix(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": f"test create ip {rand}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": f"test create ip {rand}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"]["address"], f"Result has no ip {res1['result']}"

        worker, res2 = tuple(create_2.items())[0]
        assert (
            res1["result"]["address"] == res2["result"]["address"]
        ), "Should have been same IP address"
        assert (
            res1["result"]["description"] == res2["result"]["description"]
        ), "Should have been same IP description"

Test Create Ip By Prefix Description

Source code in tests\test_netbox_service.py

def test_create_ip_by_prefix_description(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "TEST NEXT IP PREFIX",
                "description": f"test create ip {rand}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "TEST NEXT IP PREFIX",
                "description": f"test create ip {rand}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"]["address"], f"Result has no ip {res1['result']}"

        worker, res2 = tuple(create_2.items())[0]
        assert (
            res1["result"]["address"] == res2["result"]["address"]
        ), "Should have been same IP address"
        assert (
            res1["result"]["description"] == res2["result"]["description"]
        ), "Should have been same IP description"

Test Create Ip By Prefix Multiple

Source code in tests\test_netbox_service.py

def test_create_ip_by_prefix_multiple(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": f"test create ip {random.randint(1, 1000)}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": f"test create ip {random.randint(1, 1000)}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res2["failed"] == False, "Allocation failed"

        assert res1["result"]["address"] != res2["result"]["address"]
        assert res1["result"]["description"] != res2["result"]["description"]

Test Create Ip Nonexist Prefix

Source code in tests\test_netbox_service.py

def test_create_ip_nonexist_prefix(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "1.2.3.0/24",
                "description": f"test create ip {random.randint(1, 1000)}",
            },
        )
        pprint.pprint(create_1, width=200)
        worker, res1 = tuple(create_1.items())[0]
        assert res1["failed"] == True, "Allocation not failed"
        assert "Unable to source parent prefix from Netbox" in res1["messages"][0]

Test Create Ip By Prefix Device Interface

Source code in tests\test_netbox_service.py

def test_create_ip_by_prefix_device_interface(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res2["failed"] == False, "Allocation failed"

        assert (
            res1["result"]["address"] == res2["result"]["address"]
        ), "Should be same IP"

Test Create Ip By Prefix Description Device Interface

This test should allocate two different IPs since device, interface, description given

Source code in tests\test_netbox_service.py

def test_create_ip_by_prefix_description_device_interface(self, nfclient):
    "This test should allocate two different IPs since device, interface, description given"
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
                "description": "foo1",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
                "description": "foo2",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res2["failed"] == False, "Allocation failed"

        assert (
            res1["result"]["address"] != res2["result"]["address"]
        ), "Should be different IP cause description is different"

Test Create Ip With Vrf Tags Tenant Role Dnsname Comments

Source code in tests\test_netbox_service.py

def test_create_ip_with_vrf_tags_tenant_role_dnsname_comments(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
                "description": "foo1",
                "vrf": "VRF1",
                "tags": ["NORFAB", "ACCESS"],
                "tenant": "NORFAB",
                "dns_name": "foo1.lab.local",
                "role": "anycast",
                "comments": "Some comments",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"

Test Create Ip Non Existing Device

Source code in tests\test_netbox_service.py

def test_create_ip_non_existing_device(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "does_not_exist",
                "interface": "eth1",
                "description": "foo1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == True, "Allocation should have failed"

Test Create Ip Non Existing Interface

Source code in tests\test_netbox_service.py

def test_create_ip_non_existing_interface(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fce0s4",
                "interface": "does_not_exist",
                "description": "foo1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == True, "Allocation should have failed"

Test Create Ip Is Primary

Source code in tests\test_netbox_service.py

def test_create_ip_is_primary(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "ceos-spine-1",
                "interface": "Ethernet1",
                "description": "foo1",
                "is_primary": True,
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"

Test Create Ip Dry Run New Ip

Source code in tests\test_netbox_service.py

def test_create_ip_dry_run_new_ip(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        # test dry run for new ip
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "ceos-spine-1",
                "interface": "Ethernet1",
                "dry_run": True,
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={"prefix": "10.0.0.0/24", "description": "foo", "dry_run": True},
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"
        assert res1["dry_run"] is True, "No dry run flag set to true"
        assert res1["status"] == "unchanged", "Unexpected status"

        assert res2["failed"] == False, "Allocation failed"
        assert res2["result"]["address"], "No ip allocated"
        assert res2["dry_run"] is True, "No dry run flag set to true"
        assert res2["status"] == "unchanged", "Unexpected status"

Test Create Ip Dry Run Existing Ip

Source code in tests\test_netbox_service.py

def test_create_ip_dry_run_existing_ip(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": "foobar",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": "foobar",
                "role": "anycast",
                "dry_run": True,
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"
        assert res1["dry_run"] is False, "No dry run flag set to true"
        assert res1["status"] == "created", "Unexpected status"

        assert res2["failed"] == False, "Allocation failed"
        assert res2["result"]["address"], "No ip allocated"
        assert res2["dry_run"] is True, "No dry run flag set to true"
        assert res2["status"] == "unchanged", "Unexpected status"

Test Create Ip With Nb Instance

Source code in tests\test_netbox_service.py

def test_create_ip_with_nb_instance(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "description": "foobar",
                "instance": "dev",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"
        assert res1["dry_run"] is False, "No dry run flag set to true"
        assert res1["status"] == "created", "Unexpected status"
        assert "dev" in res1["resources"]

Test Create Ip With Branch

Source code in tests\test_netbox_service.py

def test_create_ip_with_branch(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)
    delete_branch("create_ip_1", nfclient)
    delete_ips("10.0.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_ip",
            workers="any",
            kwargs={
                "prefix": "10.0.0.0/24",
                "device": "fceos4",
                "interface": "eth1",
                "description": "foo1",
                "vrf": "VRF1",
                "tags": ["NORFAB", "ACCESS"],
                "tenant": "NORFAB",
                "dns_name": "foo1.lab.local",
                "role": "anycast",
                "comments": "Some comments",
                "branch": "create_ip_1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        worker, res1 = tuple(create_1.items())[0]

        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"]["address"], "No ip allocated"
        assert (
            res1["result"]["branch"] == "create_ip_1"
        ), "No branch info in results"

Test Create Ip With Mask Len

Source code in tests\test_netbox_service.py

def test_create_ip_with_mask_len(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1",
            "mask_len": 31,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)
    worker, res1 = tuple(create_1.items())[0]
    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"

Test Create Ip With Mask Len Dry Run

Source code in tests\test_netbox_service.py

def test_create_ip_with_mask_len_dry_run(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1",
            "mask_len": 31,
            "dry_run": True,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)
    worker, res1 = tuple(create_1.items())[0]
    # dry run will allocate first ip within /24 as opposed to /31
    assert res1["result"]["address"] == "10.0.0.1/24", "Wrong ip allocated"

Test Create Ip Check Create Peer Ip

Source code in tests\test_netbox_service.py

def test_create_ip_check_create_peer_ip(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
            "mask_len": 31,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)

    worker, res1 = tuple(create_1.items())[0]

    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"
    assert (
        res1["result"]["peer"]["address"] == "10.0.0.1/31"
    ), "Wrong ip allocated for peer"
    assert (
        res1["result"]["peer"]["device"] == "fceos5"
    ), "Wrong ip allocated for peer"
    assert (
        res1["result"]["peer"]["interface"] == "ae5.101"
    ), "Wrong ip allocated for peer"

Test Create Ip Check Create Peer Ip With Branch

Source code in tests\test_netbox_service.py

def test_create_ip_check_create_peer_ip_with_branch(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    delete_branch("create_ip_with_peer", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
            "mask_len": 31,
            "branch": "create_ip_with_peer",
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)

    worker, res1 = tuple(create_1.items())[0]

    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"
    assert res1["result"]["branch"] == "create_ip_with_peer", "Wrong branch"
    assert (
        res1["result"]["peer"]["address"] == "10.0.0.1/31"
    ), "Wrong ip allocated for peer"
    assert (
        res1["result"]["peer"]["device"] == "fceos5"
    ), "Wrong ip allocated for peer"
    assert (
        res1["result"]["peer"]["interface"] == "ae5.101"
    ), "Wrong ip allocated for peer"

Test Create Ip Check Skip Create Peer Ip

Source code in tests\test_netbox_service.py

def test_create_ip_check_skip_create_peer_ip(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
            "mask_len": 31,
            "create_peer_ip": False,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)

    worker, res1 = tuple(create_1.items())[0]

    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"
    assert (
        "peer" not in res1["result"]
    ), "SHould have been skipping peer ip creation"

Test Create Ip Use Peer Ip

Source code in tests\test_netbox_service.py

def test_create_ip_use_peer_ip(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
            "mask_len": 31,
            "create_peer_ip": False,
        },
    )
    create_2 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos5",
            "interface": "ae5.101",
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)
    print("create_2")
    pprint.pprint(create_2, width=200)

    worker, res1 = tuple(create_1.items())[0]
    worker, res2 = tuple(create_2.items())[0]

    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"
    assert res2["result"]["address"] == "10.0.0.1/31", "Wrong ip allocated"

Test Create Ip With Link Peer Dry Run

Source code in tests\test_netbox_service.py

def test_create_ip_with_link_peer_dry_run(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
            "mask_len": 31,
        },
    )
    create_2 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos5",
            "interface": "ae5.101",
            "dry_run": True,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)
    print("create_2")
    pprint.pprint(create_2, width=200)

    worker, res1 = tuple(create_1.items())[0]
    worker, res2 = tuple(create_2.items())[0]

    assert res1["result"]["address"] == "10.0.0.0/31", "Wrong ip allocated"
    assert res2["result"]["address"] == "10.0.0.1/31", "Wrong ip allocated"

Test Create Ip With Link Peer Within Parent

Source code in tests\test_netbox_service.py

def test_create_ip_with_link_peer_within_parent(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos4",
            "interface": "Port-Channel1.101",
        },
    )
    create_2 = nfclient.run_job(
        "netbox",
        "create_ip",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "device": "fceos5",
            "interface": "ae5.101",
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)
    print("create_2")
    pprint.pprint(create_2, width=200)

    worker, res1 = tuple(create_1.items())[0]
    worker, res2 = tuple(create_2.items())[0]

    assert res1["result"]["address"] == "10.0.0.1/24", "Wrong ip allocated"
    assert res2["result"]["address"] == "10.0.0.2/24", "Wrong ip allocated"

TestNetboxCache

Caching functionality and cache management.

Purpose: Validate cache operations for improved query performance.

Test Cache List

Source code in tests\test_netbox_service.py

def test_cache_list(self, nfclient):
    # populate the cache
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is present
    ret = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={},
    )

    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} - cache operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache list result is not a list"
        assert "get_devices::fceos5" in res["result"]
        assert "get_devices::ceos1" in res["result"]
        assert "get_devices::fceos4" in res["result"]

Test Cache List Details

Source code in tests\test_netbox_service.py

def test_cache_list_details(self, nfclient):
    # populate the cache
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is present
    ret = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"details": True},
    )

    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} - cache operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache list result is not a list"
        for item in res["result"]:
            assert all(
                key in item for key in ["age", "creation", "expires", "key"]
            ), f"{worker} - not all cache list data details returned"

Test Cache List Filter

Source code in tests\test_netbox_service.py

def test_cache_list_filter(self, nfclient):
    # populate the cache
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is present
    ret = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*ceos1*"},
    )

    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} - cache operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache list result is not a list"
        for key in res["result"]:
            assert (
                "ceos1" in key
            ), f"{worker} - key '{key}' does not contain 'ceos1' pattern "

Test Cache Clear All

Source code in tests\test_netbox_service.py

def test_cache_clear_all(self, nfclient):
    # populate the cache
    ret_populate = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # clear cache
    ret_clear = nfclient.run_job(
        "netbox",
        "cache_clear",
        workers="all",
        kwargs={"keys": "*"},
    )

    # list cache
    ret_list = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*"},
    )

    print("\nret_populate:")
    pprint.pprint(ret_populate, width=150)

    print("\nret_clear:")
    pprint.pprint(ret_clear, width=150)

    print("\nret_list:")
    pprint.pprint(ret_list, width=150)

    for worker, res in ret_populate.items():
        assert (
            res["failed"] == False
        ), f"{worker} - get_devices populate operation failed"

    for worker, res in ret_clear.items():
        assert res["failed"] == False, f"{worker} - cache clear operation failed"
        assert (
            len(res["result"]) > 0
        ), f"{worker} - did not return list of cleared keys"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache clear result is not a list"

    for worker, res in ret_list.items():
        assert res["failed"] == False, f"{worker} - cache list operation failed"
        assert len(res["result"]) == 0, f"{worker} - cache is not empty"

Test Cache Clear Key

Source code in tests\test_netbox_service.py

def test_cache_clear_key(self, nfclient):
    # populate the cache
    ret_populate = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # list cache
    ret_list_before = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*"},
    )

    # clear cache
    ret_clear = nfclient.run_job(
        "netbox",
        "cache_clear",
        workers="all",
        kwargs={"key": "get_devices::ceos1"},
    )

    # list cache
    ret_list_after = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*"},
    )

    print("\nret_populate:")
    pprint.pprint(ret_populate, width=150)

    print("\nret_list_before:")
    pprint.pprint(ret_list_before, width=150)

    print("\nret_clear:")
    pprint.pprint(ret_clear, width=150)

    print("\nret_list_after:")
    pprint.pprint(ret_list_after, width=150)

    for worker, res in ret_populate.items():
        assert (
            res["failed"] == False
        ), f"{worker} - get_devices populate operation failed"

    for worker, res in ret_list_before.items():
        assert res["failed"] == False, f"{worker} - cache list operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert (
            "get_devices::ceos1" in res["result"]
        ), f"{worker} - cache does not have get_devices::ceos1 key"

    for worker, res in ret_clear.items():
        assert res["failed"] == False, f"{worker} - cache clear operation failed"
        assert (
            len(res["result"]) > 0
        ), f"{worker} - did not return list of cleared keys"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache clear result is not a list"

    for worker, res in ret_list_after.items():
        assert res["failed"] == False, f"{worker} - cache list operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert (
            "get_devices::ceos1" not in res["result"]
        ), f"{worker} - cache still has get_devices::ceos1 key"

Test Cache Get Key

Source code in tests\test_netbox_service.py

def test_cache_get_key(self, nfclient):
    # populate the cache
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is present
    ret = nfclient.run_job(
        "netbox",
        "cache_get",
        workers="all",
        kwargs={"key": "get_devices::ceos1"},
    )

    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} - cache operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert isinstance(
            res["result"], dict
        ), f"{worker} - cache get result is not a dict"
        assert res["result"][
            "get_devices::ceos1"
        ], f"{worker} - cache get result key data is empty"

Test Cache Get Keys

Source code in tests\test_netbox_service.py

def test_cache_get_keys(self, nfclient):
    # populate the cache
    ret = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is present
    ret = nfclient.run_job(
        "netbox",
        "cache_get",
        workers="all",
        kwargs={"keys": "*ceos1*"},
    )

    pprint.pprint(ret, width=200)

    for worker, res in ret.items():
        assert res["failed"] == False, f"{worker} - cache operation failed"
        assert len(res["result"]) > 0, f"{worker} - cache is empty"
        assert isinstance(
            res["result"], dict
        ), f"{worker} - cache get result is not a dict"
        for key in res["result"].keys():
            assert (
                "ceos1" in key
            ), f"{worker} - cache key '{key}' does not contain ceos1 pattern"

Test Cache False

Source code in tests\test_netbox_service.py

def test_cache_false(self, nfclient):
    # clear cache
    ret_clear = nfclient.run_job(
        "netbox",
        "cache_clear",
        workers="all",
        kwargs={"keys": "*"},
    )

    # query data with cache set to False
    ret_query = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": False, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # verify cache is empty
    ret_list = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*"},
    )

    print("\nret_clear:")
    pprint.pprint(ret_clear, width=150)

    print("\nret_query:")
    pprint.pprint(ret_query, width=150)

    print("\nret_list:")
    pprint.pprint(ret_list, width=150)

    for worker, res in ret_clear.items():
        assert res["failed"] == False, f"{worker} - cache clear operation failed"

    for worker, res in ret_query.items():
        assert res["failed"] == False, f"{worker} - query netbox operation failed"

    for worker, res in ret_list.items():
        assert res["failed"] == False, f"{worker} - cache list operation failed"
        assert isinstance(
            res["result"], list
        ), f"{worker} - cache get result is not a dict"
        assert len(res["result"]) == 0, f"{worker} - cache is not empty"

Test Cache Refresh

Source code in tests\test_netbox_service.py

def test_cache_refresh(self, nfclient):
    # clear cache
    ret_clear = nfclient.run_job(
        "netbox",
        "cache_clear",
        workers="all",
        kwargs={"keys": "*"},
    )

    # query data with cache set to True
    ret_query_true = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": True, "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # get cache creation time
    ret_list_1st = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*", "details": True},
    )

    # query data with cache set to refresh
    ret_query_refresh = nfclient.run_job(
        "netbox",
        "get_devices",
        workers="all",
        kwargs={"cache": "refresh", "devices": ["ceos1", "fceos4", "fceos5"]},
    )

    # get cache creation time
    ret_list_2nd = nfclient.run_job(
        "netbox",
        "cache_list",
        workers="all",
        kwargs={"keys": "*", "details": True},
    )

    print("\nret_clear:")
    pprint.pprint(ret_clear, width=150)

    print("\nret_query_true:")
    pprint.pprint(ret_query_true, width=150)

    print("\nret_list_1st:")
    pprint.pprint(ret_list_1st, width=150)

    print("\nret_query_refresh:")
    pprint.pprint(ret_query_refresh, width=150)

    print("\nret_list_2nd:")
    pprint.pprint(ret_list_2nd, width=150)

    # verify no errors
    for worker, res in ret_clear.items():
        assert res["failed"] == False, f"{worker} - cache clear operation failed"

    for worker, res in ret_query_true.items():
        assert res["failed"] == False, f"{worker} - query netbox operation failed"

    for worker, res in ret_list_1st.items():
        assert res["failed"] == False, f"{worker} - ret_list_1st operation failed"

    for worker, res in ret_query_refresh.items():
        assert (
            res["failed"] == False
        ), f"{worker} - ret_query_refresh operation failed"

    for worker, res in ret_list_2nd.items():
        assert res["failed"] == False, f"{worker} - ret_list_2nd operation failed"

    # compare 2nd list items expiration time is after the 1st one
    for worker_2nd, res_2nd in ret_list_2nd.items():
        for item_2nd in res_2nd["result"]:
            for item_1st in ret_list_1st[worker_2nd]["result"]:
                if item_2nd["key"] == item_1st["key"]:
                    assert item_2nd["expires"] > item_1st["expires"]

TestGetContainerlabInventory

Containerlab topology generation from Netbox.

Purpose: Validate containerlab topology file generation from Netbox inventory.

Test Get Containerlab Inventory Devices

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_devices(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "lab_name": "foobar",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "links"
        ], f"{worker} - no clab inventory links data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        assert (
            res["result"]["name"] == "foobar"
        ), f"{worker} - clab inventory name is not foobar"

Test Get Containerlab Inventory Non Existing Devices

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_non_existing_devices(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceosabc", "fceosxyz"],
            "lab_name": "foobar",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert res["errors"], f"{worker} - received no error"
        assert res["failed"] == True, f"{worker} - should have failed"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert (
            res["result"]["topology"]["links"] == []
        ), f"{worker} - clab inventory links data returned"
        assert (
            res["result"]["topology"]["nodes"] == {}
        ), f"{worker} - clab inventory nodes data returned"

Test Get Containerlab Inventory By Tenant

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_by_tenant(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version >= (4, 3, 0):
        ret = nfclient.run_job(
            "netbox",
            "get_containerlab_inventory",
            workers="any",
            kwargs={
                "tenant": "NORFAB",
                "lab_name": "foobar",
            },
        )
    else:
        ret = nfclient.run_job(
            "netbox",
            "get_containerlab_inventory",
            workers="any",
            kwargs={
                "tenant": "norfab",
                "lab_name": "foobar",
            },
        )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "links"
        ], f"{worker} - no clab inventory links data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        assert (
            res["result"]["name"] == "foobar"
        ), f"{worker} - clab inventory name is not foobar"

Test Get Containerlab Inventory By Filters

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_by_filters(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version >= (4, 3, 0):
        ret = nfclient.run_job(
            "netbox",
            "get_containerlab_inventory",
            workers="any",
            kwargs={
                "filters": [{"name__ic": "ceos-spine", "status": "active"}],
                "lab_name": "foobar",
            },
        )
    else:
        ret = nfclient.run_job(
            "netbox",
            "get_containerlab_inventory",
            workers="any",
            kwargs={
                "filters": [{"q": "ceos-spine", "status": "active"}],
                "lab_name": "foobar",
            },
        )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "links"
        ], f"{worker} - no clab inventory links data returned"
        assert (
            len(res["result"]["topology"]["nodes"]) == 2
        ), f"{worker} - clab inventory nodes data wromg"
        assert (
            res["result"]["name"] == "foobar"
        ), f"{worker} - clab inventory name is not foobar"

Test Get Containerlab Inventory With Nb Instance

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_with_nb_instance(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "lab_name": "foobar",
            "instance": "preprod",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "links"
        ], f"{worker} - no clab inventory links data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        assert (
            res["result"]["name"] == "foobar"
        ), f"{worker} - clab inventory name is not foobar"

Test Get Containerlab Inventory With Image

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_with_image(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["ceos1"],
            "lab_name": "foobar",
            "image": "ceos:latest",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        for node_name, node_data in res["result"]["topology"]["nodes"].items():
            assert (
                node_data["image"] == "ceos:latest"
            ), f"{worker} - node {node_name} image is not ceos:latest"

Test Get Containerlab Inventory Run Out Of Ports

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_run_out_of_ports(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "lab_name": "foobar",
            "ports": [10000, 10005],
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert res["errors"], f"{worker} - received no error"
        assert res["failed"] == True, f"{worker} - should have failed"

Test Get Containerlab Inventory Run Out Of Ips

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_run_out_of_ips(self, nfclient):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5", "ceos1"],
            "lab_name": "foobar",
            "ipv4_subnet": "172.100.100.0/30",
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert res["errors"], f"{worker} - received no error"
        assert res["failed"] == True, f"{worker} - should have failed"

Test Get Containerlab Inventory With Ports Map

Source code in tests\test_netbox_service.py

def test_get_containerlab_inventory_with_ports_map(self, nfclient):
    ports_map = {
        "fceos4": [
            "10007:22/tcp",
            "10008:23/tcp",
            "10009:80/tcp",
            "10010:161/udp",
            "10011:443/tcp",
            "10012:830/tcp",
            "10013:8080/tcp",
        ]
    }
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "lab_name": "foobar",
            "ports_map": ports_map,
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        assert (
            res["result"]["topology"]["nodes"]["fceos4"]["ports"]
            == ports_map["fceos4"]
        ), f"{worker} - ports map not applied for fceos4"
        assert res["result"]["topology"]["nodes"]["fceos5"][
            "ports"
        ], f"{worker} - no ports allocated for fceos5"
        assert all(
            p.startswith("1200")
            for p in res["result"]["topology"]["nodes"]["fceos5"]["ports"]
        ), f"{worker} - ports for fceos5 allocated correctly"

Test Get Containerlab Inventory With Cache

Source code in tests\test_netbox_service.py

@pytest.mark.parametrize("cache", cache_options)
def test_get_containerlab_inventory_with_cache(self, nfclient, cache):
    ret = nfclient.run_job(
        "netbox",
        "get_containerlab_inventory",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "lab_name": "foobar",
            "cache": cache,
        },
    )
    pprint.pprint(ret)

    for worker, res in ret.items():
        assert not res["errors"], f"{worker} - received error"
        assert all(
            k in res["result"] for k in ["mgmt", "name", "topology"]
        ), f"{worker} - not all clab inventory data returned"
        assert res["result"]["topology"][
            "links"
        ], f"{worker} - no clab inventory links data returned"
        assert res["result"]["topology"][
            "nodes"
        ], f"{worker} - no clab inventory nodes data returned"
        assert (
            res["result"]["name"] == "foobar"
        ), f"{worker} - clab inventory name is not foobar"

TestCreatePrefix

IP prefix creation and management.

Purpose: Validate IP prefix allocation and creation.

Test Create Prefix

Source code in tests\test_netbox_service.py

def test_create_prefix(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"][
                "prefix"
            ], f"Result has no prefix {res1['result']}"

        worker, res2 = tuple(create_2.items())[0]
        assert (
            res1["result"]["prefix"] == res2["result"]["prefix"]
        ), "Should have been same prefix"
        assert (
            res1["result"]["description"] == res2["result"]["description"]
        ), "Should have been same prefix description"

Test Create Prefix Multiple

Source code in tests\test_netbox_service.py

def test_create_prefix_multiple(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {random.randint(1, 100)}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {random.randint(200, 300)}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"][
                "prefix"
            ], f"Result has no prefix {res1['result']}"

        worker, res2 = tuple(create_2.items())[0]
        assert (
            res1["result"]["prefix"] != res2["result"]["prefix"]
        ), "Should have been different prefix"

Test Create Prefix Non Exist Parent

Source code in tests\test_netbox_service.py

def test_create_prefix_non_exist_parent(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.123.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {random.randint(1, 100)}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == True, "Allocation not failed"
            assert (
                "Unable to source parent prefix from Netbox" in res1["messages"][0]
            ), "Result has no errors"

Test Create Prefix With Vrf

Should create single prefix and handle deduplication within vrf

Source code in tests\test_netbox_service.py

def test_create_prefix_with_vrf(self, nfclient):
    """Should create single prefix and handle deduplication within vrf"""
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.2.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "vrf": "VRF1",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "vrf": "VRF1",
            },
        )
        create_3 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand+1}",
                "vrf": "VRF1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)
        print("create_3")
        pprint.pprint(create_3, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"][
                "prefix"
            ], f"Result has no prefix {res1['result']}"

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]
        worker, res3 = tuple(create_3.items())[0]
        assert (
            res1["result"]["prefix"] == res2["result"]["prefix"]
        ), "Should have been same prefix"
        assert (
            res1["result"]["prefix"] != res3["result"]["prefix"]
        ), "Should have been different prefix"

Test Create Prefix With Parent Vrf Mismatch

Source code in tests\test_netbox_service.py

def test_create_prefix_with_parent_vrf_mismatch(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "vrf": "VRF1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == True, "Allocation not failed"
            assert "NetboxAllocationError" in res1["errors"][0]

Test Create Prefix By Parent Prefix Name

Source code in tests\test_netbox_service.py

def test_create_prefix_by_parent_prefix_name(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "TEST CREATE PREFIXES",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "TEST CREATE PREFIXES",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"][
                "prefix"
            ], f"Result has no prefix {res1['result']}"

        worker, res2 = tuple(create_2.items())[0]
        assert (
            res1["result"]["prefix"] == res2["result"]["prefix"]
        ), "Should have been same prefix"
        assert (
            res1["result"]["description"] == res2["result"]["description"]
        ), "Should have been same prefix description"

Test Create Prefix Within Vrf By Parent Prefix Name

Source code in tests\test_netbox_service.py

def test_create_prefix_within_vrf_by_parent_prefix_name(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.2.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "TEST CREATE PREFIXES WITH VRF",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "vrf": "VRF1",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "TEST CREATE PREFIXES WITH VRF",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "vrf": "VRF1",
            },
        )
        create_3 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "TEST CREATE PREFIXES WITH VRF",
                "prefixlen": 30,
                "description": f"test create prefix {rand+1}",
                "vrf": "VRF1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)
        print("create_3")
        pprint.pprint(create_3, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["result"][
                "prefix"
            ], f"Result has no prefix {res1['result']}"

        worker, res1 = tuple(create_1.items())[0]
        worker, res2 = tuple(create_2.items())[0]
        worker, res3 = tuple(create_3.items())[0]

        assert (
            res1["result"]["prefix"] == res2["result"]["prefix"]
        ), "Should have been same prefix"
        assert (
            res1["result"]["description"] == res2["result"]["description"]
        ), "Should have been same prefix description"
        assert (
            res1["result"]["prefix"] != res3["result"]["prefix"]
        ), "Should have been different prefix"

Test Create Prefix Dry Run Empty Parent

Source code in tests\test_netbox_service.py

def test_create_prefix_dry_run_empty_parent(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "dry_run": True,
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["dry_run"] == True
            assert res1["status"] == "unchanged"
            assert res1["result"]["prefix"] == "10.1.0.0/30"

Test Create Prefix Dry Run Parent Has Children

Source code in tests\test_netbox_service.py

def test_create_prefix_dry_run_parent_has_children(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 31,
                "description": f"test create prefix {rand}",
            },
        )
        create_dry_run = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand+1}",
                "dry_run": True,
            },
        )
        print("create_dry_run")
        pprint.pprint(create_dry_run, width=200)

        for worker, res1 in create_dry_run.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["dry_run"] == True
            assert res1["status"] == "unchanged"
            assert res1["result"]["prefix"] == "10.1.0.4/30"

Test Create Prefix Dry Run Prefix Exists

Source code in tests\test_netbox_service.py

def test_create_prefix_dry_run_prefix_exists(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        create_dry_run = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "dry_run": True,
            },
        )
        print("create_dry_run")
        pprint.pprint(create_dry_run, width=200)

        for worker, res1 in create_dry_run.items():
            assert res1["failed"] == False, "Allocation failed"
            assert res1["dry_run"] == True
            assert res1["status"] == "unchanged"
            assert res1["result"]["prefix"] == "10.1.0.0/30"

Test Create Prefix Test Length Mismatch

We creating first prefix, next creating prefix with same description but different prefix length

Source code in tests\test_netbox_service.py

def test_create_prefix_test_length_mismatch(self, nfclient):
    """We creating first prefix, next creating prefix with same
    description but different prefix length"""
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.1.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 31,
                "description": f"test create prefix {rand}",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.1.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        for worker, res in create_2.items():
            assert res["failed"] == True, "Allocation not failed"
            assert "NetboxAllocationError" in res["errors"][0]

Test Create Prefix With Attributes

Source code in tests\test_netbox_service.py

def test_create_prefix_with_attributes(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.2.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "site": "NORFAB-LAB",
                "tenant": "NORFAB",
                "role": "PREFIX_ROLE_1",
                "comments": "Some important comment",
                "vrf": "VRF1",
                "tags": ["NORFAB"],
                "status": "reserved",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation failed"
            assert all(
                k in res1["diff"]
                for k in [
                    "comments",
                    "description",
                    "role",
                    "site",
                    "status",
                    "tags",
                    "tenant",
                ]
            )

        # retrieve created prefix details from Netbox
        nb_prefix = nfclient.run_job(
            "netbox",
            "rest",
            workers="any",
            kwargs={
                "method": "get",
                "api": "/ipam/prefixes/",
                "params": {"prefix": res1["result"]["prefix"]},
            },
        )
        print("nb_prefix:")
        pprint.pprint(nb_prefix, width=200)
        worker, created_prefix = tuple(nb_prefix.items())[0]
        created_prefix = created_prefix["result"]["results"][0]

        assert created_prefix["role"]["name"] == "PREFIX_ROLE_1"
        assert created_prefix["scope"]["name"] == "NORFAB-LAB"
        assert created_prefix["tags"][0]["name"] == "NORFAB"
        assert created_prefix["tenant"]["name"] == "NORFAB"
        assert created_prefix["vrf"]["name"] == "VRF1"
        assert created_prefix["description"]
        assert created_prefix["comments"]

Test Create Prefix With Attributes Updates

Source code in tests\test_netbox_service.py

def test_create_prefix_with_attributes_updates(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_prefixes_within("10.2.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "site": "NORFAB-LAB",
                "tenant": "NORFAB",
                "role": "PREFIX_ROLE_1",
                "comments": "Some important comment",
                "tags": ["NORFAB"],
                "status": "reserved",
            },
        )
        create_2 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "description": f"test create prefix {rand}",
                "site": "SALTNORNIR-LAB",
                "tenant": "SALTNORNIR",
                "role": "PREFIX_ROLE_2",
                "comments": "Some important comments updates",
                "tags": ["ACCESS"],
                "status": "active",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)
        print("create_2")
        pprint.pprint(create_2, width=200)

        # verify has changes
        for worker, res1 in create_2.items():
            assert res1["failed"] == False, "Allocation failed"
            assert all(
                k in res1["diff"]
                for k in [
                    "comments",
                    "role",
                    "site",
                    "status",
                    "tags",
                    "tenant",
                ]
            )

        # retrieve created prefix details from Netbox
        nb_prefix = nfclient.run_job(
            "netbox",
            "rest",
            workers="any",
            kwargs={
                "method": "get",
                "api": "/ipam/prefixes/",
                "params": {"prefix": res1["result"]["prefix"]},
            },
        )
        print("nb_prefix:")
        pprint.pprint(nb_prefix, width=200)
        worker, created_prefix = tuple(nb_prefix.items())[0]
        created_prefix = created_prefix["result"]["results"][0]

        assert created_prefix["role"]["name"] == "PREFIX_ROLE_2"
        assert created_prefix["scope"]["name"] == "SALTNORNIR-LAB"
        for tag in created_prefix["tags"]:
            assert tag["name"] == "NORFAB" or tag["name"] == "ACCESS"
        assert created_prefix["tenant"]["name"] == "SALTNORNIR"
        assert created_prefix["vrf"]["name"] == "VRF1"
        assert created_prefix["description"] == f"test create prefix {rand}"
        assert created_prefix["comments"] == "Some important comments updates"

Test Create Prefix With Branch

Source code in tests\test_netbox_service.py

def test_create_prefix_with_branch(self, nfclient):
    if self.nb_version is None:
        self.nb_version = get_nb_version(nfclient)

    delete_branch("create_prefix_1", nfclient)
    delete_prefixes_within("10.2.0.0/24", nfclient)

    rand = random.randint(1, 1000)
    if self.nb_version[0] == 4:
        create_1 = nfclient.run_job(
            "netbox",
            "create_prefix",
            workers="any",
            kwargs={
                "parent": "10.2.0.0/24",
                "prefixlen": 30,
                "description": f"test create prefix {rand}",
                "site": "NORFAB-LAB",
                "tenant": "NORFAB",
                "role": "PREFIX_ROLE_1",
                "comments": "Some important comment",
                "vrf": "VRF1",
                "tags": ["NORFAB"],
                "status": "reserved",
                "branch": "create_prefix_1",
            },
        )
        print("create_1")
        pprint.pprint(create_1, width=200)

        for worker, res1 in create_1.items():
            assert res1["failed"] == False, "Allocation with branch failed"
            assert (
                res1["result"]["branch"] == "create_prefix_1"
            ), "No branch details in result"

TestCreateIPBulk

Bulk IP address creation operations.

Purpose: Validate efficient bulk IP creation for large datasets.

Test Create Ip Bulk

Source code in tests\test_netbox_service.py

def test_create_ip_bulk(self, nfclient):
    delete_prefixes_within("10.0.0.0/24", nfclient)
    delete_ips("10.0.0.0/24", nfclient)
    create_1 = nfclient.run_job(
        "netbox",
        "create_ip_bulk",
        workers="any",
        kwargs={
            "prefix": "10.0.0.0/24",
            "devices": ["fceos4", "fceos5"],
            "interface_regex": "eth103.0|eth11.123|Port-Channel1.101|ae5.101",
            "mask_len": 31,
        },
    )
    print("create_1")
    pprint.pprint(create_1, width=200)

    for worker, res1 in create_1.items():
        assert res1["failed"] == False, "Allocation failed"
        assert res1["result"] == {
            "fceos4": {
                "Port-Channel1.101": {
                    "address": "10.0.0.0/31",
                    "description": "",
                    "device": "fceos4",
                    "interface": "Port-Channel1.101",
                    "peer": {
                        "address": "10.0.0.1/31",
                        "description": "",
                        "device": "fceos5",
                        "interface": "ae5.101",
                        "vrf": "None",
                    },
                    "vrf": "None",
                },
                "eth103.0": {
                    "address": "10.0.0.2/31",
                    "description": "",
                    "device": "fceos4",
                    "interface": "eth103.0",
                    "peer": {
                        "address": "10.0.0.3/31",
                        "description": "",
                        "device": "fceos5",
                        "interface": "eth103",
                        "vrf": "None",
                    },
                    "vrf": "None",
                },
                "eth11.123": {
                    "address": "10.0.0.4/31",
                    "description": "",
                    "device": "fceos4",
                    "interface": "eth11.123",
                    "peer": {
                        "address": "10.0.0.5/31",
                        "description": "",
                        "device": "fceos5",
                        "interface": "eth11.123",
                        "vrf": "None",
                    },
                    "vrf": "None",
                },
            },
            "fceos5": {
                "ae5.101": {
                    "address": "10.0.0.1/31",
                    "description": "",
                    "device": "fceos5",
                    "interface": "ae5.101",
                    "vrf": "None",
                },
                "eth11.123": {
                    "address": "10.0.0.5/31",
                    "description": "",
                    "device": "fceos5",
                    "interface": "eth11.123",
                    "vrf": "None",
                },
            },
        }

Test Execution

Configuration

Netbox tests require configuration in netbox_data.py:

NB_URL = "http://netbox-instance:8000"
NB_API_TOKEN = "your-api-token"

Tests operate against three Netbox instances configured in inventory: dev, preprod, and prod.

Running Tests

# Run all Netbox tests
pytest tests/test_netbox_service.py -v

# Run specific test class
pytest tests/test_netbox_service.py::TestGetDevices -v

# Run specific test method
pytest tests/test_netbox_service.py::TestGetDevices::test_get_devices_with_filters -v

Test Utilities

Helper functions available in test file:

• get_nb_version(nfclient, instance=None) - Get Netbox version
• delete_branch(branch, nfclient) - Delete transaction branch
• delete_interfaces(nfclient, device, interface) - Delete interface
• delete_prefixes_within(prefix, nfclient) - Delete prefix subtree
• delete_ips(prefix, nfclient) - Delete IPs in prefix
• get_pynetbox(nfclient) - Get pynetbox API instance

Troubleshooting

• Connection errors: Verify Netbox is running and netbox_data.py is configured correctly
• Fixture timeout: Increase sleep time in fixture if workers need more time to start
• Cache issues: Clear cache between test runs if stale data is observed
• Branch isolation: Always delete test branches after use to prevent state pollution

Related Documentation

• Netbox Service Overview
• Netbox Service Tasks
• NORFAB Testing Framework
• Netbox Worker API Reference