You've already forked ansible-netbox-inventory
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.
This commit is contained in:
81
e2e/conftest.py
Normal file
81
e2e/conftest.py
Normal file
@@ -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
|
||||||
279
e2e/seed.py
Normal file
279
e2e/seed.py
Normal file
@@ -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_<key>.<secret>".
|
||||||
|
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 <url> <admin-user> <admin-password>."""
|
||||||
|
if len(sys.argv) != 4:
|
||||||
|
print("usage: seed.py <netbox-url> <admin-user> <admin-password>", 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())
|
||||||
165
e2e/test_e2e.py
Normal file
165
e2e/test_e2e.py
Normal file
@@ -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"])
|
||||||
Reference in New Issue
Block a user