test(sanitycheck): vm backups

pyproject.toml

@@ -70,17 +70,21 @@ exclude = '''
 '''
 
 [tool.pytest.ini_options]
-# this allows us to import pvecontrol directly from tests
 pythonpath = [
-    'src',
+    # this allows to import pvecontrol directly from tests
+    "src",
+    # this allows to import modules from the test package without relative paths
+    ".",
 ]
 
 [tool.pylint.MAIN]
-# this hook allows us to import pvecontrol directly from tests
 init-hook = '''
 import sys
 import os
+# this hook allows to import pvecontrol directly from tests
 sys.path.insert(0, str(os.getcwd() + "/src"))
+# this allows to import modules from the test package without relative paths
+sys.path.insert(0, str(os.getcwd() + "/"))
 '''
 
 [tool.pylint.'MESSAGES CONTROL']

src/pvecontrol/utils.py

@@ -3,6 +3,7 @@
 import sys
 import re
 import curses
+import os
 
 from collections import OrderedDict
 from enum import Enum
@@ -21,6 +22,9 @@ class Fonts:
 
 
 def terminal_support_colors():
+    if os.getenv("NO_COLOR"):
+        return False
+
     try:
         _stdscr = curses.initscr()
         curses.start_color()

tests/__init__.py

@@ -0,0 +1,3 @@
+import os
+
+os.environ["NO_COLOR"] = "1"

tests/fixtures/api.py

@@ -2,15 +2,19 @@
 import requests
 
 
-def mock_api_requests(nodes, vms):
-    routes = generate_routes(nodes, vms)
+def mock_api_requests(nodes, vms, backup_jobs=[], storage_content=[]):
+    routes = generate_routes(nodes, vms, backup_jobs, storage_content)
 
-    def side_effect(method, url, *args, **kwargs):
+    def side_effect(method, url, **kwargs):
         print(f"{method} {url}")
+        print(f"params: {kwargs['params']}")
         path = url.replace("https://host:8006", "")
         assert path in routes
 
-        content = json.dumps({"data": routes[path]})
+        route = routes[path]
+        data = route(method, **kwargs) if callable(route) else route
+
+        content = json.dumps({"data": data})
         print(content + "\n")
 
         res = requests.Response()
@@ -21,14 +25,16 @@ def side_effect(method, url, *args, **kwargs):
     return side_effect
 
 
-def generate_routes(nodes, vms):
+def generate_routes(nodes, vms, backup_jobs, storages_contents):
+    storage_resources = [storage_resource("s3", n["status"]["name"]) for n in nodes]
     routes = {
         "/api2/json/cluster/status": [
             {"type": "cluster", "version": 2, "quorate": 1, "nodes": len(nodes), "id": "cluster", "name": "devel"},
             *[n["status"] for n in nodes],
         ],
         "/api2/json/cluster/resources": [
             *[n["resource"] for n in nodes],
+            *storage_resources,
             *vms,
         ],
         "/api2/json/nodes": [
@@ -38,7 +44,9 @@ def generate_routes(nodes, vms):
         "/api2/json/cluster/ha/groups": [],
         "/api2/json/cluster/ha/status/manager_status": [],
         "/api2/json/cluster/ha/resources": [],
+        "/api2/json/cluster/backup": backup_jobs,
         **generate_vm_routes(nodes, vms),
+        **generate_storages_contents_routes(nodes, storage_resources, storages_contents),
     }
 
     print("ROUTES:")
@@ -107,6 +115,28 @@ def generate_vm_routes(nodes, vms):
     return routes
 
 
+def generate_storages_contents_routes(nodes, storage_resources, storages_contents):
+    routes = {}
+
+    for node in nodes:
+        node_name = node["status"]["name"]
+        for storage in storage_resources:
+
+            def my_route(method, params={}, **kwargs):
+                items = []
+                for item in storages_contents:
+                    storage_filter = item["volid"].split(":")[0] == storage["storage"]
+                    content_filter = "content" not in params or item["content"] == params["content"]
+                    if storage_filter and content_filter:
+                        items.append(item)
+                return items
+
+            storage_name = storage["storage"]
+            routes[f"/api2/json/nodes/{node_name}/storage/{storage_name}/content"] = my_route
+
+    return routes
+
+
 def node(id, local=False):
     resource_id = f"node/pve-devel-{id}"
     name = f"pve-devel-{id}"
@@ -165,3 +195,57 @@ def vm(id, node, status="running"):
         "maxcpu": 1,
         "type": "qemu",
     }
+
+
+def backup_job(job_id, vmid):
+    return {
+        "id": f"backup-d71917f0-{job_id:04x}",
+        "prune-backups": {"keep-last": "3"},
+        "storage": "local",
+        "notes-template": "{{guestname}}",
+        "schedule": "sun 01:00",
+        "fleecing": {"enabled": "0"},
+        "enabled": 1,
+        "type": "vzdump",
+        "next-run": 1735430400,
+        "mode": "snapshot",
+        "vmid": vmid,
+        "compress": "zstd",
+    }
+
+
+def storage_resource(name, node_name):
+    return {
+        "content": "snippets,images,iso,backup,rootdir,vztmpl",
+        "id": f"storage/{node_name}/{name}",
+        "disk": 0,
+        "storage": name,
+        "shared": 1,
+        "status": "available",
+        "maxdisk": 33601372160,
+        "type": "storage",
+        "node": node_name,
+        "plugintype": "s3",
+    }
+
+
+def storage_content(storage, volid, content, ctime, format, options):
+    return {
+        "volid": f"{storage}:{content}/{volid}",
+        "content": content,
+        "ctime": ctime,
+        "format": format,
+        "size": 1124800,
+        **options,
+    }
+
+
+def backup(storage, vmid, created_at):
+    created_at_str = created_at.strftime("%Y_%m_%d-%H_%M_%S")
+    volid = f"vz-dump-qemu-{vmid}-{created_at_str}.vma.zst"
+    options = {
+        "vmid": vmid,
+        "notes": f"VM {vmid}",
+        "subtype": "qemu",
+    }
+    return storage_content(storage, volid, "backup", int(created_at.timestamp()), "vma.zst", options)

tests/sanitycheck/test_vm_backups.py

@@ -0,0 +1,75 @@
+from unittest.mock import patch
+from pvecontrol.cluster import PVECluster
+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, node, vm, backup_job, backup
+from datetime import datetime, timedelta
+
+
+@patch("proxmoxer.backends.https.ProxmoxHTTPAuth")
+@patch("proxmoxer.backends.https.ProxmoxHttpSession.request")
+def test_sanitycheck_vm_backups(request, _proxmox_http_auth):
+    nodes = [
+        node(1, True),
+        node(2, True),
+    ]
+    vms = [
+        vm(100, nodes[0]),
+        vm(101, nodes[0]),
+        vm(102, nodes[1]),
+        vm(103, nodes[1]),
+    ]
+    backup_jobs = [
+        backup_job(1, "100"),
+        backup_job(2, "101"),
+        backup_job(3, "102"),
+    ]
+    storages_contents = [
+        backup("s3", 100, datetime.now() - timedelta(minutes=110)),
+        backup("s3", 101, datetime.now() - timedelta(minutes=90)),
+    ]
+
+    request.side_effect = mock_api_requests(nodes, vms, backup_jobs, storages_contents)
+
+    proxmox = PVECluster(
+        "name",
+        "host",
+        "username",
+        "password",
+        config={
+            "node": {
+                "cpufactor": 2.5,
+                "memoryminimum": 81928589934592,
+            },
+            "sanitycheck": {
+                VmBackups.id: {
+                    "max_last_backup_in_minutes": 100,
+                },
+            },
+        },
+        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")