184 lines
9.8 KiB
Python
184 lines
9.8 KiB
Python
"""Check NVDA RPC routing against private reader D-Bus services."""
|
|
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("--reader", choices=("cthulhu", "orca"))
|
|
parser.add_argument("--extension-endpoint", 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),
|
|
*(["--extension-endpoint"] if args.extension_endpoint else []),
|
|
*(["--reader", args.reader] if args.reader else [])], build, "controller_integration",
|
|
keep=args.keep, keepOnFailure=args.keepOnFailure))
|
|
|
|
busName = "org.gnome.Orca1.Service" if args.reader == "orca" else "org.stormux.Cthulhu1.Service"
|
|
interface = busName.rsplit(".", 1)[0] + ".WineAccess" if args.extension_endpoint else busName
|
|
otherInterface = "org.stormux.Cthulhu1.Service" if args.reader == "orca" else "org.gnome.Orca1.Service"
|
|
objectPath = "/" + busName.replace(".", "/") + ("/WineAccess" if args.extension_endpoint else "")
|
|
endpointArguments = ["--reader-path", objectPath, "--reader-interface", interface] if args.extension_endpoint else []
|
|
methods = {"SpeakText": ("sb", "b"), "CancelSpeech": ("", "b"),
|
|
"BrailleMessage": ("s", "b"), "SpeakSsml": ("ss", "i")}
|
|
|
|
|
|
def interface_xml(methods):
|
|
xml = '<node><interface name="' + interface + '">'
|
|
for method, (inputs, output) in methods.items():
|
|
xml += '<method name="' + method + '">'
|
|
xml += ''.join('<arg type="' + kind + '" direction="in"/>' for kind in inputs)
|
|
xml += '<arg type="' + output + '" direction="out"/></method>'
|
|
return xml + '</interface></node>'
|
|
|
|
|
|
xml = interface_xml(methods)
|
|
events = []
|
|
|
|
|
|
def record(connection, sender, path, iface, method, parameters, invocation):
|
|
events.append((method, parameters.unpack()))
|
|
invocation.return_value(GLib.Variant("(i)", (120,)) if method == "SpeakSsml"
|
|
else GLib.Variant("(b)", (True,)))
|
|
|
|
|
|
connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
|
registration = connection.register_object(objectPath, Gio.DBusNodeInfo.new_for_xml(xml).interfaces[0], record, None, None)
|
|
|
|
|
|
def name_call(method, name):
|
|
parameters = GLib.Variant("(su)", (name, 0)) if method == "RequestName" else GLib.Variant("(s)", (name,))
|
|
return connection.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus",
|
|
"org.freedesktop.DBus", method, parameters, None,
|
|
Gio.DBusCallFlags.NONE, 1000, None).unpack()[0]
|
|
|
|
|
|
def wait_for_client(client, expected=0):
|
|
deadline = time.monotonic() + 10
|
|
while client.poll() is None and time.monotonic() < deadline:
|
|
while GLib.MainContext.default().iteration(False):
|
|
pass
|
|
time.sleep(.005)
|
|
assert client.poll() == expected, f"client failed or timed out: {client.poll()}"
|
|
|
|
|
|
with wine_test_directory(prefix="wine-access-controller-") as temporary:
|
|
tempPath = Path(temporary)
|
|
env = os.environ.copy()
|
|
env.update(WINEPREFIX=str(tempPath / "prefix"), WINEDEBUG="-all",
|
|
WINEDLLOVERRIDES="mscoree,mshtml=", CTHULHU_WINE_ACCESS_MANAGED="1")
|
|
source = Path(__file__).resolve().parent.parent
|
|
# Like NVDA's clients, bind directly to the local endpoint. WIDL's default
|
|
# automatic binding is unsupported by Wine's client NDR interpreter.
|
|
clientIdl = tempPath / "nvdaController.idl"
|
|
clientIdl.write_text((source / "nvdaController.idl").read_text().replace(
|
|
"\n uuid(", "\n implicit_handle(handle_t controllerBinding),\n uuid("))
|
|
subprocess.run(["widl", "--win64", "-c", "-o", str(tempPath / "nvdaController_c.c"),
|
|
str(clientIdl)], check=True)
|
|
clientPath = tempPath / "controller-fixture.exe"
|
|
subprocess.run(["x86_64-w64-mingw32-gcc", "-Wall", "-Wextra", "-I" + str(tempPath),
|
|
"-I" + str(build / "generated"), "-o", str(clientPath),
|
|
str(source / "tests/controller_fixture.c"), "-lrpcrt4"], check=True)
|
|
helper = None
|
|
client = None
|
|
helperCommand = [str(build / "wine-access.exe"), "--no-dialog-reader",
|
|
*endpointArguments,
|
|
*(["--reader", args.reader] if args.reader else [])]
|
|
try:
|
|
subprocess.run(["wineboot", "-u"], env=env, check=True, timeout=45)
|
|
# Even when the other reader exists, the selected reader is required.
|
|
assert name_call("RequestName", otherInterface) == 1
|
|
result = subprocess.run(helperCommand, env=env, timeout=5)
|
|
assert result.returncode == 1, "helper accepted an unselected reader"
|
|
for readerArgs in (["--reader"], ["--reader", "unknown"],
|
|
["--reader-path"], ["--reader-path", ""],
|
|
["--reader-path", "relative/path"], ["--reader-path", "/bad-path"],
|
|
["--reader-interface"], ["--reader-interface", ""],
|
|
["--reader-interface", "WineAccess"],
|
|
["--reader-interface", "org.example.Bad-Interface"]):
|
|
result = subprocess.run([str(build / "wine-access.exe"), *readerArgs],
|
|
env=env, timeout=5)
|
|
assert result.returncode == 1, "invalid endpoint argument was accepted"
|
|
assert name_call("RequestName", busName) == 1
|
|
# An owned reader name is insufficient: refuse missing objects,
|
|
# interfaces, methods and incompatible signatures before starting RPC.
|
|
def reject(command):
|
|
rejected = subprocess.Popen(command, env=env, stderr=subprocess.PIPE)
|
|
try:
|
|
wait_for_client(rejected, 1)
|
|
return rejected.communicate()[1].decode()
|
|
finally:
|
|
if rejected.poll() is None:
|
|
rejected.kill()
|
|
rejected.wait()
|
|
diagnostic = reject([*helperCommand, "--reader-path", "/Missing"])
|
|
# Gio can introspect an absent object as an empty node.
|
|
assert "reader object" in diagnostic or "reader interface" in diagnostic, diagnostic
|
|
assert "reader interface" in reject([*helperCommand, "--reader-interface", "org.example.Missing"])
|
|
connection.unregister_object(registration)
|
|
for brokenXml in (
|
|
xml.replace('<method name="CancelSpeech"><arg type="b" direction="out"/></method>', ''),
|
|
xml.replace('<arg type="i" direction="out"/>', '<arg type="b" direction="out"/>'),
|
|
xml.replace('<arg type="b" direction="in"/>', '<arg type="s" direction="in"/>')):
|
|
broken = connection.register_object(objectPath, Gio.DBusNodeInfo.new_for_xml(brokenXml).interfaces[0], record, None, None)
|
|
assert "reader method" in reject(helperCommand)
|
|
connection.unregister_object(broken)
|
|
# UIA still uses direct speech when the NVDA RPC controller is disabled.
|
|
dialogMethods = {"SpeakText": ("sb", "b"), "CancelSpeech": ("", "b"),
|
|
"PresentAccessibleEvent": ("sssssssiis", "b")}
|
|
dialogCommand = ["--no-controller" if part == "--no-dialog-reader" else part
|
|
for part in helperCommand]
|
|
for method in ("SpeakText", "CancelSpeech"):
|
|
for signature in (None, ("s", "b"), (dialogMethods[method][0], "i")):
|
|
brokenMethods = dialogMethods.copy()
|
|
if signature is None:
|
|
del brokenMethods[method]
|
|
else:
|
|
brokenMethods[method] = signature
|
|
broken = connection.register_object(objectPath,
|
|
Gio.DBusNodeInfo.new_for_xml(interface_xml(brokenMethods)).interfaces[0],
|
|
record, None, None)
|
|
assert "reader method " + method in reject(dialogCommand)
|
|
connection.unregister_object(broken)
|
|
print("PASS: UIA speech methods required without the controller")
|
|
registration = connection.register_object(objectPath, Gio.DBusNodeInfo.new_for_xml(xml).interfaces[0], record, None, None)
|
|
# Controller-only interfaces must not enable control reading.
|
|
assert "PresentAccessibleEvent" in reject([part for part in helperCommand if part != "--no-dialog-reader"])
|
|
print("PASS: endpoint validation and required method signatures")
|
|
helper = subprocess.Popen(helperCommand, env=env)
|
|
# The fixture waits for the RPC server to listen before making calls.
|
|
client = subprocess.Popen(["wine", str(clientPath)], env=env)
|
|
wait_for_client(client)
|
|
assert helper.poll() is None, "helper exited during controller requests"
|
|
assert events == [
|
|
("SpeakText", ("Controller speech", False)),
|
|
("CancelSpeech", ()),
|
|
("BrailleMessage", ("Controller braille",)),
|
|
("SpeakSsml", ("wine-nvda-controller", "<speak>Controller SSML</speak>")),
|
|
], events
|
|
assert name_call("ReleaseName", busName) == 1
|
|
client = subprocess.Popen(["wine", str(clientPath), "unavailable"], env=env)
|
|
wait_for_client(client)
|
|
print(f"PASS: {interface} controller routing, SSML status, service loss, and reader validation")
|
|
finally:
|
|
for child in (client, helper):
|
|
if child is not None and 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)
|