# 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"])