From c7c92dc0fb9840ff9956a5df10d7538c5db0bc38 Mon Sep 17 00:00:00 2001 From: Robert Kaussow Date: Sun, 27 Feb 2022 17:56:08 +0100 Subject: [PATCH] chore: add unit tests for ip address gathering (#157) --- prometheuspvesd/discovery.py | 37 +++-- prometheuspvesd/test/fixtures/fixtures.py | 175 ++++++++++++++++---- prometheuspvesd/test/unit/test_discovery.py | 43 +++-- 3 files changed, 191 insertions(+), 64 deletions(-) diff --git a/prometheuspvesd/discovery.py b/prometheuspvesd/discovery.py index b81289d..32f607f 100644 --- a/prometheuspvesd/discovery.py +++ b/prometheuspvesd/discovery.py @@ -86,15 +86,6 @@ class Discovery(): return variables 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 ipv6_address = False networks = False @@ -102,9 +93,9 @@ class Discovery(): # If qemu agent is enabled, try to gather the IP address try: PVE_REQUEST_COUNT_TOTAL.inc() - if self.client.nodes(pve_node).get(pve_type, vmid, "agent", "info") is not None: - networks = self.client.nodes(pve_node).get( - "qemu", vmid, "agent", "network-get-interfaces" + if self.client.get("nodes", pve_node, pve_type, vmid, "agent", "info") is not None: + networks = self.client.get( + "nodes", pve_node, "qemu", vmid, "agent", "network-get-interfaces" )["result"] except Exception: # noqa # nosec pass @@ -120,7 +111,7 @@ class Discovery(): if not ipv4_address: try: 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(): sources = [config["net0"], config["ipconfig0"]] else: @@ -137,7 +128,7 @@ class Discovery(): if not ipv6_address: try: 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(): sources = [config["net0"], config["ipconfig0"]] else: @@ -169,17 +160,27 @@ class Discovery(): filtered.append(item.copy()) 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() def propagate(self): self.host_list.clear() 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: 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() - 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 PVE_REQUEST_COUNT_ERROR_TOTAL.inc() raise APIError(str(e)) @@ -200,7 +201,7 @@ class Discovery(): pve_type = "qemu" 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: description = (config["description"]) diff --git a/prometheuspvesd/test/fixtures/fixtures.py b/prometheuspvesd/test/fixtures/fixtures.py index 00cc469..ca17c29 100644 --- a/prometheuspvesd/test/fixtures/fixtures.py +++ b/prometheuspvesd/test/fixtures/fixtures.py @@ -1,42 +1,147 @@ """Global pytest fixtures.""" 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 def qemus(): - return [{ - "diskwrite": 0, - "vmid": "100", - "name": "100.example.com", - "cpu": 0.0202130478509556, - "diskread": 0, - "template": "", - "uptime": 3101505, - "maxdisk": 26843545600, - "maxmem": 1073741824, - "pid": "1765", - "cpus": 1, - "netin": 2856071643, - "disk": 0, - "status": "running", - "netout": 12159205236, - "mem": 496179157 - }, { - "diskwrite": 0, - "vmid": "101", - "name": "101.example.com", - "cpu": 0.0202130478509556, - "diskread": 0, - "template": "", - "uptime": 3101505, - "maxdisk": 26843545600, - "maxmem": 1073741824, - "pid": "1765", - "cpus": 1, - "netin": 2856071643, - "disk": 0, - "status": "running", - "netout": 12159205236, - "mem": 496179157 - }] + return [ + { + "diskwrite": 0, + "vmid": "100", + "name": "100.example.com", + "cpu": 0.0202130478509556, + "diskread": 0, + "template": "", + "uptime": 3101505, + "maxdisk": 26843545600, + "maxmem": 1073741824, + "pid": "1765", + "cpus": 1, + "netin": 2856071643, + "disk": 0, + "status": "running", + "netout": 12159205236, + "mem": 496179157 + }, + { + "diskwrite": 0, + "vmid": "101", + "name": "101.example.com", + "cpu": 0.0202130478509556, + "diskread": 0, + "template": "", + "uptime": 3101505, + "maxdisk": 26843545600, + "maxmem": 1073741824, + "pid": "1765", + "cpus": 1, + "netin": 2856071643, + "disk": 0, + "status": "running", + "netout": 12159205236, + "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 + } + }, + ] diff --git a/prometheuspvesd/test/unit/test_discovery.py b/prometheuspvesd/test/unit/test_discovery.py index b4a1c28..5397482 100644 --- a/prometheuspvesd/test/unit/test_discovery.py +++ b/prometheuspvesd/test/unit/test_discovery.py @@ -1,22 +1,20 @@ """Test Autostop class.""" -import pytest -from proxmoxer import ProxmoxAPI - -from prometheuspvesd import discovery - pytest_plugins = [ "prometheuspvesd.test.fixtures.fixtures", ] -@pytest.fixture -def discovery_fixture(mocker): - mocker.patch.object( - discovery.Discovery, "_auth", return_value=mocker.create_autospec(ProxmoxAPI) - ) +def get_mock(*args): + networks = args[0] + args = args[1:] - 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): @@ -26,3 +24,26 @@ def test_exclude(discovery_fixture, qemus): filtered = discovery_fixture._exclude(qemus) 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"], + )