diff --git a/AGENTS.md b/AGENTS.md index caed4dc..774539c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,9 +7,10 @@ - The end goal is to have a Python library that can be used as a VDDK replacement to retrieve VMware disk contents. - Integration tests under `tests/integration/` are a good starting point for - interacting with the VMware NBD / NFC APIs. They inherit lab credentials and - VM/disk settings from `tests.integration.base.TestBase`. The base - class creates a temporary empty VM for the run. We can make use of + interacting with the VMware NBD / NFC APIs. They take a session-scoped + `lab` fixture from `tests/integration/conftest.py` (credentials and VM + settings in `tests.integration.base.LabEnv`). The fixture creates a + temporary empty VM for the pytest session. We can make use of them to reverse engineer the VMware protocol, for example making various calls, capturing the request and replies and then trying to determine the structures used by the protocol. @@ -34,14 +35,13 @@ licensing constrains. - `tests/integration/vixdisklib.py` is a Python wrapper on top of `libvixDiskLib`, used to cross-check the replacement against native VDDK. -- Integration tests live under `tests/integration/`, use the unittest - framework, and inherit `tests.integration.base.TestBase`. Lab vCenter - credentials, datacenter, and datastore come from repo-root - `.test_config.yaml` (gitignored; sample in `README.md`). Each test - class shares a temporary empty VM with a 10 GiB disk created in - `TestBase.setUpClass` and destroyed in `tearDownClass`. Run them with +- Integration tests live under `tests/integration/` and use pytest. + Lab vCenter credentials, datacenter, and datastore come from repo-root + `.test_config.yaml` (gitignored; sample in `README.md`). A session-scoped + fixture creates one temporary empty VM with a 10 GiB disk for the whole + run and destroys it at session end. Run them with `tox -e integration` or - `.venv/bin/python -m unittest discover -s tests/integration`. + `.venv/bin/pytest tests/integration`. ## Other rules diff --git a/README.md b/README.md index 33da552..646bdd1 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,7 @@ VDDK-shaped handle. | `openvixdisklib/openvixdisklib.py` | Drop-in handle (`connect` / `open` / `read` / `write`) | | `openvixdisklib/nfc_auth.py` | VIM login, NFC ticket, authd on 902 | | `openvixdisklib/nfc_open.py` | Classic NFC handshake, AIO open, sector read/write | -| `tests/integration/` | Live unittest suite against a lab vCenter | +| `tests/integration/` | Live pytest suite against a lab vCenter | | `tests/integration/vixdisklib.py` | Native VDDK wrapper used only to cross-check | | `docs/` | Protocol notes and reverse-engineering steps | @@ -85,19 +85,19 @@ datacenter: Datacenter datastore: datastore0 ``` -`TestBase.setUpClass` creates an empty VM with a 10 GiB thin disk on -that datastore and tears it down in `tearDownClass`. Tests write known -patterns and read them back. +A session-scoped pytest fixture creates an empty VM with a 10 GiB thin +disk on that datastore and tears it down when the session ends. Tests +write known patterns and read them back. ```bash tox -e integration # or -.venv/bin/python -m unittest discover -s tests/integration +.venv/bin/pytest tests/integration ``` VDDK cross-check tests skip when `libvixDiskLib` is not loadable from `.vddk`. `tox -e integration` sets `LD_LIBRARY_PATH` to that directory -and clears `LD_PRELOAD`. For a direct unittest run, do the same. +and clears `LD_PRELOAD`. For a direct pytest run, do the same. Lint and typecheck: `tox -e pep8`, `tox -e mypy`. diff --git a/docs/nfc_auth.md b/docs/nfc_auth.md index 1ac8c37..ff7c569 100644 --- a/docs/nfc_auth.md +++ b/docs/nfc_auth.md @@ -251,7 +251,7 @@ hard-coding the lab certificate. Run: ```bash -.venv/bin/python -m unittest tests.integration.test_nfc_auth +.venv/bin/pytest tests/integration/test_nfc_auth.py ``` The test completes VIM login and the authd handshake (`200 Connect`) diff --git a/docs/nfc_open.md b/docs/nfc_open.md index 76324df..c5b9346 100644 --- a/docs/nfc_open.md +++ b/docs/nfc_open.md @@ -206,7 +206,7 @@ classic type 4 `NFC_SESSION_COMPLETE`. Run: ```bash -.venv/bin/python -m unittest tests.integration.test_nfc_open +.venv/bin/pytest tests/integration/test_nfc_open.py ``` The test opens the temporary lab VMDK, asserts an opaque handle and diff --git a/docs/nfc_read.md b/docs/nfc_read.md index 8afccca..d61205c 100644 --- a/docs/nfc_read.md +++ b/docs/nfc_read.md @@ -105,7 +105,7 @@ Writes use the same 44-byte IO payload with opcode `2`; see `openvixdisklib/nfc_open.py`. Run: ```bash -.venv/bin/python -m unittest tests.integration.test_nfc_read_write +.venv/bin/pytest tests/integration/test_nfc_read_write.py ``` The integration test writes and then reads the captured VDDK ranges diff --git a/docs/reverse_engineering_procedure.md b/docs/reverse_engineering_procedure.md index b948eb0..256b399 100644 --- a/docs/reverse_engineering_procedure.md +++ b/docs/reverse_engineering_procedure.md @@ -9,9 +9,9 @@ NFC work can follow the same loop instead of rediscovering it. Scope so far: `VixDiskLib_ConnectEx` + `VixDiskLib_Open` + `VixDiskLib_Read` + `VixDiskLib_Write` against lab vCenter 8.0.1 / -ESXi 8, transport `nbd`. Driver: `tests/integration/` (`TestBase` -creates a temporary empty VM with a 10 GiB disk in `setUpClass` and -destroys it in `tearDownClass`). +ESXi 8, transport `nbd`. Driver: `tests/integration/` (the session-scoped +`lab` fixture creates a temporary empty VM with a 10 GiB disk and +destroys it when the pytest session ends). Rule from `AGENTS.md`: reuse pyVmomi for every public VIM operation. Only reimplement what pyVmomi does not expose. diff --git a/test-requirements.txt b/test-requirements.txt index 8bcd6eb..80f6f15 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,4 +1,4 @@ coverage discover ddt -stestr +pytest diff --git a/tests/integration/base.py b/tests/integration/base.py index 532b9d8..39f21a5 100644 --- a/tests/integration/base.py +++ b/tests/integration/base.py @@ -1,17 +1,18 @@ # Copyright 2026 Cloudbase Solutions Srl # All Rights Reserved. -"""Test base classes for openvixdisklib integration tests.""" +"""Lab helpers for openvixdisklib integration tests.""" from __future__ import annotations import ctypes import os import time -import unittest import uuid +from dataclasses import dataclass from typing import Any, Optional +import pytest import yaml from pyVim.connect import Disconnect from pyVmomi import vim @@ -38,290 +39,49 @@ _TASK_POLL_S = 0.5 _TASK_TIMEOUT_S = 300 _LAB_VM_PREFIX = "ovdl-test-" +VDDK_DIR = _VDDK_DIR +SECTOR_SIZE = 512 +SECTOR_AT_1GB = (1024 * 1024 * 1024) // SECTOR_SIZE -class TestBase(unittest.TestCase): - """Shared lab vSphere settings for live NFC / VDDK integration tests.""" - HOST: str - PORT: int - USERNAME: str - PASSWORD: str - ALLOW_UNTRUSTED: bool - DATACENTER: str - DATASTORE: str - THUMBPRINT: str - VM_MOREF: str - VMX_SPEC: str - DISK_PATH: str - SECTOR_SIZE = 512 - SECTOR_AT_1GB = (1024 * 1024 * 1024) // SECTOR_SIZE - VDDK_DIR = _VDDK_DIR - _lab_refcount = 0 - _lab_vm_moref: Optional[str] = None - _lab_vm_name: Optional[str] = None +@dataclass +class LabEnv: + """vCenter settings and the temporary VM used by a pytest session.""" - @classmethod - def setUpClass(cls) -> None: - """Prepare process environment and create a temporary lab VM.""" - super().setUpClass() - os.environ.pop("LD_PRELOAD", None) - cls._ensure_vddk_library_path() - TestBase._acquire_lab() - - @classmethod - def tearDownClass(cls) -> None: - """Release the temporary lab VM when the last test class finishes.""" - TestBase._release_lab() - super().tearDownClass() - - def setUp(self) -> None: - """Reset per-test state; subclasses may reuse this.""" - super().setUp() - - @classmethod - def _ensure_vddk_library_path(cls) -> None: - current = os.environ.get("LD_LIBRARY_PATH", "") - parts = [p for p in current.split(":") if p] - if cls.VDDK_DIR not in parts: - os.environ["LD_LIBRARY_PATH"] = ( - cls.VDDK_DIR if not current else f"{cls.VDDK_DIR}:{current}") - - @classmethod - def require_vddk(cls) -> None: - """Skip when ``libvixDiskLib`` cannot be loaded from ``.vddk``.""" - cls._ensure_vddk_library_path() - try: - ctypes.CDLL(_VDDK_LIB) - except OSError as exc: - raise unittest.SkipTest( - f"VDDK library not available at {_VDDK_LIB}: {exc}") from exc - - @classmethod - def _load_test_config(cls) -> None: - """Load lab settings from the repo-root ``.test_config.yaml``.""" - if not os.path.isfile(_CONFIG_PATH): - raise unittest.SkipTest( - "integration tests need .test_config.yaml in the repo " - "root; see README.md for a sample") - with open(_CONFIG_PATH, encoding="utf-8") as config_file: - data = yaml.safe_load(config_file) or {} - missing = [key for key in _CONFIG_KEYS if key not in data] - if missing: - raise RuntimeError( - f"{_CONFIG_PATH} is missing keys: {', '.join(missing)}") - TestBase.HOST = str(data["host"]) - TestBase.PORT = int(data["port"]) - TestBase.USERNAME = str(data["username"]) - TestBase.PASSWORD = str(data["password"]) - TestBase.ALLOW_UNTRUSTED = bool(data["allow_untrusted"]) - TestBase.DATACENTER = str(data["datacenter"]) - TestBase.DATASTORE = str(data["datastore"]) - - @classmethod - def _connect_vim(cls) -> vim.ServiceInstance: - return nfc_auth.connect_vim( - cls.HOST, - cls.USERNAME, - cls.PASSWORD, - port=cls.PORT, - thumbprint=cls.THUMBPRINT, - allow_untrusted=cls.ALLOW_UNTRUSTED) - - @classmethod - def _wait_for_task(cls, task: vim.Task) -> Any: - deadline = time.monotonic() + _TASK_TIMEOUT_S - while task.info.state in ( - vim.TaskInfo.State.running, vim.TaskInfo.State.queued): - if time.monotonic() > deadline: - raise TimeoutError( - f"timed out waiting for vSphere task {task}") - time.sleep(_TASK_POLL_S) - if task.info.state != vim.TaskInfo.State.success: - raise RuntimeError(f"vSphere task failed: {task.info.error}") - return task.info.result - - @classmethod - def _find_datacenter( - cls, content: vim.ServiceInstanceContent) -> vim.Datacenter: - matches = [ - entity for entity in content.rootFolder.childEntity - if isinstance(entity, vim.Datacenter) - and entity.name == cls.DATACENTER] - if not matches: - raise RuntimeError(f"datacenter {cls.DATACENTER!r} not found") - return matches[0] - - @classmethod - def _find_datastore(cls, datacenter: vim.Datacenter) -> vim.Datastore: - matches = [ - datastore for datastore in datacenter.datastore - if datastore.name == cls.DATASTORE] - if not matches: - raise RuntimeError( - f"datastore {cls.DATASTORE!r} not found in " - f"datacenter {cls.DATACENTER!r}") - return matches[0] - - @classmethod - def _bind_lab_fields(cls) -> None: - """Copy shared lab VM fields onto the active test class.""" - cls.HOST = TestBase.HOST - cls.PORT = TestBase.PORT - cls.USERNAME = TestBase.USERNAME - cls.PASSWORD = TestBase.PASSWORD - cls.ALLOW_UNTRUSTED = TestBase.ALLOW_UNTRUSTED - cls.DATACENTER = TestBase.DATACENTER - cls.DATASTORE = TestBase.DATASTORE - cls.THUMBPRINT = TestBase.THUMBPRINT - cls.VM_MOREF = TestBase.VM_MOREF - cls.VMX_SPEC = TestBase.VMX_SPEC - cls.DISK_PATH = TestBase.DISK_PATH - - @classmethod - def _acquire_lab(cls) -> None: - if TestBase._lab_refcount == 0: - TestBase._load_test_config() - TestBase.THUMBPRINT = nfc_auth.get_ssl_cert_thumbprint( - TestBase.HOST, TestBase.PORT) - TestBase._create_lab_vm() - TestBase._lab_refcount += 1 - cls._bind_lab_fields() - - @classmethod - def _release_lab(cls) -> None: - if TestBase._lab_refcount == 0: - return - TestBase._lab_refcount -= 1 - if TestBase._lab_refcount == 0: - cls._destroy_lab_vm() - - @classmethod - def _create_lab_vm(cls) -> None: - """Create an empty VM with a 10 GiB thin disk for I/O tests.""" - si = cls._connect_vim() - vm = None - try: - content = si.RetrieveContent() - datacenter = cls._find_datacenter(content) - datastore = cls._find_datastore(datacenter) - if not datastore.host: - raise RuntimeError( - f"datastore {cls.DATASTORE!r} is not mounted on any host") - host = datastore.host[0].key - pool = host.parent.resourcePool - vm_name = _LAB_VM_PREFIX + uuid.uuid4().hex[:12] - vm = cls._wait_for_task( - datacenter.vmFolder.CreateVM_Task( - config=cls._vm_config_spec(vm_name, datastore.name), - pool=pool, - host=host)) - TestBase._lab_vm_moref = vm._moId - TestBase._lab_vm_name = vm_name - TestBase.VM_MOREF = vm._moId - TestBase.VMX_SPEC = f"moref={vm._moId}" - disks = [ - device.backing.fileName - for device in vm.config.hardware.device - if isinstance(device, vim.vm.device.VirtualDisk)] - if not disks: - raise RuntimeError( - f"temporary VM {vm_name!r} has no virtual disks") - TestBase.DISK_PATH = disks[0] - except Exception: - if vm is not None: - try: - cls._wait_for_task(vm.Destroy_Task()) - except Exception: - pass - TestBase._lab_vm_moref = None - TestBase._lab_vm_name = None - raise - finally: - Disconnect(si) - - @classmethod - def _vm_config_spec( - cls, vm_name: str, datastore_name: str) -> vim.vm.ConfigSpec: - config = vim.vm.ConfigSpec() - config.name = vm_name - config.guestId = "otherGuest64" - config.memoryMB = 128 - config.numCPUs = 1 - config.files = vim.vm.FileInfo( - vmPathName=f"[{datastore_name}]") - - controller = vim.vm.device.ParaVirtualSCSIController() - controller.key = 1000 - controller.busNumber = 0 - controller.sharedBus = ( - vim.vm.device.VirtualSCSIController.Sharing.noSharing) - controller_spec = vim.vm.device.VirtualDeviceSpec() - controller_spec.operation = ( - vim.vm.device.VirtualDeviceSpec.Operation.add) - controller_spec.device = controller - - backing = vim.vm.device.VirtualDisk.FlatVer2BackingInfo() - backing.diskMode = "persistent" - backing.thinProvisioned = True - backing.fileName = f"[{datastore_name}]" - disk = vim.vm.device.VirtualDisk() - disk.key = 2000 - disk.controllerKey = 1000 - disk.unitNumber = 0 - disk.capacityInKB = _DISK_CAPACITY_KB - disk.backing = backing - disk_spec = vim.vm.device.VirtualDeviceSpec() - disk_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.add - disk_spec.fileOperation = ( - vim.vm.device.VirtualDeviceSpec.FileOperation.create) - disk_spec.device = disk - - config.deviceChange = [controller_spec, disk_spec] - return config - - @classmethod - def _destroy_lab_vm(cls) -> None: - """Power off and delete the temporary lab VM if it still exists.""" - moref = TestBase._lab_vm_moref - TestBase._lab_vm_moref = None - TestBase._lab_vm_name = None - if not moref: - return - si = cls._connect_vim() - try: - vm = vim.VirtualMachine(moref, si._stub) - try: - vm.Reload() - except Exception: - return - if vm.runtime.powerState == vim.VirtualMachinePowerState.poweredOn: - cls._wait_for_task(vm.PowerOffVM_Task()) - cls._wait_for_task(vm.Destroy_Task()) - finally: - Disconnect(si) + host: str + port: int + username: str + password: str + allow_untrusted: bool + datacenter: str + datastore: str + thumbprint: str + vm_moref: str + vmx_spec: str + disk_path: str def authenticate(self, read_only: bool = True) -> NfcAuthSession: """Login to the lab vCenter and complete NFC authd for the temp VM.""" return nfc_auth.authenticate( - host=self.HOST, - username=self.USERNAME, - password=self.PASSWORD, - vm_moref=self.VM_MOREF, - thumbprint=self.THUMBPRINT, - allow_untrusted=self.ALLOW_UNTRUSTED, - disk_path=None if read_only else self.DISK_PATH, + host=self.host, + username=self.username, + password=self.password, + vm_moref=self.vm_moref, + thumbprint=self.thumbprint, + allow_untrusted=self.allow_untrusted, + disk_path=None if read_only else self.disk_path, read_only=read_only) def vixdisklib_connect_kwargs( self, extra: Optional[dict[str, Any]] = None) -> dict[str, Any]: """Return common ``VixDiskLib_ConnectEx`` arguments for the temp VM.""" kwargs: dict[str, Any] = { - "server_name": self.HOST, - "port": self.PORT, - "thumbprint": self.THUMBPRINT, - "username": self.USERNAME, - "password": self.PASSWORD, - "vmx_spec": self.VMX_SPEC, + "server_name": self.host, + "port": self.port, + "thumbprint": self.thumbprint, + "username": self.username, + "password": self.password, + "vmx_spec": self.vmx_spec, "transport_modes": "nbd", "read_only": False, } @@ -329,8 +89,212 @@ class TestBase(unittest.TestCase): kwargs.update(extra) return kwargs - def pattern_bytes(self, length: int, seed: bytes) -> bytes: - """Return ``length`` bytes by repeating ``seed``.""" - if not seed: - raise ValueError("seed must be non-empty") - return (seed * ((length // len(seed)) + 1))[:length] + +def pattern_bytes(length: int, seed: bytes) -> bytes: + """Return ``length`` bytes by repeating ``seed``.""" + if not seed: + raise ValueError("seed must be non-empty") + return (seed * ((length // len(seed)) + 1))[:length] + + +def ensure_vddk_library_path() -> None: + """Prepend ``.vddk`` to ``LD_LIBRARY_PATH`` if it is not already there.""" + current = os.environ.get("LD_LIBRARY_PATH", "") + parts = [p for p in current.split(":") if p] + if VDDK_DIR not in parts: + os.environ["LD_LIBRARY_PATH"] = ( + VDDK_DIR if not current else f"{VDDK_DIR}:{current}") + + +def require_vddk() -> None: + """Skip when ``libvixDiskLib`` cannot be loaded from ``.vddk``.""" + ensure_vddk_library_path() + try: + ctypes.CDLL(_VDDK_LIB) + except OSError as exc: + pytest.skip(f"VDDK library not available at {_VDDK_LIB}: {exc}") + + +def _load_test_config() -> dict[str, Any]: + if not os.path.isfile(_CONFIG_PATH): + pytest.skip( + "integration tests need .test_config.yaml in the repo " + "root; see README.md for a sample") + with open(_CONFIG_PATH, encoding="utf-8") as config_file: + data = yaml.safe_load(config_file) or {} + missing = [key for key in _CONFIG_KEYS if key not in data] + if missing: + raise RuntimeError( + f"{_CONFIG_PATH} is missing keys: {', '.join(missing)}") + return { + "host": str(data["host"]), + "port": int(data["port"]), + "username": str(data["username"]), + "password": str(data["password"]), + "allow_untrusted": bool(data["allow_untrusted"]), + "datacenter": str(data["datacenter"]), + "datastore": str(data["datastore"]), + } + + +def _connect_vim( + host: str, + username: str, + password: str, + port: int, + thumbprint: str, + allow_untrusted: bool) -> vim.ServiceInstance: + return nfc_auth.connect_vim( + host, + username, + password, + port=port, + thumbprint=thumbprint, + allow_untrusted=allow_untrusted) + + +def _wait_for_task(task: vim.Task) -> Any: + deadline = time.monotonic() + _TASK_TIMEOUT_S + while task.info.state in ( + vim.TaskInfo.State.running, vim.TaskInfo.State.queued): + if time.monotonic() > deadline: + raise TimeoutError( + f"timed out waiting for vSphere task {task}") + time.sleep(_TASK_POLL_S) + if task.info.state != vim.TaskInfo.State.success: + raise RuntimeError(f"vSphere task failed: {task.info.error}") + return task.info.result + + +def _find_datacenter( + content: vim.ServiceInstanceContent, + datacenter_name: str) -> vim.Datacenter: + matches = [ + entity for entity in content.rootFolder.childEntity + if isinstance(entity, vim.Datacenter) + and entity.name == datacenter_name] + if not matches: + raise RuntimeError(f"datacenter {datacenter_name!r} not found") + return matches[0] + + +def _find_datastore( + datacenter: vim.Datacenter, datastore_name: str) -> vim.Datastore: + matches = [ + datastore for datastore in datacenter.datastore + if datastore.name == datastore_name] + if not matches: + raise RuntimeError( + f"datastore {datastore_name!r} not found in " + f"datacenter {datacenter.name!r}") + return matches[0] + + +def _vm_config_spec(vm_name: str, datastore_name: str) -> vim.vm.ConfigSpec: + config = vim.vm.ConfigSpec() + config.name = vm_name + config.guestId = "otherGuest64" + config.memoryMB = 128 + config.numCPUs = 1 + config.files = vim.vm.FileInfo( + vmPathName=f"[{datastore_name}]") + + controller = vim.vm.device.ParaVirtualSCSIController() + controller.key = 1000 + controller.busNumber = 0 + controller.sharedBus = ( + vim.vm.device.VirtualSCSIController.Sharing.noSharing) + controller_spec = vim.vm.device.VirtualDeviceSpec() + controller_spec.operation = ( + vim.vm.device.VirtualDeviceSpec.Operation.add) + controller_spec.device = controller + + backing = vim.vm.device.VirtualDisk.FlatVer2BackingInfo() + backing.diskMode = "persistent" + backing.thinProvisioned = True + backing.fileName = f"[{datastore_name}]" + disk = vim.vm.device.VirtualDisk() + disk.key = 2000 + disk.controllerKey = 1000 + disk.unitNumber = 0 + disk.capacityInKB = _DISK_CAPACITY_KB + disk.backing = backing + disk_spec = vim.vm.device.VirtualDeviceSpec() + disk_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.add + disk_spec.fileOperation = ( + vim.vm.device.VirtualDeviceSpec.FileOperation.create) + disk_spec.device = disk + + config.deviceChange = [controller_spec, disk_spec] + return config + + +def create_lab_vm() -> LabEnv: + """Create an empty VM with a 10 GiB thin disk for I/O tests.""" + cfg = _load_test_config() + thumbprint = nfc_auth.get_ssl_cert_thumbprint(cfg["host"], cfg["port"]) + si = _connect_vim( + cfg["host"], cfg["username"], cfg["password"], cfg["port"], + thumbprint, cfg["allow_untrusted"]) + vm = None + try: + content = si.RetrieveContent() + datacenter = _find_datacenter(content, cfg["datacenter"]) + datastore = _find_datastore(datacenter, cfg["datastore"]) + if not datastore.host: + raise RuntimeError( + f"datastore {cfg['datastore']!r} is not mounted on any host") + host = datastore.host[0].key + pool = host.parent.resourcePool + vm_name = _LAB_VM_PREFIX + uuid.uuid4().hex[:12] + vm = _wait_for_task( + datacenter.vmFolder.CreateVM_Task( + config=_vm_config_spec(vm_name, datastore.name), + pool=pool, + host=host)) + disks = [ + device.backing.fileName + for device in vm.config.hardware.device + if isinstance(device, vim.vm.device.VirtualDisk)] + if not disks: + raise RuntimeError( + f"temporary VM {vm_name!r} has no virtual disks") + return LabEnv( + host=cfg["host"], + port=cfg["port"], + username=cfg["username"], + password=cfg["password"], + allow_untrusted=cfg["allow_untrusted"], + datacenter=cfg["datacenter"], + datastore=cfg["datastore"], + thumbprint=thumbprint, + vm_moref=vm._moId, + vmx_spec=f"moref={vm._moId}", + disk_path=disks[0]) + except Exception: + if vm is not None: + try: + _wait_for_task(vm.Destroy_Task()) + except Exception: + pass + raise + finally: + Disconnect(si) + + +def destroy_lab_vm(lab: LabEnv) -> None: + """Power off and delete the temporary lab VM if it still exists.""" + si = _connect_vim( + lab.host, lab.username, lab.password, lab.port, + lab.thumbprint, lab.allow_untrusted) + try: + vm = vim.VirtualMachine(lab.vm_moref, si._stub) + try: + vm.Reload() + except Exception: + return + if vm.runtime.powerState == vim.VirtualMachinePowerState.poweredOn: + _wait_for_task(vm.PowerOffVM_Task()) + _wait_for_task(vm.Destroy_Task()) + finally: + Disconnect(si) diff --git a/tests/integration/test_crosscheck.py b/tests/integration/test_crosscheck.py index 18156b6..1820832 100644 --- a/tests/integration/test_crosscheck.py +++ b/tests/integration/test_crosscheck.py @@ -7,88 +7,84 @@ from typing import Any, Optional from openvixdisklib import openvixdisklib as open_vix from tests.integration import vixdisklib -from tests.integration.base import TestBase +from tests.integration.base import ( + LabEnv, SECTOR_AT_1GB, SECTOR_SIZE, pattern_bytes) -class CrosscheckTest(TestBase): - @classmethod - def setUpClass(cls) -> None: - """Skip when the bundled VDDK shared library is not present.""" - cls.require_vddk() - super().setUpClass() +def _connect_extra(lab: LabEnv, module: Any) -> Optional[dict[str, Any]]: + """Return extra ``connect`` kwargs needed by ``module``.""" + if module is open_vix: + return {"allow_untrusted": lab.allow_untrusted} + return None - def _connect_extra(self, module: Any) -> Optional[dict[str, Any]]: - """Return extra ``connect`` kwargs needed by ``module``.""" - if module is open_vix: - return {"allow_untrusted": self.ALLOW_UNTRUSTED} - return None - def _write_sectors( - self, - module: Any, - payloads: dict[int, bytes]) -> None: - """Write one sector at each index using a vixdisklib-compatible module.""" - handle = module.VixDiskLibHandle( - vixdisklib_compatibility_version="8.0", - config_path=None) - buf = module.get_buffer(self.SECTOR_SIZE) - kwargs = self.vixdisklib_connect_kwargs(self._connect_extra(module)) - with handle.connect(**kwargs) as conn: - with handle.open(conn, self.DISK_PATH, flags=0) as disk: - for start, data in payloads.items(): - buf[:self.SECTOR_SIZE] = data - handle.write(disk, start, 1, buf) +def _write_sectors( + lab: LabEnv, + module: Any, + payloads: dict[int, bytes]) -> None: + """Write one sector at each index using a vixdisklib-compatible module.""" + handle = module.VixDiskLibHandle( + vixdisklib_compatibility_version="8.0", + config_path=None) + buf = module.get_buffer(SECTOR_SIZE) + kwargs = lab.vixdisklib_connect_kwargs(_connect_extra(lab, module)) + with handle.connect(**kwargs) as conn: + with handle.open(conn, lab.disk_path, flags=0) as disk: + for start, data in payloads.items(): + buf[:SECTOR_SIZE] = data + handle.write(disk, start, 1, buf) - def _read_sectors( - self, - module: Any, - sectors: tuple[int, ...]) -> dict[int, bytes]: - """Read one sector at each index using a vixdisklib-compatible module.""" - handle = module.VixDiskLibHandle( - vixdisklib_compatibility_version="8.0", - config_path=None) - buf = module.get_buffer(self.SECTOR_SIZE) - result: dict[int, bytes] = {} - kwargs = self.vixdisklib_connect_kwargs(self._connect_extra(module)) - with handle.connect(**kwargs) as conn: - with handle.open(conn, self.DISK_PATH, flags=0) as disk: - for start in sectors: - buf[:self.SECTOR_SIZE] = b"\xa5" * self.SECTOR_SIZE - handle.read(disk, start, 1, buf) - result[start] = buf.raw[:self.SECTOR_SIZE] - return result - def _assert_both_read( - self, - sectors: tuple[int, ...], - expected: dict[int, bytes]) -> None: - vddk = self._read_sectors(vixdisklib, sectors) - replacement = self._read_sectors(open_vix, sectors) - for start in sectors: - self.assertEqual( - vddk[start], expected[start], - f"VDDK mismatch at sector {start}") - self.assertEqual( - replacement[start], expected[start], - f"openvixdisklib mismatch at sector {start}") +def _read_sectors( + lab: LabEnv, + module: Any, + sectors: tuple[int, ...]) -> dict[int, bytes]: + """Read one sector at each index using a vixdisklib-compatible module.""" + handle = module.VixDiskLibHandle( + vixdisklib_compatibility_version="8.0", + config_path=None) + buf = module.get_buffer(SECTOR_SIZE) + result: dict[int, bytes] = {} + kwargs = lab.vixdisklib_connect_kwargs(_connect_extra(lab, module)) + with handle.connect(**kwargs) as conn: + with handle.open(conn, lab.disk_path, flags=0) as disk: + for start in sectors: + buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE + handle.read(disk, start, 1, buf) + result[start] = buf.raw[:SECTOR_SIZE] + return result - def test_openvixdisklib_matches_vddk_sectors(self) -> None: + +def _assert_both_read( + lab: LabEnv, + sectors: tuple[int, ...], + expected: dict[int, bytes]) -> None: + vddk_data = _read_sectors(lab, vixdisklib, sectors) + replacement = _read_sectors(lab, open_vix, sectors) + for start in sectors: + assert vddk_data[start] == expected[start], ( + f"VDDK mismatch at sector {start}") + assert replacement[start] == expected[start], ( + f"openvixdisklib mismatch at sector {start}") + + +class TestCrosscheck: + def test_openvixdisklib_matches_vddk_sectors( + self, lab: LabEnv, vddk: None) -> None: """Writes from either library must be visible to both readers.""" - sectors = (0, 1, self.SECTOR_AT_1GB) + sectors = (0, 1, SECTOR_AT_1GB) vddk_payloads = { - 0: self.pattern_bytes(self.SECTOR_SIZE, b"XCHK-VDDK-S0"), - 1: self.pattern_bytes(self.SECTOR_SIZE, b"XCHK-VDDK-S1"), - self.SECTOR_AT_1GB: self.pattern_bytes( - self.SECTOR_SIZE, b"XCHK-VDDK-1G"), + 0: pattern_bytes(SECTOR_SIZE, b"XCHK-VDDK-S0"), + 1: pattern_bytes(SECTOR_SIZE, b"XCHK-VDDK-S1"), + SECTOR_AT_1GB: pattern_bytes(SECTOR_SIZE, b"XCHK-VDDK-1G"), } - self._write_sectors(vixdisklib, vddk_payloads) - self._assert_both_read(sectors, vddk_payloads) + _write_sectors(lab, vixdisklib, vddk_payloads) + _assert_both_read(lab, sectors, vddk_payloads) ovdl_payloads = { - 0: self.pattern_bytes(self.SECTOR_SIZE, b"XCHK-OVDL-S0"), - 1: self.pattern_bytes(self.SECTOR_SIZE, b"XCHK-OVDL-S1"), - self.SECTOR_AT_1GB: self.pattern_bytes( - self.SECTOR_SIZE, b"XCHK-OVDL-1G"), + 0: pattern_bytes(SECTOR_SIZE, b"XCHK-OVDL-S0"), + 1: pattern_bytes(SECTOR_SIZE, b"XCHK-OVDL-S1"), + SECTOR_AT_1GB: pattern_bytes(SECTOR_SIZE, b"XCHK-OVDL-1G"), } - self._write_sectors(open_vix, ovdl_payloads) - self._assert_both_read(sectors, ovdl_payloads) + _write_sectors(lab, open_vix, ovdl_payloads) + _assert_both_read(lab, sectors, ovdl_payloads) diff --git a/tests/integration/test_nfc_auth.py b/tests/integration/test_nfc_auth.py index 96dfef6..5d46e09 100644 --- a/tests/integration/test_nfc_auth.py +++ b/tests/integration/test_nfc_auth.py @@ -3,16 +3,16 @@ """Exercise VDDK-compatible NFC authentication against the lab vCenter.""" -from tests.integration.base import TestBase +from tests.integration.base import LabEnv -class NfcAuthTest(TestBase): - def test_authd_handshake_completes(self) -> None: +class TestNfcAuth: + def test_authd_handshake_completes(self, lab: LabEnv) -> None: """Complete VIM login and authd PROXY through ``200 Connect``.""" - with self.authenticate() as session: + with lab.authenticate() as session: ticket = session.ticket - self.assertTrue(ticket.host) - self.assertTrue(ticket.port) - self.assertTrue(ticket.sessionId) - self.assertTrue(session.authd_sock.version()) - self.assertTrue(session.authd_sock.cipher()) + assert ticket.host + assert ticket.port + assert ticket.sessionId + assert session.authd_sock.version() + assert session.authd_sock.cipher() diff --git a/tests/integration/test_nfc_open.py b/tests/integration/test_nfc_open.py index 255e7d2..874c0cc 100644 --- a/tests/integration/test_nfc_open.py +++ b/tests/integration/test_nfc_open.py @@ -4,20 +4,20 @@ """Exercise NFC disk open and a one-sector write/read against the lab.""" from openvixdisklib import nfc_open -from tests.integration.base import TestBase +from tests.integration.base import LabEnv, SECTOR_SIZE, pattern_bytes -class NfcOpenTest(TestBase): - def test_open_disk_and_read_first_sector(self) -> None: +class TestNfcOpen: + def test_open_disk_and_read_first_sector(self, lab: LabEnv) -> None: """Open the temp VMDK, write sector 0, and read it back.""" - expected = self.pattern_bytes(self.SECTOR_SIZE, b"NFC-OPEN-S0") - with self.authenticate(read_only=False) as session: + expected = pattern_bytes(SECTOR_SIZE, b"NFC-OPEN-S0") + with lab.authenticate(read_only=False) as session: with nfc_open.open_disk( - session, self.DISK_PATH, read_only=False) as disk: - self.assertEqual(disk.path, self.DISK_PATH) - self.assertGreater(disk.handle, 0) - self.assertEqual(disk.sector_size, self.SECTOR_SIZE) + session, lab.disk_path, read_only=False) as disk: + assert disk.path == lab.disk_path + assert disk.handle > 0 + assert disk.sector_size == SECTOR_SIZE disk.write(0, 1, expected) got = disk.read(0, 1) - self.assertIsNot(got, expected) - self.assertEqual(got, expected) + assert got is not expected + assert got == expected diff --git a/tests/integration/test_nfc_read_write.py b/tests/integration/test_nfc_read_write.py index 427eaa1..c9922c7 100644 --- a/tests/integration/test_nfc_read_write.py +++ b/tests/integration/test_nfc_read_write.py @@ -4,11 +4,11 @@ """Exercise NFC sector writes and reads against the lab vCenter.""" from openvixdisklib import nfc_open -from tests.integration.base import TestBase +from tests.integration.base import LabEnv, SECTOR_SIZE, pattern_bytes -class NfcReadWriteTest(TestBase): - def test_sector_writes_and_reads(self) -> None: +class TestNfcReadWrite: + def test_sector_writes_and_reads(self, lab: LabEnv) -> None: """Write known patterns and read them back at several ranges.""" ranges = [ (0, 1), @@ -19,36 +19,34 @@ class NfcReadWriteTest(TestBase): (0, 129), (256, 64), ] - with self.authenticate(read_only=False) as session: + with lab.authenticate(read_only=False) as session: with nfc_open.open_disk( - session, self.DISK_PATH, read_only=False) as disk: + session, lab.disk_path, read_only=False) as disk: for start, n_sectors in ranges: - length = n_sectors * self.SECTOR_SIZE + length = n_sectors * SECTOR_SIZE seed = f"NFC-R{start}:{n_sectors}-".encode() - to_write = self.pattern_bytes(length, seed) + to_write = pattern_bytes(length, seed) disk.write(start, n_sectors, to_write) got = disk.read(start, n_sectors) - self.assertIsNot(got, to_write) - self.assertEqual(len(got), length) - self.assertEqual(got, to_write) + assert got is not to_write + assert len(got) == length + assert got == to_write two_seed = b"NFC-TWO-SECTOR" - two_to_write = self.pattern_bytes( - 2 * self.SECTOR_SIZE, two_seed) + two_to_write = pattern_bytes(2 * SECTOR_SIZE, two_seed) disk.write(0, 2, two_to_write) two_got = disk.read(0, 2) - self.assertIsNot(two_got, two_to_write) - self.assertEqual(two_got, two_to_write) - self.assertEqual( - disk.read(1, 1), two_to_write[self.SECTOR_SIZE:]) + assert two_got is not two_to_write + assert two_got == two_to_write + assert disk.read(1, 1) == two_to_write[SECTOR_SIZE:] big_seed = b"NFC-129-SECTOR-WRITE" - big_to_write = self.pattern_bytes( - 129 * self.SECTOR_SIZE, big_seed) + big_to_write = pattern_bytes( + 129 * SECTOR_SIZE, big_seed) disk.write(0, 129, big_to_write) big_got = disk.read(0, 129) - self.assertIsNot(big_got, big_to_write) - self.assertEqual(big_got, big_to_write) - self.assertEqual( - big_got[self.SECTOR_SIZE:2 * self.SECTOR_SIZE], - big_to_write[self.SECTOR_SIZE:2 * self.SECTOR_SIZE]) + assert big_got is not big_to_write + assert big_got == big_to_write + assert ( + big_got[SECTOR_SIZE:2 * SECTOR_SIZE] + == big_to_write[SECTOR_SIZE:2 * SECTOR_SIZE]) diff --git a/tests/integration/test_openvixdisklib.py b/tests/integration/test_openvixdisklib.py index b1b0f51..9831001 100644 --- a/tests/integration/test_openvixdisklib.py +++ b/tests/integration/test_openvixdisklib.py @@ -4,30 +4,30 @@ """Exercise the VDDK-compatible openvixdisklib handle against the lab.""" from openvixdisklib import openvixdisklib as vixdisklib -from tests.integration.base import TestBase +from tests.integration.base import ( + LabEnv, SECTOR_AT_1GB, SECTOR_SIZE, pattern_bytes) -class OpenVixDiskLibTest(TestBase): - def test_write_and_read_sector_zero_and_one_gib(self) -> None: +class TestOpenvixdisklib: + def test_write_and_read_sector_zero_and_one_gib(self, lab: LabEnv) -> None: """Write then read sector 0 and the sector at a 1 GiB offset.""" handle = vixdisklib.VixDiskLibHandle( vixdisklib_compatibility_version="8.0", config_path=None) - write_buf = vixdisklib.get_buffer(self.SECTOR_SIZE) - read_buf = vixdisklib.get_buffer(self.SECTOR_SIZE) - connect_kwargs = self.vixdisklib_connect_kwargs({ - "allow_untrusted": self.ALLOW_UNTRUSTED, + write_buf = vixdisklib.get_buffer(SECTOR_SIZE) + read_buf = vixdisklib.get_buffer(SECTOR_SIZE) + connect_kwargs = lab.vixdisklib_connect_kwargs({ + "allow_untrusted": lab.allow_untrusted, }) patterns = { - 0: self.pattern_bytes(self.SECTOR_SIZE, b"OVDL-S0"), - self.SECTOR_AT_1GB: self.pattern_bytes( - self.SECTOR_SIZE, b"OVDL-1GB"), + 0: pattern_bytes(SECTOR_SIZE, b"OVDL-S0"), + SECTOR_AT_1GB: pattern_bytes(SECTOR_SIZE, b"OVDL-1GB"), } with handle.connect(**connect_kwargs) as conn: - with handle.open(conn, self.DISK_PATH, flags=0) as disk: + with handle.open(conn, lab.disk_path, flags=0) as disk: for start, expected in patterns.items(): - write_buf[:self.SECTOR_SIZE] = expected + write_buf[:SECTOR_SIZE] = expected handle.write(disk, start, 1, write_buf) - read_buf[:self.SECTOR_SIZE] = b"\xa5" * self.SECTOR_SIZE + read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE handle.read(disk, start, 1, read_buf) - self.assertEqual(read_buf.raw[:self.SECTOR_SIZE], expected) + assert read_buf.raw[:SECTOR_SIZE] == expected diff --git a/tests/integration/test_vddk.py b/tests/integration/test_vddk.py index 466d7dc..2f27016 100644 --- a/tests/integration/test_vddk.py +++ b/tests/integration/test_vddk.py @@ -4,28 +4,23 @@ """Exercise native VDDK via tests.integration.vixdisklib against the lab.""" from tests.integration import vixdisklib -from tests.integration.base import TestBase +from tests.integration.base import LabEnv, SECTOR_SIZE, pattern_bytes -class VddkTest(TestBase): - @classmethod - def setUpClass(cls) -> None: - """Skip when the bundled VDDK shared library is not present.""" - cls.require_vddk() - super().setUpClass() - - def test_write_and_read_first_sector(self) -> None: +class TestVddk: + def test_write_and_read_first_sector( + self, lab: LabEnv, vddk: None) -> None: """Open the temp VMDK with VDDK, write sector 0, and read it back.""" handle = vixdisklib.VixDiskLibHandle( vixdisklib_compatibility_version="8.0", config_path=None) - write_buf = vixdisklib.get_buffer(self.SECTOR_SIZE) - read_buf = vixdisklib.get_buffer(self.SECTOR_SIZE) - expected = self.pattern_bytes(self.SECTOR_SIZE, b"VDDK-S0") - write_buf[:self.SECTOR_SIZE] = expected - with handle.connect(**self.vixdisklib_connect_kwargs()) as conn: - with handle.open(conn, self.DISK_PATH, flags=0) as disk: + write_buf = vixdisklib.get_buffer(SECTOR_SIZE) + read_buf = vixdisklib.get_buffer(SECTOR_SIZE) + expected = pattern_bytes(SECTOR_SIZE, b"VDDK-S0") + write_buf[:SECTOR_SIZE] = expected + with handle.connect(**lab.vixdisklib_connect_kwargs()) as conn: + with handle.open(conn, lab.disk_path, flags=0) as disk: handle.write(disk, 0, 1, write_buf) - read_buf[:self.SECTOR_SIZE] = b"\xa5" * self.SECTOR_SIZE + read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE handle.read(disk, 0, 1, read_buf) - self.assertEqual(read_buf.raw[:self.SECTOR_SIZE], expected) + assert read_buf.raw[:SECTOR_SIZE] == expected diff --git a/tox.ini b/tox.ini index 00a438c..82248d7 100644 --- a/tox.ini +++ b/tox.ini @@ -33,8 +33,7 @@ setenv = LD_LIBRARY_PATH={[vars]vddk_path} LD_PRELOAD= commands = - stestr run --slowest --concurrency 1 \ - --test-path tests/integration/ {posargs} + python -m pytest -v tests/integration/ {posargs} [testenv:integration-coverage] description = Integration tests with coverage report. @@ -44,11 +43,10 @@ setenv = {[testenv:integration]setenv} deps = {[testenv]deps} - pytest pytest-cov pycobertura commands = - pytest --cov=openvixdisklib --cov-report=term-missing \ + python -m pytest --cov=openvixdisklib --cov-report=term-missing \ --cov-report=xml -p no:cacheprovider tests/integration/ {posargs} pycobertura show --format html --source openvixdisklib \ coverage.xml -o coverage.html @@ -73,6 +71,7 @@ commands = description = Type checks (mypy). deps = mypy + pytest commands = mypy {[vars]all_path} @@ -89,6 +88,3 @@ ignore = E125,E251,W503,W504,E305,E731,E117,W605,F632 exclude = .venv,.git,.tox,dist,build,*.egg # Same length as ruff. max-line-length = 88 - -[stestr] -test_path = ./tests