From 51d2cc7b9920fb0e86e702f1e4f079d8f511679a Mon Sep 17 00:00:00 2001 From: Qunmi Date: Tue, 20 Oct 2026 20:26:29 +0800 Subject: [PATCH 1/2] fix(android): verify text input across layout changes --- fsq_agent/drivers/android/_uiautomator2.py | 38 +++- tests/test_uiautomator2_android_driver.py | 224 +++++++++++++++++++-- 2 files changed, 240 insertions(+), 22 deletions(-) diff --git a/fsq_agent/drivers/android/_uiautomator2.py b/fsq_agent/drivers/android/_uiautomator2.py index e8c9768..23e2128 100644 --- a/fsq_agent/drivers/android/_uiautomator2.py +++ b/fsq_agent/drivers/android/_uiautomator2.py @@ -32,7 +32,7 @@ ANDROID_STATE_ASSERTION_FIELDS = ("enabled", "checked", "selected", "clickable", "focused") ANDROID_LOCATOR_FIELDS = ("resourceId", "accessibilityId", "text", "className", "xpath") ANDROID_EDITABLE_CLASSES = frozenset({"android.widget.EditText", "android.widget.AutoCompleteTextView", "android.widget.MultiAutoCompleteTextView"}) -ANDROID_INPUT_IDENTITY_FIELDS = ("className", "resourceName", "packageName", "bounds") +ANDROID_INPUT_IDENTITY_FIELDS = ("className", "resourceName", "packageName") # Only these structural XPath forms are demonstrably independent of the text being replaced. ANDROID_TEXT_INDEPENDENT_XPATH = re.compile(r"""(?:/+(?:hierarchy|[A-Za-z_][\w.$]*)(?:\[\d+\]|\[@(?:resource-id|content-desc)=(?:'[^']*'|"[^"]*")\])?)+""") ANDROID_UI_SNAPSHOT_TEXT_LIMIT_CHARS = 50 @@ -155,7 +155,7 @@ def input_text(self, params: AndroidInputTextParams) -> dict[str, object]: class_name = before.get("className") if not isinstance(class_name, str) or class_name not in ANDROID_EDITABLE_CLASSES or before.get("enabled") is False: return self._failed("target_resolution_error", "Input target is not demonstrably editable.", metadata={"reason": "invalid_input_target"}) - if not self._input_text_is_readable(before): + if not self._input_text_is_readable(before) or not isinstance(before.get("packageName"), str) or not before["packageName"].strip(): return self._failed("action_error", "Input target text cannot be verified.", metadata={"reason": "verification_unavailable"}) try: selector.click() @@ -175,19 +175,49 @@ def input_text(self, params: AndroidInputTextParams) -> dict[str, object]: # Stale/missing targets and read failures cannot establish either success or a rejected write. except Exception: # noqa: BLE001 return self._failed("action_error", "Text entry acceptance could not be read.", metadata={"reason": "verification_unavailable"}) - if not self._input_text_is_readable(after) or any(before.get(field) != after.get(field) for field in ANDROID_INPUT_IDENTITY_FIELDS): + # Focus, keyboard visibility, and field icons may move or resize the same input. + # Only a resource-bound unique locator can tolerate geometry changes; a + # positional XPath singleton may otherwise silently resolve another input. + if ( + not self._input_text_is_readable(after) + or after.get("enabled") is False + or any(before.get(field) != after.get(field) for field in ANDROID_INPUT_IDENTITY_FIELDS) + or (before.get("bounds") != after.get("bounds") and not self._input_locator_binds_resource(data, before.get("resourceName"))) + ): return self._failed("action_error", "Text entry acceptance cannot be verified on the original target.", metadata={"reason": "verification_unavailable"}) if after["text"] != text: return self._failed("action_error", "Text entry did not match the requested replacement.", metadata={"reason": "entry_mismatch"}) return self._passed() + def _input_locator_binds_resource(self, data: dict[str, object], resource_name: object) -> bool: + locator = data.get("locator") + if not isinstance(resource_name, str) or not resource_name.strip() or not isinstance(locator, dict): + return False + xpath = locator.get("xpath") + if isinstance(xpath, str): + # The predicate must bind the final control, rather than an ancestor. + return ANDROID_TEXT_INDEPENDENT_XPATH.fullmatch(xpath) is not None and any(xpath.endswith(f"[@resource-id={quote}{resource_name}{quote}]") for quote in ("'", '"')) + return locator.get("resourceId") == resource_name + def _input_text_is_readable(self, info: dict[str, object]) -> bool: text = info.get("text") # Unknown password state and mask-only readback cannot prove observable acceptance. return info.get("password") is False and isinstance(text, str) and (not text or not set(text) <= {"*", "•", "●"}) def _input_target_info(self, selector: object) -> dict[str, object]: - info = self._selector_info(selector) + all_matches = getattr(selector, "all", None) + if callable(all_matches): + # XPath selector.info otherwise silently reads the first matching control. + # Read uniqueness and metadata together from one fresh hierarchy snapshot. + matches = all_matches() + if not isinstance(matches, list) or len(matches) != 1: + return {} + info = self._selector_info(matches[0]) + else: + count = getattr(selector, "count", None) + if type(count) is not int or count != 1: + return {} + info = self._selector_info(selector) if "password" in info: return info # Native ObjInfo omits password; XPath info supplies it. Observe the same native diff --git a/tests/test_uiautomator2_android_driver.py b/tests/test_uiautomator2_android_driver.py index 2fa81b8..ba64434 100644 --- a/tests/test_uiautomator2_android_driver.py +++ b/tests/test_uiautomator2_android_driver.py @@ -22,6 +22,13 @@ def __init__(self, device: "FakeDevice", query: dict[str, object], exists: bool self.query = query self.exists = device.exists if exists is None else exists + @property + def count(self) -> object: + self.device.calls.append(("count", self.query)) + if any(call[0] == "set_text" for call in self.device.calls): + return self.device.post_write_match_count + return self.device.match_count + def wait(self, **kwargs: object) -> bool: self.device.calls.append(("wait", self.query, kwargs)) return self.device.wait_result @@ -63,6 +70,13 @@ def info(self) -> dict[str, object]: class FakeXPathSelector(FakeSelector): + def all(self) -> list[FakeSelector]: + self.device.calls.append(("all", self.query)) + count = self.device.post_write_match_count if any(call[0] == "set_text" for call in self.device.calls) else self.device.match_count + if type(count) is not int: + raise RuntimeError("private backend payload") + return [self] * count + def wait(self, **kwargs: object) -> bool: self.device.calls.append(("xpath_wait", self.query, kwargs)) if "exists" in kwargs: @@ -86,6 +100,8 @@ def __init__( self.wait_gone_result = wait_gone_result self.hierarchy = hierarchy self.write_mode = "accept" + self.match_count: object = 1 + self.post_write_match_count: object = 1 self.post_write_info: dict[str, object] = {} self.calls: list[tuple[Any, ...]] = [] self.selector_info: dict[str, object] = { @@ -209,7 +225,7 @@ def test_uiautomator2_driver_actions_use_locator_selectors() -> None: assert driver.swipe({"direction": "up", "duration": 100})["status"] == "passed" assert driver.assert_state({"element": {"resourceId": "url"}, "text": {"contains": "hello"}})["status"] == "passed" - assert device.calls[:14] == [ + assert device.calls[:16] == [ ("select", {"resourceId": "login"}), ("wait", {"resourceId": "login"}, {"exists": True, "timeout": 10.0}), ("click", {"resourceId": "login"}), @@ -219,9 +235,11 @@ def test_uiautomator2_driver_actions_use_locator_selectors() -> None: ("long_click", {"description": "Menu"}), ("select", {"resourceId": "search"}), ("wait", {"resourceId": "search"}, {"exists": True, "timeout": 10.0}), + ("count", {"resourceId": "search"}), ("info", {"resourceId": "search"}), ("click", {"resourceId": "search"}), ("set_text", {"resourceId": "search"}, "hello"), + ("count", {"resourceId": "search"}), ("info", {"resourceId": "search"}), ("press", "back"), ] @@ -504,13 +522,154 @@ def test_uiautomator2_driver_input_text_focuses_replaces_and_verifies_text() -> assert device.calls == [ ("select", {"resourceId": "url_bar"}), ("wait", {"resourceId": "url_bar"}, {"exists": True, "timeout": 10.0}), + ("count", {"resourceId": "url_bar"}), ("info", {"resourceId": "url_bar"}), ("click", {"resourceId": "url_bar"}), ("set_text", {"resourceId": "url_bar"}, "machine learning tutorials"), + ("count", {"resourceId": "url_bar"}), ("info", {"resourceId": "url_bar"}), ] +@pytest.mark.parametrize("backend", ["selector", "xpath"]) +@pytest.mark.parametrize( + "before_bounds", + [ + {"left": 58, "top": 1209, "right": 1038, "bottom": 1346}, + {"left": 436, "top": 2056, "right": 728, "bottom": 2193}, + ], +) +def test_input_text_accepts_unique_target_layout_change_on_first_write(backend: str, before_bounds: dict[str, int]) -> None: + device = FakeDevice() + device.selector_info.update({"resourceName": "com.microsoft.emmx:id/url_bar", "packageName": "com.microsoft.emmx", "bounds": before_bounds}) + device.post_write_info = {"bounds": {"left": 186, "top": 1209, "right": 943, "bottom": 1346}} + locator = {"resourceId": "com.microsoft.emmx:id/url_bar"} if backend == "selector" else {"xpath": "//android.widget.EditText[@resource-id='com.microsoft.emmx:id/url_bar']"} + + result = UiAutomator2AndroidDriver(app_id="com.microsoft.emmx", device=device).input_text({"text": "FSQ FSQ-lab", "locator": locator}) + + assert result == {"status": "passed", "output": None} + assert sum(call[0] == "set_text" for call in device.calls) == 1 + assert not any(call[0] == "clear_text" for call in device.calls) + + +@pytest.mark.parametrize("backend", ["selector", "xpath"]) +@pytest.mark.parametrize("after_write", [False, True]) +@pytest.mark.parametrize("count", [0, 2, None, "1", True]) +def test_input_text_requires_unique_target_before_and_after_entry(backend: str, after_write: bool, count: object) -> None: + device = FakeDevice() + if after_write: + device.post_write_match_count = count + else: + device.match_count = count + locator = {"resourceId": "input"} if backend == "selector" else {"xpath": "//android.widget.EditText[1]"} + + result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": "submitted-private", "locator": locator}) + + assert result["status"] == "failed" + assert sum(call[0] == "set_text" for call in device.calls) == int(after_write) + if not after_write: + assert not any(call[0] in {"click", "clear_text"} for call in device.calls) + assert all(value not in str(result) for value in ["submitted-private", "private backend payload", "params"]) + + +@pytest.mark.parametrize("backend", ["selector", "xpath"]) +@pytest.mark.parametrize("after_write", [False, True]) +def test_input_text_match_inspection_failure_is_safe(backend: str, after_write: bool, monkeypatch: pytest.MonkeyPatch) -> None: + device = FakeDevice() + original_count = FakeSelector.count.fget + original_all = FakeXPathSelector.all + + def inspect_count(selector: FakeSelector) -> object: + if after_write == any(call[0] == "set_text" for call in device.calls): + raise RuntimeError("submitted-private private backend payload") + return original_count(selector) + + def inspect_all(selector: FakeXPathSelector) -> list[FakeSelector]: + inspect_count(selector) + return original_all(selector) + + monkeypatch.setattr(FakeSelector, "count", property(inspect_count)) + monkeypatch.setattr(FakeXPathSelector, "all", inspect_all) + locator = {"resourceId": "input"} if backend == "selector" else {"xpath": "//android.widget.EditText[1]"} + + result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": "submitted-private", "locator": locator}) + + assert result["status"] == "failed" + assert result["metadata"] == {"reason": "verification_unavailable" if after_write else "entry_backend_error"} + assert sum(call[0] == "set_text" for call in device.calls) == int(after_write) + assert all(value not in str(result) for value in ["submitted-private", "private backend payload", "params"]) + + +@pytest.mark.parametrize("resource_name", [None, ""]) +@pytest.mark.parametrize("layout_changed", [False, True]) +def test_input_text_structural_xpath_without_resource_id_requires_stable_bounds(resource_name: str | None, layout_changed: bool) -> None: + device = FakeDevice() + device.selector_info["resourceName"] = resource_name + if layout_changed: + device.post_write_info = {"bounds": {"left": 186, "top": 1209, "right": 943, "bottom": 1346}} + + result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": "Unicode 中文", "locator": {"xpath": "//android.widget.EditText[1]"}}) + + if layout_changed: + assert result["status"] == "failed" + assert result["metadata"] == {"reason": "verification_unavailable"} + else: + assert result == {"status": "passed", "output": None} + assert sum(call[0] == "set_text" for call in device.calls) == 1 + + +@pytest.mark.parametrize("resource_name", ["", "shared"]) +def test_input_text_real_xpath_positional_retarget_cannot_verify_noop_write(resource_name: str) -> None: + xpath = pytest.importorskip("uiautomator2.xpath") + + class Device(FakeDevice): + wait_timeout = 0.01 + + def xpath(self, value: str) -> object: + return xpath.XPathEntry(self)(value) + + def dump_hierarchy(self) -> str: + nodes = [ + {"content-desc": "original", "text": "old-private", "bounds": "[0,0][100,50]"}, + {"content-desc": "other", "text": "replacement-private", "bounds": "[0,100][100,150]"}, + ] + if any(call[0] == "set_text" for call in self.calls): + nodes.reverse() + root = ElementTree.Element("hierarchy") + for node in nodes: + ElementTree.SubElement(root, "node", {"class": "android.widget.EditText", "package": "app", "resource-id": resource_name, "enabled": "true", "password": "false", **node}) + return ElementTree.tostring(root, encoding="unicode") + + def clear_text(self) -> None: + self.calls.append(("clear_text",)) + + def send_keys(self, value: str) -> None: + self.calls.append(("set_text", value)) + + device = Device() + result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": "replacement-private", "locator": {"xpath": "//android.widget.EditText[1]"}}) + + assert result["status"] == "failed" + assert result["metadata"] == {"reason": "verification_unavailable"} + assert sum(call[0] == "set_text" for call in device.calls) == 1 + assert all(value not in str(result) for value in ["old-private", "replacement-private", "params"]) + # This XML is constructed entirely from the static offline fixture above. + nodes = ElementTree.fromstring(device.dump_hierarchy()).findall("node") # noqa: S314 + assert next(node for node in nodes if node.get("content-desc") == "original").get("text") == "old-private" + + +@pytest.mark.parametrize("package_name", [None, "", " "]) +def test_input_text_missing_package_identity_is_not_verifiable(package_name: str | None) -> None: + device = FakeDevice() + device.selector_info["packageName"] = package_name + + result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": "submitted-private", "locator": {"resourceId": "input"}}) + + assert result["status"] == "failed" + assert result["metadata"] == {"reason": "verification_unavailable"} + assert not any(call[0] in {"click", "clear_text", "set_text"} for call in device.calls) + + @pytest.mark.parametrize( "info", [ @@ -556,7 +715,12 @@ def test_input_text_requires_exact_replacement_readback(text: str, class_name: s ("accept", {"text": None}, "verification_unavailable"), ("accept", {"text": 123}, "verification_unavailable"), ("accept", {"resourceName": "different"}, "verification_unavailable"), + ("accept", {"resourceName": None}, "verification_unavailable"), + ("accept", {"packageName": "different"}, "verification_unavailable"), + ("accept", {"packageName": None}, "verification_unavailable"), + ("accept", {"enabled": False}, "verification_unavailable"), ("accept", {"className": "android.widget.FrameLayout"}, "verification_unavailable"), + ("accept", {"className": "android.widget.AutoCompleteTextView"}, "verification_unavailable"), ("accept", {"password": True}, "verification_unavailable"), ("accept", {"text": "••••"}, "verification_unavailable"), ], @@ -680,7 +844,15 @@ def test_input_text_supports_info_property(monkeypatch: pytest.MonkeyPatch) -> N @pytest.mark.parametrize("backend", ["selector", "xpath"]) @pytest.mark.parametrize("text", ["", "Unicode 中文 👋"]) -def test_input_text_real_uiautomator2_apis_offline(backend: str, text: str) -> None: +@pytest.mark.parametrize(("before_count", "after_count"), [(1, 1), (2, 1), (1, 2)]) +@pytest.mark.parametrize( + "before_bounds", + [ + {"left": 58, "top": 1209, "right": 1038, "bottom": 1346}, + {"left": 436, "top": 2056, "right": 728, "bottom": 2193}, + ], +) +def test_input_text_real_uiautomator2_apis_offline(backend: str, text: str, before_count: int, after_count: int, before_bounds: dict[str, int]) -> None: selectors = pytest.importorskip("uiautomator2._selector") xpath = pytest.importorskip("uiautomator2.xpath") @@ -691,9 +863,15 @@ def __init__(self, device: FakeDevice) -> None: def waitForExists(self, *_args: object, **_kwargs: object) -> bool: return True + def count(self, _selector: object) -> int: + return after_count if any(call[0] in {"set_text", "clear_text"} for call in self.device.calls) else before_count + def objInfo(self, _selector: object) -> dict[str, object]: # The shipped native ObjInfo DTO does not expose password state. - return {"text": self.device.text, **{key: value for key, value in self.device.selector_info.items() if key != "password"}} + info = {"text": self.device.text, **{key: value for key, value in self.device.selector_info.items() if key != "password"}} + if any(call[0] in {"set_text", "clear_text"} for call in self.device.calls): + info.update(self.device.post_write_info) + return info def setText(self, _selector: object, value: str) -> bool: self.device.calls.append(("set_text", value)) @@ -711,6 +889,8 @@ class Device(FakeDevice): def __init__(self) -> None: super().__init__() self.selector_info["resourceName"] = "input" + self.selector_info["bounds"] = before_bounds + self.post_write_info = {"bounds": {"left": 186, "top": 1209, "right": 943, "bottom": 1346}} self.jsonrpc = Rpc(self) def __call__(self, **query: object) -> object: @@ -720,19 +900,22 @@ def xpath(self, value: str) -> object: return xpath.XPathEntry(self)(value) def dump_hierarchy(self) -> str: + info = self.jsonrpc.objInfo(None) + bounds = info["bounds"] root = ElementTree.Element("hierarchy") - ElementTree.SubElement( - root, - "node", - { - "class": "android.widget.EditText", - "text": self.text, - "password": "false", - "resource-id": "input", - "package": "com.example", - "bounds": "[0,0][100,50]", - }, - ) + for _ in range(self.jsonrpc.count(None)): + ElementTree.SubElement( + root, + "node", + { + "class": "android.widget.EditText", + "text": self.text, + "password": "false", + "resource-id": "input", + "package": "com.example", + "bounds": f"[{bounds['left']},{bounds['top']}][{bounds['right']},{bounds['bottom']}]", + }, + ) return ElementTree.tostring(root, encoding="unicode") def clear_text(self) -> None: @@ -747,9 +930,14 @@ def send_keys(self, value: str) -> None: locator = {"resourceId": "input"} if backend == "selector" else {"xpath": "//android.widget.EditText[@resource-id='input']"} result = UiAutomator2AndroidDriver(app_id="app", device=device).input_text({"text": text, "locator": locator}) - assert result == {"status": "passed", "output": None} - assert device.text == text - assert sum(call[0] == "set_text" or (call[0] == "clear_text" and backend == "selector" and not text) for call in device.calls) == 1 + if before_count == after_count == 1: + assert result == {"status": "passed", "output": None} + assert device.text == text + else: + assert result["status"] == "failed" + if text: + assert text not in str(result) + assert sum(call[0] == "set_text" or (call[0] == "clear_text" and backend == "selector" and not text) for call in device.calls) == int(before_count == 1) @pytest.mark.parametrize( From e6c2decc4c2ab1d669646ce3d1e5dbf86c3af1f6 Mon Sep 17 00:00:00 2001 From: Qunmi Date: Tue, 20 Oct 2026 21:26:21 +0800 Subject: [PATCH 2/2] fix(android): recover initial device context read --- fsq_agent/drivers/android/_uiautomator2.py | 27 +++- tests/test_uiautomator2_android_driver.py | 169 ++++++++++++++++++++- 2 files changed, 194 insertions(+), 2 deletions(-) diff --git a/fsq_agent/drivers/android/_uiautomator2.py b/fsq_agent/drivers/android/_uiautomator2.py index 23e2128..b9b663c 100644 --- a/fsq_agent/drivers/android/_uiautomator2.py +++ b/fsq_agent/drivers/android/_uiautomator2.py @@ -83,7 +83,14 @@ def close(self) -> None: return None def context(self) -> dict[str, object]: - info = self._device_info() + try: + info = self._device_info() + except Exception as exc: + if not self._is_shared_memory_initialization_error(exc): + raise + # Some Android versions initialize this service during the first read. + # Observe once more; never repeat the prepared application action. + info = self._device_info() width = info.get("displayWidth") height = info.get("displayHeight") return { @@ -95,6 +102,24 @@ def context(self) -> dict[str, object]: }, } + @staticmethod + def _is_shared_memory_initialization_error(error: Exception) -> bool: + try: + from uiautomator2.exceptions import RPCUnknownError + except ImportError: + return False + if not isinstance(error, RPCUnknownError) or len(error.args) != 3: + return False + summary, _, stack = error.args + if summary != "Unknown RPC error: -32001 java.lang.IllegalStateException" or not isinstance(stack, str): + return False + lines = stack.splitlines() + return ( + len(lines) > 1 + and lines[0] == "java.lang.IllegalStateException: ApplicationSharedMemory not initialized" + and re.fullmatch(r"\s*at com\.android\.internal\.os\.ApplicationSharedMemory\.getInstance\(ApplicationSharedMemory\.java:\d+\)", lines[1]) is not None + ) + @_android_driver_tool("launchApp", description="Launch the configured Android app.") def launch_app(self, params: AndroidLaunchAppParams) -> dict[str, object]: data = self._param_data(params) diff --git a/tests/test_uiautomator2_android_driver.py b/tests/test_uiautomator2_android_driver.py index ba64434..ed9ffe1 100644 --- a/tests/test_uiautomator2_android_driver.py +++ b/tests/test_uiautomator2_android_driver.py @@ -4,12 +4,14 @@ import builtins from asyncio import CancelledError from io import BytesIO +from types import SimpleNamespace from typing import Any from xml.etree import ElementTree import pytest -from fsq_agent.core import AndroidDriverInterface +from fsq_agent.core import AndroidDriverInterface, CapabilityRegistry, StepRunner +from fsq_agent.core.evidence import ArtifactStore from fsq_agent.core.harness._android import AndroidHarness from fsq_agent.core.harness._driver_tools import _discover_driver_capability_definitions from fsq_agent.core.harness._uiautomator2_driver import UiAutomator2AndroidDriver @@ -213,6 +215,171 @@ def test_uiautomator2_driver_context_launch_kill_and_artifacts() -> None: ] +def _shared_memory_rpc_error() -> Exception: + errors = pytest.importorskip("uiautomator2.exceptions") + return errors.RPCUnknownError( + "Unknown RPC error: -32001 java.lang.IllegalStateException", + {}, + "java.lang.IllegalStateException: ApplicationSharedMemory not initialized\n" + "\tat com.android.internal.os.ApplicationSharedMemory.getInstance(ApplicationSharedMemory.java:85)\n" + "\tat com.wetest.uia2.stub.AutomatorServiceImpl.deviceInfo(AutomatorServiceImpl.java:158)\n", + ) + + +class ContextReadDevice(FakeDevice): + def __init__(self, outcomes: list[object], *, launch_error: Exception | None = None) -> None: + super().__init__() + self.outcomes = iter(outcomes) + self.info_reads = 0 + self.launch_error = launch_error + + @property + def info(self) -> dict[str, object]: + self.info_reads += 1 + outcome = next(self.outcomes, self.normal_info) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + @info.setter + def info(self, value: dict[str, object]) -> None: + self.normal_info = value + + def app_start(self, app_id: str, **options: object) -> None: + super().app_start(app_id, **options) + if self.launch_error is not None: + raise self.launch_error + + +def test_context_recovers_known_initialization_fault_and_returns_fresh_info() -> None: + fresh = {"displayWidth": 720, "displayHeight": 1600, "currentPackageName": "fresh.package"} + device = ContextReadDevice([_shared_memory_rpc_error(), fresh]) + driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=device) + + context = driver.context() + + assert context["screen_size"] == (720, 1600) + assert context["metadata"] == {"backend": "uiautomator2", "current_package": "fresh.package"} + assert "current_activity" not in context + assert device.info_reads == 2 + assert device.calls == [] + assert driver.context()["screen_size"] == (1080, 2400) + assert device.info_reads == 3 + + +@pytest.mark.parametrize("second_fault", ["same", "disconnect", "cancel"]) +def test_context_recovery_is_bounded_and_preserves_second_fault(second_fault: str) -> None: + error = {"same": _shared_memory_rpc_error(), "disconnect": ConnectionError("offline"), "cancel": CancelledError()}[second_fault] + device = ContextReadDevice([_shared_memory_rpc_error(), error]) + driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=device) + + with pytest.raises(type(error)) as caught: + driver.context() + + assert caught.value is error + assert device.info_reads == 2 + assert device.calls == [] + + +@pytest.mark.parametrize( + "fault", ["timeout", "disconnect", "ordinary-lookalike", "rpc-params", "rpc-wrong-code", "rpc-wrong-java", "rpc-missing-frame", "rpc-malformed", "cancel", "interrupt", "exit"] +) +def test_context_does_not_retry_other_faults_or_lookalikes(fault: str) -> None: + known = _shared_memory_rpc_error() + errors = pytest.importorskip("uiautomator2.exceptions") + summary, _, stack = known.args + error = { + "timeout": TimeoutError("offline"), + "disconnect": errors.UiAutomationNotConnectedError("offline"), + "ordinary-lookalike": RuntimeError(*known.args), + "rpc-params": errors.RPCUnknownError(summary, {"locator": stack}, "java.lang.IllegalStateException: other failure"), + "rpc-wrong-code": errors.RPCUnknownError("Unknown RPC error: -32002 java.lang.IllegalStateException", {}, stack), + "rpc-wrong-java": errors.RPCUnknownError(summary, {}, stack.replace("java.lang.IllegalStateException:", "java.lang.SecurityException:", 1)), + "rpc-missing-frame": errors.RPCUnknownError(summary, {}, "java.lang.IllegalStateException: ApplicationSharedMemory not initialized"), + "rpc-malformed": errors.RPCUnknownError(summary, {}, {"stack": stack}), + "cancel": CancelledError(), + "interrupt": KeyboardInterrupt(), + "exit": SystemExit(), + }[fault] + device = ContextReadDevice([error]) + driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=device) + + with pytest.raises(type(error)) as caught: + driver.context() + + assert caught.value is error + assert device.info_reads == 1 + assert device.calls == [] + + +def test_context_recovery_uses_actual_sdk_info_and_rpc_error_offline(monkeypatch) -> None: + sdk = pytest.importorskip("uiautomator2") + sdk_core = pytest.importorskip("uiautomator2.core") + summary, _, stack = _shared_memory_rpc_error().args + response = SimpleNamespace(text="initialization fault", json=lambda: {"error": {"code": -32001, "message": summary.removeprefix("Unknown RPC error: -32001 "), "data": stack}}) + responses = iter([response, SimpleNamespace(json=lambda: {"result": {"displayWidth": 600, "displayHeight": 900, "currentPackageName": "sdk.package"}})]) + calls = [] + + def request(*args, **kwargs): + calls.append((args, kwargs)) + return next(responses) + + class Rpc: + def deviceInfo(self, *, http_timeout): + return sdk_core._jsonrpc_call(None, 9008, "deviceInfo", {}, http_timeout, False) + + class Device: + info = sdk.Device.info + jsonrpc = Rpc() + + monkeypatch.setattr(sdk_core, "_http_request", request) + driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=Device()) + + assert driver.context()["screen_size"] == (600, 900) + assert len(calls) == 2 + assert all(call[0][4] == {"jsonrpc": "2.0", "id": 1, "method": "deviceInfo", "params": {}} for call in calls) + + +@pytest.mark.parametrize("mode", ["transient", "persistent", "launch-failure", "launch-and-evidence-failure"]) +def test_context_recovery_through_runner_never_retries_launch(mode: str, tmp_path) -> None: + outcomes = [_shared_memory_rpc_error()] + if mode == "persistent": + outcomes.append(_shared_memory_rpc_error()) + device = ContextReadDevice(outcomes, launch_error=RuntimeError("app start rejected") if mode.startswith("launch") else None) + if mode == "launch-and-evidence-failure": + + def failed_capture(*args, **kwargs): + raise RuntimeError("capture unavailable") + + device.screenshot = failed_capture + driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=device) + harness = AndroidHarness(driver=driver, artifact_store=ArtifactStore(run_dir=tmp_path)) + registry = CapabilityRegistry() + for definition in _discover_driver_capability_definitions(driver, platform="android"): + registry.register(definition) + runner = StepRunner(harness, capability_registry=registry) + step = ExecutableStep(step_id="launch", action_name="launch_app", params={}, kind="setup") + + result = runner.run_step("context-recovery", step) + + assert len([call for call in device.calls if call[0] == "app_start"]) == (0 if mode == "persistent" else 1) + if mode == "persistent": + assert device.info_reads == 2 + assert result.status == "failed" + assert [(phase.phase, phase.status) for phase in result.phase_reports] == [("prepare", "failed")] + assert "RPCUnknownError" in result.error_message + else: + assert result.phase_reports[0].status == "passed" + invoke = next(phase for phase in result.phase_reports if phase.phase == "invoke") + assert invoke.status == ("failed" if mode.startswith("launch") else "passed") + assert result.status == ("failed" if mode.startswith("launch") else "passed") + if mode.startswith("launch"): + assert "app start rejected" in result.error_message + assert result.failure_category == "harness_error" + if mode == "launch-and-evidence-failure": + assert result.evidence_errors + + def test_uiautomator2_driver_actions_use_locator_selectors() -> None: device = FakeDevice(text="bing.com") driver = UiAutomator2AndroidDriver(app_id="com.example.app", device=device)