160 lines
8.3 KiB
Python
160 lines
8.3 KiB
Python
"""Check event feedback and normal shutdown while a Wine MSAA provider stalls.
|
|
|
|
Run on a private X display. Uses a disposable prefix and a recording D-Bus
|
|
service; no speech reaches the user's reader or physical braille 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("--reader", choices=("cthulhu", "orca"), default="cthulhu")
|
|
parser.add_argument("--session", action="store_true")
|
|
parser.add_argument("--getter", action="store_true")
|
|
parser.add_argument("--background", action="store_true")
|
|
parser.add_argument("--popup", 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), "--reader", args.reader,
|
|
*(["--32-bit"] if args.is32Bit else []),
|
|
*(["--getter"] if args.getter else []),
|
|
*(["--background"] if args.background else []),
|
|
*(["--popup"] if args.popup else [])], build, "event_stall_integration",
|
|
keep=args.keep, keepOnFailure=args.keepOnFailure))
|
|
|
|
interface = "org.gnome.Orca1.Service" if args.reader == "orca" else "org.stormux.Cthulhu1.Service"
|
|
objectPath = "/" + interface.replace(".", "/")
|
|
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 = []
|
|
|
|
|
|
def record(connection, sender, path, iface, method, parameters, invocation):
|
|
events.append(parameters.unpack())
|
|
invocation.return_value(GLib.Variant("(b)", (True,)))
|
|
|
|
|
|
connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
|
connection.register_object(objectPath, 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)
|
|
|
|
|
|
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-event-stall-") as temporary:
|
|
root = Path(temporary)
|
|
env = os.environ.copy()
|
|
env.update(WINEPREFIX=str(root / "prefix"), WINEDEBUG="-all",
|
|
WINEDLLOVERRIDES="mscoree,mshtml=", CTHULHU_WINE_ACCESS_MANAGED="1")
|
|
ready, done, checked = root / "ready", root / "done", root / "checked"
|
|
for key, path in (("WINE_ACCESS_STALL_READY", ready), ("WINE_ACCESS_STALL_DONE", done),
|
|
("WINE_ACCESS_EDITOR_CHECKED", checked)):
|
|
env[key] = "Z:" + str(path).replace("/", "\\")
|
|
fixture = root / "fixture.exe"
|
|
compiler = "i686-w64-mingw32-g++" if args.is32Bit else "x86_64-w64-mingw32-g++"
|
|
subprocess.run([compiler, "-Wall", "-Wextra", "-mwindows", "-static", "-o", str(fixture),
|
|
str(Path(__file__).with_name("event_stall_fixture.cpp")), "-luser32", "-loleacc", "-lole32", "-loleaut32", "-luuid"], check=True)
|
|
children = []
|
|
try:
|
|
subprocess.run(["wineboot", "-u"], env=env, check=True, timeout=45)
|
|
helper = subprocess.Popen([str(build / "wine-access.exe"), "--no-controller",
|
|
"--reader", args.reader], env=env)
|
|
children.append(helper)
|
|
time.sleep(1)
|
|
modes = ["getter"] if args.getter else []
|
|
if args.background: modes.append("background")
|
|
if args.background or args.popup:
|
|
editor = subprocess.Popen(["wine", str(fixture), "editor", *(["popup"] if args.popup else [])], env=env)
|
|
children.append(editor)
|
|
if args.popup:
|
|
pump_until(lambda: editor.poll() is not None or any(e[2] == "Owned popup provider" for e in events), 3)
|
|
assert editor.poll() is None, f"popup fixture failed: {editor.returncode}"
|
|
assert any(e[2] == "Owned popup provider" for e in events)
|
|
print("PASS: accessible owned popup preserves foreground and native focus", flush=True)
|
|
sys.exit(0)
|
|
pump_until(lambda: any(e[1] == "text" and e[2] == "a" for e in events), 3)
|
|
stall = subprocess.Popen(["wine", str(fixture), *modes], env=env)
|
|
children.append(stall)
|
|
deadline = time.monotonic() + 2
|
|
while time.monotonic() < deadline:
|
|
while GLib.MainContext.default().iteration(False):
|
|
pass
|
|
time.sleep(.005)
|
|
assert not ready.exists(), "background provider was queried"
|
|
assert any(e[1] == "focus" and e[2] == "Responsive button" and not e[0].endswith(":standard") for e in events), "background event starved foreground MSAA"
|
|
print("PASS: background provider not queried; ordinary foreground MSAA feedback retained", flush=True)
|
|
sys.exit(0)
|
|
stall = subprocess.Popen(["wine", str(fixture), *modes], env=env)
|
|
children.append(stall)
|
|
pump_until(ready.exists, 10)
|
|
events.clear()
|
|
started = time.monotonic()
|
|
editor = subprocess.Popen(["wine", str(fixture), "editor"], env=env)
|
|
children.append(editor)
|
|
pump_until(lambda: sum(e[1] == "text" and e[2] == "a" for e in events) >= 2, 2)
|
|
elapsed = time.monotonic() - started
|
|
assert not done.exists(), "typing waited for the blocked provider"
|
|
assert helper.poll() is None, "helper exited"
|
|
print(f"PASS: repeated typing while MSAA stalled ({elapsed:.3f} s)", flush=True)
|
|
pump_until(lambda: any(e[1] == "focus" and e[2] == "Responsive button" for e in events), 2)
|
|
assert checked.exists(), "native text, caret or selection changed"
|
|
assert not done.exists(), "native button feedback waited for the blocked provider"
|
|
print("PASS: native button fallback; unchanged native text/caret/selection", flush=True)
|
|
pump_until(done.exists, 10)
|
|
# Drain enough time for an incorrectly queued late result to surface.
|
|
deadline = time.monotonic() + .5
|
|
while time.monotonic() < deadline:
|
|
while GLib.MainContext.default().iteration(False):
|
|
pass
|
|
time.sleep(.005)
|
|
assert not any(e[9] == "Stalled provider fixture" for e in events), "late stale focus escaped"
|
|
print("PASS: stale MSAA focus suppressed", flush=True)
|
|
ready.unlink(); done.unlink()
|
|
second = subprocess.Popen(["wine", str(fixture), *modes], env=env)
|
|
children.append(second)
|
|
pump_until(ready.exists, 10)
|
|
started = time.monotonic()
|
|
subprocess.run(["wine", str(fixture), "quit-helper"], env=env, check=True, timeout=3)
|
|
pump_until(lambda: helper.poll() is not None, 2)
|
|
assert helper.returncode == 0, f"normal helper shutdown failed: {helper.returncode}"
|
|
assert not done.exists(), "normal shutdown waited for the provider"
|
|
print(f"PASS: normal shutdown during MSAA stall ({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)
|