"""Explicit device acceptance for the already-explored isolated native fixture. Run after building/installing MainActivity and exploring it through ARTEMIS: uv run python tests/native_actions_fixture/verify_device.py --serial SERIAL --output DIR Uses the existing device lease, driver and actuator. Never resets app data or stops another app. Artifacts contain fixture state and actual native receipts. """ import argparse import asyncio import base64 from io import BytesIO import json import math from pathlib import Path import subprocess import time from types import SimpleNamespace from adbutils import AdbClient from PIL import Image from artemis.clients.accessibility_client import AccessibilityClient from artemis.drivers.android.adb_driver import AndroidAdbDriver from artemis.mcp.actuators.adb import AdbActuator from artemis.runtime.adb_endpoint import adb_command from artemis.runtime.device_lock import DeviceExecutionLock async def verify(serial: str, output: Path, repeats: int) -> None: output.mkdir(parents=True, exist_ok=True) lease = DeviceExecutionLock(serial, description="Native fixture acceptance") lease.acquire(blocking=False) client = AccessibilityClient(serial) records = [] checks = [] started = time.monotonic() def check(condition, label): checks.append({"check": label, "passed": bool(condition)}) if not condition: raise AssertionError(label) def state(): result = subprocess.run( adb_command( [ "-s", serial, "shell", "run-as", "com.artemis.fixture", "cat", "files/native_actions_state.json", ] ), capture_output=True, text=True, encoding="utf-8", check=True, timeout=10, ) return json.loads(result.stdout) try: await asyncio.to_thread(client.connect) session = client.session check(session is not None and session.protocol_version >= 3, "helper protocol 3") # Recreate only this isolated test Activity, so counters are reproducible. for command in ( ["shell", "am", "force-stop", "com.artemis.fixture"], ["shell", "am", "start", "-W", "-n", "com.artemis.fixture/.MainActivity"], ): subprocess.run( adb_command(["-s", serial, *command]), capture_output=True, check=True, timeout=15 ) screen = await asyncio.to_thread(client.get_screen_data) image = Image.open(BytesIO(base64.b64decode(screen.base64))) width, height = image.size driver = AndroidAdbDriver(serial, AdbClient(), client, width, height) ctx = SimpleNamespace( device=SimpleNamespace(device_width=width, device_height=height), data_engine=None ) actuator = AdbActuator( ctx, SimpleNamespace(driver=driver, native_request=driver.native_request) ) async def call(name, *, expect=True, **args): result = await getattr(actuator, name)(**args) records.append({"action": name, "args": args, "result": result.model_dump(mode="json")}) check(result.ok is expect, f"{name} expected success={expect}: {result.message}") return result.metadata async def read(desc): return (await call("read_control", selector={"content_desc": desc}))["node"] async def control(desc, action, **args): # Focus/selection can asynchronously pan the window for the IME. # Observe a settled target instead of treating a safe stale-token # rejection during that animation as successful input or retrying it. async def settled_editor(): await call( "wait_until", selector={"content_desc": desc}, state="stable", timeout_ms=3000, stable_ms=500, poll_interval_ms=100, ) if desc == "fixture.editor": await settled_editor() receipt = await call( "control_action", selector={"content_desc": desc}, action=action, **args ) if desc == "fixture.editor": # Selection handles/toolbars can disappear after a successful # write. Wait through incomplete window reads; never replay it. await settled_editor() return receipt await read("fixture.title") original = state() records.append( { "initial_fixture_state": original, "helper_version": session.version_name, "helper_code": session.version_code, } ) ambiguous = await call( "read_control", expect=False, selector={"class_name": "android.widget.TextView"} ) check(ambiguous["code"] == "AMBIGUOUS", "ambiguous selector rejected") missing = await call( "wait_until", selector={"content_desc": "fixture.missing"}, state="absent", timeout_ms=100, ) check(missing["success"], "short absent wait works") timeout = await call( "wait_until", expect=False, selector={"content_desc": "fixture.missing"}, state="visible", timeout_ms=100, ) check(timeout["code"] == "TIMEOUT", "missing visible times out") unsupported = await call( "control_action", expect=False, selector={"content_desc": "fixture.click"}, action="set_progress", value=50, ) check(unsupported["execution"] == "not_started", "unsupported node action did not start") for iteration in range(repeats): print( f"fixture iteration {iteration + 1}/{repeats}: native controls/text/wait", flush=True, ) value = [73, 21, 88, 49, 65][iteration % 5] node = await read("fixture.range") check( node["range"]["min"] == 0 and node["range"]["max"] == 100, "range endpoints exact" ) await control( "fixture.range", "set_progress", value=value, expected_token=node["token"] ) await call( "wait_until", selector={"content_desc": "fixture.range"}, state="value", expected_value=value, timeout_ms=1500, ) check(state()["progress"] == value, "range value persisted in app") before = state() await control("fixture.click", "click") await control("fixture.long_click", "long_click") after = state() check(after["click_count"] == before["click_count"] + 1, "native click delivered once") check( after["long_click_count"] == before["long_click_count"] + 1, "native long click delivered once", ) await control("fixture.editor", "set_text", text="A😀B中文") bad = await call( "control_action", expect=False, selector={"content_desc": "fixture.editor"}, action="set_selection", selection_start=2, selection_end=3, ) check( bad["execution"] == "not_started" and state()["text"] == "A😀B中文", "surrogate split rejected before edit", ) await control("fixture.editor", "set_selection", selection_start=1, selection_end=3) selected = await read("fixture.editor") check( (selected["selection_start"], selected["selection_end"]) == (1, 3), "UTF16 emoji selection exact", ) await control("fixture.editor", "replace_selection", text=" X ") edited = await read("fixture.editor") check( edited["text"] == "A X B中文" and edited["selection_start"] == edited["selection_end"] == 4, "replace only selection and place caret", ) stale = await call( "control_action", expect=False, selector={"content_desc": "fixture.editor"}, action="set_text", text="wrong", expected_token=selected["token"], ) check( stale["code"] == "STALE" and state()["text"] == "A X B中文", "stale token cannot edit changed node", ) await control("fixture.editor", "set_text", text=" 保留空白 😀 ") check( (await read("fixture.editor"))["text"] == " 保留空白 😀 ", "native set text preserves whitespace", ) editor_node = await read("fixture.editor") if not editor_node["focused"]: await control("fixture.editor", "focus") ime_before = state()["ime_enter_count"] await control("fixture.editor", "ime_enter") check(state()["ime_enter_count"] == ime_before + 1, "IME submit delivered once") ime = subprocess.run( adb_command(["-s", serial, "shell", "dumpsys", "input_method"]), capture_output=True, text=True, encoding="utf-8", errors="replace", check=True, timeout=10, ) if "mInputShown=true" in ime.stdout: check(await asyncio.to_thread(client.press_key, "back"), "dismiss visible IME") deadline = time.monotonic() + 2 while True: ime = subprocess.run( adb_command(["-s", serial, "shell", "dumpsys", "input_method"]), capture_output=True, text=True, encoding="utf-8", errors="replace", check=True, timeout=10, ) if "mInputShown=false" in ime.stdout or time.monotonic() >= deadline: break await asyncio.sleep(0.05) check("mInputShown=false" in ime.stdout, "IME no longer covers touch surface") await call( "wait_until", selector={"content_desc": "fixture.touch"}, state="stable", stable_ms=300, timeout_ms=1500, poll_interval_ms=100, ) before = state() await control("fixture.delayed", "click") waiting = await read("fixture.status") expected_version = ( int(before["status"].rsplit(":", 1)[-1]) + 1 if before["status"].startswith("status:changed:") else iteration + 1 ) # Each delayed callback increments its own counter once per iteration. await call( "wait_until", selector={"content_desc": "fixture.status"}, state="text", expected_text=f"status:changed:{expected_version}", timeout_ms=2500, poll_interval_ms=100, ) check( waiting["text"] in {"status:waiting", f"status:changed:{expected_version}"}, "condition read uses app text", ) await call( "wait_until", selector={"content_desc": "fixture.status"}, state="stable", stable_ms=150, timeout_ms=1000, poll_interval_ms=50, ) print(f"fixture iteration {iteration + 1}/{repeats}: synchronized touch", flush=True) await control("fixture.touch_reset", "click") touch = await read("fixture.touch") x1, y1, x2, y2 = touch["bounds"] left, right = max(0, x1) + 20, min(width - 1, x2) - 20 top, bottom = max(0, y1) + 20, min(height - 70, y2) - 20 check(right - left > 200 and bottom - top > 160, "touch fixture has room for gesture") center = [(left + right) // 2, (top + bottom) // 2] await call( "pinch", center=center, start_span=80, end_span=160, duration_ms=400, coordinate_space="pixel", target_description="fixture touch surface", ) touch_state = state()["touch"] check( touch_state["max_pointer_count"] == 2 and touch_state["down_count"] == touch_state["up_count"] == 1, "pinch has one contact with two pointers", ) check( touch_state["cancel_count"] == 0 and 1.85 < touch_state["scale"] < 2.15, "pinch ratio near two without cancellation", ) await control("fixture.touch_reset", "click") await call( "pinch", center=center, start_span=120, end_span=120, rotation_degrees=60, duration_ms=500, coordinate_space="pixel", target_description="fixture rotation", ) touch_state = state()["touch"] check( abs(touch_state["rotation_degrees"] - 60) < 3, "two finger rotation reaches 60 degrees", ) check( touch_state["down_count"] == touch_state["up_count"] == 1 and touch_state["cancel_count"] == 0, "rotation keeps contact across segments", ) await control("fixture.touch_reset", "click") cx, cy = center await call( "touch_gesture", frames=[ {"at_ms": 0, "points": [[cx - 60, cy]]}, {"at_ms": 250, "points": [[cx - 60, cy]]}, {"at_ms": 500, "points": [[cx, cy - 30]]}, {"at_ms": 750, "points": [[cx + 60, cy + 30]]}, ], coordinate_space="pixel", target_description="fixture hold then path", ) touch_state = state()["touch"] check( touch_state["down_count"] == touch_state["up_count"] == 1 and touch_state["move_count"] > 0 and touch_state["cancel_count"] == 0, "multi segment hold is a single continuous contact", ) events = touch_state["events"] down = next(event for event in events if event["action"] == 0) up = next(event for event in reversed(events) if event["action"] == 1) first = down["pointers"][0] moved = next( event for event in events if event["action"] == 2 and any(abs(event["pointers"][0][axis] - first[axis]) > 1 for axis in ("x", "y")) ) check( moved["time_ms"] - down["time_ms"] >= 235, "APP timestamps prove initial 250ms hold" ) check( up["time_ms"] - down["time_ms"] >= 730, "APP timestamps prove total 750ms path duration", ) records.append({"iteration": iteration + 1, "fixture_state": state()}) print("fixture scoped scroll repeated twice", flush=True) for _ in range(2): await call( "scroll_to", container={"content_desc": "fixture.scroll"}, target={"content_desc": "fixture.bottom"}, max_scrolls=15, timeout_ms=20000, settle_ms=200, ) check((await read("fixture.bottom"))["visible"], "scoped scroll found bottom") await call( "scroll_to", container={"content_desc": "fixture.scroll"}, target={"content_desc": "fixture.title"}, direction="backward", max_scrolls=15, timeout_ms=20000, settle_ms=200, ) check((await read("fixture.title"))["visible"], "scoped scroll restored top") await control("fixture.editor", "set_text", text=original["text"]) await control("fixture.range", "set_progress", value=original["progress"]) final_screen = await asyncio.to_thread(client.get_screen_data) (output / "fixture-final.jpg").write_bytes(base64.b64decode(final_screen.base64)) print(f"PASS {len(checks)} checks / {repeats} repeated fixture iterations", flush=True) finally: if client.session is not None: records.append({"final_fixture_state": state()}) (output / "result.json").write_text( json.dumps( { "checks": checks, "records": records, "elapsed_seconds": time.monotonic() - started, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) client.disconnect() lease.release() if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--serial", required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--repeats", type=int, choices=range(1, 11), default=5) options = parser.parse_args() asyncio.run(verify(options.serial, options.output, options.repeats))