added tests

Author: evakhoni

packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py

@@ -1,8 +1,10 @@
+import json
 import os
 import platform
 import sys
 import tarfile
 from pathlib import Path
+from unittest.mock import AsyncMock, patch
 
 import pytest
 import requests
@@ -13,6 +15,12 @@
 from jumpstarter.common.utils import serve
 
 
+@pytest.fixture
+def anyio_backend():
+    """Use only asyncio backend for anyio tests."""
+    return "asyncio"
+
+
 @pytest.fixture(scope="session")
 def ovmf(tmpdir_factory):
     tmp_path = tmpdir_factory.mktemp("ovmf")
@@ -91,3 +99,66 @@ def test_driver_qemu(tmp_path, ovmf):
             assert s.run("uname -r").stdout.strip() == f"6.11.4-301.fc41.{arch}"
 
         qemu.power.off()
+
+
+def _setup_resize_test(disk_size, current_size_gb):
+    """Create a Qemu driver with a sparse root disk."""
+    driver = Qemu(disk_size=disk_size)
+    root = Path(driver._tmp_dir.name) / "root"
+    root.write_bytes(b"")
+    os.truncate(root, current_size_gb * 1024**3)
+    return driver, current_size_gb * 1024**3
+
+
+def _mock_qemu_img_info(virtual_size):
+    """Return a mock for run_process that simulates qemu-img info."""
+    async def mock(cmd, **kwargs):
+        result = AsyncMock()
+        result.returncode = 0
+        result.stdout = json.dumps({"format": "raw", "virtual-size": virtual_size}).encode()
+        result.check_returncode = lambda: None
+        return result
+    return mock
+
+
+@pytest.mark.anyio
+async def test_resize_shrink_blocked():
+    """Shrinking disk should raise RuntimeError."""
+    driver, current = _setup_resize_test("10G", 20)  # requested: 10G, current: 20G
+
+    with patch("jumpstarter_driver_qemu.driver.run_process", side_effect=_mock_qemu_img_info(current)):
+        with pytest.raises(RuntimeError, match="Shrinking disk is not supported"):
+            await driver.children["power"].on()
+
+
+@pytest.mark.anyio
+async def test_resize_insufficient_space_blocked():
+    """Resize beyond available host space should raise RuntimeError."""
+    driver, current = _setup_resize_test("100G", 10)  # requested: 100G, current: 10G
+
+    mock_usage = type("U", (), {"free": 5 * 1024**3})()  # only 5G free
+
+    with patch("jumpstarter_driver_qemu.driver.run_process", side_effect=_mock_qemu_img_info(current)):
+        with patch("jumpstarter_driver_qemu.driver.shutil.disk_usage", return_value=mock_usage):
+            with pytest.raises(RuntimeError, match="Not enough disk space"):
+                await driver.children["power"].on()
+
+
+@pytest.mark.anyio
+async def test_resize_succeeds():
+    """Resize should call qemu-img resize with correct size."""
+    driver, current = _setup_resize_test("20G", 10)  # requested: 20G, current: 10G
+    mock_usage = type("U", (), {"free": 50 * 1024**3})()
+
+    with patch("jumpstarter_driver_qemu.driver.run_process", side_effect=_mock_qemu_img_info(current)) as mock_run:
+        with patch("jumpstarter_driver_qemu.driver.shutil.disk_usage", return_value=mock_usage):
+            with patch("jumpstarter_driver_qemu.driver.Popen"):
+                try:
+                    await driver.children["power"].on()
+                except Exception:
+                    pass  # Expected to fail at Popen
+
+    # Find the resize call
+    resize_calls = [c for c in mock_run.call_args_list if "resize" in c[0][0]]
+    assert resize_calls, "qemu-img resize should be called"
+    assert resize_calls[0][0][0][-1] == str(20 * 1024**3)