446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""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))
|