133 lines
6.2 KiB
Python
133 lines
6.2 KiB
Python
"""Check ordered editor feedback while the reader delays or loses replies.
|
|
|
|
Uses a private D-Bus service, disposable Wine prefix and private X display.
|
|
"""
|
|
import argparse
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
from gi.repository import Gio, GLib
|
|
|
|
from wine_test_environment import add_test_options, run_wine_test, wine_test_directory
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--build-dir", default="build")
|
|
parser.add_argument("--32-bit", dest="is32Bit", action="store_true")
|
|
parser.add_argument("--timeout", action="store_true")
|
|
parser.add_argument("--focus-error", action="store_true")
|
|
parser.add_argument("--session", action="store_true")
|
|
add_test_options(parser)
|
|
args = parser.parse_args()
|
|
build = Path(args.build_dir).resolve()
|
|
if not args.session:
|
|
sys.exit(run_wine_test(["dbus-run-session", "--", sys.executable, __file__,
|
|
"--session", "--build-dir", str(build),
|
|
*(["--32-bit"] if args.is32Bit else []),
|
|
*(["--timeout"] if args.timeout else []),
|
|
*(["--focus-error"] if args.focus_error else [])], build, "reader_stall_integration",
|
|
keep=args.keep, keepOnFailure=args.keepOnFailure))
|
|
|
|
interface = "org.stormux.Cthulhu1.Service"
|
|
object_path = "/org/stormux/Cthulhu1/Service"
|
|
xml = '<node><interface name="' + interface + '"><method name="PresentAccessibleEvent">'
|
|
xml += ''.join('<arg type="' + kind + '" direction="in"/>' for kind in ["s"] * 7 + ["i", "i", "s"])
|
|
xml += '<arg type="b" direction="out"/></method>'
|
|
for method, inputs in (("SpeakText", "sb"), ("CancelSpeech", "")):
|
|
xml += '<method name="' + method + '">'
|
|
xml += ''.join('<arg type="' + kind + '" direction="in"/>' for kind in inputs)
|
|
xml += '<arg type="b" direction="out"/></method>'
|
|
xml += '</interface></node>'
|
|
events, held = [], []
|
|
|
|
|
|
def pump_until(predicate, seconds):
|
|
deadline = time.monotonic() + seconds
|
|
while time.monotonic() < deadline:
|
|
while GLib.MainContext.default().iteration(False):
|
|
pass
|
|
if predicate():
|
|
return
|
|
time.sleep(.005)
|
|
raise AssertionError(f"condition timed out; events: {events}")
|
|
|
|
|
|
with wine_test_directory(prefix="wine-reader-stall-") as temporary:
|
|
root = Path(temporary)
|
|
waiting, typed, resume = (root / name for name in ("waiting", "typed", "resume"))
|
|
env = os.environ.copy()
|
|
env.update(WINEPREFIX=str(root / "prefix"), WINEDEBUG="-all",
|
|
WINEDLLOVERRIDES="mscoree,mshtml=", CTHULHU_WINE_ACCESS_MANAGED="1")
|
|
for key, path in (("WAITING", waiting), ("TYPED", typed), ("RESUME", resume)):
|
|
env["WINE_ACCESS_READER_" + key] = "Z:" + str(path).replace("/", "\\")
|
|
|
|
def record(connection, sender, path, iface, method, parameters, invocation):
|
|
event = parameters.unpack()
|
|
first_focus = not waiting.exists() and event[1] == "focus" and event[0].endswith(":text")
|
|
if first_focus and args.focus_error:
|
|
waiting.touch()
|
|
invocation.return_dbus_error("org.freedesktop.DBus.Error.Failed", "Reader temporarily unavailable")
|
|
return
|
|
events.append(event)
|
|
if first_focus:
|
|
held.append(invocation)
|
|
waiting.touch()
|
|
elif event[1] == "text" and event[2] == "z":
|
|
held.append(invocation)
|
|
else:
|
|
invocation.return_value(GLib.Variant("(b)", (True,)))
|
|
|
|
connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
|
connection.register_object(object_path, Gio.DBusNodeInfo.new_for_xml(xml).interfaces[0], record, None, None)
|
|
connection.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus",
|
|
"RequestName", GLib.Variant("(su)", (interface, 0)), None,
|
|
Gio.DBusCallFlags.NONE, 1000, None)
|
|
fixture = root / "fixture.exe"
|
|
compiler = "i686-w64-mingw32-gcc" if args.is32Bit else "x86_64-w64-mingw32-gcc"
|
|
subprocess.run([compiler, "-Wall", "-Wextra", "-mwindows", "-o", str(fixture),
|
|
str(Path(__file__).with_name("reader_stall_fixture.c")), "-luser32"], check=True)
|
|
children = []
|
|
try:
|
|
subprocess.run(["wineboot", "-u"], env=env, check=True, timeout=45)
|
|
helper = subprocess.Popen([str(build / "wine-access.exe"), "--no-controller"], env=env)
|
|
children.append(helper)
|
|
time.sleep(1)
|
|
app = subprocess.Popen(["wine", str(fixture)], env=env)
|
|
children.append(app)
|
|
pump_until(waiting.exists, 5)
|
|
# Longer than the hook's 50 ms deadline, shorter than the bus timeout.
|
|
if not args.focus_error:
|
|
GLib.timeout_add(700 if args.timeout else 350,
|
|
lambda: (held.pop(0).return_value(GLib.Variant("(b)", (True,))), False)[1])
|
|
pump_until(typed.exists, 3)
|
|
pump_until(lambda: len([e for e in events if e[1] == "text"]) >= 8, 3)
|
|
actions = [e[2] for e in events if e[1] == "text"]
|
|
assert actions == list("abbacdef"), f"editor feedback lost or reordered: {actions}"
|
|
focused = None
|
|
for event in events:
|
|
if event[1] == "focus":
|
|
focused = event[0]
|
|
elif event[1] == "text":
|
|
assert event[0] == focused, "text forwarded before its focus"
|
|
print("PASS: delayed reader preserves repeated typing, order, native text and caret", flush=True)
|
|
pump_until(lambda: not held, 2)
|
|
resume.touch()
|
|
pump_until(lambda: bool(held), 2)
|
|
started = time.monotonic()
|
|
quit_client = subprocess.Popen(["wine", str(fixture), "quit-helper"], env=env)
|
|
children.append(quit_client)
|
|
pump_until(lambda: helper.poll() is not None, .4)
|
|
assert helper.returncode == 0, f"helper shutdown failed: {helper.returncode}"
|
|
print(f"PASS: normal shutdown while reader withholds reply ({time.monotonic() - started:.3f} s)", flush=True)
|
|
finally:
|
|
for child in reversed(children):
|
|
if child.poll() is None:
|
|
child.terminate()
|
|
try:
|
|
child.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
child.kill(); child.wait(timeout=3)
|
|
subprocess.run(["wineserver", "-k"], env=env, timeout=5, check=False)
|