justin/OpenVixDiskLib
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: use pytest instead of unittest

Browse Source 
pytest provides session scoped fixtures, making it much easier
to reuse resources across tests.
This commit is contained in:
Lucian Petrut
2026-09-07 11:52:46 +00:00
parent 4ddf0c015d
commit 1e9c2816ef
15 changed files with 403 additions and 454 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+10 -10
View File
@@ -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
README.md
+6 -6
View File
@@ -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`.
docs/nfc_auth.md
+1 -1
View File
@@ -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`)
docs/nfc_open.md
+1 -1
View File
@@ -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
docs/nfc_read.md
+1 -1
View File
@@ -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
docs/reverse_engineering_procedure.md
+3 -3
View File
@@ -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.
test-requirements.txt
+1 -1
View File
@@ -1,4 +1,4 @@
coverage
discover
ddt
stestr
pytest
tests/integration/base.py
+242 -278
View File
@@ -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)
tests/integration/test_crosscheck.py
+68 -72
View File
@@ -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)
tests/integration/test_nfc_auth.py
+9 -9
View File
@@ -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()
tests/integration/test_nfc_open.py
+11 -11
View File
@@ -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
tests/integration/test_nfc_read_write.py
+21 -23
View File
@@ -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])
tests/integration/test_openvixdisklib.py
+14 -14
View File
@@ -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
tests/integration/test_vddk.py
+12 -17
View File
@@ -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
tox.ini
+3 -7
View File
@@ -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