github/prometheus-pve-sd
0
0
Fork 0
mirror of https://github.com/thegeeklab/prometheus-pve-sd.git synced 2024-11-25 10:50:39 +00:00
Code Issues Projects Releases Wiki Activity

chore: add unit tests for ip address gathering (#157)

Browse Source
This commit is contained in:
Robert Kaussow 2022-02-27 17:56:08 +01:00 committed by GitHub
parent 573384e6dc
commit c7c92dc0fb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 191 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
prometheuspvesd/discovery.py
View File

@ -86,15 +86,6 @@ class Discovery():
return variables return variables
def _get_ip_addresses(self, pve_type, pve_node, vmid): def _get_ip_addresses(self, pve_type, pve_node, vmid):
def validate_ip(address: object) -> object:
try:
if not ipaddress.ip_address(address).is_loopback \
and not ipaddress.ip_address(address).is_link_local:
return address
except ValueError:
return False
ipv4_address = False ipv4_address = False
ipv6_address = False ipv6_address = False
networks = False networks = False
@ -102,9 +93,9 @@ class Discovery():
# If qemu agent is enabled, try to gather the IP address # If qemu agent is enabled, try to gather the IP address
try: try:
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
if self.client.nodes(pve_node).get(pve_type, vmid, "agent", "info") is not None: if self.client.get("nodes", pve_node, pve_type, vmid, "agent", "info") is not None:
networks = self.client.nodes(pve_node).get( networks = self.client.get(
"qemu", vmid, "agent", "network-get-interfaces" "nodes", pve_node, "qemu", vmid, "agent", "network-get-interfaces"
)["result"] )["result"]
except Exception: # noqa # nosec except Exception: # noqa # nosec
pass pass
@ -120,7 +111,7 @@ class Discovery():
if not ipv4_address: if not ipv4_address:
try: try:
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
config = self.client.nodes(pve_node).get(pve_type, vmid, "config") config = self.client.get("nodes", pve_node, pve_type, vmid, "config")
if "ipconfig0" in config.keys(): if "ipconfig0" in config.keys():
sources = [config["net0"], config["ipconfig0"]] sources = [config["net0"], config["ipconfig0"]]
else: else:
@ -137,7 +128,7 @@ class Discovery():
if not ipv6_address: if not ipv6_address:
try: try:
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
config = self.client.nodes(pve_node).get(pve_type, vmid, "config") config = self.client.get("nodes", pve_node, pve_type, vmid, "config")
if "ipconfig0" in config.keys(): if "ipconfig0" in config.keys():
sources = [config["net0"], config["ipconfig0"]] sources = [config["net0"], config["ipconfig0"]]
else: else:
@ -169,17 +160,27 @@ class Discovery():
filtered.append(item.copy()) filtered.append(item.copy())
return filtered return filtered
def _validate_ip(self, address: object) -> object:
try:
if (
not ipaddress.ip_address(address).is_loopback
and not ipaddress.ip_address(address).is_link_local
):
return address
except ValueError:
return False
@PROPAGATION_TIME.time() @PROPAGATION_TIME.time()
def propagate(self): def propagate(self):
self.host_list.clear() self.host_list.clear()
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
for node in self._get_names(self.client.nodes.get(), "node"): for node in self._get_names(self.client.get("nodes"), "node"):
try: try:
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
qemu_list = self._exclude(self.client.nodes(node).qemu.get()) qemu_list = self._exclude(self.client.get("nodes", node, "qemu"))
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
container_list = self._exclude(self.client.nodes(node).lxc.get()) container_list = self._exclude(self.client.get("nodes", node, "lxc"))
except Exception as e: # noqa except Exception as e: # noqa
PVE_REQUEST_COUNT_ERROR_TOTAL.inc() PVE_REQUEST_COUNT_ERROR_TOTAL.inc()
raise APIError(str(e)) raise APIError(str(e))
@ -200,7 +201,7 @@ class Discovery():
pve_type = "qemu" pve_type = "qemu"
PVE_REQUEST_COUNT_TOTAL.inc() PVE_REQUEST_COUNT_TOTAL.inc()
config = self.client.nodes(node).get(pve_type, vmid, "config") config = self.client.get("nodes", node, pve_type, vmid, "config")
try: try:
description = (config["description"]) description = (config["description"])

111
prometheuspvesd/test/fixtures/fixtures.py vendored
View File

@ -1,11 +1,24 @@
"""Global pytest fixtures.""" """Global pytest fixtures."""
import pytest import pytest
from proxmoxer import ProxmoxAPI
from prometheuspvesd import discovery
@pytest.fixture
def discovery_fixture(mocker):
mocker.patch.object(
discovery.Discovery, "_auth", return_value=mocker.create_autospec(ProxmoxAPI)
)
return discovery.Discovery()
@pytest.fixture @pytest.fixture
def qemus(): def qemus():
return [{ return [
{
"diskwrite": 0, "diskwrite": 0,
"vmid": "100", "vmid": "100",
"name": "100.example.com", "name": "100.example.com",
@ -22,7 +35,8 @@ def qemus():
"status": "running", "status": "running",
"netout": 12159205236, "netout": 12159205236,
"mem": 496179157 "mem": 496179157
}, { },
{
"diskwrite": 0, "diskwrite": 0,
"vmid": "101", "vmid": "101",
"name": "101.example.com", "name": "101.example.com",
@ -39,4 +53,95 @@ def qemus():
"status": "running", "status": "running",
"netout": 12159205236, "netout": 12159205236,
"mem": 496179157 "mem": 496179157
}] },
]
@pytest.fixture
def addresses():
return {
"ipv4_valid": [
"192.168.0.1",
"10.0.0.1",
],
"ipv4_invalid": [
"127.0.0.1",
"169.254.1.1",
],
"ipv6_valid": [
"2001:db8:3333:4444:5555:6666:7777:8888",
"2001:db8:3333:4444:CCCC:DDDD:EEEE:FFFF",
"::",
"2001:db8::",
"::1234:5678",
"2001:db8::1234:5678",
"2001:0db8:0001:0000:0000:0ab9:C0A8:0102",
],
"ipv6_invalid": [
"::1",
"fe80::903a:1c1a:e802:11e4",
],
}
@pytest.fixture
def networks():
return [
{
"hardware-address": "00:00:00:00:00:00",
"ip-addresses": [
{
"ip-address": "127.0.0.1",
"ip-address-type": "ipv4",
"prefix": 8
},
{
"ip-address": "::1",
"ip-address-type": "ipv6",
"prefix": 128
},
],
"name": "lo",
"statistics": {
"rx-bytes": 9280,
"rx-dropped": 0,
"rx-errs": 0,
"rx-packets": 92,
"tx-bytes": 9280,
"tx-dropped": 0,
"tx-errs": 0,
"tx-packets": 92
}
},
{
"hardware-address": "92:0b:bd:c1:f8:39",
"ip-addresses": [
{
"ip-address": "10.168.0.1",
"ip-address-type": "ipv4",
"prefix": 32
},
{
"ip-address": "10.168.0.2",
"ip-address-type": "ipv4",
"prefix": 32
},
{
"ip-address": "2001:cdba:3333:4444:5555:6666:7777:8888",
"ip-address-type": "ipv6",
"prefix": 64
},
],
"name": "eth0",
"statistics": {
"rx-bytes": 2861070337,
"rx-dropped": 0,
"rx-errs": 0,
"rx-packets": 18065580,
"tx-bytes": 12185866619,
"tx-dropped": 0,
"tx-errs": 0,
"tx-packets": 14423878
}
},
]

43
prometheuspvesd/test/unit/test_discovery.py
View File

@ -1,22 +1,20 @@
"""Test Autostop class.""" """Test Autostop class."""
import pytest
from proxmoxer import ProxmoxAPI
from prometheuspvesd import discovery
pytest_plugins = [ pytest_plugins = [
"prometheuspvesd.test.fixtures.fixtures", "prometheuspvesd.test.fixtures.fixtures",
] ]
@pytest.fixture def get_mock(*args):
def discovery_fixture(mocker): networks = args[0]
mocker.patch.object( args = args[1:]
discovery.Discovery, "_auth", return_value=mocker.create_autospec(ProxmoxAPI)
)
return discovery.Discovery() if "info" in args:
return True
if "network-get-interfaces" in args:
return {"result": networks}
return False
def test_exclude(discovery_fixture, qemus): def test_exclude(discovery_fixture, qemus):
@ -26,3 +24,26 @@ def test_exclude(discovery_fixture, qemus):
filtered = discovery_fixture._exclude(qemus) filtered = discovery_fixture._exclude(qemus)
assert filtered == expected assert filtered == expected
def test_validate_ip(discovery_fixture, addresses):
# IPv4 validation
for address in addresses["ipv4_valid"]:
assert discovery_fixture._validate_ip(address)
for address in addresses["ipv4_invalid"]:
assert not discovery_fixture._validate_ip(address)
# IPv6 validation
for address in addresses["ipv6_valid"]:
assert discovery_fixture._validate_ip(address)
for address in addresses["ipv6_invalid"]:
assert not discovery_fixture._validate_ip(address)
def test_get_ip_addresses(mocker, discovery_fixture, networks):
discovery_fixture.client.get.side_effect = lambda *args: get_mock(networks, *args)
assert discovery_fixture._get_ip_addresses("qemu", "dummy", "dummy") == (
networks[1]["ip-addresses"][0]["ip-address"],
networks[1]["ip-addresses"][2]["ip-address"],
)