from contextlib import ExitStack, nullcontext import importlib.util import json import sys import types import unittest from pathlib import Path from unittest.mock import patch class FakeStateType: PROTECTED = 1 SELECTED = 2 ACTIVE = 3 SHOWING = 4 FOCUSED = 5 EDITABLE = 6 class FakeStateSet: def __init__(self, *states): self.states = set(states) def contains(self, state): return state in self.states class FakeText: @staticmethod def get_character_count(node): return len(node.value) @staticmethod def get_text(node, start, end): return node.value[start:end] @staticmethod def get_caret_offset(node): return node.caret @staticmethod def set_caret_offset(node, offset): node.caret = offset return True @staticmethod def get_n_selections(node): return 1 if node.selection else 0 @staticmethod def get_selection(node, _): return types.SimpleNamespace(start_offset=node.selection[0], end_offset=node.selection[1]) class FakeEditableText: @staticmethod def insert_text(node, position, text, length): inserted = text.encode("utf-8")[:length].decode("utf-8") node.value = node.value[:position] + inserted + node.value[position:] return True @staticmethod def delete_text(node, start, end): node.value = node.value[:start] + node.value[end:] node.selection = None return True def load_runtime(): gi = types.ModuleType("gi") repository = types.ModuleType("gi.repository") gi.require_version = lambda *_: None repository.Atspi = types.SimpleNamespace(StateType=FakeStateType, Text=FakeText, EditableText=FakeEditableText) repository.Gdk = types.SimpleNamespace() repository.GdkPixbuf = types.SimpleNamespace() gi.repository = repository sys.modules["gi"] = gi sys.modules["gi.repository"] = repository path = Path(__file__).with_name("runtime.py") spec = importlib.util.spec_from_file_location("orca_linux_runtime_test", path) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module runtime = load_runtime() class FakeAccessible: def __init__(self, role, name="", value="", children=()): self.role = role self.name = name self.value = value self.children = list(children) def get_child_count(self): return len(self.children) def get_child_at_index(self, index): return self.children[index] def get_role_name(self): return self.role def get_name(self): return self.name def get_accessible_id(self): return "" def get_toolkit_name(self): return "fake" def get_component_iface(self): return None def get_state_set(self): return None def get_n_actions(self): return 0 def is_text(self): return bool(self.value) def get_text_iface(self): return self if self.value else None def get_value_iface(self): return None class SnapshotFixture(ExitStack): def __init__(self, root, app_name="Fixture"): super().__init__() self.root = root self.app = FakeAccessible("application", app_name, children=[root]) self.app.get_process_id = lambda: 42 self.clock = 0.0 self.timeout = (800, 15000) self.delays = {} self.restore = lambda: None self.bounds = {node: runtime.Rect(100 + i * 10, 200 + i * 20, 300, 400) for i, node in enumerate(runtime.all_nodes(root))} for node in self.bounds: node.get_component_iface = lambda node=node: node def __enter__(self): for obj, name, value in ( (runtime, "desktop_apps", lambda: iter([self.app])), (runtime.time, "monotonic", lambda: self.clock), (runtime.Atspi, "set_timeout", self.set_timeout), (runtime.Atspi, "Component", types.SimpleNamespace(get_extents=self.extents)), (runtime.Atspi, "CoordType", types.SimpleNamespace(SCREEN=0)), (runtime, "capture_png", self.capture), (runtime, "restore_window", lambda *_: self.restore()), ): self.enter_context(patch.object(obj, name, value, create=True)) return self def set_timeout(self, method, startup): self.timeout = (method, startup) def extents(self, node, _): method, startup = self.timeout allowance = max(method, startup - self.clock * 1000) / 1000 delay = self.delays.get(node, 0) self.clock += min(delay, allowance) if delay > allowance: raise TimeoutError("unresponsive geometry") return self.bounds.get(node) def capture(self, bounds): self.clock += 1 return {"base64": json.dumps(bounds.to_json())} if bounds else None class RuntimeRenderTest(unittest.TestCase): def test_focus_search_keeps_transient_accessible_wrappers_alive(self): class TransientNode: def __init__(self, depth): self.depth = depth def get_child_count(self): return int(self.depth < 30) def get_child_at_index(self, _): return TransientNode(self.depth + 1) focused = runtime.first_descendant(TransientNode(0), lambda node: node.depth == 30) self.assertIsNotNone(focused) self.assertEqual(focused.depth, 30) def test_window_only_protection_preserves_completed_semantic_action(self): button = FakeAccessible("button", "Continue") effects = [] button.get_n_actions = lambda: 1 button.get_action_name = lambda _: "click" def click(_): effects.append("clicked") button.value = "Completed" return True button.do_action = click with SnapshotFixture(FakeAccessible("frame", "Editor", children=[button])) as fixture: fixture.delays[button] = 20 snapshot = runtime.make_snapshot("Fixture", False) self.assertIsNone(snapshot["elements"][1]["frame"]) fixture.app.children.append(FakeAccessible("label", "Bitwarden integration")) result = runtime.run_operation({"tool": "click", "app": "Fixture", "windowIndex": 0, "element": snapshot["elements"][1], "noScreenshot": True}) self.assertEqual(effects, ["clicked"]) self.assertTrue(result["ok"]) self.assertEqual(result["action"]["path"], "accessibility") self.assertEqual(result["snapshot"]["elements"][1]["value"], "Completed") fixture.root.name = "Bitwarden vault" fixture.bounds[fixture.root] = None with self.assertRaisesRegex(RuntimeError, "appBlocked"): runtime.run_operation({"tool": "click", "app": "Fixture"}) self.assertEqual(effects, ["clicked"]) def test_pointer_input_waits_for_activation_and_uses_restored_geometry(self): operations = ( {"tool": "click", "x": 5, "y": 6}, {"tool": "scroll", "x": 5, "y": 6, "direction": "down", "pages": 1}, {"tool": "drag", "from_x": 5, "from_y": 6, "to_x": 25, "to_y": 26}, ) for operation in operations: for activates in (False, True): with self.subTest(tool=operation["tool"], activates=activates), SnapshotFixture( FakeAccessible("frame", "Editor") ) as fixture: restored = [] pointer_calls = [] fixture.root.get_state_set = lambda: FakeStateSet( *([FakeStateType.ACTIVE] if activates and restored and fixture.clock >= 0.1 else []) ) def restore(): restored.append(True) fixture.bounds[fixture.root] = runtime.Rect(500, 200, 300, 400) fixture.restore = restore def sleep(seconds): fixture.clock += seconds with patch.object(runtime.time, "sleep", sleep), \ patch.object(runtime, "click_at", lambda *args: pointer_calls.append(args[:2])), \ patch.object(runtime, "scroll_at", lambda *args: pointer_calls.append(args[:2])), \ patch.object(runtime, "drag_between", lambda start, end: pointer_calls.extend((start, end))): request = {**operation, "app": "Fixture", "noScreenshot": True, "restoreWindow": False} if activates: result = runtime.run_operation(request) self.assertTrue(result["ok"]) self.assertEqual(pointer_calls, [(505, 206), (525, 226)] if operation["tool"] == "drag" else [(505, 206)]) self.assertGreaterEqual(fixture.clock, 0.1) else: with self.assertRaisesRegex(RuntimeError, "window_not_focused"): runtime.run_operation(request) self.assertEqual(pointer_calls, []) self.assertGreaterEqual(fixture.clock, 0.5) self.assertTrue(restored) def test_restoration_requires_selected_window_focus_before_keyboard_input(self): restore_window = runtime.restore_window cases = ( {"name": "already active", "initial": 71, "expected": True, "native": False}, {"name": "delayed AT-SPI focus", "ack": False, "delay": 0.1, "expected": True, "native": False}, {"name": "ineffective true acknowledgment", "ack": True, "expected": True, "native": True}, {"name": "unmanaged helper", "ids": "70\n71\n", "managed": [71], "expected": True, "native": True}, {"name": "two accessible windows", "windows": 2}, {"name": "two native windows", "ids": "72\n71\n"}, {"name": "missing managed stack", "managed": None}, {"name": "failed managed window identity", "managed": [71, None]}, {"name": "invalid managed window identity", "managed": [71, 0]}, {"name": "Wayland", "session": "wayland"}, {"name": "search timeout", "timeout": "search"}, {"name": "activation timeout", "timeout": "windowactivate"}, {"name": "activation refused", "refused": True}, ) for case in cases: with self.subTest(case=case["name"]), SnapshotFixture(FakeAccessible("frame", "Editor")) as fixture: active = [case.get("initial", 99)] activations, keys = [], [] if case.get("windows") == 2: fixture.app.children.append(FakeAccessible("frame", "Sibling")) def state(): if fixture.clock >= case.get("delay", float("inf")): active[0] = 71 states = [FakeStateType.SHOWING] if active[0] == 71: states.append(FakeStateType.ACTIVE) return FakeStateSet(*states) def sleep(seconds): fixture.clock += seconds def activate(window_id): activations.append(window_id) if not case.get("refused"): active[0] = window_id def native(command, **_kwargs): stage = command[1] if case.get("timeout") == stage or (case.get("timeout") == "windowactivate" and "windowactivate" in command): raise runtime.subprocess.TimeoutExpired(command, 2) if stage == "search": ids = case.get("ids", "71\n") if "windowactivate" in command: activate(int(ids.split()[0])) return types.SimpleNamespace(returncode=0, stdout=ids) if stage == "windowactivate": activate(int(command[-1])) return types.SimpleNamespace(returncode=int(bool(case.get("refused"))), stdout="") if stage == "type": keys.append(active[0]) return types.SimpleNamespace(returncode=0) raise AssertionError("unexpected native operation") fixture.root.get_state_set = state with patch.object(runtime, "restore_window", restore_window), \ patch.object(runtime.time, "sleep", sleep), \ patch.object(runtime.Atspi.Component, "grab_focus", lambda _: case.get("ack", False), create=True), \ patch.object(runtime.Atspi, "KeySynthType", types.SimpleNamespace(STRING=1), create=True), \ patch.object(runtime.Atspi, "generate_keyboard_event", lambda *_: keys.append(active[0]), create=True), \ patch.object(runtime.Gdk, "Screen", types.SimpleNamespace(get_default=lambda: types.SimpleNamespace( get_window_stack=lambda: case.get("managed", [71, 72]))), create=True), \ patch.object(runtime, "GdkX11", types.SimpleNamespace(X11Window=types.SimpleNamespace(get_xid=int)), create=True), \ patch.object(runtime.shutil, "which", lambda _: "/usr/bin/xdotool"), \ patch.object(runtime, "temporary_x11_keymap", lambda *_: nullcontext(None)), \ patch.object(runtime.subprocess, "run", native), \ patch.dict(runtime.os.environ, {"XDG_SESSION_TYPE": case.get("session", "x11")}): request = {"tool": "press_key", "key": "x", "app": "Fixture", "windowIndex": 0, "restoreWindow": True, "noScreenshot": True} if case.get("expected"): self.assertTrue(runtime.run_operation(request)["ok"]) else: with self.assertRaisesRegex(RuntimeError, "window_not_focused"): runtime.run_operation(request) self.assertEqual(keys, [71] if case.get("expected") else []) self.assertEqual(activations, [71] if case.get("native") or case.get("refused") else []) if not case.get("expected"): self.assertEqual(active[0], 99) def test_text_entry_inserts_into_the_focused_field_of_a_background_window(self): for tool in ("type_text", "paste_text"): for value in ("there", "🚀 café"): for selection in ((6, 11), None): field = FakeAccessible("text", "Message", value="Hello world") field.caret, field.selection = 11, selection field.get_state_set = lambda: FakeStateSet(FakeStateType.FOCUSED, FakeStateType.EDITABLE) field.get_editable_text_iface = lambda: field with self.subTest(tool=tool, value=value, selection=selection), SnapshotFixture(FakeAccessible("frame", "Editor", children=[field])) as fixture: calls = [] fixture.restore = lambda: calls.append("restore") with patch.object(runtime, "type_text", lambda _: calls.append("keys")), \ patch.object(runtime, "paste_text", lambda _: calls.append("clipboard")): result = runtime.run_operation({"tool": tool, "app": "Fixture", "text": value, "noScreenshot": True}) self.assertEqual(calls, []) prefix = "Hello " if selection else "Hello world" self.assertEqual((field.value, field.caret), (prefix + value, len(prefix) + len(value))) self.assertEqual(result["action"]["path"], "accessibility") self.assertEqual(result["action"]["verification"]["state"], "verified") def test_text_entry_without_an_editable_focused_field_still_requires_window_focus(self): cases = [ ("not-editable", 1, (6, 11)), ("count-error", TimeoutError("selection timed out"), (6, 11)), ("multiple", 2, (6, 11)), ("range-error", 1, TimeoutError("range timed out")), ("range-outside", 1, (6, 12)), ] for tool in ("type_text", "paste_text"): for active in (False, True): for name, count, selection in cases: field = FakeAccessible("text", "Message", value="Hello world") field.caret, field.selection = 11, (6, 11) states = [FakeStateType.FOCUSED] + ([] if name == "not-editable" else [FakeStateType.EDITABLE]) field.get_state_set = lambda: FakeStateSet(*states) field.get_editable_text_iface = lambda: field selected = types.SimpleNamespace(start_offset=selection[0], end_offset=selection[1]) if isinstance(selection, tuple) else selection with self.subTest(tool=tool, active=active, case=name), SnapshotFixture( FakeAccessible("frame", "Editor", children=[field]) ) as fixture, patch.object(runtime.Atspi.Text, "get_n_selections", return_value=count, side_effect=count if isinstance(count, Exception) else None), \ patch.object(runtime.Atspi.Text, "get_selection", return_value=selected, side_effect=selected if isinstance(selected, Exception) else None), \ patch.object(runtime.Atspi.EditableText, "insert_text", wraps=FakeEditableText.insert_text) as insert, \ patch.object(runtime, "type_text") as keys, patch.object(runtime, "paste_text") as paste: if active: fixture.root.get_state_set = lambda: FakeStateSet(FakeStateType.ACTIVE) result = runtime.run_operation({"tool": tool, "app": "Fixture", "text": "x", "noScreenshot": True}) if tool == "type_text": keys.assert_called_once_with("x", fixture.root) else: paste.assert_called_once_with("x") self.assertEqual(result["action"]["verification"]["state"], "unverified") else: with self.assertRaisesRegex(RuntimeError, "window_not_focused"): runtime.run_operation({"tool": tool, "app": "Fixture", "text": "x", "noScreenshot": True}) if not active or tool != "type_text": keys.assert_not_called() if not active or tool != "paste_text": paste.assert_not_called() insert.assert_not_called() self.assertEqual((field.value, field.selection), ("Hello world", (6, 11))) def test_partial_text_replacement_fails_without_typing_again(self): tool = "type_text" for failure in (False, True, TimeoutError("provider timed out"), "wrong-range"): field = FakeAccessible("text", "Message", value="Hello world") field.caret, field.selection = 11, (6, 11) field.get_state_set = lambda: FakeStateSet(FakeStateType.FOCUSED, FakeStateType.EDITABLE) field.get_editable_text_iface = lambda: field def delete(node, _start, _end): if failure == "wrong-range": node.value = node.value[5:] node.selection = None return True if isinstance(failure, Exception): raise failure return failure with self.subTest(tool=tool, failure=failure), SnapshotFixture( FakeAccessible("frame", "Editor", children=[field]) ) as fixture, patch.object(runtime.Atspi.EditableText, "delete_text", delete), \ patch.object(runtime, "type_text") as keys, patch.object(runtime, "paste_text") as paste: fixture.root.get_state_set = lambda: FakeStateSet(FakeStateType.ACTIVE) with self.assertRaisesRegex(RuntimeError, "text_replacement_incomplete"): runtime.run_operation({"tool": tool, "app": "Fixture", "text": "there", "noScreenshot": True}) self.assertEqual(field.value, " worldthere" if failure == "wrong-range" else "Hello worldthere") keys.assert_not_called() paste.assert_not_called() def test_unconfirmed_text_insertion_does_not_fall_back_to_another_write(self): write = FakeEditableText.insert_text failures = { "false-after-write": "foobarbar", "timeout-after-write": "foobarbar", "no-op": "foobar", "truncated": "foobbar", "wrong-text": "foobazbar", "count-unavailable": "foobarbar", } tool = "type_text" for failure, expected in failures.items(): field = FakeAccessible("text", "Message", value="foobar") field.caret, field.selection = 6, (0, 3) field.get_state_set = lambda: FakeStateSet(FakeStateType.FOCUSED, FakeStateType.EDITABLE) field.get_editable_text_iface = lambda: field def insert(node, position, text, length): if failure == "no-op": return True write(node, position, "baz" if failure == "wrong-text" else text, 1 if failure == "truncated" else length) if failure == "timeout-after-write": raise TimeoutError("reply lost after insertion") return failure != "false-after-write" def count(node): return None if failure == "count-unavailable" and node.value != "foobar" else len(node.value) with self.subTest(tool=tool, failure=failure), SnapshotFixture( FakeAccessible("frame", "Editor", children=[field]) ) as fixture, patch.object(runtime.Atspi.EditableText, "insert_text", side_effect=insert), \ patch.object(runtime.Atspi.Text, "get_character_count", side_effect=count), \ patch.object(runtime, "type_text") as keys, patch.object(runtime, "paste_text") as paste: fixture.root.get_state_set = lambda: FakeStateSet(FakeStateType.ACTIVE) with self.assertRaisesRegex(RuntimeError, "text_insertion_unconfirmed"): runtime.run_operation({"tool": tool, "app": "Fixture", "text": "bar", "noScreenshot": True}) self.assertEqual(field.value, expected) self.assertEqual(field.selection, (0, 3)) keys.assert_not_called() paste.assert_not_called() def test_unconfirmed_caret_movement_does_not_report_text_entry_success(self): tool = "type_text" for failure in (False, True, TimeoutError("provider timed out")): field = FakeAccessible("text", "Message", value="Hello world") field.caret, field.selection = 6, (6, 11) field.get_state_set = lambda: FakeStateSet(FakeStateType.FOCUSED, FakeStateType.EDITABLE) field.get_editable_text_iface = lambda: field with self.subTest(tool=tool, failure=failure), SnapshotFixture( FakeAccessible("frame", "Editor", children=[field]) ) as fixture, patch.object(runtime.Atspi.Text, "set_caret_offset", side_effect=failure if isinstance(failure, Exception) else None, return_value=failure), \ patch.object(runtime, "type_text") as keys, patch.object(runtime, "paste_text") as paste: fixture.root.get_state_set = lambda: FakeStateSet(FakeStateType.ACTIVE) with self.assertRaisesRegex(RuntimeError, "text_caret_unconfirmed"): runtime.run_operation({"tool": tool, "app": "Fixture", "text": "there", "noScreenshot": True}) self.assertEqual((field.value, field.caret), ("Hello there", 6)) keys.assert_not_called() paste.assert_not_called() def test_compacted_browser_records_keep_their_own_frames_and_indexes(self): tabs = [FakeAccessible("page tab", f"Tab {i}", value="1" if i == 9 else "") for i in range(10)] buttons = [FakeAccessible("button", f"Other {i}") for i in range(10)] root = FakeAccessible("frame", "Browser", children=[ FakeAccessible("page tab list", "Tabs", children=tabs), *buttons, ]) with SnapshotFixture(root, "chrome") as fixture: fixture.delays = {tab: 20 for tab in tabs[:-1]} snapshot = runtime.make_snapshot("chrome", False) by_name = {item["name"]: item for item in snapshot["elements"]} for node in (tabs[-1], buttons[8]): self.assertEqual(by_name[node.name]["frame"]["x"], fixture.bounds[node].x - 100) self.assertNotIn("Tab 0", by_name) indexes = [item["index"] for item in snapshot["elements"]] self.assertEqual(len(indexes), len(set(indexes))) if __name__ == "__main__": unittest.main()