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 IP
192.0.2.20 named vm01.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) 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",
        }
        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"}

    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"}

    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"}


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"]
        assert vm01["ansible_host"] == "192.0.2.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"}


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