mirror of
https://github.com/thegeeklab/prometheus-pve-sd.git
synced 2024-11-01 08:00:38 +00:00
135 lines
3.9 KiB
Python
135 lines
3.9 KiB
Python
"""Test Discovery class."""
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
from proxmoxer import ProxmoxAPI
|
|
|
|
from prometheuspvesd.client import ProxmoxClient
|
|
from prometheuspvesd.discovery import Discovery
|
|
|
|
pytest_plugins = [
|
|
"prometheuspvesd.test.fixtures.fixtures",
|
|
]
|
|
|
|
|
|
def records_to_messages(records):
|
|
return [r.getMessage() for r in records]
|
|
|
|
|
|
@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, local_caplog):
|
|
discovery.config.config["exclude_tags"] = ["unmonitored"]
|
|
|
|
with local_caplog(level=logging.DEBUG) as caplog:
|
|
filtered = discovery._filter(qemus)
|
|
|
|
assert (
|
|
"vmid 100: discovered tags: ['unmonitored', 'excluded', 'postgres']"
|
|
in records_to_messages(caplog.records)
|
|
)
|
|
assert "vmid 100: excluded by tags: ['unmonitored']"
|
|
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
|