diff --git a/openvixdisklib/nfc_auth.py b/openvixdisklib/nfc_auth.py index bd8d75c..e2b835e 100644 --- a/openvixdisklib/nfc_auth.py +++ b/openvixdisklib/nfc_auth.py @@ -17,10 +17,11 @@ with pyVmomi's type system and invoked through the same SOAP stub. from __future__ import annotations +import contextlib import hashlib import socket import ssl -from typing import Optional +from typing import Self from pyVim.connect import Disconnect, SmartConnect from pyVmomi import vim @@ -45,12 +46,10 @@ def _register_nfc_types() -> None: global _NFC_TYPES_REGISTERED if _NFC_TYPES_REGISTERED: return - try: + with contextlib.suppress(Exception): GetVmodlType("vim.NfcService") _NFC_TYPES_REGISTERED = True return - except Exception: - pass CreateManagedType( "vim.NfcService", @@ -136,7 +135,7 @@ def connect_vim( username: str, password: str, port: int = 443, - thumbprint: Optional[str] = None, + thumbprint: str | None = None, allow_untrusted: bool = False, ) -> vim.ServiceInstance: """Login to vCenter or ESXi using pyVim.connect.SmartConnect. @@ -186,10 +185,10 @@ def _virtual_disk_key(vm: vim.VirtualMachine, disk_path: str) -> int: def get_nfc_ticket( si: vim.ServiceInstance, vm: vim.VirtualMachine, - disk_device_key: Optional[int] = None, - host_for_access: Optional[vim.HostSystem] = None, + disk_device_key: int | None = None, + host_for_access: vim.HostSystem | None = None, read_only: bool = True, - disk_path: Optional[str] = None, + disk_path: str | None = None, ) -> vim.HostServiceTicket: """Return a one-time NFC HostServiceTicket for ``vm``. @@ -235,7 +234,7 @@ def get_ssl_cert_thumbprint( host: str, port: int = 443, digest_algorithm: str = "sha1", - ssl_context: Optional[ssl.SSLContext] = None, + ssl_context: ssl.SSLContext | None = None, timeout: float = 30.0, ) -> str: """Return the TLS certificate thumbprint of ``host``:``port``. @@ -257,9 +256,11 @@ def get_ssl_cert_thumbprint( """ if ssl_context is None: ssl_context = _ssl_client_context(verify=False) - with socket.create_connection((host, port), timeout=timeout) as sock: - with ssl_context.wrap_socket(sock, server_hostname=host) as ssock: - cert = ssock.getpeercert(binary_form=True) + with ( + socket.create_connection((host, port), timeout=timeout) as sock, + ssl_context.wrap_socket(sock, server_hostname=host) as ssock, + ): + cert = ssock.getpeercert(binary_form=True) if not cert: raise ConnectionError(f"no peer certificate from {host}:{port}") return _format_thumbprint(hashlib.new(digest_algorithm, cert).digest()) @@ -341,7 +342,10 @@ def connect_authd( try: if not allow_untrusted and ticket.sslThumbprint: - peer = _sha1_thumbprint(ssock.getpeercert(True)) + der_cert = ssock.getpeercert(True) + if not der_cert: + raise ConnectionError(f"no peer certificate from {host}:{port}") + peer = _sha1_thumbprint(der_cert) if _normalize_thumbprint(peer) != _normalize_thumbprint( ticket.sslThumbprint ): @@ -391,7 +395,7 @@ class NfcAuthSession: finally: Disconnect(self.si) - def __enter__(self) -> "NfcAuthSession": + def __enter__(self) -> Self: return self def __exit__(self, exc_type, exc, tb) -> None: @@ -404,10 +408,10 @@ def authenticate( password: str, vm_moref: str, port: int = 443, - thumbprint: Optional[str] = None, + thumbprint: str | None = None, allow_untrusted: bool = False, - disk_device_key: Optional[int] = None, - disk_path: Optional[str] = None, + disk_device_key: int | None = None, + disk_path: str | None = None, read_only: bool = True, nfc_ssl: bool = True, ) -> NfcAuthSession: diff --git a/openvixdisklib/nfc_open.py b/openvixdisklib/nfc_open.py index 1ea5676..8bccd73 100644 --- a/openvixdisklib/nfc_open.py +++ b/openvixdisklib/nfc_open.py @@ -25,6 +25,7 @@ import os import socket import ssl import struct +from typing import Self from openvixdisklib import fastlz from openvixdisklib.nfc_auth import NfcAuthSession, _ssl_client_context @@ -384,7 +385,7 @@ class NfcDisk: except OSError: pass - def __enter__(self) -> "NfcDisk": + def __enter__(self) -> Self: return self def __exit__(self, exc_type, exc, tb) -> None: @@ -485,6 +486,7 @@ def open_disk( raise NotImplementedError( f"NFC compression type {compression} is not supported" ) + sock: socket.socket if session.nfc_ssl: sock = wrap_nfcssl_socket(session.authd_sock, session.ticket.host) else: diff --git a/openvixdisklib/openvixdisklib.py b/openvixdisklib/openvixdisklib.py index a66cc2b..66dc906 100644 --- a/openvixdisklib/openvixdisklib.py +++ b/openvixdisklib/openvixdisklib.py @@ -19,7 +19,7 @@ import contextlib import ctypes import logging import os -from typing import Iterator, Optional, Union +from collections.abc import Iterator from pyVim.connect import Disconnect from pyVmomi import vim @@ -70,7 +70,7 @@ def get_buffer(size: int): return ctypes.create_string_buffer(size) -def _parse_vm_moref(vmx_spec: Optional[str]) -> str: +def _parse_vm_moref(vmx_spec: str | None) -> str: if not vmx_spec: raise ValueError("vmx_spec is required (for example 'moref=vm-13098')") if "=" in vmx_spec: @@ -81,7 +81,7 @@ def _parse_vm_moref(vmx_spec: Optional[str]) -> str: return vmx_spec -def _select_transport(transport_modes: Optional[str]) -> str: +def _select_transport(transport_modes: str | None) -> str: """Return the first requested transport this replacement implements. ``None`` defaults to ``nbdssl``. A colon-separated list (VDDK @@ -105,8 +105,8 @@ class _Connection: self, si: vim.ServiceInstance, vm_moref: str, - snapshot_ref: Optional[str], - thumbprint: Optional[str], + snapshot_ref: str | None, + thumbprint: str | None, allow_untrusted: bool, read_only: bool, transport_mode: str, @@ -134,8 +134,8 @@ class VixDiskLibHandle: def __init__( self, - config_path: Optional[str] = None, - vixdisklib_compatibility_version: Optional[str] = None, + config_path: str | None = None, + vixdisklib_compatibility_version: str | None = None, ) -> None: """Accept the VDDK wrapper constructor; no native library is loaded. @@ -166,7 +166,7 @@ class VixDiskLibHandle: break if not version_used: - raise Exception( + raise RuntimeError( "Could not initialize vixDiskLib with any of the following " "versions: %s" % target_versions ) @@ -194,13 +194,13 @@ class VixDiskLibHandle: def connect( self, server_name: str, - thumbprint: Optional[str], + thumbprint: str | None, username: str, password: str, - vmx_spec: Optional[str] = None, - snapshot_ref: Optional[str] = None, + vmx_spec: str | None = None, + snapshot_ref: str | None = None, read_only: bool = True, - transport_modes: Optional[str] = None, + transport_modes: str | None = None, port: int = 443, allow_untrusted: bool = False, ) -> Iterator[_Connection]: @@ -318,7 +318,7 @@ class VixDiskLibHandle: disk_handle: _DiskHandle, start_sector: int, num_sectors: int, - buf: Union[ctypes.Array, bytearray, memoryview], + buf: ctypes.Array | bytearray | memoryview, ) -> None: """Read ``num_sectors`` from ``start_sector`` into ``buf``. @@ -331,7 +331,7 @@ class VixDiskLibHandle: data = disk_handle.disk.read(start_sector, num_sectors) if isinstance(buf, (bytearray, memoryview)): if len(buf) < len(data): - raise Exception(f"read buffer is {len(buf)} bytes, need {len(data)}") + raise RuntimeError(f"read buffer is {len(buf)} bytes, need {len(data)}") buf[: len(data)] = data return ctypes.memmove(buf, data, len(data)) @@ -341,7 +341,7 @@ class VixDiskLibHandle: disk_handle: _DiskHandle, start_sector: int, num_sectors: int, - buf: Union[ctypes.Array, bytes, bytearray, memoryview], + buf: ctypes.Array | bytes | bytearray | memoryview, ) -> None: """Write ``num_sectors`` from ``buf`` starting at ``start_sector``. diff --git a/tests/integration/base.py b/tests/integration/base.py index f2648ce..a5ce4f0 100644 --- a/tests/integration/base.py +++ b/tests/integration/base.py @@ -5,12 +5,13 @@ from __future__ import annotations +import contextlib import ctypes import os import time import uuid from dataclasses import dataclass -from typing import Any, Optional +from typing import Any import pytest import yaml @@ -76,7 +77,7 @@ class LabEnv: ) def vixdisklib_connect_kwargs( - self, extra: Optional[dict[str, Any]] = None + self, extra: dict[str, Any] | None = None ) -> dict[str, Any]: """Return common ``VixDiskLib_ConnectEx`` arguments for the temp VM.""" kwargs: dict[str, Any] = { @@ -283,10 +284,8 @@ def create_lab_vm() -> LabEnv: ) except Exception: if vm is not None: - try: + with contextlib.suppress(Exception): _wait_for_task(vm.Destroy_Task()) - except Exception: - pass raise finally: Disconnect(si) diff --git a/tests/integration/test_crosscheck.py b/tests/integration/test_crosscheck.py index b8edb81..d4ff621 100644 --- a/tests/integration/test_crosscheck.py +++ b/tests/integration/test_crosscheck.py @@ -3,7 +3,7 @@ """Compare writes and reads from VDDK with openvixdisklib.""" -from typing import Any, Optional +from typing import Any import pytest @@ -12,7 +12,7 @@ from tests.integration import vixdisklib from tests.integration.base import SECTOR_AT_1GB, SECTOR_SIZE, LabEnv, pattern_bytes -def _connect_extra(lab: LabEnv, module: Any) -> Optional[dict[str, Any]]: +def _connect_extra(lab: LabEnv, module: Any) -> dict[str, Any] | None: """Return extra ``connect`` kwargs needed by ``module``.""" if module is open_vix: return {"allow_untrusted": lab.allow_untrusted} @@ -28,11 +28,13 @@ def _write_sectors( ) 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=flags) as disk: - for start, data in payloads.items(): - buf[:SECTOR_SIZE] = data - handle.write(disk, start, 1, buf) + with ( + handle.connect(**kwargs) as conn, + handle.open(conn, lab.disk_path, flags=flags) as disk, + ): + for start, data in payloads.items(): + buf[:SECTOR_SIZE] = data + handle.write(disk, start, 1, buf) def _read_sectors( @@ -45,12 +47,14 @@ def _read_sectors( 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=flags) 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] + with ( + handle.connect(**kwargs) as conn, + handle.open(conn, lab.disk_path, flags=flags) 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 diff --git a/tests/integration/test_nfc_open.py b/tests/integration/test_nfc_open.py index 80bc4d1..9d9beb1 100644 --- a/tests/integration/test_nfc_open.py +++ b/tests/integration/test_nfc_open.py @@ -21,14 +21,16 @@ class TestNfcOpen: ) -> None: """Open the temp VMDK, write sector 0, and read it back.""" expected = pattern_bytes(SECTOR_SIZE, b"NFC-OPEN-S0") - with lab.authenticate(read_only=False, nfc_ssl=nfc_ssl) as session: - with nfc_open.open_disk( + with ( + lab.authenticate(read_only=False, nfc_ssl=nfc_ssl) as session, + nfc_open.open_disk( session, lab.disk_path, read_only=False, compression=compression - ) 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) - assert got is not expected - assert got == expected + ) 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) + 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 a28c0ca..ff27728 100644 --- a/tests/integration/test_nfc_read_write.py +++ b/tests/integration/test_nfc_read_write.py @@ -30,38 +30,40 @@ class TestNfcReadWrite: (0, 129), (256, 64), ] - with lab.authenticate(read_only=False) as session: - with nfc_open.open_disk( + with ( + lab.authenticate(read_only=False) as session, + nfc_open.open_disk( session, lab.disk_path, read_only=False, compression=compression - ) as disk: - for start, n_sectors in ranges: - length = n_sectors * SECTOR_SIZE - seed = f"NFC-R{start}:{n_sectors}-".encode() - to_write = pattern_bytes(length, seed) - disk.write(start, n_sectors, to_write) - got = disk.read(start, n_sectors) - assert got is not to_write - assert len(got) == length - assert got == to_write + ) as disk, + ): + for start, n_sectors in ranges: + length = n_sectors * SECTOR_SIZE + seed = f"NFC-R{start}:{n_sectors}-".encode() + to_write = pattern_bytes(length, seed) + disk.write(start, n_sectors, to_write) + got = disk.read(start, n_sectors) + assert got is not to_write + assert len(got) == length + assert got == to_write - two_seed = b"NFC-TWO-SECTOR" - two_to_write = pattern_bytes(2 * SECTOR_SIZE, two_seed) - disk.write(0, 2, two_to_write) - two_got = disk.read(0, 2) - assert two_got is not two_to_write - assert two_got == two_to_write - assert disk.read(1, 1) == two_to_write[SECTOR_SIZE:] + two_seed = b"NFC-TWO-SECTOR" + two_to_write = pattern_bytes(2 * SECTOR_SIZE, two_seed) + disk.write(0, 2, two_to_write) + two_got = disk.read(0, 2) + 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 = pattern_bytes(129 * SECTOR_SIZE, big_seed) - disk.write(0, 129, big_to_write) - big_got = disk.read(0, 129) - 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] - ) + big_seed = b"NFC-129-SECTOR-WRITE" + big_to_write = pattern_bytes(129 * SECTOR_SIZE, big_seed) + disk.write(0, 129, big_to_write) + big_got = disk.read(0, 129) + 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] + ) @pytest.mark.parametrize( "compression", @@ -72,12 +74,14 @@ class TestNfcReadWrite: """Write 32 MiB (512 AIO chunks) and read it back in one request.""" n_sectors = _32MIB // SECTOR_SIZE to_write = os.urandom(_32MIB) - with lab.authenticate(read_only=False) as session: - with nfc_open.open_disk( + with ( + lab.authenticate(read_only=False) as session, + nfc_open.open_disk( session, lab.disk_path, read_only=False, compression=compression - ) as disk: - disk.write(0, n_sectors, to_write) - got = disk.read(0, n_sectors) - assert got is not to_write - assert len(got) == _32MIB - assert got == to_write + ) as disk, + ): + disk.write(0, n_sectors, to_write) + got = disk.read(0, n_sectors) + assert got is not to_write + assert len(got) == _32MIB + assert got == to_write diff --git a/tests/integration/test_openvixdisklib.py b/tests/integration/test_openvixdisklib.py index a186460..39b130b 100644 --- a/tests/integration/test_openvixdisklib.py +++ b/tests/integration/test_openvixdisklib.py @@ -55,15 +55,17 @@ class TestOpenvixdisklib: SECTOR_AT_1GB: pattern_bytes(SECTOR_SIZE, b"OVDL-1GB"), } assert handle.get_transport_modes() == ["nbdssl", "nbd"] - with handle.connect(**connect_kwargs) as conn: - with handle.open(conn, lab.disk_path, flags=open_flags) as disk: - assert handle.get_transport_mode(disk) == transport_mode - for start, expected in patterns.items(): - write_buf[:SECTOR_SIZE] = expected - handle.write(disk, start, 1, write_buf) - read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE - handle.read(disk, start, 1, read_buf) - assert read_buf.raw[:SECTOR_SIZE] == expected + with ( + handle.connect(**connect_kwargs) as conn, + handle.open(conn, lab.disk_path, flags=open_flags) as disk, + ): + assert handle.get_transport_mode(disk) == transport_mode + for start, expected in patterns.items(): + write_buf[:SECTOR_SIZE] = expected + handle.write(disk, start, 1, write_buf) + 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. @@ -92,15 +94,19 @@ class TestOpenvixdisklib: 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) + with ( + handle.connect(**read_kwargs) as conn, + 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) + with ( + handle.connect(**write_kwargs) as conn, + handle.open(conn, lab.disk_path, flags=0) as disk, + ): + handle.write(disk, 0, 1, write_buf) assert read_sector(lab.disk_path) == expected diff --git a/tests/integration/test_vddk.py b/tests/integration/test_vddk.py index 1f102e7..f7eb3e9 100644 --- a/tests/integration/test_vddk.py +++ b/tests/integration/test_vddk.py @@ -17,9 +17,11 @@ class TestVddk: 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[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE - handle.read(disk, 0, 1, read_buf) - assert read_buf.raw[:SECTOR_SIZE] == expected + with ( + handle.connect(**lab.vixdisklib_connect_kwargs()) as conn, + handle.open(conn, lab.disk_path, flags=0) as disk, + ): + handle.write(disk, 0, 1, write_buf) + read_buf[:SECTOR_SIZE] = b"\xa5" * SECTOR_SIZE + handle.read(disk, 0, 1, read_buf) + assert read_buf.raw[:SECTOR_SIZE] == expected diff --git a/tests/integration/vixdisklib.py b/tests/integration/vixdisklib.py old mode 100755 new mode 100644 index 00de693..061611f --- a/tests/integration/vixdisklib.py +++ b/tests/integration/vixdisklib.py @@ -82,7 +82,7 @@ def get_buffer(size): return ctypes.create_string_buffer(size) -class VixDiskLibHandle(object): +class VixDiskLibHandle: """Class which acts as a proxy for vixDiskLib-related operations:""" def __init__(self, config_path=None, vixdisklib_compatibility_version=None): @@ -130,7 +130,7 @@ class VixDiskLibHandle(object): ) if not version_used: - raise Exception( + raise RuntimeError( "Could not initialize vixDiskLib with any of the following " "versions: %s" % target_versions ) @@ -234,7 +234,9 @@ class VixDiskLibHandle(object): self._vix_disklib.VixDiskLib_Exit.argtypes = [] self._vix_disklib.VixDiskLib_Exit.restype = None - def _check_err(self, err, allowed_values=[vix_disklib_errors.VIX_OK]): + def _check_err(self, err, allowed_values=None): + if allowed_values is None: + allowed_values = [vix_disklib_errors.VIX_OK] if err not in allowed_values: err_msg = self._vix_disklib.VixDiskLib_GetErrorText(err, None) err_msg_copy = str(ctypes.cast(err_msg, ctypes.c_char_p).value.decode()) @@ -279,7 +281,7 @@ class VixDiskLibHandle(object): LOG.debug("Original vixDiskLib error message: %s", err_msg_copy) err_msg = msg - raise Exception(err_msg) + raise RuntimeError(err_msg) def get_transport_modes(self): transport_modes = self._vix_disklib.VixDiskLib_ListTransportModes() diff --git a/tests/perf/test_compare.py b/tests/perf/test_compare.py index 1a595f0..cb75f91 100644 --- a/tests/perf/test_compare.py +++ b/tests/perf/test_compare.py @@ -43,15 +43,17 @@ def _time_write_read( read_buf = module.get_buffer(len(payload)) write_buf[: len(payload)] = payload kwargs = lab.vixdisklib_connect_kwargs(_connect_extra(lab, module, transport_mode)) - with handle.connect(**kwargs) as conn: - with handle.open(conn, lab.disk_path, flags=flags) as disk: - started = time.perf_counter() - handle.write(disk, 0, n_sectors, write_buf) - write_s = time.perf_counter() - started - read_buf[: len(payload)] = b"\xa5" * len(payload) - started = time.perf_counter() - handle.read(disk, 0, n_sectors, read_buf) - read_s = time.perf_counter() - started + with ( + handle.connect(**kwargs) as conn, + handle.open(conn, lab.disk_path, flags=flags) as disk, + ): + started = time.perf_counter() + handle.write(disk, 0, n_sectors, write_buf) + write_s = time.perf_counter() - started + read_buf[: len(payload)] = b"\xa5" * len(payload) + started = time.perf_counter() + handle.read(disk, 0, n_sectors, read_buf) + read_s = time.perf_counter() - started assert read_buf.raw[: len(payload)] == payload return write_s, read_s diff --git a/tox.ini b/tox.ini index 6e549be..3528df9 100644 --- a/tox.ini +++ b/tox.ini @@ -88,12 +88,14 @@ commands = sitepackages = True deps = flake8 -commands = flake8 {posargs} +commands = flake8 {[vars]all_path} {posargs} allowlist_externals = flake8 [flake8] -ignore = E125,E251,W503,W504,E305,E731,E117,W605,F632 -exclude = .venv,.git,.tox,dist,build,*.egg +ignore = E125,E251,W503,W504,E305,E731,E117,W605,F632,E203 +exclude = .venv,.git,.tox,dist,build,*.egg,docs,._* # Same length as ruff. max-line-length = 88 +per-file-ignores = + tests/integration/vix_disklib_errors.py:E501