diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 84d5e2a..72ea920 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,6 +55,31 @@ jobs: - name: Run test suite run: python -m pytest -q + dxgi-contract: + name: DXGI contract · Windows · Python 3.13 + runs-on: windows-latest + timeout-minutes: 15 + + steps: + - name: Check out repository + uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: pyproject.toml + + - name: Upgrade packaging tools + run: python -m pip install --upgrade pip setuptools wheel + + - name: Install performance and test dependencies + run: python -m pip install -e ".[dev,performance]" + + - name: Verify DXCam private integration contract + run: python -m pytest -q tests/test_dxgi_dxcam_contract.py tests/test_dxgi_metadata.py + coverage: name: Coverage runs-on: ubuntu-latest @@ -91,7 +116,7 @@ jobs: build: name: Build package - needs: test + needs: [test, dxgi-contract] runs-on: ubuntu-latest timeout-minutes: 15 diff --git a/benchmarks/benchmark_dxgi_metadata.py b/benchmarks/benchmark_dxgi_metadata.py new file mode 100644 index 0000000..4533d68 --- /dev/null +++ b/benchmarks/benchmark_dxgi_metadata.py @@ -0,0 +1,107 @@ +"""Inspect DXGI dirty/move metadata quality on a real Windows desktop. + +Run on Windows with the performance extra installed: + + python -m pip install -e ".[performance]" + python benchmarks/benchmark_dxgi_metadata.py --frames 300 --fps 60 + +The benchmark is diagnostic rather than a synthetic score. It reports how +often native Desktop Duplication metadata was available, rectangle counts and +the approximate fraction of the framebuffer affected by each frame. +""" + +from __future__ import annotations + +import argparse +import statistics +import time + +from vnc_lib.screen_capture import ScreenCapture + + +NATIVE_BGR0 = { + "bits_per_pixel": 32, + "depth": 24, + "big_endian_flag": 0, + "true_colour_flag": 1, + "red_max": 255, + "green_max": 255, + "blue_max": 255, + "red_shift": 16, + "green_shift": 8, + "blue_shift": 0, +} + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark DXGI dirty/move metadata") + parser.add_argument("--frames", type=int, default=300) + parser.add_argument("--fps", type=float, default=60.0) + args = parser.parse_args() + + frames = max(1, args.frames) + fps = max(1.0, args.fps) + interval = 1.0 / fps + capture = ScreenCapture(backend_preference="dxcam") + if capture.get_backend_name() != "dxcam": + raise SystemExit( + "DXCam backend is unavailable; install the performance extra on Windows" + ) + + capture.set_cache_frame_rate(fps) + native_frames = 0 + fallback_frames = 0 + dirty_counts: list[int] = [] + move_counts: list[int] = [] + changed_ratios: list[float] = [] + capture_ms: list[float] = [] + + next_tick = time.perf_counter() + try: + for _ in range(frames): + frame = capture.capture_frame(NATIVE_BGR0) + result = frame.result + metadata = frame.metadata + if result.pixel_data is None: + continue + + capture_ms.append(result.capture_time * 1000.0) + if metadata.dirty_regions is None: + fallback_frames += 1 + else: + native_frames += 1 + dirty_counts.append(len(metadata.dirty_regions)) + move_counts.append(len(metadata.move_rects)) + changed_area = sum(w * h for _, _, w, h in metadata.dirty_regions) + changed_ratios.append( + min(1.0, changed_area / max(1, result.width * result.height)) + ) + + next_tick += interval + delay = next_tick - time.perf_counter() + if delay > 0: + time.sleep(delay) + else: + next_tick = time.perf_counter() + finally: + capture.close_current_thread_sessions() + + total = native_frames + fallback_frames + print(f"backend: {capture.get_backend_name()}") + print(f"frames sampled: {total}") + print(f"native metadata: {native_frames} ({native_frames / max(1, total):.1%})") + print(f"metadata fallbacks: {fallback_frames}") + if capture_ms: + ordered = sorted(capture_ms) + p95 = ordered[int((len(ordered) - 1) * 0.95)] + print(f"capture avg: {statistics.mean(capture_ms):.2f} ms") + print(f"capture p95: {p95:.2f} ms") + if dirty_counts: + print(f"dirty rects avg: {statistics.mean(dirty_counts):.2f}") + print(f"move rects avg: {statistics.mean(move_counts):.2f}") + print(f"changed area avg: {statistics.mean(changed_ratios):.2%}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/performance.md b/docs/performance.md index 293b5b0..5f4931d 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -8,7 +8,7 @@ PyVNCServer optimizes the path from desktop capture to encoded rectangle rather | Backend | Platform | Role | | --- | --- | --- | -| DXCam / DXGI | Windows | optional fast Desktop Duplication capture path | +| DXCam / DXGI | Windows | optional fast Desktop Duplication capture path with native dirty/move metadata in v3.3 | | MSS | cross-platform | portable primary fallback | | Pillow ImageGrab | platform dependent | fallback capture path | @@ -18,16 +18,29 @@ Install the performance extras: python -m pip install -e ".[performance]" ``` +PyVNCServer 3.3 requires DXCam `0.3.0+` for the native metadata integration. + ## Shared capture producer A server-wide producer captures once and distributes framebuffer generations to sessions. This avoids N clients causing N independent desktop captures. -## Changed regions +## Native DXGI changed regions + +On an unscaled, unrotated full-output DXCam capture, v3.3 reads Desktop Duplication metadata directly from the active `IDXGIOutputDuplication` object: + +- `GetFrameDirtyRects` identifies framebuffer regions whose pixel contents changed; +- `GetFrameMoveRects` identifies regions copied from another framebuffer location and exposes them as CopyRect hints; +- a DXGI timeout/no-new-frame is treated as an authoritative empty update; +- any metadata read/validation failure returns to the existing software change detector instead of assuming the screen is unchanged. -Incremental requests benefit from limiting work to changed regions. When a backend cannot provide native dirty metadata, PyVNCServer performs change detection above the backend. +Native metadata is deliberately disabled for scaled, rotated or cropped DXCam captures until coordinate translation is implemented. MSS and Pillow also continue to use software change detection. -!!! note - Native DXGI dirty/move rectangle harvesting is not yet implemented in the current DXCam integration. CopyRect/dirty-region opportunities therefore depend on metadata available to the higher layers. +!!! important + Native metadata is an optimization, never a correctness requirement. An ambiguous native result becomes `dirty_regions = None`, which explicitly activates the software differ. + +### Move-rectangle safety in v3.3 + +Move destinations are currently included in the dirty pixel list even when a CopyRect hint is emitted. This is intentionally conservative: clients without CopyRect support and clients that skip capture generations still converge to the correct framebuffer. Once the real-client interoperability matrix has broader coverage, the redundant pixel update can be removed for clients that are exactly one generation behind and advertise CopyRect. ## Request coalescing @@ -59,6 +72,8 @@ Relevant LAN settings include zlib/ZRLE compression levels, raw thresholds and J ## Benchmarks +Encoder/capture benchmarks: + ```bash PYTHONPATH=src python benchmarks/benchmark_encoders.py PYTHONPATH=src python benchmarks/benchmark_screen_capture.py @@ -66,9 +81,25 @@ PYTHONPATH=src python benchmarks/benchmark_screen_capture_methods.py PYTHONPATH=src python benchmarks/benchmark_lan_latency.py ``` +### DXGI metadata benchmark + +On a real Windows desktop with the performance extra installed: + +```powershell +python benchmarks/benchmark_dxgi_metadata.py --frames 300 --fps 60 +``` + +The diagnostic reports: + +- native metadata hit rate vs software-diff fallback rate; +- average and p95 capture time; +- average dirty/move rectangle count; +- approximate changed framebuffer area. + For meaningful numbers: - benchmark on the target OS/GPU/display setup; +- test idle desktop, text editing, window movement, scrolling and video separately; - separate capture time from encode time; - test full-screen and small-region updates; - include the actual viewer over the intended network path; diff --git a/docs/ultravnc.md b/docs/ultravnc.md index ae25897..2c65c26 100644 --- a/docs/ultravnc.md +++ b/docs/ultravnc.md @@ -51,6 +51,56 @@ These are useful control encodings when isolating a compatibility problem: - **Zlib**: compressed full rectangles; - **ZRLE**: tiled zlib/RLE, usually a good general-purpose option. +## v3.3 interoperability regression suite + +Version 3.3 adds a real TCP/RFB regression test that keeps one connection open while changing the advertised encoding set. The test parses the actual rectangle payload rather than only checking that the socket remained connected. + +The automated sequence is: + +```text +Raw → RRE → Hextile → Zlib → ZRLE → Tight → Raw +``` + +For each step it verifies the rectangle header encoding ID and the corresponding wire format. This specifically guards against failures such as advertising RRE while sending Raw bytes, stale zlib state after an encoding switch, or a connection reset caused by stream desynchronization. + +This test is not a substitute for running the UltraVNC binary itself: client-specific decoder behavior still needs validation before a release. It does, however, make the protocol-level failure modes reproducible in Linux and Windows CI. + +## Real UltraVNC Viewer smoke test + +UltraVNC Viewer exposes a command-line `-encoding` option, so v3.3 also ships a PowerShell smoke-test for the actual `vncviewer.exe` binary. + +Start PyVNCServer in one terminal: + +```powershell +pyvncserver serve --log-level DEBUG +``` + +Then run in another PowerShell window: + +```powershell +.\scripts\ultravnc_smoke.ps1 ` + -ViewerPath 'C:\Program Files\uvnc bvba\UltraVNC\vncviewer.exe' +``` + +The default sequence is: + +```text +raw → rre → hextile → zlib → tight +``` + +For every encoding the script launches a fresh view-only UltraVNC process, keeps it connected for a short observation period and treats an early viewer exit as failure. Viewer logs are written under `%TEMP%\pyvncserver-ultravnc`. + +A longer run is useful when validating a release candidate: + +```powershell +.\scripts\ultravnc_smoke.ps1 ` + -SecondsPerEncoding 10 ` + -Encodings raw,rre,hextile,zlib,tight +``` + +!!! note + The script proves that the real UltraVNC process can negotiate and keep the session alive. Visual correctness should still be checked while moving windows, scrolling and changing screen content, especially for Tight and RRE. + ## What to log Run: @@ -68,9 +118,9 @@ Useful lines include: - selected encoding per update/region; - socket reset/timeout messages. -## Suggested compatibility matrix +## Suggested manual compatibility matrix -After a code change to an encoder, test a single connection while switching in this order: +After a code change to an encoder, test a single UltraVNC connection while switching in this order: ```text Raw → Hextile → Zlib → ZRLE → Tight → RRE → Auto diff --git a/pyproject.toml b/pyproject.toml index 3c8a399..57fbc76 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ dependencies = [ [project.optional-dependencies] performance = [ "numpy>=1.26", - "dxcam>=0.0.5; platform_system == 'Windows'", + "dxcam>=0.3.0; platform_system == 'Windows'", ] h264 = ["av>=12.0"] dev = ["pytest>=8.0"] diff --git a/scripts/ultravnc_smoke.ps1 b/scripts/ultravnc_smoke.ps1 new file mode 100644 index 0000000..f2caeb5 --- /dev/null +++ b/scripts/ultravnc_smoke.ps1 @@ -0,0 +1,128 @@ +<# +.SYNOPSIS + Smoke-test PyVNCServer with the real UltraVNC Viewer executable. + +.DESCRIPTION + Opens a fresh UltraVNC Viewer connection for each requested encoding and + keeps it alive for a short observation window. A viewer that exits before + the window ends is treated as a failed interoperability check. + + Run PyVNCServer separately before starting this script. The default target + is the safe local listener at 127.0.0.1:5900 and the viewer is always + launched in view-only mode. + +.EXAMPLE + .\scripts\ultravnc_smoke.ps1 ` + -ViewerPath 'C:\Program Files\uvnc bvba\UltraVNC\vncviewer.exe' + +.EXAMPLE + .\scripts\ultravnc_smoke.ps1 -Encodings raw,rre,hextile,zlib,tight -SecondsPerEncoding 5 +#> + +[CmdletBinding()] +param( + [Parameter(Mandatory = $false)] + [string]$ViewerPath = "$env:ProgramFiles\uvnc bvba\UltraVNC\vncviewer.exe", + + [Parameter(Mandatory = $false)] + [string]$ServerHost = "127.0.0.1", + + [Parameter(Mandatory = $false)] + [ValidateRange(1, 65535)] + [int]$Port = 5900, + + [Parameter(Mandatory = $false)] + [ValidateRange(1, 120)] + [int]$SecondsPerEncoding = 4, + + [Parameter(Mandatory = $false)] + [ValidateSet("raw", "rre", "corre", "hextile", "zlib", "zlibhex", "tight", "ultra")] + [string[]]$Encodings = @("raw", "rre", "hextile", "zlib", "tight"), + + [Parameter(Mandatory = $false)] + [string]$LogDirectory = (Join-Path $env:TEMP "pyvncserver-ultravnc") +) + +$ErrorActionPreference = "Stop" + +if (-not (Test-Path -LiteralPath $ViewerPath -PathType Leaf)) { + throw "UltraVNC Viewer not found: $ViewerPath" +} + +New-Item -ItemType Directory -Force -Path $LogDirectory | Out-Null + +# UltraVNC accepts host::port for an explicit TCP port. Keep the short host +# form for the standard 5900 port to match the most common local setup. +$Target = if ($Port -eq 5900) { $ServerHost } else { "${ServerHost}::${Port}" } +$Failures = @() + +Write-Host "PyVNCServer / UltraVNC interoperability smoke test" +Write-Host "Viewer: $ViewerPath" +Write-Host "Target: $Target" +Write-Host "Encodings: $($Encodings -join ', ')" +Write-Host "" + +foreach ($Encoding in $Encodings) { + $Timestamp = Get-Date -Format "yyyyMMdd-HHmmssfff" + $ViewerLog = Join-Path $LogDirectory "ultravnc-$Encoding-$Timestamp.log" + $Arguments = @( + "-connect", $Target, + "-encoding", $Encoding, + "-viewonly", + "-nocursorshape", + "-noremotecursor", + "-loglevel", "10", + "-logfile", $ViewerLog + ) + + Write-Host "[$Encoding] connecting..." -ForegroundColor Cyan + $Process = Start-Process ` + -FilePath $ViewerPath ` + -ArgumentList $Arguments ` + -PassThru + + try { + $Deadline = (Get-Date).AddSeconds($SecondsPerEncoding) + $ExitedEarly = $false + while ((Get-Date) -lt $Deadline) { + Start-Sleep -Milliseconds 200 + $Process.Refresh() + if ($Process.HasExited) { + $ExitedEarly = $true + break + } + } + + if ($ExitedEarly) { + $ExitCode = $Process.ExitCode + $Failures += $Encoding + Write-Host "[$Encoding] FAILED: viewer exited early (code $ExitCode)" -ForegroundColor Red + if (Test-Path -LiteralPath $ViewerLog) { + Write-Host "[$Encoding] UltraVNC log: $ViewerLog" + } + continue + } + + Write-Host "[$Encoding] OK: connection stayed alive for $SecondsPerEncoding s" -ForegroundColor Green + } + finally { + $Process.Refresh() + if (-not $Process.HasExited) { + Stop-Process -Id $Process.Id -Force -ErrorAction SilentlyContinue + $Process.WaitForExit(3000) | Out-Null + } + } + + Start-Sleep -Milliseconds 400 +} + +Write-Host "" +if ($Failures.Count -gt 0) { + Write-Host "FAILED encodings: $($Failures -join ', ')" -ForegroundColor Red + Write-Host "Logs: $LogDirectory" + exit 1 +} + +Write-Host "All UltraVNC smoke checks passed." -ForegroundColor Green +Write-Host "Logs: $LogDirectory" +exit 0 diff --git a/src/pyvncserver/_version.py b/src/pyvncserver/_version.py index d5a083c..df50b22 100644 --- a/src/pyvncserver/_version.py +++ b/src/pyvncserver/_version.py @@ -1,4 +1,4 @@ """Single source of truth for the project version.""" -__version__ = "3.2.1" +__version__ = "3.3.0" SERVER_NAME = f"PyVNCServer {__version__}" diff --git a/src/vnc_lib/capture_backends.py b/src/vnc_lib/capture_backends.py index 59b6f41..432edfe 100644 --- a/src/vnc_lib/capture_backends.py +++ b/src/vnc_lib/capture_backends.py @@ -1,9 +1,9 @@ """ Capture backend abstractions and metadata hints. -This module does not implement full DXGI dirty/move rect harvesting yet. -It provides a stable interface so the server runtime can consume backend- -supplied metadata as soon as a Windows backend starts exposing it. +The DXCam adapter can harvest native DXGI Desktop Duplication dirty/move +rectangles through the optional metadata hook while MSS/Pillow retain the +portable software-diff fallback. """ from __future__ import annotations @@ -11,6 +11,8 @@ from dataclasses import dataclass, field from typing import Any +from .dxgi_metadata import DXGIMetadataHook + Rectangle = tuple[int, int, int, int] @@ -179,22 +181,120 @@ class DXCamCaptureBackend(BaseCaptureBackend): supports_bgra=True, supports_rgb=True, supports_pil_image=False, - # The runtime is now ready for these metadata hints, but dxcam does not - # provide them in this implementation yet. - supports_dirty_regions=False, - supports_move_rects=False, + supports_dirty_regions=True, + supports_move_rects=True, ) + def __init__(self, owner: Any): + super().__init__(owner) + self._metadata_hook = DXGIMetadataHook(logger=getattr(owner, "logger", None)) + def is_available(self) -> bool: return bool(getattr(self.owner, "_dxcam_available", False)) + def _get_camera(self) -> Any: + camera = self.owner._get_dxcam_session() + if camera is not None: + self._metadata_hook.ensure_attached(camera) + return camera + + def _metadata_is_usable(self, camera: Any) -> bool: + if not bool(getattr(self.owner, "enable_dxgi_metadata", True)): + return False + if camera is None or not self._metadata_hook.is_available: + return False + if float(getattr(self.owner, "scale_factor", 1.0)) != 1.0: + return False + if int(getattr(camera, "rotation_angle", 0) or 0) != 0: + return False + + region = getattr(camera, "region", None) + camera_width = int(getattr(camera, "width", 0) or 0) + camera_height = int(getattr(camera, "height", 0) or 0) + if region is not None and camera_width > 0 and camera_height > 0: + if tuple(region) != (0, 0, camera_width, camera_height): + return False + return True + def healthcheck(self) -> bool: + # Do not create a DXCamera during backend probing. DXCam keeps one + # singleton camera per (device, output, backend), while PyVNCServer's + # capture producer runs on a dedicated thread. Creating the camera here + # on the server thread makes the producer call dxcam.create() again and + # triggers an "instance already exists" warning before first capture. + # The first real grab initializes the camera on the capture thread; any + # initialization failure is handled by the normal backend failover path. + return self.is_available() + + def build_metadata(self, width: int, height: int) -> CaptureMetadata: + # Capability probes use a 0x0 size and must be side-effect free. In + # particular, do not call dxcam.create() here: the real camera belongs + # to the capture-producer thread. Creating it during server startup + # would make the producer request the same DXCam singleton again. + if width <= 0 or height <= 0: + supported = ( + bool(getattr(self.owner, "enable_dxgi_metadata", True)) + and float(getattr(self.owner, "scale_factor", 1.0)) == 1.0 + and self.is_available() + ) + return CaptureMetadata( + backend_name=( + "dxcam+dxgi-metadata" if supported else self.name + ), + dirty_regions=None, + move_rects=[], + supports_dirty_regions=supported, + supports_move_rects=supported, + ) + try: - return self.is_available() and self.owner._get_dxcam_session() is not None + camera = self._get_camera() except Exception: - return False + camera = None + + supported = self._metadata_is_usable(camera) + if not supported: + return CaptureMetadata( + backend_name=self.name, + dirty_regions=None, + move_rects=[], + supports_dirty_regions=False, + supports_move_rects=False, + ) + + hints = self._metadata_hook.consume(width, height) + if hints is None: + return CaptureMetadata( + backend_name="dxcam+dxgi-fallback", + dirty_regions=None, + move_rects=[], + supports_dirty_regions=True, + supports_move_rects=True, + ) + + # Conservative v3.3 safety rule: move destinations are included in the + # pixel-dirty list as well as exposed as CopyRect hints. This guarantees + # correctness for clients without CopyRect and for clients that skip + # producer generations. A later optimization may suppress the redundant + # pixel rectangle for clients that are exactly one generation behind + # and advertise CopyRect. + dirty_regions = list(hints.dirty_regions) + dirty_regions.extend( + (move.dst_x, move.dst_y, move.width, move.height) + for move in hints.move_rects + ) + dirty_regions = list(dict.fromkeys(dirty_regions)) + + return CaptureMetadata( + backend_name="dxcam+dxgi-metadata", + dirty_regions=dirty_regions, + move_rects=list(hints.move_rects), + supports_dirty_regions=True, + supports_move_rects=True, + ) def grab_bgra(self) -> tuple[bytes | None, int, int]: + self._get_camera() frame_bytes, width, height, channels, color_hint = self.owner._grab_dxcam_frame() if frame_bytes is None: return None, 0, 0 @@ -206,6 +306,7 @@ def grab_bgra(self) -> tuple[bytes | None, int, int]: return bgra, width, height def grab_rgb(self) -> tuple[bytes | None, int, int]: + self._get_camera() frame_bytes, width, height, channels, color_hint = self.owner._grab_dxcam_frame() if frame_bytes is None: return None, 0, 0 diff --git a/src/vnc_lib/dxgi_metadata.py b/src/vnc_lib/dxgi_metadata.py new file mode 100644 index 0000000..6ecdd17 --- /dev/null +++ b/src/vnc_lib/dxgi_metadata.py @@ -0,0 +1,322 @@ +"""DXGI Desktop Duplication dirty/move rectangle metadata support. + +DXCam already owns the IDXGIOutputDuplication object used for capture. The +public DXCam API exposes pixels rather than frame metadata, so PyVNCServer +attaches a small optional hook around DXCam's duplicator to read +GetFrameDirtyRects/GetFrameMoveRects while the acquired frame is still held. + +The hook is deliberately fail-safe for correctness: when metadata cannot be +read safely, callers receive ``None`` and fall back to the software change +detector instead of assuming that the framebuffer did not change. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import ctypes +import logging +import os +from typing import Any, Protocol, TYPE_CHECKING + +Rectangle = tuple[int, int, int, int] + +if TYPE_CHECKING: + from .capture_backends import CaptureMoveRect + + +_GET_FRAME_DIRTY_RECTS_INDEX = 9 +_GET_FRAME_MOVE_RECTS_INDEX = 10 +_MAX_METADATA_BYTES = 8 * 1024 * 1024 +_MAX_METADATA_RECTS = 65535 + +_UINT = ctypes.c_uint32 +_HRESULT = ctypes.c_int32 +_WINFUNCTYPE = getattr(ctypes, "WINFUNCTYPE", ctypes.CFUNCTYPE) + + +class _Point(ctypes.Structure): + _fields_ = [("x", ctypes.c_int32), ("y", ctypes.c_int32)] + + +class _Rect(ctypes.Structure): + _fields_ = [ + ("left", ctypes.c_int32), + ("top", ctypes.c_int32), + ("right", ctypes.c_int32), + ("bottom", ctypes.c_int32), + ] + + +class _MoveRect(ctypes.Structure): + _fields_ = [("source_point", _Point), ("destination_rect", _Rect)] + + +@dataclass(frozen=True, slots=True) +class DXGIFrameMetadata: + """Validated Desktop Duplication metadata for one acquired frame.""" + + dirty_regions: tuple[Rectangle, ...] + move_rects: tuple[CaptureMoveRect, ...] + + +class DXGIMetadataReaderProtocol(Protocol): + def is_supported(self) -> bool: ... + def read(self, duplicator_pointer: Any) -> DXGIFrameMetadata | None: ... + + +class DXGIMetadataReader: + """Read dirty and move rectangles from an IDXGIOutputDuplication pointer.""" + + def __init__(self, logger: logging.Logger | None = None) -> None: + self.logger = logger or logging.getLogger(__name__) + + def is_supported(self) -> bool: + return os.name == "nt" + + def read(self, duplicator_pointer: Any) -> DXGIFrameMetadata | None: + if not self.is_supported() or not duplicator_pointer: + return None + try: + move_buffer = self._read_buffer( + duplicator_pointer, + _GET_FRAME_MOVE_RECTS_INDEX, + ctypes.sizeof(_MoveRect), + ) + dirty_buffer = self._read_buffer( + duplicator_pointer, + _GET_FRAME_DIRTY_RECTS_INDEX, + ctypes.sizeof(_Rect), + ) + if move_buffer is None or dirty_buffer is None: + return None + move_rects = self._decode_moves(move_buffer) + dirty_regions = self._decode_dirty(dirty_buffer) + if move_rects is None or dirty_regions is None: + return None + return DXGIFrameMetadata(tuple(dirty_regions), tuple(move_rects)) + except Exception as exc: + self.logger.debug("Unable to read DXGI frame metadata: %s", exc) + return None + + def _read_buffer(self, duplicator_pointer: Any, method_index: int, + item_size: int) -> bytes | None: + method, this_pointer = self._bind_metadata_method( + duplicator_pointer, method_index + ) + required = _UINT(0) + try: + probe_hr = int(method(this_pointer, 0, None, ctypes.byref(required))) + except Exception: + return None + + size = int(required.value) + if size == 0: + return b"" if probe_hr >= 0 else None + if size < item_size or size > _MAX_METADATA_BYTES: + return None + if size // item_size > _MAX_METADATA_RECTS: + return None + + buffer = ctypes.create_string_buffer(size) + returned = _UINT(size) + hr = int(method( + this_pointer, + size, + ctypes.cast(buffer, ctypes.c_void_p), + ctypes.byref(returned), + )) + if hr < 0: + return None + + used = int(returned.value) + if used < 0 or used > size or used % item_size != 0: + return None + return bytes(buffer.raw[:used]) + + @staticmethod + def _bind_metadata_method(duplicator_pointer: Any, method_index: int): + this_pointer = ctypes.cast(duplicator_pointer, ctypes.c_void_p) + if not this_pointer.value: + raise ValueError("null IDXGIOutputDuplication pointer") + vtable_ptr = ctypes.cast( + duplicator_pointer, + ctypes.POINTER(ctypes.POINTER(ctypes.c_void_p)), + ).contents + function_address = vtable_ptr[method_index] + if isinstance(function_address, ctypes.c_void_p): + function_address = function_address.value + if not function_address: + raise ValueError("missing IDXGIOutputDuplication vtable entry") + prototype = _WINFUNCTYPE( + _HRESULT, + ctypes.c_void_p, + _UINT, + ctypes.c_void_p, + ctypes.POINTER(_UINT), + ) + return prototype(function_address), this_pointer + + @staticmethod + def _decode_dirty(buffer: bytes) -> list[Rectangle] | None: + if not buffer: + return [] + item_size = ctypes.sizeof(_Rect) + if len(buffer) % item_size: + return None + values = (_Rect * (len(buffer) // item_size)).from_buffer_copy(buffer) + regions: list[Rectangle] = [] + for rect in values: + width = int(rect.right - rect.left) + height = int(rect.bottom - rect.top) + if width > 0 and height > 0: + regions.append((int(rect.left), int(rect.top), width, height)) + return regions + + @staticmethod + def _decode_moves(buffer: bytes) -> list[CaptureMoveRect] | None: + if not buffer: + return [] + item_size = ctypes.sizeof(_MoveRect) + if len(buffer) % item_size: + return None + from .capture_backends import CaptureMoveRect + values = (_MoveRect * (len(buffer) // item_size)).from_buffer_copy(buffer) + moves: list[CaptureMoveRect] = [] + for item in values: + dst = item.destination_rect + width = int(dst.right - dst.left) + height = int(dst.bottom - dst.top) + if width > 0 and height > 0: + moves.append(CaptureMoveRect( + src_x=int(item.source_point.x), + src_y=int(item.source_point.y), + dst_x=int(dst.left), + dst_y=int(dst.top), + width=width, + height=height, + )) + return moves + + +class DXGIMetadataHook: + """Attach metadata collection to the private DXCam DXGI duplicator.""" + + def __init__(self, reader: DXGIMetadataReaderProtocol | None = None, + logger: logging.Logger | None = None) -> None: + self.logger = logger or logging.getLogger(__name__) + self.reader = reader or DXGIMetadataReader(self.logger) + self._attached_duplicator: Any = None + self._pending: DXGIFrameMetadata | None = None + self._pending_valid = False + + @property + def is_available(self) -> bool: + return bool(self._attached_duplicator is not None and self.reader.is_supported()) + + def ensure_attached(self, camera: Any) -> bool: + if not self.reader.is_supported(): + return False + if getattr(camera, "backend", "dxgi") != "dxgi": + return False + duplicator = getattr(camera, "_duplicator", None) + if duplicator is None or getattr(duplicator, "duplicator", None) is None: + return False + if duplicator is self._attached_duplicator: + return True + original_update = getattr(duplicator, "update_frame", None) + if not callable(original_update): + return False + hook = self + + def update_frame_with_metadata(*args: Any, **kwargs: Any) -> bool: + ok = bool(original_update(*args, **kwargs)) + hook._pending = None + hook._pending_valid = False + if not ok: + return False + if not bool(getattr(duplicator, "updated", False)): + hook._pending = DXGIFrameMetadata((), ()) + hook._pending_valid = True + return True + metadata = hook.reader.read(getattr(duplicator, "duplicator", None)) + if metadata is not None: + hook._pending = metadata + hook._pending_valid = True + return True + + duplicator.update_frame = update_frame_with_metadata + self._attached_duplicator = duplicator + self._pending = None + self._pending_valid = False + return True + + def consume(self, width: int, height: int) -> DXGIFrameMetadata | None: + if not self._pending_valid: + return None + pending = self._pending + self._pending = None + self._pending_valid = False + if pending is None: + return None + + dirty = tuple( + region for region in ( + _clip_rectangle(rect, width, height) + for rect in pending.dirty_regions + ) if region is not None + ) + clipped_moves: list[CaptureMoveRect] = [] + for move in pending.move_rects: + clipped = _clip_move_rect(move, width, height) + if clipped is not None: + clipped_moves.append(clipped) + continue + destination = _clip_rectangle( + (move.dst_x, move.dst_y, move.width, move.height), width, height + ) + if destination is not None: + return None + return DXGIFrameMetadata(dirty, tuple(clipped_moves)) + + +def _clip_rectangle(rect: Rectangle, width: int, height: int) -> Rectangle | None: + x, y, rect_width, rect_height = map(int, rect) + x1 = max(0, x) + y1 = max(0, y) + x2 = min(int(width), x + rect_width) + y2 = min(int(height), y + rect_height) + if x1 >= x2 or y1 >= y2: + return None + return x1, y1, x2 - x1, y2 - y1 + + +def _clip_move_rect(move: "CaptureMoveRect", width: int, + height: int) -> "CaptureMoveRect | None": + from .capture_backends import CaptureMoveRect + dst_x1 = max(0, int(move.dst_x)) + dst_y1 = max(0, int(move.dst_y)) + dst_x2 = min(int(width), int(move.dst_x + move.width)) + dst_y2 = min(int(height), int(move.dst_y + move.height)) + if dst_x1 >= dst_x2 or dst_y1 >= dst_y2: + return None + + offset_x = dst_x1 - int(move.dst_x) + offset_y = dst_y1 - int(move.dst_y) + src_x = int(move.src_x) + offset_x + src_y = int(move.src_y) + offset_y + clipped_width = dst_x2 - dst_x1 + clipped_height = dst_y2 - dst_y1 + if ( + src_x < 0 or src_y < 0 + or src_x + clipped_width > width + or src_y + clipped_height > height + ): + return None + return CaptureMoveRect( + src_x=src_x, + src_y=src_y, + dst_x=dst_x1, + dst_y=dst_y1, + width=clipped_width, + height=clipped_height, + ) diff --git a/tests/test_dxgi_dxcam_contract.py b/tests/test_dxgi_dxcam_contract.py new file mode 100644 index 0000000..44b0da5 --- /dev/null +++ b/tests/test_dxgi_dxcam_contract.py @@ -0,0 +1,32 @@ +"""Contract tests for the optional DXCam integration used by v3.3. + +These tests do not create a Desktop Duplication session, so they can run on a +headless GitHub-hosted Windows runner. Their purpose is to detect an upstream +DXCam private-API change before it silently disables native metadata capture. +""" + +from __future__ import annotations + +import inspect +import os + +import pytest + + +@pytest.mark.skipif(os.name != "nt", reason="DXCam contract is Windows-only") +def test_dxcam_private_metadata_contract(): + pytest.importorskip("dxcam") + + from dxcam.core.dxgi_duplicator import DXGIDuplicator + from dxcam.dxcam import DXCamera + from dxcam._libs.dxgi import IDXGIOutputDuplication + + assert hasattr(DXGIDuplicator, "update_frame") + assert "_duplicator" in inspect.getsource(DXCamera.__init__) + + method_names = { + getattr(method, "name", None) + for method in getattr(IDXGIOutputDuplication, "_methods_", ()) + } + assert "GetFrameDirtyRects" in method_names + assert "GetFrameMoveRects" in method_names diff --git a/tests/test_dxgi_metadata.py b/tests/test_dxgi_metadata.py new file mode 100644 index 0000000..9e0db54 --- /dev/null +++ b/tests/test_dxgi_metadata.py @@ -0,0 +1,262 @@ +"""DXGI Desktop Duplication metadata hook tests.""" + +from __future__ import annotations + +import ctypes +from dataclasses import dataclass +import logging + +import vnc_lib.dxgi_metadata as dxgi_metadata +from vnc_lib.capture_backends import CaptureMoveRect, DXCamCaptureBackend +from vnc_lib.dxgi_metadata import DXGIFrameMetadata, DXGIMetadataHook, DXGIMetadataReader + + +class _FakeReader: + def __init__(self, values): + self.values = list(values) + self.calls = 0 + + def is_supported(self) -> bool: + return True + + def read(self, _duplicator_pointer): + self.calls += 1 + return self.values.pop(0) if self.values else None + + +class _FakeDuplicator: + def __init__(self): + self.duplicator = object() + self.updated = True + + def update_frame(self, *_args, **_kwargs): + return True + + +@dataclass +class _FakeCamera: + _duplicator: _FakeDuplicator + backend: str = "dxgi" + width: int = 100 + height: int = 100 + region: tuple[int, int, int, int] = (0, 0, 100, 100) + rotation_angle: int = 0 + + +def test_decode_dirty_rectangles(): + values = (dxgi_metadata._Rect * 2)( + dxgi_metadata._Rect(10, 20, 40, 60), + dxgi_metadata._Rect(100, 200, 150, 220), + ) + assert DXGIMetadataReader._decode_dirty(bytes(values)) == [ + (10, 20, 30, 40), + (100, 200, 50, 20), + ] + + +def test_decode_move_rectangles(): + values = (dxgi_metadata._MoveRect * 1)( + dxgi_metadata._MoveRect( + dxgi_metadata._Point(5, 7), + dxgi_metadata._Rect(20, 30, 60, 80), + ) + ) + assert DXGIMetadataReader._decode_moves(bytes(values)) == [ + CaptureMoveRect(src_x=5, src_y=7, dst_x=20, dst_y=30, width=40, height=50) + ] + + +def test_hook_captures_metadata_once_per_frame(): + reader = _FakeReader([ + DXGIFrameMetadata( + dirty_regions=((1, 2, 3, 4),), + move_rects=(CaptureMoveRect(0, 0, 10, 10, 5, 5),), + ) + ]) + duplicator = _FakeDuplicator() + hook = DXGIMetadataHook(reader=reader) + assert hook.ensure_attached(_FakeCamera(duplicator)) + assert duplicator.update_frame() + + metadata = hook.consume(100, 100) + assert metadata == DXGIFrameMetadata( + dirty_regions=((1, 2, 3, 4),), + move_rects=(CaptureMoveRect(0, 0, 10, 10, 5, 5),), + ) + assert hook.consume(100, 100) is None + assert reader.calls == 1 + + +def test_hook_timeout_is_authoritative_empty_metadata(): + reader = _FakeReader([]) + duplicator = _FakeDuplicator() + duplicator.updated = False + hook = DXGIMetadataHook(reader=reader) + assert hook.ensure_attached(_FakeCamera(duplicator)) + assert duplicator.update_frame() + assert hook.consume(100, 100) == DXGIFrameMetadata((), ()) + assert reader.calls == 0 + + +def test_failed_metadata_read_requests_software_fallback(): + reader = _FakeReader([None]) + duplicator = _FakeDuplicator() + hook = DXGIMetadataHook(reader=reader) + assert hook.ensure_attached(_FakeCamera(duplicator)) + duplicator.update_frame() + assert hook.consume(100, 100) is None + + +def test_hook_clips_metadata_to_framebuffer(): + reader = _FakeReader([ + DXGIFrameMetadata( + dirty_regions=((-10, -5, 30, 20), (200, 200, 10, 10)), + move_rects=(CaptureMoveRect(5, 5, -2, -3, 10, 10),), + ) + ]) + duplicator = _FakeDuplicator() + hook = DXGIMetadataHook(reader=reader) + assert hook.ensure_attached(_FakeCamera(duplicator)) + duplicator.update_frame() + + metadata = hook.consume(100, 100) + assert metadata is not None + assert metadata.dirty_regions == ((0, 0, 20, 15),) + assert metadata.move_rects == ( + CaptureMoveRect(src_x=7, src_y=8, dst_x=0, dst_y=0, width=8, height=7), + ) + + +def test_invalid_move_source_invalidates_native_metadata(): + reader = _FakeReader([ + DXGIFrameMetadata( + dirty_regions=(), + move_rects=(CaptureMoveRect(-5, 0, 20, 20, 10, 10),), + ) + ]) + duplicator = _FakeDuplicator() + hook = DXGIMetadataHook(reader=reader) + assert hook.ensure_attached(_FakeCamera(duplicator)) + duplicator.update_frame() + assert hook.consume(100, 100) is None + + +def test_raw_vtable_metadata_buffer_reader(): + item = dxgi_metadata._Rect(1, 2, 11, 22) + payload = bytes(item) + callback_type = dxgi_metadata._WINFUNCTYPE( + dxgi_metadata._HRESULT, + ctypes.c_void_p, + dxgi_metadata._UINT, + ctypes.c_void_p, + ctypes.POINTER(dxgi_metadata._UINT), + ) + + def callback(_this, buffer_size, buffer, required): + required[0] = len(payload) + if int(buffer_size) < len(payload) or not buffer: + return -1 + ctypes.memmove(buffer, payload, len(payload)) + return 0 + + callback_fn = callback_type(callback) + vtable = (ctypes.c_void_p * 15)() + vtable[dxgi_metadata._GET_FRAME_DIRTY_RECTS_INDEX] = ctypes.cast( + callback_fn, ctypes.c_void_p + ).value + + class _FakeComObject(ctypes.Structure): + _fields_ = [("vtable", ctypes.POINTER(ctypes.c_void_p))] + + obj = _FakeComObject(ctypes.cast(vtable, ctypes.POINTER(ctypes.c_void_p))) + reader = DXGIMetadataReader() + raw = reader._read_buffer( + ctypes.pointer(obj), + dxgi_metadata._GET_FRAME_DIRTY_RECTS_INDEX, + ctypes.sizeof(dxgi_metadata._Rect), + ) + assert raw == payload + assert reader._decode_dirty(raw) == [(1, 2, 10, 20)] + + +def test_dxcam_backend_keeps_move_destination_dirty_for_safe_fallback(): + move = CaptureMoveRect(1, 2, 20, 30, 10, 12) + + class FakeMetadataHook: + is_available = True + + def ensure_attached(self, _camera): + return True + + def consume(self, _width, _height): + return DXGIFrameMetadata(((4, 5, 6, 7),), (move,)) + + class FakeOwner: + logger = logging.getLogger("test.dxgi.backend") + scale_factor = 1.0 + enable_dxgi_metadata = True + _dxcam_available = True + + def __init__(self): + self.camera = _FakeCamera(_FakeDuplicator()) + + def _get_dxcam_session(self): + return self.camera + + backend = DXCamCaptureBackend(FakeOwner()) + backend._metadata_hook = FakeMetadataHook() + metadata = backend.build_metadata(100, 100) + + assert metadata.dirty_regions == [(4, 5, 6, 7), (20, 30, 10, 12)] + assert metadata.move_rects == [move] + assert metadata.supports_dirty_regions is True + assert metadata.supports_move_rects is True + + +def test_dxcam_healthcheck_does_not_create_camera(): + class FakeOwner: + logger = logging.getLogger("test.dxgi.healthcheck") + _dxcam_available = True + + def _get_dxcam_session(self): + raise AssertionError("healthcheck must not create a DXCamera") + + backend = DXCamCaptureBackend(FakeOwner()) + + assert backend.healthcheck() is True + + +def test_dxcam_capability_probe_does_not_create_camera(): + class ProbeOwner: + logger = logging.getLogger("test.dxgi.capability-probe") + scale_factor = 1.0 + enable_dxgi_metadata = True + _dxcam_available = True + + def _get_dxcam_session(self): + raise AssertionError("capability probe must not create a DXCamera") + + backend = DXCamCaptureBackend(ProbeOwner()) + metadata = backend.build_metadata(0, 0) + + assert metadata.backend_name == "dxcam+dxgi-metadata" + assert metadata.supports_dirty_regions is True + assert metadata.supports_move_rects is True + + +def test_dxcam_capability_probe_respects_metadata_disable_flag(): + class ProbeOwner: + logger = logging.getLogger("test.dxgi.capability-probe-disabled") + scale_factor = 1.0 + enable_dxgi_metadata = False + _dxcam_available = True + + def _get_dxcam_session(self): + raise AssertionError("capability probe must not create a DXCamera") + + backend = DXCamCaptureBackend(ProbeOwner()) + metadata = backend.build_metadata(0, 0) + + assert metadata.backend_name == "dxcam" + assert metadata.supports_dirty_regions is False + assert metadata.supports_move_rects is False diff --git a/tests/test_e2e_encoding_switch.py b/tests/test_e2e_encoding_switch.py new file mode 100644 index 0000000..b431ece --- /dev/null +++ b/tests/test_e2e_encoding_switch.py @@ -0,0 +1,221 @@ +"""Real socket regression tests for switching RFB encodings mid-session.""" + +from __future__ import annotations + +import socket +import struct +import threading + +from pyvncserver.app import server as server_module +from vnc_lib.capture_backends import CaptureBackendCapabilities, CaptureFrame, CaptureMetadata +from vnc_lib.screen_capture import CaptureResult + + +class _SolidCapture: + def __init__(self, scale_factor=1.0, monitor=0, backend_preference="auto"): + self.scale_factor = scale_factor + self.monitor = monitor + self.backend_preference = backend_preference + self._backend = self + self._pixels = bytes([10, 20, 30, 0] * 16) + + def set_cache_frame_rate(self, _fps): + return None + + def get_backend_name(self): + return "fake" + + def get_backend_capabilities(self): + return CaptureBackendCapabilities( + name="fake", supports_bgra=True, supports_rgb=True, + supports_pil_image=False, supports_dirty_regions=True, + supports_move_rects=False, + ) + + def healthcheck(self): + return True + + def capture_frame(self, _pixel_format): + return CaptureFrame( + result=CaptureResult(self._pixels, None, 4, 4, 0.0001), + metadata=CaptureMetadata( + backend_name="fake", dirty_regions=[(0, 0, 4, 4)], + supports_dirty_regions=True, + ), + ) + + def capture_fast(self, pixel_format): + return self.capture_frame(pixel_format).result + + def convert_native_bgr0(self, pixel_data, _width, _height, _pixel_format): + return pixel_data + + def close_current_thread_sessions(self): + return None + + +class _FakeInputHandler: + def __init__(self, scale_factor=1.0): + self.scale_factor = scale_factor + + def handle_pointer_event(self, *_args, **_kwargs): + return None + + def handle_key_event(self, *_args, **_kwargs): + return None + + +def _recv_exact(sock: socket.socket, size: int) -> bytes: + data = bytearray() + while len(data) < size: + chunk = sock.recv(size - len(data)) + if not chunk: + raise AssertionError(f"connection closed after {len(data)}/{size} bytes") + data.extend(chunk) + return bytes(data) + + +def _free_tcp_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def _handshake(client: socket.socket): + version = _recv_exact(client, 12) + assert version == b"RFB 003.008\n" + client.sendall(version) + count = _recv_exact(client, 1)[0] + security_types = _recv_exact(client, count) + assert 1 in security_types + client.sendall(b"\x01") + assert _recv_exact(client, 4) == b"\x00\x00\x00\x00" + client.sendall(b"\x01") + width, height = struct.unpack(">HH", _recv_exact(client, 4)) + _recv_exact(client, 16) + name_len = struct.unpack(">I", _recv_exact(client, 4))[0] + _recv_exact(client, name_len) + assert (width, height) == (4, 4) + + +def _set_encodings(client: socket.socket, encodings: list[int]): + payload = struct.pack(">BBH", 2, 0, len(encodings)) + payload += b"".join(struct.pack(">i", value) for value in encodings) + client.sendall(payload) + + +def _request_full(client: socket.socket): + client.sendall(struct.pack(">BBHHHH", 3, 0, 0, 0, 4, 4)) + + +def _read_rect_header(client: socket.socket): + message_type, count = struct.unpack(">BxH", _recv_exact(client, 4)) + assert message_type == 0 + assert count == 1 + return struct.unpack(">HHHHi", _recv_exact(client, 12)) + + +def test_switch_rre_to_tight_without_disconnect(tmp_path, monkeypatch): + monkeypatch.setattr(server_module, "ScreenCapture", _SolidCapture) + monkeypatch.setattr(server_module, "InputHandler", _FakeInputHandler) + + port = _free_tcp_port() + config = tmp_path / "server.toml" + config.write_text( + f'''[server]\nhost = "127.0.0.1"\nport = {port}\nframe_rate = 30\nlan_frame_rate = 30\nnetwork_profile_override = "auto"\nmax_connections = 2\nmax_connections_per_ip = 2\nmax_unauthenticated_connections = 2\nhandshake_timeout = 2.0\nclient_socket_timeout = 2.0\n\n[features]\nenable_metrics = false\nenable_health_checks = false\nenable_capture_producer = false\nenable_parallel_encoding = false\nenable_region_detection = false\nenable_websocket = false\nenable_tight_extensions = false\nenable_tight_encoding = true\nenable_jpeg_encoding = false\nenable_zrle_encoding = false\nenable_copyrect_encoding = false\n''', + encoding="utf-8", + ) + + server = server_module.VNCServerV3(config) + thread = threading.Thread(target=server.start, name="test-vnc-encoding-switch") + thread.start() + client = socket.create_connection(("127.0.0.1", port), timeout=2.0) + client.settimeout(2.0) + try: + _handshake(client) + + _set_encodings(client, [2, 0]) + _request_full(client) + x, y, width, height, encoding = _read_rect_header(client) + assert (x, y, width, height, encoding) == (0, 0, 4, 4, 2) + subrect_count = struct.unpack(">I", _recv_exact(client, 4))[0] + background = _recv_exact(client, 4) + assert subrect_count == 0 + assert background == bytes([10, 20, 30, 0]) + + _set_encodings(client, [7, 0]) + _request_full(client) + x, y, width, height, encoding = _read_rect_header(client) + assert (x, y, width, height, encoding) == (0, 0, 4, 4, 7) + control = _recv_exact(client, 1)[0] + assert control >> 4 == 8 # Tight fill + assert _recv_exact(client, 3) == bytes([30, 20, 10]) + finally: + client.close() + server.shutdown_handler.shutdown() + thread.join(timeout=5.0) + assert not thread.is_alive() + + +def _read_and_validate_payload(client: socket.socket, expected_encoding: int): + import zlib + + x, y, width, height, encoding = _read_rect_header(client) + assert (x, y, width, height) == (0, 0, 4, 4) + assert encoding == expected_encoding + + if encoding == 0: # Raw + assert _recv_exact(client, 64) == bytes([10, 20, 30, 0] * 16) + elif encoding == 2: # RRE solid background + assert struct.unpack(">I", _recv_exact(client, 4))[0] == 0 + assert _recv_exact(client, 4) == bytes([10, 20, 30, 0]) + elif encoding == 5: # Hextile uses raw tile in current implementation + assert _recv_exact(client, 1) == b"\x01" + assert _recv_exact(client, 64) == bytes([10, 20, 30, 0] * 16) + elif encoding == 6: # Zlib + length = struct.unpack(">I", _recv_exact(client, 4))[0] + compressed = _recv_exact(client, length) + decoder = zlib.decompressobj() + assert decoder.decompress(compressed) == bytes([10, 20, 30, 0] * 16) + elif encoding == 16: # ZRLE solid tile: subencoding 1 + CPIXEL + length = struct.unpack(">I", _recv_exact(client, 4))[0] + compressed = _recv_exact(client, length) + decoder = zlib.decompressobj() + assert decoder.decompress(compressed) == bytes([1, 10, 20, 30]) + elif encoding == 7: # Tight fill + control = _recv_exact(client, 1)[0] + assert control >> 4 == 8 + assert _recv_exact(client, 3) == bytes([30, 20, 10]) + else: + raise AssertionError(f"unsupported test encoding {encoding}") + + +def test_live_session_encoding_matrix(tmp_path, monkeypatch): + monkeypatch.setattr(server_module, "ScreenCapture", _SolidCapture) + monkeypatch.setattr(server_module, "InputHandler", _FakeInputHandler) + + port = _free_tcp_port() + config = tmp_path / "server-matrix.toml" + config.write_text( + f'''[server]\nhost = "127.0.0.1"\nport = {port}\nframe_rate = 30\nlan_frame_rate = 30\nnetwork_profile_override = "auto"\nmax_connections = 2\nmax_connections_per_ip = 2\nmax_unauthenticated_connections = 2\nhandshake_timeout = 2.0\nclient_socket_timeout = 2.0\n\n[features]\nenable_metrics = false\nenable_health_checks = false\nenable_capture_producer = false\nenable_parallel_encoding = false\nenable_region_detection = false\nenable_websocket = false\nenable_tight_extensions = false\nenable_tight_encoding = true\nenable_jpeg_encoding = false\nenable_zrle_encoding = true\nenable_copyrect_encoding = false\n''', + encoding="utf-8", + ) + + server = server_module.VNCServerV3(config) + thread = threading.Thread(target=server.start, name="test-vnc-encoding-matrix") + thread.start() + client = socket.create_connection(("127.0.0.1", port), timeout=2.0) + client.settimeout(2.0) + try: + _handshake(client) + for encoding in (0, 2, 5, 6, 16, 7, 0): + _set_encodings(client, [encoding, 0] if encoding != 0 else [0]) + _request_full(client) + _read_and_validate_payload(client, encoding) + finally: + client.close() + server.shutdown_handler.shutdown() + thread.join(timeout=5.0) + assert not thread.is_alive()