ansible-netbox-inventory/e2e/test_e2e.py

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