improve tests robustness

Author: evakhoni

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

@@ -21,7 +21,7 @@
 from jumpstarter_driver_opendal.driver import FlasherInterface
 from jumpstarter_driver_power.driver import PowerInterface, PowerReading
 from jumpstarter_driver_pyserial.driver import PySerial
-from pydantic import BaseModel, ByteSize, Field, TypeAdapter, validate_call
+from pydantic import BaseModel, ByteSize, Field, TypeAdapter, ValidationError, validate_call
 from qemu.qmp import QMPClient
 from qemu.qmp.protocol import ConnectError, Runstate
 
@@ -183,9 +183,7 @@ async def on(self) -> None:  # noqa: C901
 
             # Resize disk if configured
             if self.parent.disk_size:
-                # Convert QEMU binary format (20G) to Pydantic format (20GiB)
-                s = self.parent.disk_size
-                requested = int(TypeAdapter(ByteSize).validate_python(s + "iB" if s[-1] in "kmgtKMGT" else s))
+                requested = self.parent._parse_size(self.parent.disk_size)
 
                 if requested < current_virtual_size:
                     raise RuntimeError(
@@ -406,24 +404,23 @@ def get_username(self) -> str:
     def get_password(self) -> str:
         return self.password
 
-    def _validate_size(self, size: str) -> None:
-        """Validate size string (e.g., '20G', '512M')."""
+    def _parse_size(self, size: str) -> int:
+        """Parse size string (e.g., '20G') to bytes."""
         try:
-            s = size
-            TypeAdapter(ByteSize).validate_python(s + "iB" if s[-1] in "kmgtKMGT" else s)
-        except Exception:
+            return int(TypeAdapter(ByteSize).validate_python(size + "iB" if size[-1] in "kmgtKMGT" else size))
+        except (ValidationError, IndexError):
             raise ValueError(f"Invalid size: '{size}'. Use e.g. '20G', '512M', '2T'") from None
 
     @export
     @validate_call(validate_return=True)
     def set_disk_size(self, size: str) -> None:
         """Set the disk size for resizing before boot."""
-        self._validate_size(size)
+        self._parse_size(size)  # Validate
         self.disk_size = size
 
     @export
     @validate_call(validate_return=True)
     def set_memory_size(self, size: str) -> None:
         """Set the memory size for next boot."""
-        self._validate_size(size)
+        self._parse_size(size)  # Validate
         self.mem = size

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

@@ -4,6 +4,7 @@
 import sys
 import tarfile
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -146,7 +147,7 @@ async def test_resize_insufficient_space_blocked(resize_test):
     """Resize beyond available host space should raise RuntimeError."""
     driver, current = resize_test("100G", 10)  # requested: 100G, current: 10G
 
-    mock_usage = type("U", (), {"free": 5 * 1024**3})()  # only 5G free
+    mock_usage = SimpleNamespace(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):
@@ -158,20 +159,20 @@ async def test_resize_insufficient_space_blocked(resize_test):
 async def test_resize_succeeds(resize_test):
     """Resize should call qemu-img resize with correct size."""
     driver, current = resize_test("20G", 10)  # requested: 20G, current: 10G
-    mock_usage = type("U", (), {"free": 50 * 1024**3})()
+    mock_usage = SimpleNamespace(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:
+            # Mock Popen to stop before actually starting QEMU VM
+            with patch("jumpstarter_driver_qemu.driver.Popen", side_effect=RuntimeError("mock popen")):
+                with pytest.raises(RuntimeError, match="mock popen"):
                     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]]
+    # Find the resize call and verify size argument
+    resize_calls = [c for c in mock_run.call_args_list if "resize" in c.args[0]]
     assert resize_calls, "qemu-img resize should be called"
-    assert resize_calls[0][0][0][-1] == str(20 * 1024**3)
+    resize_cmd = resize_calls[0].args[0]  # ['qemu-img', 'resize', path, size]
+    assert resize_cmd[-1] == str(20 * 1024**3)
 
 
 def test_set_disk_size_valid():