From c3d3165a5482562772eeb2e7abf7f74f6f72b07f Mon Sep 17 00:00:00 2001 From: Bryan Joshua Pedini Date: Tue, 25 Aug 2026 13:47:18 +0200 Subject: [PATCH] test: end-to-end scenarios against a live netbox seed.py populates the stack through the API: a read-only user with a self-provisioned token, tenants, sites, clusters, a VRF, virtual machines (one with a primary IP) and standalone addresses; the tests run the real inventory script against every source/filter combination, every error path and the ansible --list/--host protocol. --- e2e/conftest.py | 81 ++++++++++++++ e2e/seed.py | 279 ++++++++++++++++++++++++++++++++++++++++++++++++ e2e/test_e2e.py | 165 ++++++++++++++++++++++++++++ 3 files changed, 525 insertions(+) create mode 100644 e2e/conftest.py create mode 100644 e2e/seed.py create mode 100644 e2e/test_e2e.py diff --git a/e2e/conftest.py b/e2e/conftest.py new file mode 100644 index 0000000..d833a4b --- /dev/null +++ b/e2e/conftest.py @@ -0,0 +1,81 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# Copyright (C) 2026 Bryan Joshua Pedini +"""Fixtures for the end-to-end tests against a live NetBox stack.""" + +import json +import os +import subprocess + +import pytest +import seed + +E2E_DIR = os.path.dirname(os.path.abspath(__file__)) +REPO_ROOT = os.path.dirname(E2E_DIR) +SCRIPT = os.path.join(REPO_ROOT, "netbox_inventory.py") + + +def read_env_file() -> dict[str, str]: + """Read e2e/.env (or env.example as fallback) into a dict.""" + path = os.path.join(E2E_DIR, ".env") + if not os.path.exists(path): + path = os.path.join(E2E_DIR, "env.example") + values: dict[str, str] = {} + with open(path, encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + values[key] = value.strip('"') + return values + + +@pytest.fixture(scope="session") +def stack() -> dict[str, str]: + """Wait for the compose stack, seed it, return url + read-only token.""" + env_file = read_env_file() + url = f"http://127.0.0.1:{env_file['NETBOX_HTTP_PORT']}" + try: + seed.wait_for_api(url, timeout=300) + admin_token = seed.provision_token( + url, env_file["NETBOX_SUPERUSER_NAME"], env_file["NETBOX_SUPERUSER_PASS"] + ) + seeded = seed.seed(url, admin_token) + except seed.SeedError as exc: + pytest.fail(f"e2e stack not usable, run 'make e2e-up' first: {exc}") + return {"url": url, "token": seeded["token"], "admin_token": admin_token} + + +@pytest.fixture +def run_inventory(stack): + """ + Run the real inventory script as Ansible would and return the result. + + Returns a callable taking source/filter/value keyword arguments; the + parsed JSON ends up in result.inventory when the run succeeds. + """ + + def runner( + source: str = "", + filter_name: str = "", + filter_value: str = "", + args: tuple[str, ...] = ("--list",), + token: str | None = None, + ): + env = { + **os.environ, + # os.devnull keeps a developer's real config.yaml out of the run + "NETBOX_INVENTORY_CONFIG": os.devnull, + "NETBOX_URL": stack["url"], + "NETBOX_TOKEN": token if token is not None else stack["token"], + "NETBOX_INVENTORY_SOURCE": source or "ip-addresses", + "NETBOX_INVENTORY_FILTER": filter_name, + "NETBOX_INVENTORY_FILTER_VALUE": filter_value, + } + result = subprocess.run( + [SCRIPT, *args], capture_output=True, text=True, timeout=120, env=env + ) + result.inventory = json.loads(result.stdout) if result.returncode == 0 else None + return result + + return runner diff --git a/e2e/seed.py b/e2e/seed.py new file mode 100644 index 0000000..2fc387d --- /dev/null +++ b/e2e/seed.py @@ -0,0 +1,279 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# Copyright (C) 2026 Bryan Joshua Pedini +""" +Seed a throwaway NetBox instance with the end-to-end test dataset. + +Everything goes through the REST API with the superuser token: a read-only +user with its own token, two tenants, two sites, two clusters, a VRF, three +virtual machines (one with a primary IP) and a handful of IP addresses. All +operations are get-or-create so seeding is idempotent. +""" + +import sys +import time +from typing import Any + +import requests + + +class SeedError(Exception): + """Raised when the NetBox instance cannot be seeded.""" + + +class AdminApi: + """Minimal admin client for creating objects through the NetBox API.""" + + def __init__(self, url: str, token: str, timeout: int = 15): + self._url = url.rstrip("/") + self._timeout = timeout + self._session = requests.Session() + scheme = "Bearer" if token.startswith("nbt_") else "Token" + self._session.headers.update( + {"Authorization": f"{scheme} {token}", "Accept": "application/json"} + ) + + def _check(self, response: requests.Response) -> Any: + if response.status_code not in (200, 201, 204): + raise SeedError( + f"{response.request.method} {response.url}: HTTP {response.status_code}: {response.text[:300]}" + ) + return response.json() if response.status_code != 204 else None + + def get(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: + response = self._session.get( + f"{self._url}/api/{path}/", params=params, timeout=self._timeout + ) + return self._check(response)["results"] + + def post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + response = self._session.post( + f"{self._url}/api/{path}/", json=payload, timeout=self._timeout + ) + return self._check(response) + + def patch(self, path: str, object_id: int, payload: dict[str, Any]) -> dict[str, Any]: + response = self._session.patch( + f"{self._url}/api/{path}/{object_id}/", json=payload, timeout=self._timeout + ) + return self._check(response) + + def delete(self, path: str, object_id: int) -> None: + response = self._session.delete( + f"{self._url}/api/{path}/{object_id}/", timeout=self._timeout + ) + self._check(response) + + def ensure(self, path: str, lookup: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: + """Return the object matching lookup, creating it from payload if missing.""" + matches = self.get(path, params=lookup) + if matches: + return matches[0] + return self.post(path, payload) + + +def wait_for_api(url: str, timeout: int = 300) -> None: + """ + Poll /api/status/ until NetBox answers or the timeout expires. + + A 403 counts as up: it means the application is serving requests and only + authentication is missing. + """ + deadline = time.monotonic() + timeout + last_error = "no attempt made" + while time.monotonic() < deadline: + try: + response = requests.get(f"{url.rstrip('/')}/api/status/", timeout=5) + if response.status_code in (200, 403): + return + last_error = f"HTTP {response.status_code}" + except requests.exceptions.RequestException as exc: + last_error = str(exc) + time.sleep(3) + raise SeedError(f"NetBox at {url} did not come up within {timeout}s: {last_error}") + + +def provision_token(url: str, username: str, password: str) -> str: + """Create a fresh API token from user credentials and return its key.""" + try: + response = requests.post( + f"{url.rstrip('/')}/api/users/tokens/provision/", + json={"username": username, "password": password}, + timeout=15, + ) + except requests.exceptions.RequestException as exc: + raise SeedError(f"Token provisioning failed: {exc}") from exc + if response.status_code != 201: + raise SeedError( + f"Token provisioning for {username} failed: HTTP {response.status_code}: {response.text[:300]}" + ) + return _token_secret(response.json()) + + +def _token_secret(data: dict[str, Any]) -> str: + """ + Return the usable credential of a freshly created token. + + NetBox 4.5 (token v2) shows the secret once in "token" and keeps only the + public identifier in "key"; the full credential is "nbt_.". + Older releases expose the whole secret as "key" directly. + """ + if data.get("token"): + return f"nbt_{data['key']}.{data['token']}" + return data["key"] + + +def seed(url: str, admin_token: str) -> dict[str, str]: + """Create the whole e2e dataset and return the read-only user's token.""" + api = AdminApi(url, admin_token) + + customer1 = api.ensure( + "tenancy/tenants", {"name": "customer1"}, {"name": "customer1", "slug": "customer1"} + ) + customer2 = api.ensure( + "tenancy/tenants", {"name": "customer2"}, {"name": "customer2", "slug": "customer2"} + ) + milan = api.ensure("dcim/sites", {"name": "milan"}, {"name": "milan", "slug": "milan"}) + turin = api.ensure("dcim/sites", {"name": "turin"}, {"name": "turin", "slug": "turin"}) + cluster_type = api.ensure( + "virtualization/cluster-types", {"name": "proxmox"}, {"name": "proxmox", "slug": "proxmox"} + ) + cluster1 = api.ensure( + "virtualization/clusters", + {"name": "cluster1"}, + { + "name": "cluster1", + "type": cluster_type["id"], + "scope_type": "dcim.site", + "scope_id": milan["id"], + }, + ) + cluster2 = api.ensure( + "virtualization/clusters", + {"name": "cluster2"}, + { + "name": "cluster2", + "type": cluster_type["id"], + "scope_type": "dcim.site", + "scope_id": turin["id"], + }, + ) + internal = api.ensure("ipam/vrfs", {"name": "internal"}, {"name": "internal", "rd": "65000:1"}) + + vm01 = api.ensure( + "virtualization/virtual-machines", + {"name": "vm01"}, + { + "name": "vm01", + "status": "active", + "cluster": cluster1["id"], + "site": milan["id"], + "tenant": customer1["id"], + }, + ) + api.ensure( + "virtualization/virtual-machines", + {"name": "vm02"}, + { + "name": "vm02", + "status": "active", + "cluster": cluster1["id"], + "site": milan["id"], + "tenant": customer2["id"], + }, + ) + api.ensure( + "virtualization/virtual-machines", + {"name": "vm03.example.com"}, + { + "name": "vm03.example.com", + "status": "active", + "cluster": cluster2["id"], + "site": turin["id"], + }, + ) + + # give vm01 a primary IP: interface -> assigned address -> primary_ip4 + eth0 = api.ensure( + "virtualization/interfaces", + {"virtual_machine_id": vm01["id"], "name": "eth0"}, + {"virtual_machine": vm01["id"], "name": "eth0"}, + ) + vm01_ip = api.ensure( + "ipam/ip-addresses", + {"address": "192.0.2.20/24"}, + { + "address": "192.0.2.20/24", + "dns_name": "vm01.example.com", + "assigned_object_type": "virtualization.vminterface", + "assigned_object_id": eth0["id"], + }, + ) + api.patch("virtualization/virtual-machines", vm01["id"], {"primary_ip4": vm01_ip["id"]}) + + standalone_addresses = [ + {"address": "192.0.2.10/24", "dns_name": "web.example.com", "tenant": customer1["id"]}, + {"address": "192.0.2.11/24", "vrf": internal["id"]}, + {"address": "192.0.2.12/24", "dns_name": "web.example.com"}, + { + "address": "192.0.2.13/24", + "dns_name": "db.other.org", + "tenant": customer2["id"], + "vrf": internal["id"], + }, + ] + for payload in standalone_addresses: + api.ensure("ipam/ip-addresses", {"address": payload["address"]}, payload) + + # dedicated read-only user for the inventory, with a fresh token every run + # (token secrets are only revealed on self-provisioning, never to admins) + password = "Ansible-e2e-Passw0rd!" + user = api.ensure( + "users/users", + {"username": "ansible-inventory"}, + {"username": "ansible-inventory", "password": password}, + ) + api.ensure( + "users/permissions", + {"name": "ansible-inventory read"}, + { + "name": "ansible-inventory read", + "enabled": True, + "actions": ["view"], + "object_types": [ + "tenancy.tenant", + "dcim.site", + "ipam.ipaddress", + "ipam.vrf", + "virtualization.cluster", + "virtualization.virtualmachine", + ], + "users": [user["id"]], + }, + ) + for token in api.get("users/tokens", params={"user_id": user["id"]}): + api.delete("users/tokens", token["id"]) + return { + "url": url, + "token": provision_token(url, user["username"], password), + "user": user["username"], + } + + +def main() -> int: + """Seed from the command line: seed.py .""" + if len(sys.argv) != 4: + print("usage: seed.py ", file=sys.stderr) + return 2 + try: + wait_for_api(sys.argv[1]) + admin_token = provision_token(sys.argv[1], sys.argv[2], sys.argv[3]) + result = seed(sys.argv[1], admin_token) + except SeedError as exc: + print(f"seed: {exc}", file=sys.stderr) + return 1 + print(f"seeded, inventory token for {result['user']}: {result['token']}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/e2e/test_e2e.py b/e2e/test_e2e.py new file mode 100644 index 0000000..9643aad --- /dev/null +++ b/e2e/test_e2e.py @@ -0,0 +1,165 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# Copyright (C) 2026 Bryan Joshua Pedini +""" +End-to-end tests: every source/filter scenario against a live NetBox. + +The dataset these expectations rest on is created by seed.py: +tenants customer1/customer2, sites milan/turin, clusters cluster1 (milan) and +cluster2 (turin), VRF internal, VMs vm01 (cluster1, customer1, primary IP +192.0.2.20 named vm01.example.com), vm02 (cluster1, customer2, no primary IP) +and vm03.example.com (cluster2), plus standalone addresses 192.0.2.10 +(web.example.com, customer1), 192.0.2.11 (unnamed, internal), 192.0.2.12 +(web.example.com duplicate) and 192.0.2.13 (db.other.org, customer2, internal). +""" + + +def hosts(result) -> set[str]: + return set(result.inventory["netbox"]["hosts"]) + + +def hostvars(result) -> dict: + return result.inventory["_meta"]["hostvars"] + + +class TestIpAddressSource: + def test_all_addresses(self, run_inventory): + result = run_inventory(source="ip-addresses") + assert result.returncode == 0, result.stderr + # the duplicate DNS name 192.0.2.12 falls back to its bare address + assert hosts(result) == { + "web.example.com", + "192.0.2.11", + "192.0.2.12", + "db.other.org", + "vm01.example.com", + } + web = hostvars(result)["web.example.com"] + assert web["ansible_host"] == "192.0.2.10" + assert web["netbox_tenant"] == "customer1" + assert hostvars(result)["192.0.2.11"]["netbox_vrf"] == "internal" + + def test_tenant_filter(self, run_inventory): + result = run_inventory( + source="ip-addresses", filter_name="tenant", filter_value="customer1" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"web.example.com"} + + def test_named_filter_deduplicates(self, run_inventory): + result = run_inventory(source="ip-addresses", filter_name="named") + assert result.returncode == 0, result.stderr + assert hosts(result) == {"web.example.com", "db.other.org", "vm01.example.com"} + + def test_named_false_keeps_unnamed(self, run_inventory): + result = run_inventory(source="ip-addresses", filter_name="named", filter_value="false") + assert result.returncode == 0, result.stderr + assert "192.0.2.11" in hosts(result) + + def test_vrf_filter(self, run_inventory): + result = run_inventory(source="ip-addresses", filter_name="vrf", filter_value="internal") + assert result.returncode == 0, result.stderr + assert hosts(result) == {"192.0.2.11", "db.other.org"} + + def test_domain_filter(self, run_inventory): + result = run_inventory( + source="ip-addresses", filter_name="domain", filter_value="example.com" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"web.example.com", "192.0.2.12", "vm01.example.com"} + + def test_domain_filter_other_domain(self, run_inventory): + result = run_inventory( + source="ip-addresses", filter_name="domain", filter_value="other.org" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"db.other.org"} + + +class TestVirtualMachineSource: + def test_all_virtual_machines(self, run_inventory): + result = run_inventory(source="virtual-machines") + assert result.returncode == 0, result.stderr + assert hosts(result) == {"vm01", "vm02", "vm03.example.com"} + vm01 = hostvars(result)["vm01"] + assert vm01["ansible_host"] == "192.0.2.20" + assert vm01["netbox_cluster"] == "cluster1" + assert vm01["netbox_site"] == "milan" + assert vm01["netbox_tenant"] == "customer1" + assert vm01["netbox_status"] == "active" + # no primary IP -> no ansible_host, Ansible resolves the name + assert "ansible_host" not in hostvars(result)["vm02"] + + def test_tenant_filter(self, run_inventory): + result = run_inventory( + source="virtual-machines", filter_name="tenant", filter_value="customer2" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"vm02"} + + def test_cluster_filter(self, run_inventory): + result = run_inventory( + source="virtual-machines", filter_name="cluster", filter_value="cluster1" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"vm01", "vm02"} + + def test_site_filter(self, run_inventory): + result = run_inventory(source="virtual-machines", filter_name="site", filter_value="turin") + assert result.returncode == 0, result.stderr + assert hosts(result) == {"vm03.example.com"} + + def test_domain_filter(self, run_inventory): + result = run_inventory( + source="virtual-machines", filter_name="domain", filter_value="example.com" + ) + assert result.returncode == 0, result.stderr + assert hosts(result) == {"vm03.example.com"} + + +class TestErrorScenarios: + def test_unknown_tenant_name(self, run_inventory): + result = run_inventory(source="ip-addresses", filter_name="tenant", filter_value="ghost") + assert result.returncode == 1 + assert 'No tenant named "ghost"' in result.stderr + + def test_filter_source_mismatch(self, run_inventory): + result = run_inventory( + source="virtual-machines", filter_name="vrf", filter_value="internal" + ) + assert result.returncode == 1 + assert "does not apply" in result.stderr + + def test_unknown_filter(self, run_inventory): + result = run_inventory(source="ip-addresses", filter_name="rack", filter_value="r1") + assert result.returncode == 1 + assert "Unknown filter" in result.stderr + + def test_unknown_source(self, run_inventory): + result = run_inventory(source="devices") + assert result.returncode == 1 + assert "Unknown source" in result.stderr + + def test_missing_filter_value(self, run_inventory): + result = run_inventory(source="virtual-machines", filter_name="cluster") + assert result.returncode == 1 + assert "requires a value" in result.stderr + + def test_invalid_token(self, run_inventory): + result = run_inventory(source="ip-addresses", token="wrong-token") + assert result.returncode == 1 + assert "NetBox API error" in result.stderr + + +class TestInventoryProtocol: + def test_host_returns_empty_vars(self, run_inventory): + result = run_inventory(args=("--host", "web.example.com")) + assert result.returncode == 0 + assert result.inventory == {} + + def test_list_structure(self, run_inventory): + result = run_inventory(source="ip-addresses") + assert result.returncode == 0, result.stderr + inventory = result.inventory + assert set(inventory) == {"netbox", "_meta"} + assert inventory["netbox"]["hosts"] == sorted(inventory["netbox"]["hosts"]) + assert set(inventory["netbox"]["hosts"]) == set(inventory["_meta"]["hostvars"])