tests: mock api, run tests in CI and add test_PVECluster_find_node

.github/workflows/tests.yml

@@ -23,3 +23,24 @@ jobs:
         pip install black
     - name: Run black
       run: python -m black . --check
+
+  # Run tests
+  code:
+    runs-on: ubuntu-latest
+    concurrency: tests
+
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 0
+
+    - name: Set up Python
+      uses: actions/setup-python@v3
+      with:
+        python-version: '3.10'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt -r requirements-dev.txt
+    - name: Run black
+      run: python -m pytest

requirements-dev.txt

@@ -0,0 +1 @@
+pytest

tests/fixtures/api.py

@@ -0,0 +1,167 @@
+import json
+import requests
+
+
+def mock_api_requests(nodes, vms):
+    routes = generate_routes(nodes, vms)
+
+    def side_effect(method, url, *args, **kwargs):
+        print(f"{method} {url}")
+        path = url.replace("https://host:8006", "")
+        assert path in routes
+
+        content = json.dumps({"data": routes[path]})
+        print(content + "\n")
+
+        res = requests.Response()
+        res.status_code = 200
+        res._content = content.encode("utf-8")
+        return res
+
+    return side_effect
+
+
+def generate_routes(nodes, vms):
+    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],
+            *vms,
+        ],
+        "/api2/json/nodes": [
+            *[n["resource"] for n in nodes],
+        ],
+        "/api2/json/cluster/tasks": [],
+        "/api2/json/cluster/ha/groups": [],
+        "/api2/json/cluster/ha/status/manager_status": [],
+        "/api2/json/cluster/ha/resources": [],
+        **generate_vm_routes(nodes, vms),
+    }
+
+    print("ROUTES:")
+    for route_path in routes.keys():
+        print(route_path)
+    print("")
+
+    return routes
+
+
+def generate_vm_routes(nodes, vms):
+    routes = {}
+
+    for node in nodes:
+        name = node["status"]["name"]
+        routes[f"/api2/json/nodes/{name}/qemu"] = []
+
+    for vm in vms:
+        node_name = vm["node"]
+        id = vm["vmid"]
+        routes[f"/api2/json/nodes/{node_name}/qemu/{id}/config"] = {
+            "memory": "1024",
+            "vmgenid": "00000000-0000-0000-0000-000000000000",
+            "template": 1,
+            "scsihw": "virtio-scsi-single",
+            "serial0": "socket",
+            "balloon": 0,
+            "onboot": 1,
+            "ide2": "local:9012/vm-9012-cloudinit.qcow2,media=cdrom",
+            "agent": "1",
+            "cores": 1,
+            "numa": 1,
+            "digest": "0000000000000000000000000000000000000000",
+            "smbios1": "uuid=00000000-0000-0000-0000-000000000000",
+            "boot": "order=scsi0;net0",
+            "ostype": "l26",
+            "sockets": 1,
+            "machine": "q35",
+            "net0": "virtio=00:00:00:00:00:00,bridge=vmbr0",
+            "cpu": "x86-64-v2-AES",
+            "rng0": "source=/dev/urandom",
+            "scsi0": "local:9012/base-9012-disk-0.qcow2,size=2G,ssd=1",
+            "name": "template.debian-12-bookworm-amd64",
+        }
+        routes[f"/api2/json/nodes/{node_name}/qemu"].append(
+            {
+                "name": vm["name"],
+                "maxmem": vm["maxmem"],
+                "uptime": vm["uptime"],
+                "vmid": vm["vmid"],
+                "mem": vm["mem"],
+                "disk": vm["disk"],
+                "cpu": vm["cpu"],
+                "maxdisk": vm["maxdisk"],
+                "diskread": vm["diskread"],
+                "netout": vm["netout"],
+                "netin": vm["netin"],
+                "diskwrite": vm["diskwrite"],
+                "status": vm["status"],
+                "serial": 1,
+                "pid": 454971,
+                "cpus": 1,
+            }
+        )
+
+    return routes
+
+
+def node(id, local=False):
+    resource_id = f"node/pve-devel-{id}"
+    name = f"pve-devel-{id}"
+    return {
+        "status": {
+            "id": resource_id,
+            "nodeid": id,
+            "name": name,
+            "ip": f"10.42.24.{id}",
+            "local": 1 if local else 0,
+            "type": "node",
+            "online": 1,
+            "level": "",
+        },
+        "resource": {
+            "id": resource_id,
+            "node": name,
+            "maxmem": 202758361088,
+            "disk": 20973391872,
+            "mem": 4133466112,
+            "uptime": 1987073,
+            "maxdisk": 33601372160,
+            "cpu": 0.00572599602193961,
+            "type": "node",
+            "status": "online",
+            "level": "",
+            "maxcpu": 32,
+            # only in /api2/json/cluster/resources
+            "cgroup-mode": 2,
+            "hastate": "online",
+            # only in /api2/json/nodes
+            "ssl_fingerprint": ":".join(["00"] * 32),
+        },
+    }
+
+
+def vm(id, node, status="running"):
+    return {
+        "id": f"qemu/{id}",
+        "vmid": id,
+        "name": f"vm-{id}",
+        "node": node["status"]["name"],
+        "status": status,
+        "diskread": 0,
+        "mem": 292823173,
+        "disk": 0,
+        "maxmem": 1073741824,
+        "maxdisk": 2147483648,
+        "uptime": 869492,
+        "diskwrite": 2405421056,
+        "netout": 4896058,
+        "cpu": 0.00581464923852424,
+        "netin": 7215771,
+        "template": 0,
+        "hastate": "started",
+        "maxcpu": 1,
+        "type": "qemu",
+    }

tests/test_cluster.py

@@ -0,0 +1,31 @@
+from unittest.mock import patch
+from pvecontrol.cluster import PVECluster
+from tests.fixtures.api import mock_api_requests, node, vm
+
+
+@patch("proxmoxer.backends.https.ProxmoxHttpSession.request")
+@patch("proxmoxer.backends.https.ProxmoxHTTPAuth")
+def test_PVECluster_find_node(ProxmoxHTTPAuth, request):
+    nodes = [
+        node(1, True),
+        node(2, True),
+    ]
+    vms = [
+        vm(100, nodes[0]),
+        vm(101, nodes[0]),
+        vm(102, nodes[1]),
+    ]
+
+    request.side_effect = mock_api_requests(nodes, vms)
+
+    cluster = PVECluster("name", "host", "username", "password", None)
+    cluster_vms = cluster.vms()
+
+    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"]

src/pvecontrol/test_pvecontrol.py → tests/test_pvecontrol.py

@@ -1,7 +1,7 @@
 from pvecontrol.utils import filter_keys
 
 
-def testfilter_keys():
+def test_filter_keys():
     input = {"test": "toto", "none": "noninclude"}
     filter = ["test"]
     assert filter_keys(input, filter) == {"test": "toto"}