mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 10:27:33 +00:00
Kelvin Amoaba
d6ca3aae25
* test(computer): run the Linux bridge against an unreachable accessibility bus * fix(computer): report an unreachable AT-SPI bus instead of aborting the Linux bridge * test(computer): read the bridge response past the registry banner * test(computer): keep the healthy-bus check off the desktop's X display
53 lines
2.6 KiB
Python
53 lines
2.6 KiB
Python
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
RUNTIME = Path(__file__).with_name("runtime.py")
|
|
|
|
|
|
@unittest.skipUnless(sys.platform.startswith("linux"), "drives the real libatspi")
|
|
class AccessibilityBusTest(unittest.TestCase):
|
|
"""Runs the bridge against real libatspi in a private runtime dir, never the desktop's buses."""
|
|
|
|
def run_list_apps(self, runtime_dir, env, prefix=()):
|
|
operation = os.path.join(runtime_dir, "operation.json")
|
|
with open(operation, "w", encoding="utf-8") as handle:
|
|
json.dump({"tool": "list_apps"}, handle)
|
|
env = {**env, "XDG_RUNTIME_DIR": runtime_dir}
|
|
# Why files, not pipes: bus daemons the session activates inherit them and can outlive the bridge.
|
|
with tempfile.TemporaryFile("w+") as stdout, tempfile.TemporaryFile("w+") as stderr:
|
|
result = subprocess.run([*prefix, sys.executable, str(RUNTIME), operation],
|
|
env=env, stdout=stdout, stderr=stderr, timeout=60)
|
|
stdout.seek(0)
|
|
stderr.seek(0)
|
|
return subprocess.CompletedProcess(result.args, result.returncode, stdout.read(), stderr.read())
|
|
|
|
def test_unreachable_bus_returns_an_error_instead_of_aborting(self):
|
|
with tempfile.TemporaryDirectory() as runtime_dir:
|
|
missing = "unix:path=" + os.path.join(runtime_dir, "missing-bus")
|
|
result = self.run_list_apps(runtime_dir, {**os.environ, "DBUS_SESSION_BUS_ADDRESS": missing,
|
|
"AT_SPI_BUS_ADDRESS": missing})
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
response = json.loads(result.stdout)
|
|
self.assertFalse(response["ok"])
|
|
self.assertTrue(response["error"].startswith("unsupported_capability:"), response["error"])
|
|
|
|
@unittest.skipUnless(shutil.which("dbus-run-session"), "needs dbus-run-session")
|
|
def test_reachable_bus_lists_apps(self):
|
|
env = {name: value for name, value in os.environ.items()
|
|
if name not in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "DISPLAY", "WAYLAND_DISPLAY")}
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as runtime_dir:
|
|
result = self.run_list_apps(runtime_dir, env, ("dbus-run-session", "--"))
|
|
# Why the last line: the registry daemon the session activates shares stdout and prints a banner.
|
|
response = result.stdout.strip().splitlines()[-1:]
|
|
self.assertEqual((result.returncode, response), (0, ['{"ok":true,"apps":[]}']), result.stdout + result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|