From 362e71bff1f12ee6c4584f6ced0c71ad2ac20f22 Mon Sep 17 00:00:00 2001 From: Bryan Joshua Pedini Date: Tue, 25 Aug 2026 13:11:20 +0200 Subject: [PATCH] test: unit tests for config, filters, client and inventory building pytest suite with mocked NetBox API responses, plus a makefile driving venv setup, tests, linting and formatting. --- conftest.py | 3 + makefile | 35 +++++ netbox_inventory.py | 8 +- pytest.ini | 5 + tests/test_inventory.py | 310 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 355 insertions(+), 6 deletions(-) create mode 100644 conftest.py create mode 100644 makefile create mode 100644 pytest.ini create mode 100644 tests/test_inventory.py diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..012d1af --- /dev/null +++ b/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.""" diff --git a/makefile b/makefile new file mode 100644 index 0000000..842cbfa --- /dev/null +++ b/makefile @@ -0,0 +1,35 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# Copyright (C) 2026 Bryan Joshua Pedini +.PHONY: help dev-environment test lint lint-fix format clean + +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" + +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 . + +clean: + rm -rf $(VENV) .pytest_cache .ruff_cache __pycache__ tests/__pycache__ diff --git a/netbox_inventory.py b/netbox_inventory.py index ecccdd9..0bea7b5 100755 --- a/netbox_inventory.py +++ b/netbox_inventory.py @@ -100,9 +100,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: @@ -235,9 +233,7 @@ def validate_options(source: str, filter_name: str, filter_value: str) -> None: raise InventoryError(f'Filter "{filter_name}" requires a value.') -def server_side_params( - client: NetBoxClient, filter_name: str, filter_value: str -) -> dict[str, Any]: +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")} diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..7f8332b --- /dev/null +++ b/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 diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..e28cbd8 --- /dev/null +++ b/tests/test_inventory.py @@ -0,0 +1,310 @@ +# 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 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", "") + + +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 + + +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_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_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