120 lines
5.8 KiB
Python
120 lines
5.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 tempfile
|
|
import time
|
|
|
|
from gi.repository import Gio, GLib
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--build-dir", default="build")
|
|
parser.add_argument("--reader", choices=("cthulhu", "orca"))
|
|
parser.add_argument("--session", action="store_true")
|
|
args = parser.parse_args()
|
|
build = Path(args.build_dir).resolve()
|
|
if not args.session:
|
|
sys.exit(subprocess.call(["dbus-run-session", "--", sys.executable, __file__,
|
|
"--session", "--build-dir", str(build),
|
|
*(["--reader", args.reader] if args.reader else [])]))
|
|
|
|
interface = "org.gnome.Orca1.Service" if args.reader == "orca" else "org.stormux.Cthulhu1.Service"
|
|
otherInterface = "org.stormux.Cthulhu1.Service" if args.reader == "orca" else "org.gnome.Orca1.Service"
|
|
objectPath = "/" + interface.replace(".", "/")
|
|
methods = {"SpeakText": ("sb", "b"), "CancelSpeech": ("", "b"),
|
|
"BrailleMessage": ("s", "b"), "SpeakSsml": ("ss", "i")}
|
|
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>'
|
|
xml += '</interface></node>'
|
|
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)
|
|
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):
|
|
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() == 0, f"controller client failed or timed out: {client.poll()}"
|
|
|
|
|
|
with tempfile.TemporaryDirectory(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 / "cthulhu-wine-access.exe"), "--no-dialog-reader",
|
|
*(["--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"]):
|
|
result = subprocess.run([str(build / "cthulhu-wine-access.exe"), *readerArgs],
|
|
env=env, timeout=5)
|
|
assert result.returncode == 1, "invalid reader argument was accepted"
|
|
assert name_call("RequestName", interface) == 1
|
|
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", interface) == 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)
|