fdo-mirrors/xserver
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/xorg/xserver.git synced 2026-06-07 02:58:22 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

pyxtest: rework the request handling to avoid to_bytes() invocations

Browse source 
xclient.send_request() should just take a Request object and handle
to_bytes with the right byte order. This avoids typos/copy-paste errors
in tests when the byte order changes between tests.

Part-of: <https://gitlab.freedesktop.org/xorg/xserver/-/merge_requests/2216>
This commit is contained in:
Peter Hutterer 2026-05-07 15:01:34 +10:00 • committed by Marge Bot
parent 7e22a5cfb5
commit 2409bfda88
14 changed files with 94 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
test/pyxtest/README.md
View file

@ -165,7 +165,9 @@ AddressSanitizer. This is set automatically by meson when
return xclient
```
4. Build protocol requests using dataclasses from `proto/`:
4. Build protocol requests using dataclasses from `proto/` and send them
with `send_request()`. The byte order is handled automatically based
on the connection type (native or swapped):
```python
from proto import xi
@ -175,7 +177,7 @@ AddressSanitizer. This is set automatically by meson when
num_changes=1,
changes_data=change_data,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
```
5. If a new extension module is needed, create `proto/myext.py` with

11
test/pyxtest/proto/xkb.py
View file

@ -77,17 +77,6 @@ class UseExtensionRequest:
)
# Keep as a module-level function for backward compatibility with
# xclient.py which calls xkb.use_extension() directly.
def use_extension(
opcode: int, *, major: int = 1, minor: int = 0, byte_order: str = "<"
) -> bytes:
"""Build XkbUseExtension request (minor opcode 0)."""
return UseExtensionRequest(opcode=opcode, major=major, minor=minor).to_bytes(
byte_order
)
@dataclass
class GetMapRequest:
"""xkbGetMapReq (minor opcode 8). 28 bytes."""

14
test/pyxtest/test_present.py
View file

@ -5,7 +5,7 @@
import pytest
from proto import present
from xclient import Extension, X11Error
from xclient import BadWindow, Extension, X11Error
class TestPresentSelectInput:
@ -27,7 +27,7 @@ class TestPresentSelectInput:
pytest.skip("Present extension not available")
req = present.QueryVersionRequest(opcode=ext.opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
conn.recv_response(timeout=5.0)
win = conn.create_window()
@ -43,7 +43,7 @@ class TestPresentSelectInput:
window=win,
event_mask=PresentConfigureNotifyMask,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
responses = conn.flush_responses(timeout=1.0)
assert xserver.is_alive, "Server crashed"
@ -93,7 +93,7 @@ class TestPresentNotify:
pytest.skip("Present extension not available")
req = present.QueryVersionRequest(opcode=ext.opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
conn.recv_response(timeout=5.0)
win = conn.create_window()
@ -112,7 +112,7 @@ class TestPresentNotify:
serial=0,
notifies=[notify],
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
responses = conn.flush_responses(timeout=1.0)
assert xserver.is_alive, "Server crashed"
@ -123,7 +123,9 @@ class TestPresentNotify:
# Without the fix: BadWindow because the notify's window ID
# was not byte-swapped.
bad_window_errors = [
r for r in responses if isinstance(r, X11Error) and r.error_code == 3
r
for r in responses
if isinstance(r, X11Error) and r.error_code == BadWindow
]
assert len(bad_window_errors) == 0, (
f"PresentPixmap returned BadWindow error(s): "

22
test/pyxtest/test_randr.py
View file

@ -16,7 +16,7 @@ def _get_first_output(xclient, opcode):
opcode=opcode,
window=xclient.root_window,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resp = xclient.recv_response(timeout=5.0)
if not isinstance(resp, X11Reply) or len(resp.data) < 32:
pytest.skip("Failed to get RandR screen resources")
@ -38,7 +38,7 @@ def randr_xclient(xclient):
pytest.skip("RANDR extension not available")
req = randr.QueryVersionRequest(opcode=ext.opcode)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
xclient.recv_response(timeout=5.0)
output_id = _get_first_output(xclient, ext.opcode)
@ -53,7 +53,7 @@ def randr_xclient_swapped(xclient_swapped):
pytest.skip("RANDR extension not available")
req = randr.QueryVersionRequest(opcode=ext.opcode)
xclient_swapped.send_request(req.to_bytes(">"))
xclient_swapped.send_request(req)
xclient_swapped.recv_response(timeout=5.0)
return xclient_swapped, ext.opcode
@ -97,7 +97,7 @@ class TestRandROutputProperty:
mode=randr.PropModeReplace,
data=initial_data,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
xclient.flush_responses(timeout=0.5)
# Step 2: Prepend values [1, 2]
@ -111,7 +111,7 @@ class TestRandROutputProperty:
mode=randr.PropModePrepend,
data=prepend_data,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
xclient.flush_responses(timeout=0.5)
assert xserver.is_alive, (
@ -125,7 +125,7 @@ class TestRandROutputProperty:
property_atom=prop_atom,
type_atom=type_atom,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resp = xclient.recv_response(timeout=2.0)
assert isinstance(resp, X11Reply), f"Expected a reply, got {resp}"
@ -168,7 +168,7 @@ class TestRandROutputProperty:
num_items=0x40000000,
data=b"",
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resp = xclient.recv_response(timeout=2.0)
assert xserver.is_alive, (
@ -209,7 +209,7 @@ class TestRandRSetScreenConfig:
opcode=opcode,
window=conn.root_window,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), f"Expected reply, got {resp}"
@ -228,7 +228,7 @@ class TestRandRSetScreenConfig:
size_id=0,
rotation=1,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"
@ -275,7 +275,7 @@ class TestRandRCreateLease:
opcode=opcode,
window=conn.root_window,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), f"Expected reply, got {resp}"
@ -297,7 +297,7 @@ class TestRandRCreateLease:
crtcs=crtcs,
outputs=outputs,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"

4
test/pyxtest/test_record.py
View file

@ -18,7 +18,7 @@ def record_xclient_swapped(xclient_swapped):
pytest.skip("RECORD extension not available")
req = record.QueryVersionRequest(opcode=ext.opcode)
xclient_swapped.send_request(req.to_bytes(">"))
xclient_swapped.send_request(req)
xclient_swapped.recv_response(timeout=5.0)
return xclient_swapped, ext.opcode
@ -59,7 +59,7 @@ class TestRecordCreateContext:
n_clients_override=100, # Claim 100 clients
n_ranges_override=1,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
time.sleep(0.5)
assert xserver.is_alive, (

8
test/pyxtest/test_render.py
View file

@ -18,7 +18,7 @@ def render_xclient_swapped(xclient_swapped):
pytest.skip("RENDER extension not available")
req = render.QueryVersionRequest(opcode=ext.opcode)
xclient_swapped.send_request(req.to_bytes(">"))
xclient_swapped.send_request(req)
xclient_swapped.recv_response(timeout=5.0)
return xclient_swapped, ext.opcode
@ -49,7 +49,7 @@ class TestRenderSetPictureFilter:
# followed by numFormats xPictFormInfo entries (28 bytes each):
# [0] id(4) [4] type(1) [5] depth(1) ...
req = render.QueryPictFormatsRequest(opcode=opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), "QueryPictFormats failed"
@ -78,7 +78,7 @@ class TestRenderSetPictureFilter:
drawable=pix,
format_id=format_id,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
errors = conn.flush_responses(timeout=0.5)
create_errors = [r for r in errors if isinstance(r, X11Error)]
assert len(create_errors) == 0, f"CreatePicture failed: {create_errors}"
@ -98,7 +98,7 @@ class TestRenderSetPictureFilter:
0x00010000,
],
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
responses = conn.flush_responses(timeout=1.0)
assert xserver.is_alive, "Server crashed"

2
test/pyxtest/test_screensaver.py
View file

@ -38,7 +38,7 @@ class TestScreenSaverSuspend:
opcode=ext.opcode,
suspend=1,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
time.sleep(0.5)
assert xserver.is_alive, (

4
test/pyxtest/test_shm.py
View file

@ -37,7 +37,7 @@ class TestShmCreateSegment:
# Query SHM version
req = shm.QueryVersionRequest(opcode=ext.opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
if not isinstance(resp, X11Reply):
@ -61,7 +61,7 @@ class TestShmCreateSegment:
shmseg=shmseg_id,
size=4096,
read_only=False,
).to_bytes(">")
)
)
# Read the 32-byte reply header. The fd arrives as

8
test/pyxtest/test_vidmode.py
View file

@ -38,7 +38,7 @@ class TestVidModeSwitchToMode:
pytest.skip("XF86-VidModeExtension not available")
req = vidmode.QueryVersionRequest(opcode=ext.opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
if not isinstance(resp, X11Reply):
pytest.skip("VidMode QueryVersion failed")
@ -48,14 +48,14 @@ class TestVidModeSwitchToMode:
opcode=ext.opcode,
major=vidmode_version,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
# Get the current mode line so we can echo it back.
req = vidmode.GetModeLineRequest(
opcode=ext.opcode,
screen=0,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
if isinstance(resp, X11Error):
@ -141,7 +141,7 @@ class TestVidModeSwitchToMode:
flags=flags,
privsize=0,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=2.0)
assert xserver.is_alive, "Server crashed"

24
test/pyxtest/test_xi.py
View file

@ -18,7 +18,7 @@ def xi_xclient(xclient):
pytest.skip("XInput extension not available")
req = xi.XIQueryVersionRequest(opcode=ext.opcode)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
xclient.recv_response(timeout=5.0)
return xclient
@ -31,7 +31,7 @@ def xi_xclient_swapped(xclient_swapped):
pytest.skip("XInput extension not available")
req = xi.XIQueryVersionRequest(opcode=ext.opcode)
xclient_swapped.send_request(req.to_bytes(">"))
xclient_swapped.send_request(req)
xclient_swapped.recv_response(timeout=5.0)
return xclient_swapped
@ -65,7 +65,7 @@ class TestXIPassiveGrab:
detail=256, # OOB: valid range is 0-255
grab_type=xi.XIGrabtypeButton,
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
resp = xi_xclient.recv_response(timeout=2.0)
assert xserver.is_alive, (
@ -108,7 +108,7 @@ class TestXIPassiveGrab:
detail=256, # OOB: valid range is 0-255
grab_type=xi.XIGrabtypeButton,
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
resp = xi_xclient.recv_response(timeout=2.0)
assert xserver.is_alive, (
@ -157,7 +157,7 @@ class TestXIChangeProperty:
num_items=0x40000000,
data=b"", # no actual data
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
resp = xi_xclient.recv_response(timeout=2.0)
assert xserver.is_alive, (
@ -208,7 +208,7 @@ class TestXIChangeProperty:
mode=xi.PropModeReplace,
data=initial_data,
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
xi_xclient.flush_responses(timeout=0.5)
# Step 2: Prepend values [1, 2]
@ -222,7 +222,7 @@ class TestXIChangeProperty:
mode=xi.PropModePrepend,
data=prepend_data,
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
xi_xclient.flush_responses(timeout=0.5)
assert xserver.is_alive, (
@ -236,7 +236,7 @@ class TestXIChangeProperty:
property_atom=prop_atom,
type_atom=type_atom,
)
xi_xclient.send_request(req.to_bytes())
xi_xclient.send_request(req)
resp = xi_xclient.recv_response(timeout=2.0)
assert isinstance(resp, X11Reply), f"Expected a reply, got {resp}"
@ -293,7 +293,7 @@ class TestXIChangeProperty:
mode=xi.PropModeReplace,
data=data,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
conn.flush_responses(timeout=0.5)
assert xserver.is_alive, "Server crashed during ChangeProperty"
@ -304,7 +304,7 @@ class TestXIChangeProperty:
property_atom=prop_atom,
type_atom=type_atom,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=2.0)
assert isinstance(resp, X11Reply), f"Expected a reply, got {resp}"
@ -359,7 +359,7 @@ class TestXIChangeDeviceControl:
pytest.skip("XInput extension not available")
req = xi.XIQueryVersionRequest(opcode=ext.opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
conn.recv_response(timeout=5.0)
ctl = xi.DeviceResolutionCtl(
@ -375,7 +375,7 @@ class TestXIChangeDeviceControl:
deviceid=xi.VirtualCorePointer,
control_data=ctl_bytes,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=2.0)
assert xserver.is_alive, "Server crashed"

6
test/pyxtest/test_xinerama.py
View file

@ -28,7 +28,7 @@ class TestPseudoramiXGetState:
opcode=ext.opcode,
window=conn.root_window,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"
@ -58,7 +58,7 @@ class TestPseudoramiXGetScreenCount:
opcode=ext.opcode,
window=conn.root_window,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"
@ -90,7 +90,7 @@ class TestPseudoramiXGetScreenSize:
window=conn.root_window,
screen=0,
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"

22
test/pyxtest/test_xkb.py
View file

@ -71,7 +71,7 @@ class TestXkbSetMapOverflows:
max_key_code=255,
payload=b"\x00" * 8, # Way too little data
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
time.sleep(0.5)
assert xserver.is_alive, "Server crashed - missing length checks in SetMap"
@ -98,7 +98,7 @@ class TestXkbSetMapOverflows:
# Step 1: Query the server's key sym map to learn symsPerKey
get_map = xkb.GetMapRequest(opcode=opcode, full=xkb.XkbKeySymsMask)
xclient.send_request(get_map.to_bytes())
xclient.send_request(get_map)
resp = xclient.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), f"Expected GetMap reply, got {resp}"
@ -147,7 +147,7 @@ class TestXkbSetMapOverflows:
total_acts=0, # Mismatch: real sum is real_n_acts
payload=payload,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resps = xclient.flush_responses(timeout=0.5)
errors = [r for r in resps if isinstance(r, X11Error)]
@ -208,7 +208,7 @@ class TestXkbSetGeometry:
name_atom=name_atom,
payload=color_data + shape_data,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resp = xclient.recv_response(timeout=2.0)
assert xserver.is_alive, "Server crashed - truncated sections in SetGeometry"
@ -259,7 +259,7 @@ class TestXkbSetGeometry:
name_atom=name_atom,
payload=payload,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resps = xclient.flush_responses(timeout=0.5)
errors = [r for r in resps if isinstance(r, X11Error)]
@ -313,7 +313,7 @@ class TestXkbSetGeometry:
name_atom=name_atom,
payload=label_font + color_data + shape_data,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resps = xclient.flush_responses(timeout=0.5)
errors = [r for r in resps if isinstance(r, X11Error)]
@ -377,7 +377,7 @@ class TestXkbSetGeometry:
name_atom=name_atom,
payload=payload,
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
resps = xclient.flush_responses(timeout=0.5)
errors = [r for r in resps if isinstance(r, X11Error)]
@ -422,7 +422,7 @@ class TestXkbSetCompatMap:
truncate_si=1,
payload=payload1,
)
xclient.send_request(req1.to_bytes())
xclient.send_request(req1)
xclient.flush_responses(timeout=0.5)
# Step 2: Truncate to 2 (sets num_si=2, size_si stays 10)
@ -434,7 +434,7 @@ class TestXkbSetCompatMap:
truncate_si=1,
payload=payload2,
)
xclient.send_request(req2.to_bytes())
xclient.send_request(req2)
xclient.flush_responses(timeout=0.5)
# Step 3: Write at index 8-11 without truncation.
@ -448,7 +448,7 @@ class TestXkbSetCompatMap:
truncate_si=0,
payload=payload3,
)
xclient.send_request(req3.to_bytes())
xclient.send_request(req3)
time.sleep(0.5)
assert xserver.is_alive, (
@ -486,7 +486,7 @@ class TestXkbSetNames:
n_radio_groups=10, # 10 radio group names expected
payload=b"\x00" * 12, # Way too little for all those atoms
)
xclient.send_request(req.to_bytes())
xclient.send_request(req)
time.sleep(0.5)
assert xserver.is_alive, "Server crashed - truncated atoms in SetNames"

8
test/pyxtest/test_xres.py
View file

@ -18,7 +18,7 @@ def xres_xclient_swapped(xclient_swapped):
pytest.skip("X-Resource extension not available")
req = xres.QueryVersionRequest(opcode=ext.opcode)
xclient_swapped.send_request(req.to_bytes(">"))
xclient_swapped.send_request(req)
resp = xclient_swapped.recv_response(timeout=5.0)
if not isinstance(resp, X11Reply):
pytest.skip("XRes QueryVersion failed")
@ -48,7 +48,7 @@ class TestXResQueryClientIds:
opcode=opcode,
specs=[(0, 1)], # mask=1 (XResClientXIDMask)
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert xserver.is_alive, "Server crashed"
@ -87,7 +87,7 @@ class TestXResQueryClientIds:
# Get the list of clients
req = xres.QueryClientsRequest(opcode=opcode)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), f"Expected reply, got {resp}"
@ -119,7 +119,7 @@ class TestXResQueryClientIds:
opcode=opcode,
specs=[(target_xid, 1)], # XResClientXIDMask
)
conn.send_request(req.to_bytes(">"))
conn.send_request(req)
resp = conn.recv_response(timeout=5.0)
assert isinstance(resp, X11Reply), f"Expected reply, got {resp}"

37
test/pyxtest/xclient.py
View file

@ -8,6 +8,7 @@ import struct
import time
from dataclasses import dataclass
from enum import StrEnum
from typing import Protocol, runtime_checkable
from proto.bigrequests import BigRequestsEnableRequest
from proto.x11 import (
@ -34,6 +35,16 @@ class XExtensionData:
first_error: int
@runtime_checkable
class X11Request(Protocol):
"""Protocol for X11 request objects that can be serialized to wire format.
All ``@dataclass`` request builders in ``proto/`` satisfy this protocol.
"""
def to_bytes(self, byte_order: str = "<") -> bytes: ...
class Extension(StrEnum):
"""X11 extension wire names as used in QueryExtension requests."""
@ -243,7 +254,16 @@ class RawX11Connection:
self._next_resource_id += 1
return xid
def send_request(self, data: bytes) -> int:
def send_request(self, data: bytes | X11Request) -> int:
"""Send a raw or structured X11 request.
*data* may be a ``bytes`` object (sent as-is) or any object that
implements :class:`X11Request` (i.e. has a ``to_bytes(byte_order)``
method). In the latter case ``to_bytes`` is called automatically
with the connection's byte order.
"""
if isinstance(data, X11Request):
data = data.to_bytes(self._byte_order)
self.seq += 1
assert self.sock
self.sock.sendall(data)
@ -325,7 +345,7 @@ class RawX11Connection:
return self._extensions[name]
req = QueryExtensionRequest(name=name)
self.send_request(req.to_bytes(self._byte_order))
self.send_request(req)
resp = self.recv_response(timeout=5.0)
if resp is None or isinstance(resp, X11Error):
return None
@ -345,11 +365,10 @@ class RawX11Connection:
if not ext:
raise X11ConnectionError("XKB extension not available")
req = xkb.use_extension(
ext.opcode,
req = xkb.UseExtensionRequest(
opcode=ext.opcode,
major=major_version,
minor=minor_version,
byte_order=self._byte_order,
)
self.send_request(req)
@ -364,7 +383,7 @@ class RawX11Connection:
raise X11ConnectionError("BIG-REQUESTS not available")
req = BigRequestsEnableRequest(opcode=ext.opcode)
self.send_request(req.to_bytes(self._byte_order))
self.send_request(req)
resp = self.recv_response(timeout=5.0)
if isinstance(resp, X11Error):
@ -402,7 +421,7 @@ class RawX11Connection:
height=height,
depth=depth,
)
self.send_request(req.to_bytes(self._byte_order))
self.send_request(req)
self.flush_responses(timeout=0.2)
return wid
@ -429,13 +448,13 @@ class RawX11Connection:
height=height,
depth=depth,
)
self.send_request(req.to_bytes(self._byte_order))
self.send_request(req)
self.flush_responses(timeout=0.2)
return pid
def intern_atom(self, name: str, only_if_exists: bool = False) -> int:
req = InternAtomRequest(name=name, only_if_exists=only_if_exists)
self.send_request(req.to_bytes(self._byte_order))
self.send_request(req)
resp = self.recv_response(timeout=5.0)
if isinstance(resp, X11Error) or resp is None: