prometheus-pve-sd/prometheuspvesd/test/unit/test_discovery.py

"""Test Discovery class."""

import pytest
from proxmoxer import ProxmoxAPI

from prometheuspvesd.client import ProxmoxClient
from prometheuspvesd.discovery import Discovery

pytest_plugins = [
    "prometheuspvesd.test.fixtures.fixtures",
]


@pytest.fixture
def discovery(mocker):
    mocker.patch.object(ProxmoxClient, "_auth", return_value=mocker.create_autospec(ProxmoxAPI))

    return Discovery()


def test_exclude_vmid(discovery, qemus):
    discovery.config.config["exclude_vmid"] = ["100", "101", "102"]
    filtered = discovery._filter(qemus)

    assert len(filtered) == 0


def test_exclude_state(discovery, qemus):
    discovery.config.config["exclude_state"] = ["prelaunch"]
    filtered = discovery._filter(qemus)

    assert len(filtered) == 2


def test_exclude_tags(discovery, qemus):
    discovery.config.config["exclude_tags"] = ["unmonitored"]
    filtered = discovery._filter(qemus)

    assert len(filtered) == 2


@pytest.mark.parametrize(
    "testinput,expected", [
        (["monitored"], 1),
        (["monitored", "postgres"], 2),
        ([], 3),
    ]
)
def test_include_tags(discovery, qemus, testinput, expected):
    discovery.config.config["include_tags"] = testinput
    filtered = discovery._filter(qemus)

    assert len(filtered) == expected


@pytest.mark.parametrize("testinput,expected", [
    (["101", "100"], 2),
    ([], 3),
])
def test_include_vmid(discovery, qemus, testinput, expected):
    discovery.config.config["include_vmid"] = testinput
    filtered = discovery._filter(qemus)

    assert len(filtered) == expected


def test_include_and_exclude_tags(discovery, qemus):
    discovery.config.config["include_tags"] = ["postgres"]
    discovery.config.config["exclude_tags"] = ["unmonitored"]
    filtered = discovery._filter(qemus)

    assert len(filtered) == 0


def test_validate_ip(discovery, addresses):
    # IPv4 validation
    for address in addresses["ipv4_valid"]:
        assert discovery._validate_ip(address)
    for address in addresses["ipv4_invalid"]:
        assert not discovery._validate_ip(address)

    # IPv6 validation
    for address in addresses["ipv6_valid"]:
        assert discovery._validate_ip(address)
    for address in addresses["ipv6_invalid"]:
        assert not discovery._validate_ip(address)


def test_get_ip_addresses(mocker, discovery, networks):
    mocker.patch.object(ProxmoxClient, "get_network_interfaces", return_value=networks)

    assert discovery._get_ip_addresses("qemu", "dummy", "dummy") == (
        networks[1]["ip-addresses"][0]["ip-address"],
        networks[1]["ip-addresses"][2]["ip-address"],
    )


def test_get_ip_addresses_from_instance_config(mocker, discovery, instance_config):
    mocker.patch.object(ProxmoxClient, "get_network_interfaces", return_value=[])
    mocker.patch.object(ProxmoxClient, "get_instance_config", return_value=instance_config)

    assert discovery._get_ip_addresses("qemu", "dummy", "dummy") == (
        "192.0.2.25",
        "2001:db8::666:77:8888",
    )


def test_propagate(
    mocker, discovery, nodes, qemus, instance_config, agent_info, networks, inventory
):
    mocker.patch.object(ProxmoxClient, "get_nodes", return_value=nodes)
    mocker.patch.object(ProxmoxClient, "get_all_vms", return_value=qemus)
    mocker.patch.object(ProxmoxClient, "get_instance_config", return_value=instance_config)
    mocker.patch.object(ProxmoxClient, "get_agent_info", return_value=agent_info)
    mocker.patch.object(ProxmoxClient, "get_network_interfaces", return_value=networks)

    assert discovery.propagate() == inventory