Make d-bus available to orca if someone decides they want to implement it.

tests/controller_integration.py

@@ -0,0 +1,119 @@
+"""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)