diff --git a/tests/integration/test_openvixdisklib.py b/tests/integration/test_openvixdisklib.py index 5212170..08b8a33 100644 --- a/tests/integration/test_openvixdisklib.py +++ b/tests/integration/test_openvixdisklib.py @@ -3,11 +3,34 @@ """Exercise the VDDK-compatible openvixdisklib handle against the lab.""" +from pyVim.connect import Disconnect +from pyVmomi import vim import pytest +from openvixdisklib import nfc_auth from openvixdisklib import openvixdisklib as vixdisklib from tests.integration.base import ( - LabEnv, SECTOR_AT_1GB, SECTOR_SIZE, pattern_bytes) + LabEnv, SECTOR_AT_1GB, SECTOR_SIZE, _wait_for_task, pattern_bytes) + + +def _connect_lab_vim(lab: LabEnv) -> vim.ServiceInstance: + """Login to the lab vCenter with the session VM's credentials.""" + return nfc_auth.connect_vim( + lab.host, + lab.username, + lab.password, + port=lab.port, + thumbprint=lab.thumbprint, + allow_untrusted=lab.allow_untrusted) + + +def _virtual_disk_backing( + vm: vim.VirtualMachine) -> vim.vm.device.VirtualDevice.BackingInfo: + """Return the lab VM's first virtual disk backing.""" + for device in vm.config.hardware.device: + if isinstance(device, vim.vm.device.VirtualDisk): + return device.backing + raise AssertionError(f"{vm._moId} has no virtual disk") class TestOpenvixdisklib: @@ -42,3 +65,58 @@ class TestOpenvixdisklib: read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE handle.read(disk, start, 1, read_buf) assert read_buf.raw[:SECTOR_SIZE] == expected + + def test_read_only_open_snapshot_parent(self, lab: LabEnv) -> None: + """Read-only Open uses NfcGetVmFiles, including a snapshot parent path. + + After a snapshot the attached leaf is a new delta (``…-000001.vmdk``) + while backup tools open the parent file. That path is not + ``device.backing.fileName``; VDDK still opens it with a VM-only + ticket and NFC ``OPEN_FILE``. + """ + handle = vixdisklib.VixDiskLibHandle( + vixdisklib_compatibility_version="8.0") + expected = pattern_bytes(SECTOR_SIZE, b"OVDL-RO") + write_buf = vixdisklib.get_buffer(SECTOR_SIZE) + read_buf = vixdisklib.get_buffer(SECTOR_SIZE) + write_buf[:SECTOR_SIZE] = expected + write_kwargs = lab.vixdisklib_connect_kwargs({ + "allow_untrusted": lab.allow_untrusted, + }) + read_kwargs = lab.vixdisklib_connect_kwargs({ + "allow_untrusted": lab.allow_untrusted, + "read_only": True, + }) + read_flags = vixdisklib.VIXDISKLIB_FLAG_OPEN_READ_ONLY + + def read_sector(path: str) -> bytes: + with handle.connect(**read_kwargs) as conn: + with handle.open(conn, path, flags=read_flags) as disk: + read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE + handle.read(disk, 0, 1, read_buf) + return read_buf.raw[:SECTOR_SIZE] + + with handle.connect(**write_kwargs) as conn: + with handle.open(conn, lab.disk_path, flags=0) as disk: + handle.write(disk, 0, 1, write_buf) + + assert read_sector(lab.disk_path) == expected + + si = _connect_lab_vim(lab) + try: + vm = vim.VirtualMachine(lab.vm_moref, si._stub) + _wait_for_task( + vm.CreateSnapshot_Task("ovdl-readonly", "", False, False)) + backing = _virtual_disk_backing(vm) + parent = getattr(backing, "parent", None) + assert parent is not None + assert parent.fileName == lab.disk_path + assert backing.fileName != parent.fileName + assert read_sector(parent.fileName) == expected + finally: + try: + vm = vim.VirtualMachine(lab.vm_moref, si._stub) + if vm.snapshot is not None: + _wait_for_task(vm.RemoveAllSnapshots_Task()) + finally: + Disconnect(si)