You've already forked ansible-netbox-inventory
Bryan Joshua Pedini
927bcabc97
vm01 gains a primary IPv6 next to its IPv4, one standalone IPv6 address joins the dataset; covers ip_version on both sources, combination with filters, the IPv6-preferring behaviour of netbox's combined primary IP, and the invalid value error path.
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
# 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 = "",
|
|
ip_version: 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,
|
|
"NETBOX_INVENTORY_IP_VERSION": ip_version or "both",
|
|
}
|
|
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
|