PVEControlTestcase

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
 

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")

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

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,
+            )