You've already forked ansible-netbox-inventory
Bryan Joshua Pedini
362e71bff1
pytest suite with mocked NetBox API responses, plus a makefile driving venv setup, tests, linting and formatting.
311 lines
12 KiB
Python
311 lines
12 KiB
Python
# 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
|