docs: ip_version option documentation

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.

.gitignore

@@ -9,6 +9,7 @@ __pycache__/
 
 # Configuration
 config.yaml
+e2e/.env
 
 # IDE
 .idea/

README.md

@@ -1 +1,189 @@
 # Ansible NetBox Inventory
+
+An Ansible **dynamic inventory script** that sources hosts from NetBox, either
+from the IP address list or from the virtual machine list, with a single
+configurable filter option on top.
+
+Contract in one sentence: **one source, one optional filter, one value** — set
+in a config file, overridable per-run through environment variables.
+
+---
+
+## Features
+
+- Two host sources: **IP addresses** (`ipam/ip-addresses`) and **virtual
+  machines** (`virtualization/virtual-machines`)
+- Filter by tenant name, VRF name, cluster name, site name, hostname domain,
+  or restrict IP addresses to the ones that have a DNS name (deduplicated)
+- Restrict either source to one IP family (`ip_version: v4` or `v6`),
+  combinable with any filter
+- Name-based filters are resolved server-side (name → id) so NetBox does the
+  narrowing, not the script
+- Hosts land in a single `netbox` group with host variables delivered through
+  `_meta`, so Ansible calls the script exactly once
+- Plain Ansible inventory script protocol (`--list` / `--host`), no plugin
+  installation, no collections
+
+## Requirements
+
+- Python 3.10+
+- `requests` and `PyYAML` (PyYAML ships with Ansible anyway)
+- A NetBox API token with read permissions on the objects you inventory
+
+## Installation
+
+```sh
+git clone https://git.bjphoster.com/source/ansible-netbox-inventory.git
+cd ansible-netbox-inventory
+cp config.yaml.example config.yaml
+$EDITOR config.yaml
+```
+
+Point Ansible at the script directly:
+
+```sh
+ansible-inventory -i netbox_inventory.py --list
+ansible all -i netbox_inventory.py -m ping
+ansible-playbook -i netbox_inventory.py site.yml
+```
+
+## Configuration
+
+Everything lives in `config.yaml` next to the script (gitignored, see
+`config.yaml.example`):
+
+```yaml
+netbox:
+  url: "https://netbox.example.com"
+  token: "0123456789abcdef0123456789abcdef01234567"
+  verify_ssl: true
+  timeout: 30
+
+inventory:
+  source: "ip-addresses"
+  filter: "tenant"
+  filter_value: "customer1"
+```
+
+| Key | Default | Meaning |
+|---|---|---|
+| `netbox.url` | — (required) | Base URL of the NetBox instance |
+| `netbox.token` | — (required) | API token, both legacy tokens (sent as `Authorization: Token ...`) and NetBox 4.5+ v2 tokens (`nbt_...`, sent as `Authorization: Bearer ...`) work |
+| `netbox.verify_ssl` | `true` | Verify the TLS certificate |
+| `netbox.timeout` | `30` | Per-request timeout in seconds |
+| `inventory.source` | `ip-addresses` | `ip-addresses` or `virtual-machines` |
+| `inventory.filter` | none | Filter option, see the table below |
+| `inventory.filter_value` | none | Value for the filter option |
+| `inventory.ip_version` | `both` | `both`, `v4` or `v6`; on IP addresses only that family is fetched, on virtual machines it selects which primary IP becomes `ansible_host` |
+
+Every relevant key can be overridden through the environment, which is how you
+run several differently-filtered inventories off one config file:
+
+| Variable | Overrides |
+|---|---|
+| `NETBOX_INVENTORY_CONFIG` | Path to the config file |
+| `NETBOX_URL` | `netbox.url` |
+| `NETBOX_TOKEN` | `netbox.token` |
+| `NETBOX_INVENTORY_SOURCE` | `inventory.source` |
+| `NETBOX_INVENTORY_FILTER` | `inventory.filter` |
+| `NETBOX_INVENTORY_FILTER_VALUE` | `inventory.filter_value` |
+| `NETBOX_INVENTORY_IP_VERSION` | `inventory.ip_version` |
+
+## Filters
+
+One filter at a time, each takes exactly one value:
+
+| Filter | Sources | Value | Behaviour |
+|---|---|---|---|
+| `tenant` | both | tenant name | Objects assigned to that tenant |
+| `domain` | both | domain | Hostname ends with the domain, on a label boundary (`web.example.com` matches `example.com`, `notexample.com` does not) |
+| `named` | ip-addresses | `true`/`false` (empty = `true`) | Only addresses with a DNS name, deduplicated by hostname (first one wins) |
+| `vrf` | ip-addresses | VRF name | Addresses in that VRF |
+| `cluster` | virtual-machines | cluster name | Virtual machines in that cluster |
+| `site` | virtual-machines | site name | Virtual machines at that site |
+
+Examples:
+
+```sh
+# every IP address of one tenant
+NETBOX_INVENTORY_FILTER=tenant NETBOX_INVENTORY_FILTER_VALUE=customer1 \
+    ansible-inventory -i netbox_inventory.py --list
+
+# only IPs that resolve to something, one host per DNS name
+NETBOX_INVENTORY_FILTER=named \
+    ansible-inventory -i netbox_inventory.py --list
+
+# all virtual machines of one cluster
+NETBOX_INVENTORY_SOURCE=virtual-machines \
+NETBOX_INVENTORY_FILTER=cluster NETBOX_INVENTORY_FILTER_VALUE=proxmox01 \
+    ansible-inventory -i netbox_inventory.py --list
+
+# every host in a DNS domain
+NETBOX_INVENTORY_FILTER=domain NETBOX_INVENTORY_FILTER_VALUE=example.com \
+    ansible-inventory -i netbox_inventory.py --list
+```
+
+## Behaviour
+
+- **IP addresses**: the inventory hostname is the DNS name when set, the bare
+  address otherwise; `ansible_host` is always the bare address (no prefix
+  length). Host variables: `netbox_ip_address`, `netbox_dns_name`,
+  `netbox_tenant`, `netbox_vrf`, `netbox_description`.
+- **Virtual machines**: the inventory hostname is the VM name; `ansible_host`
+  is set to the primary IP when the VM has one, otherwise Ansible falls back
+  to resolving the name. Host variables: `netbox_status`, `netbox_cluster`,
+  `netbox_site`, `netbox_tenant`.
+- **IP version**: with `ip_version: both` (the default) virtual machines use
+  NetBox's combined primary IP, which **prefers IPv6** when a VM has both
+  (NetBox's `PREFER_IPV4` setting); set `v4` or `v6` to pin the family. On the
+  IP address source the family is filtered server-side, and a VM without a
+  primary IP of the requested family simply gets no `ansible_host`.
+- Duplicate inventory names (two IPs sharing a DNS name, outside the `named`
+  filter) keep both hosts: the second one falls back to its bare address as
+  the inventory name.
+- Pagination is followed transparently; `config_context` is excluded from the
+  virtual machine query to keep responses small.
+- Errors (unreachable NetBox, bad token, unknown tenant/cluster/site/VRF name,
+  invalid source/filter combination) print one readable line on stderr and
+  exit non-zero, so `ansible-inventory` fails loudly instead of running
+  against an empty host list.
+
+## Development
+
+```sh
+make dev-environment  # venv + dev dependencies
+make test             # unit tests (mocked API, no NetBox needed)
+make lint             # ruff checks
+make format           # ruff formatter
+```
+
+End-to-end tests run against a real NetBox in docker, modelled on the
+production deployment but exposed straight on `127.0.0.1:8800` with throwaway
+credentials (see `e2e/env.example`) and no persistence:
+
+```sh
+make e2e-up       # start the stack, first boot takes a few minutes
+make test-e2e     # seed via the API and test every source/filter scenario
+make e2e-down     # tear down and delete all data
+```
+
+Seeding creates a read-only `ansible-inventory` user with a self-provisioned
+v2 token, two tenants, two sites, two clusters, a VRF, three virtual machines
+and a handful of IP addresses, then the tests run the actual inventory script
+as an executable against every filter, every error path and the Ansible
+`--list` / `--host` protocol. Seeding is idempotent, `make test-e2e` can be
+re-run at will.
+
+## Known limitations
+
+- One filter option per run, filters do not combine; run the script twice with
+  different environments if you need an intersection
+- No caching: every run hits the NetBox API
+- Devices (`dcim/devices`) are not a source, only IP addresses and virtual
+  machines
+- Name lookups (tenant, VRF, cluster, site) take the first match when NetBox
+  returns several objects with the same name
+
+## License
+
+GPL-2.0-or-later. See [LICENSE](LICENSE) for more information.

config.yaml.example

@@ -13,3 +13,5 @@ inventory:
   # optional filter, see README.md for the full list
   filter: ""
   filter_value: ""
+  # restrict to one IP family: both | v4 | v6
+  ip_version: "both"

conftest.py

@@ -0,0 +1,3 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+"""Root conftest so that tests can import netbox_inventory from the repo root."""

e2e/conftest.py

@@ -0,0 +1,83 @@
+# 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

e2e/docker-compose.yml

@@ -0,0 +1,105 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+---
+services:
+  netbox: &netbox
+    image: netboxcommunity/netbox:${NETBOX_VERSION}-${NETBOX_DOCKER_VERSION}
+    depends_on:
+      postgres:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+      redis-cache:
+        condition: service_healthy
+    environment:
+      - API_TOKEN_PEPPER_1=${NETBOX_API_TOKEN_PEPPER_1}
+      - DB_HOST=postgres
+      - DB_NAME=${PSQL_NAME}
+      - DB_USER=${PSQL_USER}
+      - DB_PASSWORD=${PSQL_PASS}
+      - REDIS_HOST=redis
+      - REDIS_DATABASE=0
+      - REDIS_PASSWORD=${REDIS_PASS}
+      - REDIS_SSL=false
+      - REDIS_CACHE_HOST=redis-cache
+      - REDIS_CACHE_DATABASE=1
+      - REDIS_CACHE_PASSWORD=${REDIS_CACHE_PASS}
+      - REDIS_CACHE_SSL=false
+      - SECRET_KEY=${NETBOX_SECRET_KEY}
+      - SKIP_SUPERUSER=false
+      - SUPERUSER_NAME=${NETBOX_SUPERUSER_NAME}
+      - SUPERUSER_EMAIL=${NETBOX_SUPERUSER_EMAIL}
+      - SUPERUSER_PASSWORD=${NETBOX_SUPERUSER_PASS}
+    healthcheck:
+      test: curl -f http://localhost:8080/login/ || exit 1
+      # first boot runs every migration, give it plenty of headroom
+      start_period: 600s
+      timeout: 3s
+      interval: 15s
+    ports:
+      - "127.0.0.1:${NETBOX_HTTP_PORT}:8080"
+    networks:
+      - internal
+
+  netbox-worker:
+    <<: *netbox
+    command:
+      - /opt/netbox/venv/bin/python
+      - /opt/netbox/netbox/manage.py
+      - rqworker
+    depends_on:
+      netbox:
+        condition: service_healthy
+    healthcheck:
+      test: ps -aux | grep -v grep | grep -q rqworker || exit 1
+      start_period: 20s
+      timeout: 3s
+      interval: 15s
+    ports: []
+
+  postgres:
+    image: postgres:${PSQL_VERSION}
+    environment:
+      - POSTGRES_DB=${PSQL_NAME}
+      - POSTGRES_USER=${PSQL_USER}
+      - POSTGRES_PASSWORD=${PSQL_PASS}
+    healthcheck:
+      test: pg_isready -q -t 2 -d $$POSTGRES_DB -U $$POSTGRES_USER
+      start_period: 20s
+      timeout: 30s
+      interval: 10s
+      retries: 5
+    networks:
+      - internal
+
+  redis:
+    image: redis:${REDIS_VERSION}
+    command:
+      - sh
+      - -c
+      - redis-server --appendonly yes --requirepass $$REDIS_PASSWORD
+    environment:
+      - REDIS_PASSWORD=${REDIS_PASS}
+    healthcheck: &redis-healthcheck
+      test: '[ $$(redis-cli --pass "$${REDIS_PASSWORD}" ping) = ''PONG'' ]'
+      start_period: 5s
+      timeout: 3s
+      interval: 1s
+      retries: 5
+    networks:
+      - internal
+
+  redis-cache:
+    image: redis:${REDIS_VERSION}
+    command:
+      - sh
+      - -c
+      - redis-server --requirepass $$REDIS_PASSWORD
+    environment:
+      - REDIS_PASSWORD=${REDIS_CACHE_PASS}
+    healthcheck: *redis-healthcheck
+    networks:
+      - internal
+
+networks:
+  internal:

e2e/env.example

@@ -0,0 +1,24 @@
+# Throwaway credentials for the local e2e stack only, never reuse them.
+
+# NetBox
+NETBOX_DOCKER_VERSION=4.0.2
+NETBOX_VERSION=v4.5.8
+NETBOX_HTTP_PORT=8800
+NETBOX_API_TOKEN_PEPPER_1="e2e-only-pepper-0123456789abcdefghijklmnopqrstuvwxyz-0123456789"
+NETBOX_SECRET_KEY="e2e-only-secret-key-0123456789abcdefghijklmnopqrstuvwxyz"
+NETBOX_SUPERUSER_NAME=admin
+NETBOX_SUPERUSER_EMAIL=admin@example.com
+NETBOX_SUPERUSER_PASS=admin-e2e-password
+
+# PostgreSQL
+PSQL_VERSION=18.3-alpine3.22
+PSQL_NAME=netbox
+PSQL_USER=netbox
+PSQL_PASS=netbox-e2e-password
+
+# Redis
+REDIS_VERSION=8.6.2-alpine3.23
+REDIS_PASS=redis-e2e-password
+
+# Redis Cache
+REDIS_CACHE_PASS=redis-cache-e2e-password

e2e/seed.py

@@ -0,0 +1,294 @@
+# 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 primary IPs: interface -> assigned addresses -> primary_ip4/6
+    eth0 = api.ensure(
+        "virtualization/interfaces",
+        {"virtual_machine_id": vm01["id"], "name": "eth0"},
+        {"virtual_machine": vm01["id"], "name": "eth0"},
+    )
+    vm01_ip4 = 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"],
+        },
+    )
+    vm01_ip6 = api.ensure(
+        "ipam/ip-addresses",
+        {"address": "2001:db8::20/64"},
+        {
+            "address": "2001:db8::20/64",
+            "dns_name": "vm01-v6.example.com",
+            "assigned_object_type": "virtualization.vminterface",
+            "assigned_object_id": eth0["id"],
+        },
+    )
+    api.patch(
+        "virtualization/virtual-machines",
+        vm01["id"],
+        {"primary_ip4": vm01_ip4["id"], "primary_ip6": vm01_ip6["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": "2001:db8::10/64", "dns_name": "v6.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())

e2e/test_e2e.py

@@ -0,0 +1,227 @@
+# 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 IPs
+192.0.2.20 vm01.example.com and 2001:db8::20 vm01-v6.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), 2001:db8::10
+(v6.example.com) 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",
+            "v6.example.com",
+            "vm01-v6.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",
+            "v6.example.com",
+            "vm01-v6.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",
+            "v6.example.com",
+            "vm01-v6.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"}
+
+    def test_ip_version_v4(self, run_inventory):
+        result = run_inventory(source="ip-addresses", ip_version="v4")
+        assert result.returncode == 0, result.stderr
+        assert hosts(result) == {
+            "web.example.com",
+            "192.0.2.11",
+            "192.0.2.12",
+            "db.other.org",
+            "vm01.example.com",
+        }
+
+    def test_ip_version_v6(self, run_inventory):
+        result = run_inventory(source="ip-addresses", ip_version="v6")
+        assert result.returncode == 0, result.stderr
+        assert hosts(result) == {"v6.example.com", "vm01-v6.example.com"}
+        assert hostvars(result)["v6.example.com"]["ansible_host"] == "2001:db8::10"
+
+    def test_ip_version_combines_with_filters(self, run_inventory):
+        result = run_inventory(
+            source="ip-addresses",
+            filter_name="domain",
+            filter_value="example.com",
+            ip_version="v4",
+        )
+        assert result.returncode == 0, result.stderr
+        assert hosts(result) == {"web.example.com", "192.0.2.12", "vm01.example.com"}
+
+
+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"]
+        # with both families set, NetBox's primary_ip prefers IPv6 by default
+        assert vm01["ansible_host"] == "2001:db8::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"}
+
+    def test_ip_version_v4_uses_primary_ip4(self, run_inventory):
+        result = run_inventory(source="virtual-machines", ip_version="v4")
+        assert result.returncode == 0, result.stderr
+        assert hosts(result) == {"vm01", "vm02", "vm03.example.com"}
+        assert hostvars(result)["vm01"]["ansible_host"] == "192.0.2.20"
+
+    def test_ip_version_v6_uses_primary_ip6(self, run_inventory):
+        result = run_inventory(source="virtual-machines", ip_version="v6")
+        assert result.returncode == 0, result.stderr
+        assert hostvars(result)["vm01"]["ansible_host"] == "2001:db8::20"
+        # VMs without a primary IP of that family keep no ansible_host
+        assert "ansible_host" not in hostvars(result)["vm02"]
+        assert "ansible_host" not in hostvars(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_ip_version(self, run_inventory):
+        result = run_inventory(source="ip-addresses", ip_version="ipv4")
+        assert result.returncode == 1
+        assert "Unknown ip_version" 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"])

makefile

@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+.PHONY: help dev-environment test lint lint-fix format clean e2e-up e2e-down test-e2e
+
+VENV ?= .venv
+PYTHON ?= $(VENV)/bin/python
+
+help:
+	@echo "Available targets:"
+	@echo "  dev-environment  create the virtualenv and install dev dependencies"
+	@echo "  test             run the unit test suite"
+	@echo "  lint             run ruff checks"
+	@echo "  lint-fix         run ruff checks and fix what is auto-fixable"
+	@echo "  format           run the ruff formatter"
+	@echo "  clean            remove the virtualenv and caches"
+	@echo "  e2e-up           start the throwaway NetBox stack for e2e tests"
+	@echo "  e2e-down         stop the e2e stack and delete its data"
+	@echo "  test-e2e         run the end-to-end tests against the e2e stack"
+
+dev-environment:
+	python3 -m venv $(VENV)
+	$(PYTHON) -m pip install --upgrade pip
+	$(PYTHON) -m pip install requests PyYAML pytest responses ruff
+
+test:
+	$(PYTHON) -m pytest
+
+lint:
+	$(PYTHON) -m ruff check .
+
+lint-fix:
+	$(PYTHON) -m ruff check --fix .
+
+format:
+	$(PYTHON) -m ruff format .
+
+e2e-up:
+	test -f e2e/.env || cp e2e/env.example e2e/.env
+	docker compose --project-directory e2e --env-file e2e/.env up -d --wait
+
+e2e-down:
+	docker compose --project-directory e2e --env-file e2e/.env down -v
+
+test-e2e:
+	$(PYTHON) -m pytest e2e
+
+clean:
+	rm -rf $(VENV) .pytest_cache .ruff_cache __pycache__ tests/__pycache__

netbox_inventory.py

@@ -25,6 +25,23 @@ SOURCE_IP_ADDRESSES = "ip-addresses"
 SOURCE_VIRTUAL_MACHINES = "virtual-machines"
 SOURCES = (SOURCE_IP_ADDRESSES, SOURCE_VIRTUAL_MACHINES)
 
+IP_VERSION_BOTH = "both"
+IP_VERSIONS = (IP_VERSION_BOTH, "v4", "v6")
+# ip_version -> family query parameter (ip-addresses source)
+IP_FAMILIES = {"v4": "4", "v6": "6"}
+# ip_version -> primary IP field (virtual-machines source)
+PRIMARY_IP_FIELDS = {IP_VERSION_BOTH: "primary_ip", "v4": "primary_ip4", "v6": "primary_ip6"}
+
+# filter option -> sources it applies to
+FILTERS = {
+    "tenant": SOURCES,
+    "domain": SOURCES,
+    "named": (SOURCE_IP_ADDRESSES,),
+    "vrf": (SOURCE_IP_ADDRESSES,),
+    "cluster": (SOURCE_VIRTUAL_MACHINES,),
+    "site": (SOURCE_VIRTUAL_MACHINES,),
+}
+
 
 class InventoryError(Exception):
     """Raised when the inventory cannot be built (configuration or API problems)."""
@@ -56,6 +73,18 @@ def _extract_error_message(response: requests.Response) -> str:
     return f"NetBox API error: HTTP {response.status_code}"
 
 
+def auth_header(token: str) -> str:
+    """
+    Build the Authorization header value for a NetBox API token.
+
+    Version 2 tokens (NetBox 4.5+, "nbt_<key>.<secret>") use the Bearer
+    scheme, legacy tokens the Token scheme.
+    """
+    if token.startswith("nbt_"):
+        return f"Bearer {token}"
+    return f"Token {token}"
+
+
 class NetBoxClient:
     """Thin wrapper around the NetBox REST API."""
 
@@ -66,7 +95,7 @@ class NetBoxClient:
         self._session.verify = verify_ssl
         self._session.headers.update(
             {
-                "Authorization": f"Token {token}",
+                "Authorization": auth_header(token),
                 "Accept": "application/json",
             }
         )
@@ -90,9 +119,7 @@ class NetBoxClient:
         except requests.exceptions.RequestException as exc:
             raise NetBoxAPIError(f"NetBox API request failed: {exc}") from exc
         if response.status_code != 200:
-            raise NetBoxAPIError(
-                _extract_error_message(response), status_code=response.status_code
-            )
+            raise NetBoxAPIError(_extract_error_message(response), status_code=response.status_code)
         try:
             return response.json()
         except ValueError as exc:
@@ -111,6 +138,18 @@ class NetBoxClient:
             params = None  # the "next" URL already carries the query string
         return results
 
+    def resolve_id(self, path: str, name: str, label: str) -> int:
+        """
+        Look up an object by name and return its numeric id.
+
+        Raises:
+            NetBoxAPIError: If no object with that name exists.
+        """
+        matches = self.get_all(path, params={"name": name})
+        if not matches:
+            raise NetBoxAPIError(f'No {label} named "{name}" found in NetBox.')
+        return matches[0]["id"]
+
 
 def load_config(path: str | None = None) -> dict[str, Any]:
     """
@@ -149,6 +188,11 @@ def load_config(path: str | None = None) -> dict[str, Any]:
         "filter_value": os.environ.get(
             "NETBOX_INVENTORY_FILTER_VALUE", inventory.get("filter_value") or ""
         ),
+        "ip_version": os.environ.get(
+            "NETBOX_INVENTORY_IP_VERSION", inventory.get("ip_version") or IP_VERSION_BOTH
+        )
+        .strip()
+        .lower(),
     }
     try:
         merged["timeout"] = int(merged["timeout"])
@@ -162,6 +206,99 @@ def load_config(path: str | None = None) -> dict[str, Any]:
     return merged
 
 
+def parse_bool(value: str) -> bool:
+    """
+    Parse a yes/no filter value, an empty string counts as yes.
+
+    Raises:
+        InventoryError: If the value is not recognisable as a boolean.
+    """
+    lowered = value.strip().lower()
+    if lowered in ("", "1", "true", "yes", "on"):
+        return True
+    if lowered in ("0", "false", "no", "off"):
+        return False
+    raise InventoryError(f'Filter value "{value}" is not a valid boolean.')
+
+
+def domain_matches(hostname: str, domain: str) -> bool:
+    """
+    Check whether a hostname belongs to a domain.
+
+    The hostname must end with the domain on a label boundary, so
+    "web.example.com" matches "example.com" but "notexample.com" does not.
+    """
+    if not hostname:
+        return False
+    domain = domain.lstrip(".").rstrip(".").lower()
+    hostname = hostname.rstrip(".").lower()
+    return hostname == domain or hostname.endswith("." + domain)
+
+
+def validate_options(
+    source: str, filter_name: str, filter_value: str, ip_version: str = IP_VERSION_BOTH
+) -> None:
+    """
+    Validate the source/filter/ip_version combination before touching the API.
+
+    Raises:
+        InventoryError: On unknown sources, filters or IP versions, a filter
+            that does not apply to the source, or a missing filter value.
+    """
+    if source not in SOURCES:
+        raise InventoryError(f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.')
+    if ip_version not in IP_VERSIONS:
+        raise InventoryError(
+            f'Unknown ip_version "{ip_version}", valid values: {", ".join(IP_VERSIONS)}.'
+        )
+    if not filter_name:
+        return
+    if filter_name not in FILTERS:
+        raise InventoryError(
+            f'Unknown filter "{filter_name}", valid filters: {", ".join(FILTERS)}.'
+        )
+    if source not in FILTERS[filter_name]:
+        raise InventoryError(f'Filter "{filter_name}" does not apply to source "{source}".')
+    if filter_name != "named" and not filter_value:
+        raise InventoryError(f'Filter "{filter_name}" requires a value.')
+
+
+def server_side_params(client: NetBoxClient, filter_name: str, filter_value: str) -> dict[str, Any]:
+    """Translate name-based filters into NetBox API query parameters."""
+    if filter_name == "tenant":
+        return {"tenant_id": client.resolve_id("tenancy/tenants", filter_value, "tenant")}
+    if filter_name == "vrf":
+        return {"vrf_id": client.resolve_id("ipam/vrfs", filter_value, "VRF")}
+    if filter_name == "cluster":
+        return {"cluster_id": client.resolve_id("virtualization/clusters", filter_value, "cluster")}
+    if filter_name == "site":
+        return {"site_id": client.resolve_id("dcim/sites", filter_value, "site")}
+    return {}
+
+
+def apply_client_side_filters(
+    entries: list[tuple[str, dict[str, Any]]], filter_name: str, filter_value: str
+) -> list[tuple[str, dict[str, Any]]]:
+    """
+    Apply the filters that cannot be expressed as API query parameters.
+
+    "domain" keeps hostnames belonging to the given domain, "named" keeps only
+    IP addresses with a DNS name and deduplicates them by hostname.
+    """
+    if filter_name == "domain":
+        return [(name, host) for name, host in entries if domain_matches(name, filter_value)]
+    if filter_name == "named" and parse_bool(filter_value):
+        seen: set[str] = set()
+        named = []
+        for name, host in entries:
+            if not host.get("netbox_dns_name") or name in seen:
+                continue
+            seen.add(name)
+            named.append((name, host))
+        return named
+    return entries
+
+
 def strip_prefix(address: str) -> str:
     """Return a bare IP address without the CIDR prefix length."""
     return address.split("/", 1)[0]
@@ -187,12 +324,15 @@ def ip_host_entry(ip: dict[str, Any]) -> tuple[str, dict[str, Any]]:
     return name, hostvars
 
 
-def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+def vm_host_entry(
+    vm: dict[str, Any], ip_version: str = IP_VERSION_BOTH
+) -> tuple[str, dict[str, Any]]:
     """
     Turn a NetBox virtual machine record into an inventory (name, hostvars) pair.
 
-    ansible_host is only set when the virtual machine has a primary IP, so that
-    Ansible falls back to resolving the name otherwise.
+    ansible_host is only set when the virtual machine has a primary IP of the
+    requested version, so that Ansible falls back to resolving the name
+    otherwise. With ip_version "both" NetBox picks the preferred family.
     """
     name = vm.get("name") or ""
     hostvars = {
@@ -201,19 +341,27 @@ def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
         "netbox_site": (vm.get("site") or {}).get("name") or "",
         "netbox_tenant": (vm.get("tenant") or {}).get("name") or "",
     }
-    primary = vm.get("primary_ip") or {}
+    primary = vm.get(PRIMARY_IP_FIELDS[ip_version]) or {}
     if primary.get("address"):
         hostvars["ansible_host"] = strip_prefix(primary["address"])
     return name, hostvars
 
 
-def fetch_entries(client: NetBoxClient, source: str) -> list[tuple[str, dict[str, Any]]]:
+def fetch_entries(
+    client: NetBoxClient,
+    source: str,
+    params: dict[str, Any] | None = None,
+    ip_version: str = IP_VERSION_BOTH,
+) -> list[tuple[str, dict[str, Any]]]:
     """Fetch all hosts of the configured source from NetBox."""
     if source == SOURCE_IP_ADDRESSES:
-        records = client.get_all("ipam/ip-addresses")
+        if ip_version != IP_VERSION_BOTH:
+            params = {**(params or {}), "family": IP_FAMILIES[ip_version]}
+        records = client.get_all("ipam/ip-addresses", params)
         return [ip_host_entry(record) for record in records if record.get("address")]
-    records = client.get_all("virtualization/virtual-machines", {"exclude": "config_context"})
-    return [vm_host_entry(record) for record in records if record.get("name")]
+    params = {"exclude": "config_context", **(params or {})}
+    records = client.get_all("virtualization/virtual-machines", params)
+    return [vm_host_entry(record, ip_version) for record in records if record.get("name")]
 
 
 def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, Any]:
@@ -221,14 +369,16 @@ def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, A
     Build the full --list inventory structure for Ansible.
 
     Raises:
-        InventoryError: On an unknown source or NetBox API failures.
+        InventoryError: On invalid source/filter options or NetBox API failures.
     """
     source = config["source"]
-    if source not in SOURCES:
-        raise InventoryError(
-            f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.'
-        )
-    entries = fetch_entries(client, source)
+    filter_name = config["filter"]
+    filter_value = config["filter_value"]
+    ip_version = config.get("ip_version") or IP_VERSION_BOTH
+    validate_options(source, filter_name, filter_value, ip_version)
+    params = server_side_params(client, filter_name, filter_value) if filter_name else {}
+    entries = fetch_entries(client, source, params, ip_version)
+    entries = apply_client_side_filters(entries, filter_name, filter_value)
     hostvars: dict[str, dict[str, Any]] = {}
     for name, host in entries:
         if name in hostvars:

pytest.ini

@@ -0,0 +1,5 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+[pytest]
+testpaths = tests
+addopts = -ra -q

tests/test_inventory.py

@@ -0,0 +1,364 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+"""Unit tests for the Ansible NetBox inventory script."""
+
+import json
+
+import pytest
+import responses
+
+import netbox_inventory as inv
+
+NETBOX_URL = "https://netbox.example.com"
+
+
+def make_client() -> inv.NetBoxClient:
+    return inv.NetBoxClient(NETBOX_URL, "token", verify_ssl=True, timeout=5)
+
+
+class TestAuthHeader:
+    def test_legacy_token_uses_token_scheme(self):
+        assert inv.auth_header("0123456789abcdef") == "Token 0123456789abcdef"
+
+    def test_v2_token_uses_bearer_scheme(self):
+        assert inv.auth_header("nbt_abc.def") == "Bearer nbt_abc.def"
+
+
+class TestStripPrefix:
+    def test_strips_prefix_length(self):
+        assert inv.strip_prefix("192.0.2.10/24") == "192.0.2.10"
+
+    def test_leaves_bare_address_alone(self):
+        assert inv.strip_prefix("192.0.2.10") == "192.0.2.10"
+
+
+class TestDomainMatches:
+    def test_matches_subdomain(self):
+        assert inv.domain_matches("web.example.com", "example.com")
+
+    def test_matches_domain_itself(self):
+        assert inv.domain_matches("example.com", "example.com")
+
+    def test_rejects_suffix_without_label_boundary(self):
+        assert not inv.domain_matches("notexample.com", "example.com")
+
+    def test_is_case_insensitive(self):
+        assert inv.domain_matches("Web.Example.COM", "example.com")
+
+    def test_accepts_leading_dot_and_trailing_dot(self):
+        assert inv.domain_matches("web.example.com.", ".example.com")
+
+    def test_rejects_empty_hostname(self):
+        assert not inv.domain_matches("", "example.com")
+
+
+class TestParseBool:
+    def test_empty_string_counts_as_yes(self):
+        assert inv.parse_bool("") is True
+
+    def test_common_spellings(self):
+        assert inv.parse_bool("true") is True
+        assert inv.parse_bool("no") is False
+
+    def test_garbage_raises(self):
+        with pytest.raises(inv.InventoryError):
+            inv.parse_bool("maybe")
+
+
+class TestValidateOptions:
+    def test_valid_combination_passes(self):
+        inv.validate_options("ip-addresses", "vrf", "internal")
+
+    def test_no_filter_passes(self):
+        inv.validate_options("virtual-machines", "", "")
+
+    def test_unknown_source(self):
+        with pytest.raises(inv.InventoryError, match="Unknown source"):
+            inv.validate_options("devices", "", "")
+
+    def test_unknown_filter(self):
+        with pytest.raises(inv.InventoryError, match="Unknown filter"):
+            inv.validate_options("ip-addresses", "platform", "linux")
+
+    def test_filter_source_mismatch(self):
+        with pytest.raises(inv.InventoryError, match="does not apply"):
+            inv.validate_options("virtual-machines", "vrf", "internal")
+
+    def test_missing_value(self):
+        with pytest.raises(inv.InventoryError, match="requires a value"):
+            inv.validate_options("virtual-machines", "cluster", "")
+
+    def test_named_without_value_is_fine(self):
+        inv.validate_options("ip-addresses", "named", "")
+
+    def test_valid_ip_versions_pass(self):
+        for ip_version in ("both", "v4", "v6"):
+            inv.validate_options("ip-addresses", "", "", ip_version)
+
+    def test_unknown_ip_version(self):
+        with pytest.raises(inv.InventoryError, match="Unknown ip_version"):
+            inv.validate_options("ip-addresses", "", "", "ipv4")
+
+
+class TestHostEntries:
+    def test_ip_with_dns_name(self):
+        name, host = inv.ip_host_entry(
+            {
+                "address": "192.0.2.10/24",
+                "dns_name": "web.example.com",
+                "tenant": {"name": "customer1"},
+                "vrf": {"name": "internal"},
+                "description": "web server",
+            }
+        )
+        assert name == "web.example.com"
+        assert host["ansible_host"] == "192.0.2.10"
+        assert host["netbox_tenant"] == "customer1"
+        assert host["netbox_vrf"] == "internal"
+
+    def test_ip_without_dns_name_uses_address(self):
+        name, host = inv.ip_host_entry({"address": "192.0.2.11/24"})
+        assert name == "192.0.2.11"
+        assert host["netbox_dns_name"] == ""
+
+    def test_vm_with_primary_ip(self):
+        name, host = inv.vm_host_entry(
+            {
+                "name": "vm01",
+                "status": {"value": "active"},
+                "cluster": {"name": "cluster1"},
+                "site": {"name": "site1"},
+                "tenant": {"name": "customer1"},
+                "primary_ip": {"address": "192.0.2.20/24"},
+            }
+        )
+        assert name == "vm01"
+        assert host["ansible_host"] == "192.0.2.20"
+        assert host["netbox_cluster"] == "cluster1"
+        assert host["netbox_site"] == "site1"
+
+    def test_vm_without_primary_ip_has_no_ansible_host(self):
+        _, host = inv.vm_host_entry({"name": "vm02"})
+        assert "ansible_host" not in host
+
+    def test_vm_ip_version_selects_primary_ip_field(self):
+        vm = {
+            "name": "vm01",
+            "primary_ip": {"address": "2001:db8::20/64"},
+            "primary_ip4": {"address": "192.0.2.20/24"},
+            "primary_ip6": {"address": "2001:db8::20/64"},
+        }
+        assert inv.vm_host_entry(vm, "both")[1]["ansible_host"] == "2001:db8::20"
+        assert inv.vm_host_entry(vm, "v4")[1]["ansible_host"] == "192.0.2.20"
+        assert inv.vm_host_entry(vm, "v6")[1]["ansible_host"] == "2001:db8::20"
+
+    def test_vm_without_requested_family_has_no_ansible_host(self):
+        vm = {"name": "vm01", "primary_ip4": {"address": "192.0.2.20/24"}}
+        assert "ansible_host" not in inv.vm_host_entry(vm, "v6")[1]
+
+
+class TestClientSideFilters:
+    def entries(self):
+        return [
+            (
+                "web.example.com",
+                {"netbox_dns_name": "web.example.com", "ansible_host": "192.0.2.1"},
+            ),
+            (
+                "web.example.com",
+                {"netbox_dns_name": "web.example.com", "ansible_host": "192.0.2.2"},
+            ),
+            ("db.other.org", {"netbox_dns_name": "db.other.org", "ansible_host": "192.0.2.3"}),
+            ("192.0.2.4", {"netbox_dns_name": "", "ansible_host": "192.0.2.4"}),
+        ]
+
+    def test_named_keeps_only_dns_names_deduplicated(self):
+        result = inv.apply_client_side_filters(self.entries(), "named", "true")
+        assert [name for name, _ in result] == ["web.example.com", "db.other.org"]
+
+    def test_named_false_keeps_everything(self):
+        result = inv.apply_client_side_filters(self.entries(), "named", "false")
+        assert len(result) == 4
+
+    def test_domain_filters_by_suffix(self):
+        result = inv.apply_client_side_filters(self.entries(), "domain", "example.com")
+        assert [name for name, _ in result] == ["web.example.com", "web.example.com"]
+
+    def test_no_filter_keeps_everything(self):
+        result = inv.apply_client_side_filters(self.entries(), "", "")
+        assert len(result) == 4
+
+
+class TestLoadConfig:
+    def write_config(self, tmp_path, content):
+        path = tmp_path / "config.yaml"
+        path.write_text(content, encoding="utf-8")
+        return str(path)
+
+    def test_reads_file(self, tmp_path, monkeypatch):
+        monkeypatch.delenv("NETBOX_URL", raising=False)
+        path = self.write_config(
+            tmp_path,
+            "netbox:\n  url: https://nb.example.com\n  token: abc\n"
+            "inventory:\n  source: virtual-machines\n  filter: cluster\n  filter_value: c1\n",
+        )
+        config = inv.load_config(path)
+        assert config["url"] == "https://nb.example.com"
+        assert config["source"] == "virtual-machines"
+        assert config["filter"] == "cluster"
+        assert config["filter_value"] == "c1"
+        assert config["verify_ssl"] is True
+        assert config["timeout"] == 30
+
+    def test_environment_overrides_file(self, tmp_path, monkeypatch):
+        path = self.write_config(tmp_path, "netbox:\n  url: https://nb.example.com\n  token: abc\n")
+        monkeypatch.setenv("NETBOX_INVENTORY_FILTER", "tenant")
+        monkeypatch.setenv("NETBOX_INVENTORY_FILTER_VALUE", "customer1")
+        config = inv.load_config(path)
+        assert config["filter"] == "tenant"
+        assert config["filter_value"] == "customer1"
+
+    def test_missing_url_and_token_raise(self, tmp_path, monkeypatch):
+        monkeypatch.delenv("NETBOX_URL", raising=False)
+        monkeypatch.delenv("NETBOX_TOKEN", raising=False)
+        path = self.write_config(tmp_path, "netbox: {}\n")
+        with pytest.raises(inv.InventoryError, match="url and token"):
+            inv.load_config(path)
+
+    def test_ip_version_defaults_to_both(self, tmp_path):
+        path = self.write_config(tmp_path, "netbox:\n  url: https://nb.example.com\n  token: abc\n")
+        assert inv.load_config(path)["ip_version"] == "both"
+
+    def test_ip_version_env_override_is_normalised(self, tmp_path, monkeypatch):
+        path = self.write_config(tmp_path, "netbox:\n  url: https://nb.example.com\n  token: abc\n")
+        monkeypatch.setenv("NETBOX_INVENTORY_IP_VERSION", " V4 ")
+        assert inv.load_config(path)["ip_version"] == "v4"
+
+    def test_invalid_timeout_raises(self, tmp_path):
+        path = self.write_config(
+            tmp_path,
+            "netbox:\n  url: https://nb.example.com\n  token: abc\n  timeout: soon\n",
+        )
+        with pytest.raises(inv.InventoryError, match="timeout"):
+            inv.load_config(path)
+
+
+class TestNetBoxClient:
+    @responses.activate
+    def test_get_all_follows_pagination(self):
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/",
+            json={
+                "next": f"{NETBOX_URL}/api/ipam/ip-addresses/?offset=1",
+                "results": [{"id": 1}],
+            },
+        )
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/?offset=1",
+            json={"next": None, "results": [{"id": 2}]},
+        )
+        results = make_client().get_all("ipam/ip-addresses")
+        assert [r["id"] for r in results] == [1, 2]
+        assert responses.calls[0].request.headers["Authorization"] == "Token token"
+
+    @responses.activate
+    def test_error_response_raises(self):
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/",
+            json={"detail": "Invalid token."},
+            status=403,
+        )
+        with pytest.raises(inv.NetBoxAPIError, match="Invalid token") as excinfo:
+            make_client().get_all("ipam/ip-addresses")
+        assert excinfo.value.status_code == 403
+
+    @responses.activate
+    def test_resolve_id_returns_first_match(self):
+        responses.get(
+            f"{NETBOX_URL}/api/tenancy/tenants/",
+            json={"next": None, "results": [{"id": 7, "name": "customer1"}]},
+        )
+        assert make_client().resolve_id("tenancy/tenants", "customer1", "tenant") == 7
+
+    @responses.activate
+    def test_resolve_id_unknown_name_raises(self):
+        responses.get(f"{NETBOX_URL}/api/tenancy/tenants/", json={"next": None, "results": []})
+        with pytest.raises(inv.NetBoxAPIError, match="No tenant named"):
+            make_client().resolve_id("tenancy/tenants", "nope", "tenant")
+
+
+class TestBuildInventory:
+    @responses.activate
+    def test_full_ip_inventory(self):
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/",
+            json={
+                "next": None,
+                "results": [
+                    {"address": "192.0.2.10/24", "dns_name": "web.example.com"},
+                    {"address": "192.0.2.11/24", "dns_name": ""},
+                ],
+            },
+        )
+        config = {"source": "ip-addresses", "filter": "", "filter_value": ""}
+        inventory = inv.build_inventory(config, make_client())
+        assert inventory["netbox"]["hosts"] == ["192.0.2.11", "web.example.com"]
+        hostvars = inventory["_meta"]["hostvars"]
+        assert hostvars["web.example.com"]["ansible_host"] == "192.0.2.10"
+        assert json.dumps(inventory)  # must be JSON serialisable
+
+    @responses.activate
+    def test_duplicate_dns_names_fall_back_to_address(self):
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/",
+            json={
+                "next": None,
+                "results": [
+                    {"address": "192.0.2.10/24", "dns_name": "web.example.com"},
+                    {"address": "192.0.2.11/24", "dns_name": "web.example.com"},
+                ],
+            },
+        )
+        config = {"source": "ip-addresses", "filter": "", "filter_value": ""}
+        inventory = inv.build_inventory(config, make_client())
+        assert inventory["netbox"]["hosts"] == ["192.0.2.11", "web.example.com"]
+
+    @responses.activate
+    def test_ip_version_becomes_family_parameter(self):
+        responses.get(
+            f"{NETBOX_URL}/api/ipam/ip-addresses/",
+            json={
+                "next": None,
+                "results": [{"address": "2001:db8::10/64", "dns_name": "v6.example.com"}],
+            },
+        )
+        config = {"source": "ip-addresses", "filter": "", "filter_value": "", "ip_version": "v6"}
+        inventory = inv.build_inventory(config, make_client())
+        assert inventory["netbox"]["hosts"] == ["v6.example.com"]
+        assert "family=6" in responses.calls[0].request.url
+
+    @responses.activate
+    def test_vm_inventory_with_cluster_filter(self):
+        responses.get(
+            f"{NETBOX_URL}/api/virtualization/clusters/",
+            json={"next": None, "results": [{"id": 3, "name": "cluster1"}]},
+        )
+        responses.get(
+            f"{NETBOX_URL}/api/virtualization/virtual-machines/",
+            json={
+                "next": None,
+                "results": [
+                    {
+                        "name": "vm01",
+                        "cluster": {"name": "cluster1"},
+                        "primary_ip": {"address": "192.0.2.20/24"},
+                    }
+                ],
+            },
+        )
+        config = {"source": "virtual-machines", "filter": "cluster", "filter_value": "cluster1"}
+        inventory = inv.build_inventory(config, make_client())
+        assert inventory["netbox"]["hosts"] == ["vm01"]
+        vm_request = responses.calls[1].request
+        assert "cluster_id=3" in vm_request.url
+        assert "exclude=config_context" in vm_request.url