From 0929400cf8d3141ff6037749558b9c4c5a22221d Mon Sep 17 00:00:00 2001 From: Paul Laffitte Date: Fri, 7 Feb 2025 16:41:19 +0100 Subject: [PATCH] PVEControlTestcase --- src/tests/fixtures/api.py | 13 ++- src/tests/sanitycheck/test_vm_backups.py | 104 ++++++++-------------- src/tests/test_cluster.py | 108 +++++------------------ src/tests/testcase.py | 63 +++++++++++++ 4 files changed, 130 insertions(+), 158 deletions(-) create mode 100644 src/tests/testcase.py diff --git a/src/tests/fixtures/api.py b/src/tests/fixtures/api.py index e409547..075e24b 100644 --- a/src/tests/fixtures/api.py +++ b/src/tests/fixtures/api.py @@ -32,14 +32,19 @@ def side_effect(method, url, **kwargs): return side_effect -def create_response_wrapper(nodes, vms, backup_jobs=None, storage_content=None): +def create_response_wrapper(nodes, vms, backup_jobs, storage_content): routes = generate_routes(nodes, vms, backup_jobs, storage_content) - def wrapper(path, **kwargs): + def wrapper(path, data=None, **kwargs): kwargs["params"] = kwargs.get("params", {}) url = "https://host:8006" + path - content = execute_route(routes, "GET", url, **kwargs) - responses.get(url, body=content) + + if data is None: + body = execute_route(routes, "GET", url, **kwargs) + else: + body = json.dumps({"data": data}) + + responses.get(url, body=body) return wrapper diff --git a/src/tests/sanitycheck/test_vm_backups.py b/src/tests/sanitycheck/test_vm_backups.py index ad519ac..57c7f29 100644 --- a/src/tests/sanitycheck/test_vm_backups.py +++ b/src/tests/sanitycheck/test_vm_backups.py @@ -1,72 +1,40 @@ from unittest.mock import patch -from datetime import datetime, timedelta -from pvecontrol.cluster import PVECluster + +import responses + from pvecontrol.sanitycheck.tests.vm_backups import VmBackups from pvecontrol.sanitycheck import SanityCheck from pvecontrol.sanitycheck.checks import CheckCode -from tests.fixtures.api import mock_api_requests, fake_node, fake_vm, fake_backup_job, fake_backup - - -@patch("proxmoxer.backends.https.ProxmoxHTTPAuth") -@patch("proxmoxer.backends.https.ProxmoxHttpSession.request") -def test_sanitycheck_vm_backups(request, _proxmox_http_auth): - nodes = [ - fake_node(3, True), - fake_node(4, True), - ] - vms = [ - fake_vm(100, nodes[0]), - fake_vm(101, nodes[0]), - fake_vm(102, nodes[1]), - fake_vm(103, nodes[1]), - ] - backup_jobs = [ - fake_backup_job(1, "100"), - fake_backup_job(2, "101"), - fake_backup_job(3, "102"), - ] - storages_contents = [ - fake_backup("s3", 100, datetime.now() - timedelta(minutes=110)), - fake_backup("s3", 101, datetime.now() - timedelta(minutes=90)), - ] - - request.side_effect = mock_api_requests(nodes, vms, backup_jobs, storages_contents) - - proxmox = PVECluster( - "name", - "host", - config={ - "node": { - "cpufactor": 2.5, - "memoryminimum": 81928589934592, - }, - "vm": { - "max_last_backup": 100, - }, - }, - **{"user": "user", "password": "password"}, - timeout=1, - ) - - vm_backups_check = VmBackups(proxmox) - vm_backups_check.run() - - def assert_message(message, expected_code, *message_contains): - assert message.code == expected_code - for string in message_contains: - assert string in message.message - - sc = SanityCheck(proxmox) - with patch.object(sc, "_checks", new=[vm_backups_check]): - exitcode = sc.get_exit_code() - sc.display() - - assert exitcode == 1 - assert len(vm_backups_check.messages) == 7 - assert_message(vm_backups_check.messages[0], CheckCode.OK, "vm-100", "is associated") - assert_message(vm_backups_check.messages[1], CheckCode.OK, "vm-101", "is associated") - assert_message(vm_backups_check.messages[2], CheckCode.OK, "vm-102", "is associated") - assert_message(vm_backups_check.messages[3], CheckCode.CRIT, "vm-103", "not associated") - assert_message(vm_backups_check.messages[4], CheckCode.WARN, "vm-100", "more than") - assert_message(vm_backups_check.messages[5], CheckCode.OK, "vm-101", "less than") - assert_message(vm_backups_check.messages[6], CheckCode.WARN, "vm-102", "never") +from tests.testcase import PVEControlTestcase + + +class PVEClusterTestcase(PVEControlTestcase): + + @responses.activate + def test_check(self): + self.responses_get("/api2/json/cluster/backup") + self.responses_get("/api2/json/nodes/pve-devel-1/storage/s3/content", params={"content": "backup"}) + + vm_backups_check = VmBackups(self.cluster) + vm_backups_check.run() + + def assert_message(message, expected_code, *message_contains): + assert message.code == expected_code + for string in message_contains: + assert string in message.message + + sc = SanityCheck(self.cluster) + with patch.object(sc, "_checks", new=[vm_backups_check]): + exitcode = sc.get_exit_code() + sc.display() + + assert exitcode == 1 + assert len(vm_backups_check.messages) == 8 + assert_message(vm_backups_check.messages[0], CheckCode.OK, "vm-100", "is associated") + assert_message(vm_backups_check.messages[1], CheckCode.OK, "vm-101", "is associated") + assert_message(vm_backups_check.messages[2], CheckCode.OK, "vm-102", "is associated") + assert_message(vm_backups_check.messages[3], CheckCode.CRIT, "vm-103", "not associated") + assert_message(vm_backups_check.messages[4], CheckCode.CRIT, "vm-104", "not associated") + assert_message(vm_backups_check.messages[5], CheckCode.WARN, "vm-100", "more than") + assert_message(vm_backups_check.messages[6], CheckCode.OK, "vm-101", "less than") + assert_message(vm_backups_check.messages[7], CheckCode.WARN, "vm-102", "never") diff --git a/src/tests/test_cluster.py b/src/tests/test_cluster.py index f80b10a..f1d18ec 100644 --- a/src/tests/test_cluster.py +++ b/src/tests/test_cluster.py @@ -1,94 +1,30 @@ -from unittest.mock import patch +from tests.testcase import PVEControlTestcase -import responses -from pvecontrol.cluster import PVECluster -from tests.fixtures.api import fake_node, fake_vm -from tests.fixtures.api import create_response_wrapper, mock_api_requests +class PVEClusterTestcase(PVEControlTestcase): + def test_find_node(self): + assert len(self.cluster.nodes) == len(self.nodes) + assert len(self.cluster.vms) == len(self.vms) + assert len(self.cluster.nodes[0].vms) == 2 + assert len(self.cluster.nodes[1].vms) == 3 -@patch("proxmoxer.backends.https.ProxmoxHTTPAuth") -@responses.activate -def test_pvecluster_find_node(_proxmox_http_auth): - nodes = [ - fake_node(1, True), - fake_node(2, True), - ] - vms = [ - fake_vm(100, nodes[0]), - fake_vm(101, nodes[0]), - fake_vm(102, nodes[1]), - ] + for n in self.nodes: + node_object = self.cluster.find_node(n["status"]["name"]) + assert node_object.node == n["status"]["name"] - responses_get = create_response_wrapper(nodes, vms) + def test_find_nodes(self): + node_objects = self.cluster.find_nodes("*devel-1") + assert len(node_objects) == 1 + assert node_objects[0].node == "pve-devel-1" - responses_get("/api2/json/cluster/status") - responses_get("/api2/json/cluster/resources") + node_objects = self.cluster.find_nodes("pve-devel-*") + assert len(node_objects) == len(self.nodes) - cluster = PVECluster( - "name", "host", config={"node": "node"}, verify_ssl=False, **{"user": "user", "password": "password"}, timeout=1 - ) - cluster_vms = cluster.vms + node_objects = self.cluster.find_nodes("*pve-devel-[13]") + assert len(node_objects) == 2 + assert node_objects[0].node == "pve-devel-1" + assert node_objects[1].node == "pve-devel-3" - assert len(cluster.nodes) == len(nodes) - assert len(cluster_vms) == len(vms) - assert len(cluster.nodes[0].vms) == 2 - assert len(cluster.nodes[1].vms) == 1 - - for n in nodes: - node_object = cluster.find_node(n["status"]["name"]) - assert node_object.node == n["status"]["name"] - - -@patch("proxmoxer.backends.https.ProxmoxHTTPAuth") -@patch("proxmoxer.backends.https.ProxmoxHttpSession.request") -def test_pvecluster_find_nodes(request, _proxmox_http_auth): - nodes = [ - fake_node(1, True), - fake_node(2, True), - fake_node(3, True), - ] - vms = [] - - request.side_effect = mock_api_requests(nodes, vms) - - cluster = PVECluster( - "name", "host", config={"node": "node"}, verify_ssl=False, **{"user": "user", "password": "password"}, timeout=1 - ) - - node_objects = cluster.find_nodes("*devel-1") - assert len(node_objects) == 1 - assert node_objects[0].node == "pve-devel-1" - - node_objects = cluster.find_nodes("pve-devel-*") - assert len(node_objects) == len(nodes) - - node_objects = cluster.find_nodes("*pve-devel-[13]") - assert len(node_objects) == 2 - assert node_objects[0].node == "pve-devel-1" - assert node_objects[1].node == "pve-devel-3" - - node_objects = cluster.find_nodes("*prod*") - assert len(node_objects) == 0 - - -@patch("proxmoxer.backends.https.ProxmoxHTTPAuth") -@patch("proxmoxer.backends.https.ProxmoxHttpSession.request") -def test_pvecluster_http_call_made_on_initstatus(request, _proxmox_http_auth): - nodes = [ - fake_node(1, True), - fake_node(2, True), - ] - vms = [ - fake_vm(100, nodes[0]), - fake_vm(101, nodes[0]), - fake_vm(102, nodes[1]), - ] - - request.side_effect = mock_api_requests(nodes, vms) - - _cluster = PVECluster( - "name", "host", config={"node": "node"}, verify_ssl=False, **{"user": "user", "password": "password"}, timeout=1 - ) - - assert request.call_count == 2 # status and ressources + node_objects = self.cluster.find_nodes("*prod*") + assert len(node_objects) == 0 diff --git a/src/tests/testcase.py b/src/tests/testcase.py new file mode 100644 index 0000000..c88916e --- /dev/null +++ b/src/tests/testcase.py @@ -0,0 +1,63 @@ +import unittest + +from unittest.mock import patch +from datetime import datetime, timedelta + +import responses + +from tests.fixtures.api import create_response_wrapper, fake_node, fake_vm, fake_backup, fake_backup_job +from pvecontrol.cluster import PVECluster + + +class PVEControlTestcase(unittest.TestCase): + + @responses.activate + def setUp(self): + nodes = [ + fake_node(1, True), + fake_node(2, True), + fake_node(3, True), + ] + self.nodes = nodes + self.vms = [ + fake_vm(100, nodes[0]), + fake_vm(101, nodes[0]), + fake_vm(102, nodes[1]), + fake_vm(103, nodes[1]), + fake_vm(104, nodes[1]), + ] + self.backup_jobs = [ + fake_backup_job(1, "100"), + fake_backup_job(2, "101"), + fake_backup_job(3, "102"), + ] + self.storages_contents = [ + fake_backup("s3", 100, datetime.now() - timedelta(minutes=110)), + fake_backup("s3", 101, datetime.now() - timedelta(minutes=90)), + ] + + self.responses_get = create_response_wrapper(self.nodes, self.vms, self.backup_jobs, self.storages_contents) + + self.responses_get("/api2/json/cluster/status") + self.responses_get("/api2/json/cluster/resources") + + with patch("proxmoxer.backends.https.ProxmoxHTTPAuth") as mock_auth: + mock_auth_instance = mock_auth.return_value + mock_auth_instance.timeout = 1 + + self.cluster = PVECluster( + "name", + "host", + config={ + "node": { + "cpufactor": 2.5, + "memoryminimum": 81928589934592, + }, + "vm": { + "max_last_backup": 100, + }, + }, + verify_ssl=False, + **{"user": "user", "password": "password"}, + timeout=mock_auth_instance.timeout, + )