mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-11-10 18:47:04 +01:00
phone_guard: lease or block raw adb; android_ui drives a session's own virtual display
A raw `adb shell input tap` took no lease, so one session's tap landed in another session's app (2026-09-25). python_pkg.phone_guard classifies every adb call in a Bash command and, as the Claude Code PreToolUse hook, takes the matching phone_lease (one app / real screen / whole phone) or exits 2: key/text input without -d, input on another session's virtual display, or a lease still held elsewhere after 40 s. android_ui gains --display self|<id>: it drives only this session's phone_vd display under that app's lease, and every input carries -d. uiautomator sees display 0 only, so virtual-display trees come from a tiny instrumentation APK (A11yDump.java) built from source without Gradle. Prunes the oldest evidence artifact to stay under the 60-file cap. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WpA6wW2DKiovQADTdjWJoH
This commit is contained in:
parent
bb9389864e
commit
aa37852296
14
docs/superpowers/contracts/phone-guard-2026-09-25.json
Normal file
14
docs/superpowers/contracts/phone-guard-2026-09-25.json
Normal file
@ -0,0 +1,14 @@
|
||||
{
|
||||
"title": "phone_guard: lease or block raw adb; android_ui drives a session's own virtual display",
|
||||
"objective": "No Claude session's adb call can land in another session's app: raw adb in a Bash call takes the matching phone_lease or is blocked, and android_ui can tap by element on the session's own virtual display.",
|
||||
"acceptance_criteria": [
|
||||
"Reads (logcat, dumpsys, pull, screencap) pass with no lease.",
|
||||
"Input or launches on display 0 take the pkg:__screen__ lease; install/force-stop take that app's lease; reboot, Wi-Fi/airplane/Doze and settings put take the whole-phone lease.",
|
||||
"input text/keyevent without -d, and input aimed at another session's virtual display, exit 2 with a reason.",
|
||||
"A lease held by another session is waited on for 40 s, then the call exits 2 naming the holder.",
|
||||
"android_ui --display self|<id> drives only this session's display under that app's lease; every input call carries -d.",
|
||||
"Virtual-display trees come from the com.kuhy.a11ydump instrumentation helper, built from source and installed with adb install -r -t."
|
||||
],
|
||||
"out_of_scope": ["phone_lease scopes themselves (c0c51036)", "the ~/.claude hook wrapper, phone_deploy.sh and the phone-focus-mode whitelist (separate repos)"],
|
||||
"verifier": "pytest python_pkg/android_ui python_pkg/phone_guard at 100% branch coverage; pre-commit on the changed files; a real helper APK build; the live two-session run on the Pixel follows this commit."
|
||||
}
|
||||
18
docs/superpowers/evidence/phone-guard-2026-09-25.json
Normal file
18
docs/superpowers/evidence/phone-guard-2026-09-25.json
Normal file
@ -0,0 +1,18 @@
|
||||
{
|
||||
"intent": "Several Claude sessions act on one phone at once without interrupting each other; a raw `adb shell input tap` (no lease) had landed in another session's app on 2026-09-25.",
|
||||
"scope": ["python_pkg/phone_guard (new: classify, _remote, _need, vd_state, hook, CLI, tests)", "python_pkg/android_ui (--display, _a11y_helper + A11yDump.java, -d on every input, tests split under the 250-line cap)"],
|
||||
"changes": [
|
||||
"phone_guard.classify maps every adb call in a shell command to a Need: nothing, screen, one app, whole phone, or a block",
|
||||
"phone_guard.hook applies it as the Claude Code PreToolUse hook: acquires as this session, waits 40 s, exits 2 on refusal",
|
||||
"android_ui --display self|<id> resolves the session's phone_vd display and takes its package lease; a foreign display is refused (exit 5)",
|
||||
"_a11y_helper builds a tiny instrumentation APK (javac, d8, aapt2, apksigner; no Gradle), cached by source hash, reading getWindowsOnAllDisplays()",
|
||||
"The helper source sits flat beside the module (A11yDump.java, a11y_dump_manifest.xml) to respect the two-directory path cap"
|
||||
],
|
||||
"verification": [
|
||||
{"command": "pytest python_pkg/android_ui python_pkg/phone_guard --cov --cov-branch", "result": "pass", "evidence": "152 passed; every android_ui and phone_guard module 100% line and branch"},
|
||||
{"command": "python3 -c 'from python_pkg.android_ui import _a11y_helper as h; print(h.build())'", "result": "pass", "evidence": "~/.cache/android_ui/a11ydump-fb2baa8888080a3c.apk built with the real SDK"},
|
||||
{"command": "pre-commit run --files <changed-files>", "result": "pass", "evidence": "all hooks passed"}
|
||||
],
|
||||
"risks": ["The hook passes unknown commands: it guards against interrupting other sessions, it is not a sandbox.", "The helper must stay in phone-focus-mode's whitelist or the focus-owner sweep hides it."],
|
||||
"rollback": ["git revert the commit", "Remove the phone_guard_pretool.sh entry from ~/.claude/settings.json PreToolUse"]
|
||||
}
|
||||
@ -1,57 +0,0 @@
|
||||
{
|
||||
"intent": "Test ubuntu_perf_fixes.sh to 100% line coverage and take it off the shell coverage allowlist, leaving one untested lib in fixes/lib.",
|
||||
"scope": [
|
||||
"linux_configuration/scripts/single_use/fixes/lib/ubuntu_perf_fixes.sh (four test seams)",
|
||||
"linux_configuration/scripts/single_use/fixes/lib/tests/test_ubuntu_perf_fixes.sh and test_ubuntu_perf_fixes_oom.sh (new)",
|
||||
"linux_configuration/scripts/single_use/fixes/lib/tests/ubuntu_perf_harness.sh (extended, not created -- it already existed)",
|
||||
"meta/shell-coverage-allowlist.txt (one line removed)"
|
||||
],
|
||||
"changes": [
|
||||
"Added SYSCTL_DROPIN_DIR, SYSTEMD_UNIT_DIR (twice) and EARLYOOM_CONF_FILE seams, each defaulting to the real path so production behaviour is unchanged",
|
||||
"Deliberately left the add_undo lines with their literal /etc paths: the undo script runs on the real system, not in the sandbox, so seaming them would have written a broken undo",
|
||||
"Extended the existing ubuntu_perf_harness.sh with the three new overrides, three more default stubs (nvidia-smi, nvidia-persistenced, dpkg) and the _t_unit / _t_sysctl_file readers",
|
||||
"Added 59 assertions over 14 cases across two files covering all four functions",
|
||||
"Removed ubuntu_perf_fixes.sh from meta/shell-coverage-allowlist.txt, taking it from 86 to 85 entries"
|
||||
],
|
||||
"verification": [
|
||||
{
|
||||
"command": "shell_coverage_jail.sh --timeout 600s --measure ubuntu_perf_fixes.sh --min 100 --fail-on-case-error",
|
||||
"result": "pass",
|
||||
"evidence": "ubuntu_perf_fixes.sh: 81/81 lines = 100.00%, exit 0, no case timeout."
|
||||
},
|
||||
{
|
||||
"command": "shell_coverage_jail.sh --timeout 600s --measure bt_audio.sh --min 100 (canary, with the two new test files present)",
|
||||
"result": "pass",
|
||||
"evidence": "bt_audio.sh: 127/127 lines = 100.00%. This is the lib a previous test file silently broke, so it is re-measured after every addition to this suite."
|
||||
},
|
||||
{
|
||||
"command": "bash meta/scripts/check_shell_coverage.sh .../ubuntu_perf_fixes.sh, AFTER deleting its allowlist line",
|
||||
"result": "pass",
|
||||
"evidence": "'every checked shell library meets the coverage bar', EXIT 0."
|
||||
},
|
||||
{
|
||||
"command": "ls -la /etc/sysctl.d/99-performance-tuning.conf /etc/default/earlyoom",
|
||||
"result": "pass",
|
||||
"evidence": "Both 'No such file or directory' after the suite ran, confirming the seams kept every write inside the throwaway tmpdir."
|
||||
},
|
||||
{
|
||||
"command": "bash linux_configuration/scripts/single_use/fixes/lib/tests/run_all.sh",
|
||||
"result": "pass",
|
||||
"evidence": "All fixes lib test files passed; 31 + 28 = 59 new assertions, 0 failed."
|
||||
},
|
||||
{
|
||||
"command": "shellcheck + shfmt on the lib, the harness and both test files",
|
||||
"result": "pass",
|
||||
"evidence": "No findings, zero suppressions. Lib is 219 lines, under the 250-line cap."
|
||||
}
|
||||
],
|
||||
"risks": [
|
||||
"fix_earlyoom writes a config on a machine that may already have one; the test asserts the .bak backup is taken before the overwrite, which is the behaviour that makes that safe.",
|
||||
"The undo lines are asserted to contain the REAL /etc paths, so a future seam applied to them would fail these tests rather than silently producing an undo script that removes nothing.",
|
||||
"fix_failed_sssd's detection loop breaks at the first failed unit while its masking loop re-checks each one; case 7 covers the single-failure arm specifically, since a naive implementation would mask all five."
|
||||
],
|
||||
"rollback": [
|
||||
"git revert the commit, then re-add the ubuntu_perf_fixes.sh line to meta/shell-coverage-allowlist.txt",
|
||||
"Re-run check_shell_coverage.sh to confirm the gate passes again with the exemption restored"
|
||||
]
|
||||
}
|
||||
133
python_pkg/android_ui/A11yDump.java
Normal file
133
python_pkg/android_ui/A11yDump.java
Normal file
@ -0,0 +1,133 @@
|
||||
package com.kuhy.a11ydump;
|
||||
|
||||
import android.accessibilityservice.AccessibilityServiceInfo;
|
||||
import android.app.Activity;
|
||||
import android.app.Instrumentation;
|
||||
import android.app.UiAutomation;
|
||||
import android.graphics.Rect;
|
||||
import android.os.Bundle;
|
||||
import android.os.SystemClock;
|
||||
import android.util.SparseArray;
|
||||
import android.view.accessibility.AccessibilityNodeInfo;
|
||||
import android.view.accessibility.AccessibilityWindowInfo;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Dumps one display's accessibility tree as uiautomator-style XML.
|
||||
*
|
||||
* <p>Run: {@code am instrument -w -e display <id> com.kuhy.a11ydump/.A11yDump}. The
|
||||
* XML comes back on one line as the {@code xml} result key, in the same shape
|
||||
* {@code uiautomator dump} writes, so the Python side parses both with one
|
||||
* parser. Coordinates are the display's own, which is what {@code input -d}
|
||||
* expects.
|
||||
*/
|
||||
public final class A11yDump extends Instrumentation {
|
||||
private static final int ATTEMPTS = 15;
|
||||
private static final long RETRY_MS = 200;
|
||||
|
||||
private Bundle arguments;
|
||||
|
||||
@Override
|
||||
public void onCreate(Bundle args) {
|
||||
arguments = args;
|
||||
start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onStart() {
|
||||
Bundle out = new Bundle();
|
||||
try {
|
||||
int display = Integer.parseInt(arguments.getString("display", "0"));
|
||||
UiAutomation automation = getUiAutomation(
|
||||
UiAutomation.FLAG_DONT_SUPPRESS_ACCESSIBILITY_SERVICES);
|
||||
AccessibilityServiceInfo info = automation.getServiceInfo();
|
||||
info.flags |= AccessibilityServiceInfo.FLAG_RETRIEVE_INTERACTIVE_WINDOWS
|
||||
| AccessibilityServiceInfo.FLAG_INCLUDE_NOT_IMPORTANT_VIEWS
|
||||
| AccessibilityServiceInfo.FLAG_REPORT_VIEW_IDS;
|
||||
automation.setServiceInfo(info);
|
||||
out.putString("xml", dump(automation, display));
|
||||
finish(Activity.RESULT_OK, out);
|
||||
} catch (Throwable t) {
|
||||
out.putString("error", String.valueOf(t));
|
||||
finish(Activity.RESULT_CANCELED, out);
|
||||
}
|
||||
}
|
||||
|
||||
private static String dump(UiAutomation automation, int display) {
|
||||
// Windows appear a beat after the service info changes; an empty
|
||||
// list on the first read means "not yet", not "nothing there".
|
||||
List<AccessibilityWindowInfo> windows = null;
|
||||
for (int i = 0; i < ATTEMPTS; i++) {
|
||||
SparseArray<List<AccessibilityWindowInfo>> all =
|
||||
automation.getWindowsOnAllDisplays();
|
||||
windows = all.get(display);
|
||||
if (windows != null && !windows.isEmpty()) break;
|
||||
SystemClock.sleep(RETRY_MS);
|
||||
}
|
||||
StringBuilder xml = new StringBuilder(
|
||||
"<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>"
|
||||
+ "<hierarchy rotation=\"0\" display=\"" + display + "\">");
|
||||
if (windows != null) {
|
||||
// Bottom layer first, like uiautomator, so a dialog's nodes come
|
||||
// after (and so win over) the screen it covers.
|
||||
for (int w = windows.size() - 1; w >= 0; w--) {
|
||||
AccessibilityNodeInfo root = windows.get(w).getRoot();
|
||||
if (root != null) node(root, 0, xml);
|
||||
}
|
||||
}
|
||||
return xml.append("</hierarchy>").toString();
|
||||
}
|
||||
|
||||
private static void node(AccessibilityNodeInfo n, int index, StringBuilder xml) {
|
||||
Rect r = new Rect();
|
||||
n.getBoundsInScreen(r);
|
||||
xml.append("<node index=\"").append(index).append('"');
|
||||
attr(xml, "text", n.getText());
|
||||
attr(xml, "resource-id", n.getViewIdResourceName());
|
||||
attr(xml, "class", n.getClassName());
|
||||
attr(xml, "package", n.getPackageName());
|
||||
attr(xml, "content-desc", n.getContentDescription());
|
||||
attr(xml, "hint", n.getHintText());
|
||||
flag(xml, "checkable", n.isCheckable());
|
||||
flag(xml, "checked", n.isChecked());
|
||||
flag(xml, "clickable", n.isClickable());
|
||||
flag(xml, "enabled", n.isEnabled());
|
||||
flag(xml, "focusable", n.isFocusable());
|
||||
flag(xml, "focused", n.isFocused());
|
||||
flag(xml, "scrollable", n.isScrollable());
|
||||
flag(xml, "password", n.isPassword());
|
||||
flag(xml, "selected", n.isSelected());
|
||||
xml.append(" bounds=\"[").append(r.left).append(',').append(r.top)
|
||||
.append("][").append(r.right).append(',').append(r.bottom)
|
||||
.append("]\">");
|
||||
for (int i = 0; i < n.getChildCount(); i++) {
|
||||
AccessibilityNodeInfo child = n.getChild(i);
|
||||
if (child != null) node(child, i, xml);
|
||||
}
|
||||
xml.append("</node>");
|
||||
}
|
||||
|
||||
private static void flag(StringBuilder xml, String name, boolean value) {
|
||||
xml.append(' ').append(name).append("=\"").append(value).append('"');
|
||||
}
|
||||
|
||||
private static void attr(StringBuilder xml, String name, CharSequence value) {
|
||||
xml.append(' ').append(name).append("=\"");
|
||||
if (value != null) {
|
||||
for (int i = 0; i < value.length(); i++) {
|
||||
char c = value.charAt(i);
|
||||
switch (c) {
|
||||
case '&': xml.append("&"); break;
|
||||
case '<': xml.append("<"); break;
|
||||
case '>': xml.append(">"); break;
|
||||
case '"': xml.append("""); break;
|
||||
// Newlines would split the one-line result `am` prints.
|
||||
case '\n': xml.append(" "); break;
|
||||
case '\r': xml.append(" "); break;
|
||||
default: xml.append(c);
|
||||
}
|
||||
}
|
||||
}
|
||||
xml.append('"');
|
||||
}
|
||||
}
|
||||
@ -90,9 +90,30 @@ ui.wait_for("Connected.", timeout=30)
|
||||
|
||||
## One phone, many sessions
|
||||
|
||||
Every command first takes this session's lease on the phone
|
||||
(`python_pkg.phone_lease`): the owner is the nearest ancestor `claude`
|
||||
process, the lease lasts 180 s past the last call, and a call from another
|
||||
Sessions share the phone by working on separate displays. By default every
|
||||
command drives the real screen (display 0) and takes this session's
|
||||
`pkg:__screen__` lease (`python_pkg.phone_lease`: owner = the nearest
|
||||
ancestor `claude` process, 180 s past the last call). A call from another
|
||||
session waits up to 60 s, then exits 3 naming the holder. Do not retry in a
|
||||
loop — use the phone's virtual display (`~/.claude/scripts/phone_vd.sh`) or
|
||||
the headless emulator (`~/.claude/scripts/phone_emu.sh`), or ask the user.
|
||||
loop.
|
||||
|
||||
`--display self` drives this session's virtual display instead
|
||||
(`~/.claude/scripts/phone_vd.sh start <package>` creates it) and takes only
|
||||
that app's lease, so several sessions can act at once:
|
||||
|
||||
android-ui --display self dump
|
||||
android-ui --display self tap "Save"
|
||||
|
||||
`uiautomator dump` sees display 0 only, so on a virtual display the tree
|
||||
comes from the `com.kuhy.a11ydump` instrumentation helper (`A11yDump.java`,
|
||||
`_a11y_helper.py`): built once into `~/.cache/android_ui/` and installed
|
||||
with `adb install -r -t` on first use. Its package must stay in
|
||||
phone-focus-mode's whitelist, or the focus-owner sweep hides it. Every
|
||||
`input` call carries `-d` (0 for the real screen), because input without it
|
||||
goes to whichever display last took focus. A numeric `--display` that is
|
||||
another session's virtual display is refused.
|
||||
|
||||
Raw `adb` from a Claude Bash call is covered too: the
|
||||
`~/.claude/hooks/phone_guard_pretool.sh` hook (`python_pkg.phone_guard`)
|
||||
takes the matching lease or blocks the call. See the `phone-deploy` skill,
|
||||
section 3, for the full rules.
|
||||
|
||||
225
python_pkg/android_ui/_a11y_helper.py
Normal file
225
python_pkg/android_ui/_a11y_helper.py
Normal file
@ -0,0 +1,225 @@
|
||||
"""Accessibility-tree dumps of any display, via a tiny instrumentation APK.
|
||||
|
||||
``uiautomator dump`` only ever sees display 0, so an app on a session's
|
||||
virtual display (``phone_vd.sh``) could only be tapped by coordinates. The
|
||||
APK built from ``A11yDump.java`` runs under ``am instrument``, where a UiAutomation can
|
||||
read ``getWindowsOnAllDisplays()`` and print one display's tree in the same
|
||||
XML shape ``uiautomator`` writes.
|
||||
|
||||
It is built from source here (javac → d8 → aapt2 → apksigner, no Gradle),
|
||||
cached by source hash under ``~/.cache/android_ui/``, and (re)installed with
|
||||
``adb install -r -t`` whenever the phone holds a different build. The APK is
|
||||
never committed: the repo's binary gate forbids it, and the source is enough.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import zipfile
|
||||
|
||||
from python_pkg.android_ui._elements import UiAutomationError
|
||||
|
||||
PACKAGE = "com.kuhy.a11ydump"
|
||||
_RUNNER = f"{PACKAGE}/.A11yDump"
|
||||
# Flat beside this module: the repo caps paths at two directories, and javac
|
||||
# does not need the package's directory layout for an explicitly named file.
|
||||
_JAVA = Path(__file__).with_name("A11yDump.java")
|
||||
_MANIFEST = Path(__file__).with_name("a11y_dump_manifest.xml")
|
||||
_MIN_API = "30" # getWindowsOnAllDisplays()
|
||||
|
||||
# (adb args...) -> stdout, raising UiAutomationError on failure.
|
||||
AdbRun = Callable[..., str]
|
||||
|
||||
|
||||
def source_hash() -> str:
|
||||
"""Hash of every source file: the build's identity and versionName."""
|
||||
digest = hashlib.sha256()
|
||||
for path in (_JAVA, _MANIFEST):
|
||||
digest.update(path.name.encode())
|
||||
digest.update(path.read_bytes())
|
||||
return digest.hexdigest()[:16]
|
||||
|
||||
|
||||
def cache_dir() -> Path:
|
||||
"""Where built APKs are kept."""
|
||||
return Path(
|
||||
os.environ.get("ANDROID_UI_CACHE", Path.home() / ".cache" / "android_ui")
|
||||
)
|
||||
|
||||
|
||||
def sdk_root() -> Path:
|
||||
"""The Android SDK, from $ANDROID_HOME / $ANDROID_SDK_ROOT."""
|
||||
for var in ("ANDROID_HOME", "ANDROID_SDK_ROOT"):
|
||||
value = os.environ.get(var)
|
||||
if value and Path(value).is_dir():
|
||||
return Path(value)
|
||||
fallback = Path.home() / "sdk" / "Android" / "Sdk"
|
||||
if fallback.is_dir():
|
||||
return fallback
|
||||
msg = "no Android SDK: set ANDROID_HOME (needs build-tools and a platform)"
|
||||
raise UiAutomationError(msg)
|
||||
|
||||
|
||||
def _newest(parent: Path, prefix: str = "") -> Path:
|
||||
candidates = sorted(
|
||||
(p for p in parent.iterdir() if p.name.startswith(prefix) and p.is_dir()),
|
||||
key=lambda p: [
|
||||
int(x) if x.isdigit() else 0 for x in p.name.split("-")[-1].split(".")
|
||||
],
|
||||
)
|
||||
if not candidates:
|
||||
msg = f"nothing matching {prefix!r} under {parent}"
|
||||
raise UiAutomationError(msg)
|
||||
return candidates[-1]
|
||||
|
||||
|
||||
def build_tool(name: str) -> Path:
|
||||
"""Absolute path of an SDK build tool (``aapt2``, ``d8``, ...), newest version."""
|
||||
return _newest(sdk_root() / "build-tools") / name
|
||||
|
||||
|
||||
def _tool(*cmd: str | Path) -> None:
|
||||
done = subprocess.run(
|
||||
[str(c) for c in cmd], capture_output=True, text=True, check=False
|
||||
)
|
||||
if done.returncode != 0:
|
||||
msg = f"building the a11y helper failed: {cmd[0]}\n{done.stderr.strip()}"
|
||||
raise UiAutomationError(msg)
|
||||
|
||||
|
||||
def build() -> Path:
|
||||
"""Return the helper APK for the current source, building it if needed."""
|
||||
version = source_hash()
|
||||
apk = cache_dir() / f"a11ydump-{version}.apk"
|
||||
if apk.is_file():
|
||||
return apk
|
||||
sdk = sdk_root()
|
||||
tools = _newest(sdk / "build-tools")
|
||||
android_jar = _newest(sdk / "platforms", "android-") / "android.jar"
|
||||
keystore = Path.home() / ".android" / "debug.keystore"
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
work = Path(tmp)
|
||||
classes, dex = work / "classes", work / "dex"
|
||||
classes.mkdir()
|
||||
dex.mkdir()
|
||||
_tool(
|
||||
"javac",
|
||||
"--release",
|
||||
"11",
|
||||
"-classpath",
|
||||
android_jar,
|
||||
"-d",
|
||||
classes,
|
||||
_JAVA,
|
||||
)
|
||||
class_files = sorted(str(p) for p in classes.rglob("*.class"))
|
||||
_tool(
|
||||
tools / "d8",
|
||||
"--release",
|
||||
"--min-api",
|
||||
_MIN_API,
|
||||
"--lib",
|
||||
android_jar,
|
||||
"--output",
|
||||
dex,
|
||||
*class_files,
|
||||
)
|
||||
unsigned = work / "unsigned.apk"
|
||||
_tool(
|
||||
tools / "aapt2",
|
||||
"link",
|
||||
"--manifest",
|
||||
_MANIFEST,
|
||||
"-I",
|
||||
android_jar,
|
||||
"--version-code",
|
||||
"1",
|
||||
"--version-name",
|
||||
version,
|
||||
"-o",
|
||||
unsigned,
|
||||
)
|
||||
with zipfile.ZipFile(unsigned, "a") as zf:
|
||||
zf.write(dex / "classes.dex", "classes.dex")
|
||||
aligned = work / "aligned.apk"
|
||||
_tool(tools / "zipalign", "-f", "4", unsigned, aligned)
|
||||
if not keystore.is_file():
|
||||
_tool(
|
||||
"keytool",
|
||||
"-genkeypair",
|
||||
"-keystore",
|
||||
keystore,
|
||||
"-storepass",
|
||||
"android",
|
||||
"-alias",
|
||||
"androiddebugkey",
|
||||
"-keypass",
|
||||
"android",
|
||||
"-keyalg",
|
||||
"RSA",
|
||||
"-validity",
|
||||
"10000",
|
||||
"-dname",
|
||||
"CN=Android Debug,O=Android,C=US",
|
||||
)
|
||||
_tool(
|
||||
tools / "apksigner",
|
||||
"sign",
|
||||
"--ks",
|
||||
keystore,
|
||||
"--ks-pass",
|
||||
"pass:android",
|
||||
"--key-pass",
|
||||
"pass:android",
|
||||
"--out",
|
||||
work / "signed.apk",
|
||||
aligned,
|
||||
)
|
||||
apk.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(work / "signed.apk", apk)
|
||||
return apk
|
||||
|
||||
|
||||
def ensure_installed(run: AdbRun) -> None:
|
||||
"""Install the helper unless the phone already holds this exact build."""
|
||||
version = source_hash()
|
||||
info = run("shell", "dumpsys", "package", PACKAGE)
|
||||
if f"versionName={version}" in info:
|
||||
return
|
||||
run("install", "-r", "-t", str(build()), timeout=120.0)
|
||||
|
||||
|
||||
def parse_result(output: str) -> str:
|
||||
"""Pull the XML out of ``am instrument -w`` output, or raise its error."""
|
||||
for line in output.splitlines():
|
||||
if line.startswith("INSTRUMENTATION_RESULT: xml="):
|
||||
return line.split("=", 1)[1]
|
||||
for line in output.splitlines():
|
||||
if line.startswith("INSTRUMENTATION_RESULT: error="):
|
||||
msg = f"a11y helper failed: {line.split('=', 1)[1]}"
|
||||
raise UiAutomationError(msg)
|
||||
msg = f"a11y helper returned no tree:\n{output.strip()[-500:]}"
|
||||
raise UiAutomationError(msg)
|
||||
|
||||
|
||||
def dump_display(run: AdbRun, display: int) -> str:
|
||||
"""Return ``display``'s accessibility tree as uiautomator-style XML."""
|
||||
ensure_installed(run)
|
||||
output = run(
|
||||
"shell",
|
||||
"am",
|
||||
"instrument",
|
||||
"-w",
|
||||
"-e",
|
||||
"display",
|
||||
str(display),
|
||||
_RUNNER,
|
||||
timeout=45.0,
|
||||
)
|
||||
return parse_result(output)
|
||||
@ -33,6 +33,10 @@ class TextEntryMixin(ABC):
|
||||
def _run(self, *args: str, timeout: float = 30.0) -> str:
|
||||
"""Supplied by the concrete driver: run one adb command."""
|
||||
|
||||
@abstractmethod
|
||||
def _input(self, *args: str) -> str:
|
||||
"""Supplied by the concrete driver: ``input`` on the driven display."""
|
||||
|
||||
@abstractmethod
|
||||
def dump(self, *, retries: int = 4) -> list[UiElement]:
|
||||
"""Supplied by the concrete driver: read the current UI tree."""
|
||||
@ -49,9 +53,9 @@ class TextEntryMixin(ABC):
|
||||
return
|
||||
# KEYCODE_MOVE_END (123) then a run of deletes is more reliable across
|
||||
# IMEs than CTRL+A, which not every keyboard honours.
|
||||
self._run("shell", "input", "keyevent", "123")
|
||||
self._input("keyevent", "123")
|
||||
for _ in range(length + 2):
|
||||
self._run("shell", "input", "keyevent", "67")
|
||||
self._input("keyevent", "67")
|
||||
time.sleep(self._settle)
|
||||
|
||||
def editable_fields(self) -> list[UiElement]:
|
||||
@ -78,14 +82,14 @@ class TextEntryMixin(ABC):
|
||||
target = fields[index]
|
||||
before = target.text
|
||||
x, y = target.center
|
||||
self._run("shell", "input", "tap", str(x), str(y))
|
||||
self._input("tap", str(x), str(y))
|
||||
time.sleep(max(self._settle, 0.8))
|
||||
# REPLACE, don't append. `input text` inserts at the cursor, so typing
|
||||
# into a field that already holds something silently concatenates --
|
||||
# producing e.g. "old@example.comnew@example.com", which is accepted by
|
||||
# the widget, passes a "did the text change?" check, and is wrong.
|
||||
self._clear_focused_field(len(before))
|
||||
self._run("shell", "input", "text", _escape(text))
|
||||
self._input("text", _escape(text))
|
||||
time.sleep(self._settle)
|
||||
after = self.editable_fields()
|
||||
changed = any(
|
||||
@ -118,7 +122,7 @@ class TextEntryMixin(ABC):
|
||||
for keyevent in ("111", "4"):
|
||||
if not self.keyboard_is_up():
|
||||
return
|
||||
self._run("shell", "input", "keyevent", keyevent)
|
||||
self._input("keyevent", keyevent)
|
||||
time.sleep(max(self._settle, 0.6))
|
||||
if self.keyboard_is_up():
|
||||
msg = (
|
||||
|
||||
12
python_pkg/android_ui/a11y_dump_manifest.xml
Normal file
12
python_pkg/android_ui/a11y_dump_manifest.xml
Normal file
@ -0,0 +1,12 @@
|
||||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<!-- A test-only APK with no app code of its own: `am instrument` runs A11yDump
|
||||
with a UiAutomation, which can read the accessibility tree of every
|
||||
display, virtual ones included (`uiautomator dump` sees display 0 only). -->
|
||||
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
|
||||
package="com.kuhy.a11ydump">
|
||||
<uses-sdk android:minSdkVersion="30" android:targetSdkVersion="36" />
|
||||
<application android:label="a11ydump" android:hasCode="true" />
|
||||
<instrumentation
|
||||
android:name="com.kuhy.a11ydump.A11yDump"
|
||||
android:targetPackage="com.kuhy.a11ydump" />
|
||||
</manifest>
|
||||
@ -9,6 +9,7 @@ successful no-op instead of guessing from a screenshot.
|
||||
android-ui tap "Connect Firebase"
|
||||
android-ui type "Sync account email" kuhy@example.com
|
||||
android-ui wait "Connected to Firebase." --timeout 30
|
||||
android-ui --display self tap "Save" # this session's phone_vd display
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -17,7 +18,24 @@ import argparse
|
||||
import sys
|
||||
|
||||
from python_pkg.android_ui.driver import AndroidUi, UiAutomationError
|
||||
from python_pkg.phone_lease import PhoneBusyError, acquire
|
||||
from python_pkg.phone_guard.vd_state import display_owner, own_display
|
||||
from python_pkg.phone_lease import (
|
||||
NoPhoneError,
|
||||
PhoneBusyError,
|
||||
acquire,
|
||||
owner_id,
|
||||
resolve_serial,
|
||||
scope_of,
|
||||
)
|
||||
|
||||
# Exit codes shared with phone_lease: 3 held elsewhere, 4 no phone.
|
||||
_EXIT_BUSY = 3
|
||||
_EXIT_NO_PHONE = 4
|
||||
_EXIT_BAD_DISPLAY = 5
|
||||
|
||||
|
||||
class _TargetError(RuntimeError):
|
||||
"""The requested display cannot be driven by this session."""
|
||||
|
||||
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
@ -27,6 +45,11 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
description="Drive an Android app by element, never by coordinates.",
|
||||
)
|
||||
parser.add_argument("-s", "--serial", help="target device serial")
|
||||
parser.add_argument(
|
||||
"--display",
|
||||
help="drive a virtual display instead of the real screen: its logical "
|
||||
"id, or 'self' for the one phone_vd.sh started for this session",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--exact",
|
||||
action="store_true",
|
||||
@ -56,16 +79,43 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
return parser
|
||||
|
||||
|
||||
def _lease_phone(serial: str | None, command: str) -> bool:
|
||||
"""Refresh this session's lease on the phone; False when another holds it.
|
||||
def _target(serial: str, requested: str | None) -> tuple[int | None, str]:
|
||||
"""Return (display, lease scope) for ``--display``.
|
||||
|
||||
Every command goes through here first, so a second session's taps wait
|
||||
instead of landing in whatever this one has in front (2026-09-20: a
|
||||
sandbox tap opened another session's manga reader). Serial-less calls
|
||||
share one key: adb picks one device anyway.
|
||||
The real screen is shared, so it takes the screen lease. A virtual display
|
||||
must be this session's own (another session's app would be driven blind)
|
||||
and takes the lease on its app instead, which is what lets two sessions
|
||||
work at the same time.
|
||||
"""
|
||||
if requested is None or requested == "0":
|
||||
return None, scope_of() # the real screen: pkg:__screen__
|
||||
mine = own_display(serial, owner_id())
|
||||
if requested == "self":
|
||||
if mine is None:
|
||||
msg = "no virtual display for this session: run phone_vd.sh start <package>"
|
||||
raise _TargetError(msg)
|
||||
return mine.display, scope_of(mine.package)
|
||||
if not requested.isdigit():
|
||||
msg = f"--display takes a display id or 'self', not {requested!r}"
|
||||
raise _TargetError(msg)
|
||||
display = int(requested)
|
||||
if mine is None or mine.display != display:
|
||||
other = display_owner(serial, display)
|
||||
whose = f"owned by {other.owner_dir}" if other else "not this session's"
|
||||
msg = f"display {display} is {whose}; drive only your own (--display self)"
|
||||
raise _TargetError(msg)
|
||||
return display, scope_of(mine.package)
|
||||
|
||||
|
||||
def _lease_phone(serial: str, command: str, scope: str) -> bool:
|
||||
"""Refresh this session's lease; False when another session holds it.
|
||||
|
||||
Keyed by the resolved serial, never a placeholder: serial-less calls used
|
||||
to lease "default" while phone_deploy leased the real serial, so neither
|
||||
saw the other and one session's tap landed in another's app (2026-09-25).
|
||||
"""
|
||||
try:
|
||||
acquire(serial or "default", f"android_ui {command}")
|
||||
acquire(serial, f"android_ui {command}", scope=scope)
|
||||
except PhoneBusyError as exc:
|
||||
sys.stderr.write(f"android-ui: {exc}\n")
|
||||
return False
|
||||
@ -97,10 +147,19 @@ def _run(ui: AndroidUi, args: argparse.Namespace) -> None:
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
"""Run the CLI. Returns a process exit code."""
|
||||
args = _build_parser().parse_args(argv)
|
||||
if not _lease_phone(args.serial, args.command):
|
||||
return 3
|
||||
try:
|
||||
_run(AndroidUi(serial=args.serial), args)
|
||||
serial = resolve_serial(args.serial)
|
||||
display, scope = _target(serial, args.display)
|
||||
except NoPhoneError as exc:
|
||||
sys.stderr.write(f"android-ui: {exc}\n")
|
||||
return _EXIT_NO_PHONE
|
||||
except _TargetError as exc:
|
||||
sys.stderr.write(f"android-ui: {exc}\n")
|
||||
return _EXIT_BAD_DISPLAY
|
||||
if not _lease_phone(serial, args.command, scope):
|
||||
return _EXIT_BUSY
|
||||
try:
|
||||
_run(AndroidUi(serial=serial, display=display), args)
|
||||
except UiAutomationError as exc:
|
||||
# Loud and non-zero on purpose: a silent no-op here is exactly the
|
||||
# failure this package exists to eliminate. The message already names
|
||||
|
||||
@ -8,6 +8,7 @@ import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
from python_pkg.android_ui import _a11y_helper
|
||||
from python_pkg.android_ui._elements import (
|
||||
AmbiguousElementError,
|
||||
ElementNotFoundError,
|
||||
@ -38,11 +39,19 @@ class AndroidUi(TextEntryMixin):
|
||||
*,
|
||||
adb: str = "adb",
|
||||
settle_seconds: float = 0.4,
|
||||
display: int | None = None,
|
||||
) -> None:
|
||||
"""Create a driver. ``serial`` targets one of several devices."""
|
||||
"""Create a driver. ``serial`` targets one of several devices.
|
||||
|
||||
``display`` drives one display (a session's virtual display from
|
||||
``phone_vd.sh``) instead of the real screen: input goes to it with
|
||||
``input -d`` and its tree comes from the a11y helper, because
|
||||
``uiautomator dump`` cannot see any display but 0.
|
||||
"""
|
||||
self._adb = adb
|
||||
self._serial = serial
|
||||
self._settle = settle_seconds
|
||||
self._display = display
|
||||
|
||||
def _run(self, *args: str, timeout: float = 30.0) -> str:
|
||||
"""Run an adb command, returning stdout."""
|
||||
@ -69,6 +78,17 @@ class AndroidUi(TextEntryMixin):
|
||||
raise UiAutomationError(msg)
|
||||
return done.stdout
|
||||
|
||||
def _input(self, *args: str) -> str:
|
||||
"""Run ``input`` on the display this driver targets.
|
||||
|
||||
Always with ``-d``, 0 meaning the real screen: without it, key and
|
||||
text events go to whichever display last had focus, and a session's
|
||||
virtual display takes focus the moment its app launches -- so another
|
||||
session's typing lands in that app.
|
||||
"""
|
||||
display = 0 if self._display is None else self._display
|
||||
return self._run("shell", "input", "-d", str(display), *args)
|
||||
|
||||
# ── Reading the screen ────────────────────────────────────────────────
|
||||
|
||||
def dump(self, *, retries: int = 4) -> list[UiElement]:
|
||||
@ -96,6 +116,8 @@ class AndroidUi(TextEntryMixin):
|
||||
|
||||
def _dump_once(self) -> list[UiElement]:
|
||||
"""Pull one accessibility-tree snapshot."""
|
||||
if self._display is not None:
|
||||
return _parse_tree(_a11y_helper.dump_display(self._run, self._display))
|
||||
self._run("shell", "uiautomator", "dump", _REMOTE_DUMP, timeout=45.0)
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
local = Path(tmp) / "ui.xml"
|
||||
@ -161,7 +183,7 @@ class AndroidUi(TextEntryMixin):
|
||||
self.dismiss_keyboard()
|
||||
element = self.wait_for(query, timeout=timeout, exact=exact)
|
||||
x, y = element.center
|
||||
self._run("shell", "input", "tap", str(x), str(y))
|
||||
self._input("tap", str(x), str(y))
|
||||
time.sleep(self._settle)
|
||||
return element
|
||||
|
||||
@ -182,10 +204,10 @@ class AndroidUi(TextEntryMixin):
|
||||
element = self.wait_for(query, timeout=timeout, exact=exact)
|
||||
before = element.text
|
||||
x, y = element.center
|
||||
self._run("shell", "input", "tap", str(x), str(y))
|
||||
self._input("tap", str(x), str(y))
|
||||
# The field has to actually take focus before keystrokes mean anything.
|
||||
time.sleep(max(self._settle, 0.8))
|
||||
self._run("shell", "input", "text", _escape(text))
|
||||
self._input("text", _escape(text))
|
||||
time.sleep(self._settle)
|
||||
|
||||
for candidate in self.dump():
|
||||
|
||||
85
python_pkg/android_ui/tests/conftest.py
Normal file
85
python_pkg/android_ui/tests/conftest.py
Normal file
@ -0,0 +1,85 @@
|
||||
"""Shared fixtures and a scripted adb stand-in for the android_ui tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
MOD = "python_pkg.android_ui.cli"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def phone() -> Iterator[None]:
|
||||
"""One resolved phone, and no virtual display unless a test adds one."""
|
||||
with (
|
||||
patch(f"{MOD}.resolve_serial", side_effect=lambda s=None: s or "SER1"),
|
||||
patch(f"{MOD}.owner_id", return_value="claude:1"),
|
||||
patch(f"{MOD}.own_display", return_value=None),
|
||||
patch(f"{MOD}.display_owner", return_value=None),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
def node(text: str = "", **attrs: str) -> str:
|
||||
"""Render one ``<node>`` element for a fake dump.
|
||||
|
||||
``attrs`` accepts ``desc``, ``hint``, ``res``, ``cls``, ``bounds``,
|
||||
``enabled`` and ``focused``; each falls back to a representative default.
|
||||
"""
|
||||
return (
|
||||
f'<node text="{text}" content-desc="{attrs.get("desc", "")}" '
|
||||
f'hint="{attrs.get("hint", "")}" '
|
||||
f'resource-id="{attrs.get("res", "")}" '
|
||||
f'class="{attrs.get("cls", "android.widget.TextView")}" '
|
||||
f'bounds="{attrs.get("bounds", "[0,0][100,50]")}" '
|
||||
f'enabled="{attrs.get("enabled", "true")}" '
|
||||
f'focused="{attrs.get("focused", "false")}"/>'
|
||||
)
|
||||
|
||||
|
||||
def tree(*nodes: str) -> str:
|
||||
"""Wrap ``nodes`` in a hierarchy document."""
|
||||
return f"<?xml version='1.0'?><hierarchy rotation='0'>{''.join(nodes)}</hierarchy>"
|
||||
|
||||
|
||||
class FakeDevice:
|
||||
"""Scripted stand-in for adb, recording every command it is given."""
|
||||
|
||||
def __init__(self, trees: list[str] | None = None) -> None:
|
||||
self.trees = trees or [tree(node("Connect"))]
|
||||
self.calls: list[tuple[str, ...]] = []
|
||||
self.keyboard_shown = False
|
||||
self.pulled = 0
|
||||
|
||||
def run(self, *args: str, timeout: float = 30.0) -> str:
|
||||
"""Answer an adb invocation."""
|
||||
del timeout
|
||||
self.calls.append(args)
|
||||
if args[:2] == ("shell", "dumpsys") and args[2] == "input_method":
|
||||
return f"mInputShown={'true' if self.keyboard_shown else 'false'}"
|
||||
if args[:2] == ("shell", "dumpsys") and args[2] == "window":
|
||||
return "mCurrentFocus=Window{ab12 u0 com.example/com.example.Main}"
|
||||
if args[0] == "pull":
|
||||
index = min(self.pulled, len(self.trees) - 1)
|
||||
self.pulled += 1
|
||||
Path(args[2]).write_text(self.trees[index], encoding="utf-8")
|
||||
if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in {
|
||||
"111",
|
||||
"4",
|
||||
}:
|
||||
self.keyboard_shown = False
|
||||
return ""
|
||||
|
||||
def taps(self) -> list[tuple[int, int]]:
|
||||
"""Return every tap coordinate, in order."""
|
||||
return [
|
||||
(int(c[5]), int(c[6]))
|
||||
for c in self.calls
|
||||
if c[:5] == ("shell", "input", "-d", "0", "tap")
|
||||
]
|
||||
205
python_pkg/android_ui/tests/test_a11y_helper.py
Normal file
205
python_pkg/android_ui/tests/test_a11y_helper.py
Normal file
@ -0,0 +1,205 @@
|
||||
"""Tests for the a11y dump helper: build caching, install, result parsing.
|
||||
|
||||
The SDK tools are faked: each fake writes the file the real tool would, so the
|
||||
build pipeline's wiring (what feeds what, what gets cached) is what is tested.
|
||||
The real build and a real dump were verified on the Pixel 6a on 2026-09-25.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.android_ui import _a11y_helper as helper
|
||||
from python_pkg.android_ui._elements import UiAutomationError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
MOD = "python_pkg.android_ui._a11y_helper"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def source(tmp_path: Path) -> Iterator[Path]:
|
||||
"""A tiny helper source: one Java file and its manifest."""
|
||||
java = tmp_path / "A11yDump.java"
|
||||
java.write_text("class A11yDump {}")
|
||||
manifest = tmp_path / "a11y_dump_manifest.xml"
|
||||
manifest.write_text("<manifest/>")
|
||||
with patch(f"{MOD}._JAVA", java), patch(f"{MOD}._MANIFEST", manifest):
|
||||
yield tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sdk(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""A fake SDK layout, with the cache and HOME inside tmp_path."""
|
||||
root = tmp_path / "sdk"
|
||||
for d in (
|
||||
"build-tools/35.0.0",
|
||||
"build-tools/36.0.0",
|
||||
"platforms/android-9",
|
||||
"platforms/android-36",
|
||||
"platforms/android-37.2",
|
||||
):
|
||||
(root / d).mkdir(parents=True)
|
||||
monkeypatch.setenv("ANDROID_HOME", str(root))
|
||||
monkeypatch.setenv("ANDROID_UI_CACHE", str(tmp_path / "cache"))
|
||||
monkeypatch.setenv("HOME", str(tmp_path / "home"))
|
||||
return root
|
||||
|
||||
|
||||
def fake_tool(calls: list[list[str]]) -> MagicMock:
|
||||
"""subprocess.run stand-in that produces each tool's output file."""
|
||||
|
||||
def run(cmd: list[str], **_: object) -> subprocess.CompletedProcess[str]:
|
||||
calls.append(cmd)
|
||||
name = Path(cmd[0]).name
|
||||
if name == "d8":
|
||||
out = Path(cmd[cmd.index("--output") + 1])
|
||||
(out / "classes.dex").write_bytes(b"dex")
|
||||
elif name == "aapt2":
|
||||
with zipfile.ZipFile(cmd[cmd.index("-o") + 1], "w") as zf:
|
||||
zf.writestr("AndroidManifest.xml", "m")
|
||||
elif name == "zipalign":
|
||||
shutil.copy(cmd[-2], cmd[-1])
|
||||
elif name == "apksigner":
|
||||
shutil.copy(cmd[-1], cmd[cmd.index("--out") + 1])
|
||||
elif name == "keytool":
|
||||
Path(cmd[cmd.index("-keystore") + 1]).parent.mkdir(parents=True)
|
||||
Path(cmd[cmd.index("-keystore") + 1]).write_bytes(b"ks")
|
||||
return subprocess.CompletedProcess(cmd, 0, "", "")
|
||||
|
||||
return MagicMock(side_effect=run)
|
||||
|
||||
|
||||
class TestIdentity:
|
||||
def test_hash_follows_the_source(self, source: Path) -> None:
|
||||
first = helper.source_hash()
|
||||
assert first == helper.source_hash()
|
||||
(source / "a11y_dump_manifest.xml").write_text("<manifest x=''/>")
|
||||
assert helper.source_hash() != first
|
||||
|
||||
def test_cache_dir_defaults_under_home(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("ANDROID_UI_CACHE", raising=False)
|
||||
assert helper.cache_dir() == Path.home() / ".cache" / "android_ui"
|
||||
|
||||
|
||||
class TestSdk:
|
||||
def test_env_wins(self, sdk: Path) -> None:
|
||||
assert helper.sdk_root() == sdk
|
||||
|
||||
def test_falls_back_to_home_sdk(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("ANDROID_HOME", raising=False)
|
||||
monkeypatch.setenv("ANDROID_SDK_ROOT", str(tmp_path / "missing"))
|
||||
fallback = tmp_path / "sdk" / "Android" / "Sdk"
|
||||
fallback.mkdir(parents=True)
|
||||
with patch(f"{MOD}.Path.home", return_value=tmp_path):
|
||||
assert helper.sdk_root() == fallback
|
||||
|
||||
def test_no_sdk_raises(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("ANDROID_HOME", raising=False)
|
||||
monkeypatch.delenv("ANDROID_SDK_ROOT", raising=False)
|
||||
with (
|
||||
patch(f"{MOD}.Path.home", return_value=tmp_path),
|
||||
pytest.raises(UiAutomationError, match="no Android SDK"),
|
||||
):
|
||||
helper.sdk_root()
|
||||
|
||||
def test_build_tool_is_in_the_newest_build_tools(self, sdk: Path) -> None:
|
||||
assert helper.build_tool("aapt2") == sdk / "build-tools" / "36.0.0" / "aapt2"
|
||||
|
||||
def test_newest_version_wins_and_empty_raises(self, sdk: Path) -> None:
|
||||
assert helper._newest(sdk / "build-tools").name == "36.0.0"
|
||||
assert helper._newest(sdk / "platforms", "android-").name == "android-37.2"
|
||||
with pytest.raises(UiAutomationError, match="nothing matching"):
|
||||
helper._newest(sdk / "platforms", "nope-")
|
||||
|
||||
|
||||
class TestBuild:
|
||||
def test_builds_signs_and_caches(self, source: Path, sdk: Path) -> None:
|
||||
del source, sdk
|
||||
calls: list[list[str]] = []
|
||||
with patch(f"{MOD}.subprocess.run", fake_tool(calls)):
|
||||
apk = helper.build()
|
||||
assert apk.name == f"a11ydump-{helper.source_hash()}.apk"
|
||||
with zipfile.ZipFile(apk) as zf:
|
||||
assert "classes.dex" in zf.namelist()
|
||||
tools = [Path(c[0]).name for c in calls]
|
||||
assert tools == ["javac", "d8", "aapt2", "zipalign", "keytool", "apksigner"]
|
||||
assert f"{Path.home()}/.android/debug.keystore" in calls[4]
|
||||
calls.clear()
|
||||
assert helper.build() == apk
|
||||
assert calls == [], "a cached build runs no tool"
|
||||
|
||||
def test_an_existing_keystore_is_reused(self, source: Path, sdk: Path) -> None:
|
||||
del source, sdk
|
||||
keystore = Path.home() / ".android" / "debug.keystore"
|
||||
keystore.parent.mkdir(parents=True)
|
||||
keystore.write_bytes(b"ks")
|
||||
calls: list[list[str]] = []
|
||||
with patch(f"{MOD}.subprocess.run", fake_tool(calls)):
|
||||
helper.build()
|
||||
assert "keytool" not in [Path(c[0]).name for c in calls]
|
||||
|
||||
def test_a_failing_tool_is_reported(self, source: Path, sdk: Path) -> None:
|
||||
del source, sdk
|
||||
failed = subprocess.CompletedProcess(["javac"], 1, "", "boom")
|
||||
with (
|
||||
patch(f"{MOD}.subprocess.run", return_value=failed),
|
||||
pytest.raises(UiAutomationError, match=r"javac\nboom"),
|
||||
):
|
||||
helper.build()
|
||||
|
||||
|
||||
class TestDevice:
|
||||
def test_skips_install_when_the_phone_has_this_build(self, source: Path) -> None:
|
||||
del source
|
||||
run = MagicMock(return_value=f"versionName={helper.source_hash()}\n")
|
||||
helper.ensure_installed(run)
|
||||
run.assert_called_once_with("shell", "dumpsys", "package", helper.PACKAGE)
|
||||
|
||||
def test_installs_when_missing_or_stale(self, source: Path, tmp_path: Path) -> None:
|
||||
del source
|
||||
run = MagicMock(return_value="versionName=old\n")
|
||||
with patch(f"{MOD}.build", return_value=tmp_path / "a.apk"):
|
||||
helper.ensure_installed(run)
|
||||
run.assert_called_with(
|
||||
"install", "-r", "-t", str(tmp_path / "a.apk"), timeout=120.0
|
||||
)
|
||||
|
||||
def test_parse_result(self) -> None:
|
||||
out = "INSTRUMENTATION_RESULT: xml=<hierarchy/>\nINSTRUMENTATION_CODE: -1\n"
|
||||
assert helper.parse_result(out) == "<hierarchy/>"
|
||||
with pytest.raises(UiAutomationError, match=r"failed: java\.lang\.X"):
|
||||
helper.parse_result("INSTRUMENTATION_RESULT: error=java.lang.X\n")
|
||||
with pytest.raises(UiAutomationError, match="no tree"):
|
||||
helper.parse_result("INSTRUMENTATION_STATUS: nothing\n")
|
||||
|
||||
def test_dump_display(self) -> None:
|
||||
run = MagicMock(return_value="INSTRUMENTATION_RESULT: xml=<h/>\n")
|
||||
with patch(f"{MOD}.ensure_installed") as ensure:
|
||||
assert helper.dump_display(run, 9) == "<h/>"
|
||||
ensure.assert_called_once_with(run)
|
||||
run.assert_called_once_with(
|
||||
"shell",
|
||||
"am",
|
||||
"instrument",
|
||||
"-w",
|
||||
"-e",
|
||||
"display",
|
||||
"9",
|
||||
"com.kuhy.a11ydump/.A11yDump",
|
||||
timeout=45.0,
|
||||
)
|
||||
@ -17,12 +17,15 @@ from python_pkg.android_ui.driver import (
|
||||
ElementNotFoundError,
|
||||
UiElement,
|
||||
)
|
||||
from python_pkg.phone_lease import PhoneBusyError
|
||||
from python_pkg.phone_lease import NoPhoneError, PhoneBusyError
|
||||
|
||||
SCREEN_SCOPE = "pkg:__screen__"
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
MOD = "python_pkg.android_ui.cli"
|
||||
pytestmark = pytest.mark.usefixtures("phone")
|
||||
|
||||
|
||||
def _element(label: str = "Connect Firebase") -> UiElement:
|
||||
@ -48,15 +51,22 @@ def ui() -> Iterator[MagicMock]:
|
||||
|
||||
|
||||
class TestLease:
|
||||
def test_every_command_refreshes_the_lease_first(self) -> None:
|
||||
with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire:
|
||||
assert cli.main(["-s", "SER1", "focus"]) == 0
|
||||
acquire.assert_called_once_with("SER1", "android_ui focus")
|
||||
def test_every_command_refreshes_the_screen_lease_first(self) -> None:
|
||||
with patch(f"{MOD}.AndroidUi") as factory, patch(f"{MOD}.acquire") as acquire:
|
||||
assert cli.main(["-s", "SER2", "focus"]) == 0
|
||||
acquire.assert_called_once_with("SER2", "android_ui focus", scope=SCREEN_SCOPE)
|
||||
factory.assert_called_once_with(serial="SER2", display=None)
|
||||
|
||||
def test_serial_less_calls_share_one_key(self) -> None:
|
||||
def test_serial_less_calls_lease_the_resolved_serial(self) -> None:
|
||||
# Leasing a "default" key let two sessions drive one phone blind.
|
||||
with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire:
|
||||
cli.main(["focus"])
|
||||
assert acquire.call_args.args[0] == "default"
|
||||
assert acquire.call_args.args[0] == "SER1"
|
||||
|
||||
def test_no_phone_exits_4(self, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
with patch(f"{MOD}.resolve_serial", side_effect=NoPhoneError("none")):
|
||||
assert cli.main(["focus"]) == 4
|
||||
assert "none" in capsys.readouterr().err
|
||||
|
||||
def test_a_foreign_lease_stops_the_command(
|
||||
self, capsys: pytest.CaptureFixture[str]
|
||||
|
||||
83
python_pkg/android_ui/tests/test_cli_display.py
Normal file
83
python_pkg/android_ui/tests/test_cli_display.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""Tests for ``android_ui --display``: which display a call may drive.
|
||||
|
||||
The real screen is shared, so it takes the screen lease; a session's own
|
||||
virtual display takes only its app's lease; anyone else's display is refused
|
||||
before the driver or a lease is touched.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.android_ui import cli
|
||||
from python_pkg.phone_guard.vd_state import SessionDisplay
|
||||
|
||||
MOD = "python_pkg.android_ui.cli"
|
||||
SCREEN_SCOPE = "pkg:__screen__"
|
||||
pytestmark = pytest.mark.usefixtures("phone")
|
||||
|
||||
|
||||
def _mine(display: int = 9, package: str = "dev.kuhy.todo") -> SessionDisplay:
|
||||
return SessionDisplay("claude_1", display, "sf", package)
|
||||
|
||||
|
||||
class TestDisplay:
|
||||
def test_self_drives_this_sessions_display_under_its_app_lease(self) -> None:
|
||||
with (
|
||||
patch(f"{MOD}.own_display", return_value=_mine()),
|
||||
patch(f"{MOD}.AndroidUi") as factory,
|
||||
patch(f"{MOD}.acquire") as acquire,
|
||||
):
|
||||
assert cli.main(["--display", "self", "focus"]) == 0
|
||||
factory.assert_called_once_with(serial="SER1", display=9)
|
||||
assert acquire.call_args.kwargs["scope"] == "pkg:dev.kuhy.todo"
|
||||
|
||||
def test_an_explicit_id_of_ones_own_display_works(self) -> None:
|
||||
with (
|
||||
patch(f"{MOD}.own_display", return_value=_mine()),
|
||||
patch(f"{MOD}.AndroidUi") as factory,
|
||||
patch(f"{MOD}.acquire"),
|
||||
):
|
||||
assert cli.main(["--display", "9", "focus"]) == 0
|
||||
assert factory.call_args.kwargs["display"] == 9
|
||||
|
||||
def test_display_0_is_the_real_screen(self) -> None:
|
||||
with patch(f"{MOD}.AndroidUi") as factory, patch(f"{MOD}.acquire") as acquire:
|
||||
assert cli.main(["--display", "0", "focus"]) == 0
|
||||
assert factory.call_args.kwargs["display"] is None
|
||||
assert acquire.call_args.kwargs["scope"] == SCREEN_SCOPE
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("requested", "mine", "other", "needle"),
|
||||
[
|
||||
("self", None, None, "phone_vd.sh start"),
|
||||
("abc", None, None, "display id or 'self'"),
|
||||
("7", None, None, "not this session's"),
|
||||
(
|
||||
"7",
|
||||
_mine(9),
|
||||
SessionDisplay("claude_2", 7, "sf", "x"),
|
||||
"owned by claude_2",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_refuses_what_is_not_this_sessions(
|
||||
self,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
requested: str,
|
||||
mine: SessionDisplay | None,
|
||||
other: SessionDisplay | None,
|
||||
needle: str,
|
||||
) -> None:
|
||||
with (
|
||||
patch(f"{MOD}.own_display", return_value=mine),
|
||||
patch(f"{MOD}.display_owner", return_value=other),
|
||||
patch(f"{MOD}.AndroidUi") as factory,
|
||||
patch(f"{MOD}.acquire") as acquire,
|
||||
):
|
||||
assert cli.main(["--display", requested, "focus"]) == 5
|
||||
factory.assert_not_called()
|
||||
acquire.assert_not_called()
|
||||
assert needle in capsys.readouterr().err
|
||||
@ -11,72 +11,18 @@ because each of them looked like success at the time:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import NamedTuple
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.android_ui import driver as drv
|
||||
from python_pkg.android_ui.tests.conftest import FakeDevice
|
||||
from python_pkg.android_ui.tests.conftest import node as _node
|
||||
from python_pkg.android_ui.tests.conftest import tree as _tree
|
||||
|
||||
MOD = "python_pkg.android_ui.driver"
|
||||
|
||||
|
||||
def _node(text: str = "", **attrs: str) -> str:
|
||||
"""Render one ``<node>`` element for a fake dump.
|
||||
|
||||
``attrs`` accepts ``desc``, ``hint``, ``res``, ``cls``, ``bounds``,
|
||||
``enabled`` and ``focused``; each falls back to a representative default.
|
||||
"""
|
||||
return (
|
||||
f'<node text="{text}" content-desc="{attrs.get("desc", "")}" '
|
||||
f'hint="{attrs.get("hint", "")}" '
|
||||
f'resource-id="{attrs.get("res", "")}" '
|
||||
f'class="{attrs.get("cls", "android.widget.TextView")}" '
|
||||
f'bounds="{attrs.get("bounds", "[0,0][100,50]")}" '
|
||||
f'enabled="{attrs.get("enabled", "true")}" '
|
||||
f'focused="{attrs.get("focused", "false")}"/>'
|
||||
)
|
||||
|
||||
|
||||
def _tree(*nodes: str) -> str:
|
||||
"""Wrap ``nodes`` in a hierarchy document."""
|
||||
return f"<?xml version='1.0'?><hierarchy rotation='0'>{''.join(nodes)}</hierarchy>"
|
||||
|
||||
|
||||
class FakeDevice:
|
||||
"""Scripted stand-in for adb, recording every command it is given."""
|
||||
|
||||
def __init__(self, trees: list[str] | None = None) -> None:
|
||||
self.trees = trees or [_tree(_node("Connect"))]
|
||||
self.calls: list[tuple[str, ...]] = []
|
||||
self.keyboard_shown = False
|
||||
self.pulled = 0
|
||||
|
||||
def run(self, *args: str, timeout: float = 30.0) -> str:
|
||||
"""Answer an adb invocation."""
|
||||
del timeout
|
||||
self.calls.append(args)
|
||||
if args[:2] == ("shell", "dumpsys") and args[2] == "input_method":
|
||||
return f"mInputShown={'true' if self.keyboard_shown else 'false'}"
|
||||
if args[:2] == ("shell", "dumpsys") and args[2] == "window":
|
||||
return "mCurrentFocus=Window{ab12 u0 com.example/com.example.Main}"
|
||||
if args[0] == "pull":
|
||||
index = min(self.pulled, len(self.trees) - 1)
|
||||
self.pulled += 1
|
||||
Path(args[2]).write_text(self.trees[index], encoding="utf-8")
|
||||
if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}:
|
||||
self.keyboard_shown = False
|
||||
return ""
|
||||
|
||||
def taps(self) -> list[tuple[int, int]]:
|
||||
"""Return every tap coordinate, in order."""
|
||||
return [
|
||||
(int(c[3]), int(c[4]))
|
||||
for c in self.calls
|
||||
if c[:3] == ("shell", "input", "tap")
|
||||
]
|
||||
|
||||
|
||||
class Harness(NamedTuple):
|
||||
"""A driver plus the fake device backing it.
|
||||
|
||||
@ -207,7 +153,9 @@ class TestTap:
|
||||
ui.device.trees = [_tree(_node("Go", bounds="[0,0][100,100]"))]
|
||||
ui.ui.tap("Go")
|
||||
keyevents = [
|
||||
c for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent")
|
||||
c
|
||||
for c in ui.device.calls
|
||||
if c[:5] == ("shell", "input", "-d", "0", "keyevent")
|
||||
]
|
||||
assert keyevents, "expected the keyboard to be dismissed first"
|
||||
|
||||
@ -236,3 +184,26 @@ def test_real_text_outranks_the_hint_once_typed() -> None:
|
||||
(element,) = drv._parse_tree(_tree(dump))
|
||||
|
||||
assert element.label == "17:30"
|
||||
|
||||
|
||||
class TestVirtualDisplay:
|
||||
def test_reads_the_tree_through_the_helper_and_taps_with_d(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
device = FakeDevice()
|
||||
instance = drv.AndroidUi(serial="S", display=9, settle_seconds=0.0)
|
||||
monkeypatch.setattr(instance, "_run", device.run)
|
||||
monkeypatch.setattr(f"{MOD}.time.sleep", lambda _s: None)
|
||||
xml = _tree(_node("Go", bounds="[0,0][100,100]"))
|
||||
calls: list[int] = []
|
||||
|
||||
def dump(run: object, display: int) -> str:
|
||||
del run
|
||||
calls.append(display)
|
||||
return xml
|
||||
|
||||
monkeypatch.setattr(f"{MOD}._a11y_helper.dump_display", dump)
|
||||
instance.tap("Go")
|
||||
assert calls
|
||||
assert set(calls) == {9}
|
||||
assert ("shell", "input", "-d", "9", "tap", "50", "50") in device.calls
|
||||
|
||||
@ -54,16 +54,19 @@ class FakeDevice:
|
||||
index = min(self.pulled, len(self.trees) - 1)
|
||||
self.pulled += 1
|
||||
Path(args[2]).write_text(self.trees[index], encoding="utf-8")
|
||||
if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}:
|
||||
if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in {
|
||||
"111",
|
||||
"4",
|
||||
}:
|
||||
self.keyboard_shown = False
|
||||
return ""
|
||||
|
||||
def taps(self) -> list[tuple[int, int]]:
|
||||
"""Return every tap coordinate, in order."""
|
||||
return [
|
||||
(int(c[3]), int(c[4]))
|
||||
(int(c[5]), int(c[6]))
|
||||
for c in self.calls
|
||||
if c[:3] == ("shell", "input", "tap")
|
||||
if c[:5] == ("shell", "input", "-d", "0", "tap")
|
||||
]
|
||||
|
||||
|
||||
|
||||
@ -54,16 +54,19 @@ class FakeDevice:
|
||||
index = min(self.pulled, len(self.trees) - 1)
|
||||
self.pulled += 1
|
||||
Path(args[2]).write_text(self.trees[index], encoding="utf-8")
|
||||
if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}:
|
||||
if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in {
|
||||
"111",
|
||||
"4",
|
||||
}:
|
||||
self.keyboard_shown = False
|
||||
return ""
|
||||
|
||||
def taps(self) -> list[tuple[int, int]]:
|
||||
"""Return every tap coordinate, in order."""
|
||||
return [
|
||||
(int(c[3]), int(c[4]))
|
||||
(int(c[5]), int(c[6]))
|
||||
for c in self.calls
|
||||
if c[:3] == ("shell", "input", "tap")
|
||||
if c[:5] == ("shell", "input", "-d", "0", "tap")
|
||||
]
|
||||
|
||||
|
||||
@ -98,7 +101,9 @@ class TestKeyboard:
|
||||
def test_dismiss_is_a_no_op_when_already_closed(self, ui: Harness) -> None:
|
||||
ui.ui.dismiss_keyboard()
|
||||
assert not [
|
||||
c for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent")
|
||||
c
|
||||
for c in ui.device.calls
|
||||
if c[:5] == ("shell", "input", "-d", "0", "keyevent")
|
||||
]
|
||||
|
||||
def test_closes_on_the_second_keyevent_when_escape_is_ignored(
|
||||
@ -110,7 +115,9 @@ class TestKeyboard:
|
||||
monkeypatch.setattr(ui.ui, "keyboard_is_up", lambda: next(states, False))
|
||||
ui.ui.dismiss_keyboard()
|
||||
sent = [
|
||||
c[3] for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent")
|
||||
c[5]
|
||||
for c in ui.device.calls
|
||||
if c[:5] == ("shell", "input", "-d", "0", "keyevent")
|
||||
]
|
||||
assert sent == ["111", "4"]
|
||||
|
||||
@ -152,7 +159,7 @@ class TestTyping:
|
||||
deletes = [
|
||||
c
|
||||
for c in ui.device.calls
|
||||
if c[:3] == ("shell", "input", "keyevent") and c[3] == "67"
|
||||
if c[:5] == ("shell", "input", "-d", "0", "keyevent") and c[5] == "67"
|
||||
]
|
||||
assert deletes, "expected the field to be cleared first"
|
||||
|
||||
|
||||
9
python_pkg/phone_guard/__init__.py
Normal file
9
python_pkg/phone_guard/__init__.py
Normal file
@ -0,0 +1,9 @@
|
||||
"""Keep several Claude sessions from interrupting each other on one phone.
|
||||
|
||||
Two parts: :mod:`vd_state` answers "which virtual display is whose" from the
|
||||
state ``phone_vd.sh`` keeps, and :mod:`hook` is the PreToolUse hook that stops
|
||||
a raw ``adb`` command from touching the real screen (or another session's app)
|
||||
without the matching :mod:`python_pkg.phone_lease` lease.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
10
python_pkg/phone_guard/__main__.py
Normal file
10
python_pkg/phone_guard/__main__.py
Normal file
@ -0,0 +1,10 @@
|
||||
"""``python3 -m python_pkg.phone_guard``: the PreToolUse hook entry point."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from python_pkg.phone_guard.hook import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
18
python_pkg/phone_guard/_need.py
Normal file
18
python_pkg/phone_guard/_need.py
Normal file
@ -0,0 +1,18 @@
|
||||
"""The answer :mod:`classify` gives for one adb call."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
SCREEN = "screen"
|
||||
WHOLE_PHONE = "whole-phone"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Need:
|
||||
"""What one adb call needs: a lease scope, a block, or nothing."""
|
||||
|
||||
scope: str | None = None # SCREEN, WHOLE_PHONE, or "pkg:<name>"
|
||||
block: str | None = None # why it may not run at all
|
||||
display: int | None = None # the virtual display it targets, if any
|
||||
serial: str | None = None # from ``-s``
|
||||
119
python_pkg/phone_guard/_remote.py
Normal file
119
python_pkg/phone_guard/_remote.py
Normal file
@ -0,0 +1,119 @@
|
||||
"""Per-tool rules for ``adb shell`` commands, for :mod:`classify`.
|
||||
|
||||
Split out of ``classify.py`` for the 250-line cap. Each handler takes the
|
||||
remote command's arguments (after the tool name) and returns the
|
||||
:class:`~python_pkg.phone_guard.classify.Need` without a serial; the caller
|
||||
stamps the serial on.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
_PM_PACKAGE = {
|
||||
"clear",
|
||||
"uninstall",
|
||||
"disable",
|
||||
"disable-user",
|
||||
"enable",
|
||||
"hide",
|
||||
"unhide",
|
||||
"suspend",
|
||||
"unsuspend",
|
||||
"grant",
|
||||
"revoke",
|
||||
}
|
||||
_IDLE_WRITE = {"force-idle", "step", "unforce", "enable", "disable", "force-inactive"}
|
||||
# `cmd connectivity set-package-networking-enabled <bool> <package>`
|
||||
_NETWORKING_PKG_INDEX = 3
|
||||
|
||||
|
||||
def _display_flag(args: list[str], flag: str) -> int | None:
|
||||
if flag in args:
|
||||
i = args.index(flag)
|
||||
if i + 1 < len(args) and args[i + 1].isdigit():
|
||||
return int(args[i + 1])
|
||||
return None
|
||||
|
||||
|
||||
# `pm` options that take a value, so the value is not mistaken for the package.
|
||||
_PM_VALUED = {"--user", "-u"}
|
||||
|
||||
|
||||
def _package_after(args: list[str], verb: str) -> str | None:
|
||||
rest = args[args.index(verb) + 1 :]
|
||||
i = 0
|
||||
while i < len(rest):
|
||||
if rest[i] in _PM_VALUED:
|
||||
i += 2
|
||||
elif rest[i].startswith("-"):
|
||||
i += 1
|
||||
else:
|
||||
return rest[i]
|
||||
return None
|
||||
|
||||
|
||||
def _am(rest: list[str]) -> Need:
|
||||
first = rest[0] if rest else ""
|
||||
if first in {"force-stop", "kill", "stop-app"} and len(rest) > 1:
|
||||
return Need(scope=f"pkg:{rest[-1]}")
|
||||
if first in {"start", "start-activity", "startservice", "start-foreground-service"}:
|
||||
display = _display_flag(rest, "--display")
|
||||
return Need(display=display) if display else Need(scope=SCREEN)
|
||||
return Need()
|
||||
|
||||
|
||||
def _pm(rest: list[str]) -> Need:
|
||||
if not rest or rest[0] not in _PM_PACKAGE:
|
||||
return Need()
|
||||
package = _package_after(rest, rest[0])
|
||||
return Need(scope=f"pkg:{package}" if package else SCREEN)
|
||||
|
||||
|
||||
def _cmd(rest: list[str]) -> Need:
|
||||
if rest[:2] != ["connectivity", "set-package-networking-enabled"]:
|
||||
is_set = (
|
||||
rest[:1] == ["connectivity"]
|
||||
and rest[1:2] != []
|
||||
and rest[1].startswith(("set-", "airplane-mode"))
|
||||
)
|
||||
return Need(scope=WHOLE_PHONE) if is_set else Need()
|
||||
package = rest[3] if len(rest) > _NETWORKING_PKG_INDEX else None
|
||||
return Need(scope=f"pkg:{package}") if package else Need(scope=WHOLE_PHONE)
|
||||
|
||||
|
||||
def _dumpsys(rest: list[str]) -> Need:
|
||||
writes = (rest[:1] == ["deviceidle"] and rest[1:2] and rest[1] in _IDLE_WRITE) or (
|
||||
rest[:1] == ["battery"] and rest[1:2] and rest[1] in {"set", "unplug", "reset"}
|
||||
)
|
||||
return Need(scope=WHOLE_PHONE) if writes else Need()
|
||||
|
||||
|
||||
def _settings(rest: list[str]) -> Need:
|
||||
return Need(scope=WHOLE_PHONE) if rest[:1] == ["put"] else Need()
|
||||
|
||||
|
||||
def _wm(rest: list[str]) -> Need:
|
||||
changes = rest[:1] in (["size"], ["density"]) and len(rest) > 1
|
||||
return Need(scope=SCREEN) if changes else Need()
|
||||
|
||||
|
||||
REMOTE: dict[str, Callable[[list[str]], Need]] = {
|
||||
"am": _am,
|
||||
"pm": _pm,
|
||||
"cmd": _cmd,
|
||||
"dumpsys": _dumpsys,
|
||||
"settings": _settings,
|
||||
"wm": _wm,
|
||||
"monkey": lambda _rest: Need(scope=SCREEN),
|
||||
"svc": lambda _rest: Need(scope=WHOLE_PHONE),
|
||||
"reboot": lambda _rest: Need(scope=WHOLE_PHONE),
|
||||
"setprop": lambda _rest: Need(scope=WHOLE_PHONE),
|
||||
"stop": lambda _rest: Need(scope=WHOLE_PHONE),
|
||||
"start": lambda _rest: Need(scope=WHOLE_PHONE),
|
||||
}
|
||||
165
python_pkg/phone_guard/classify.py
Normal file
165
python_pkg/phone_guard/classify.py
Normal file
@ -0,0 +1,165 @@
|
||||
"""Decide what a shell command's ``adb`` calls need before they may run.
|
||||
|
||||
Pure: no adb, no filesystem. The hook feeds it the Bash tool's command and
|
||||
applies the answer. Every ``adb`` invocation in the command gets one
|
||||
:class:`Need`:
|
||||
|
||||
* nothing -- reads (``logcat``, ``dumpsys``, ``getprop``, ``pull``,
|
||||
``screencap``), file transfer, and anything aimed at the caller's own
|
||||
virtual display;
|
||||
* a lease on the real screen -- input or launches on display 0;
|
||||
* a lease on one app -- ``force-stop``, ``install``, per-app network cuts;
|
||||
* the whole-phone lease -- reboot, Wi-Fi/airplane/Doze/battery tricks,
|
||||
``settings put``: everything that changes the phone for every session;
|
||||
* a block -- key/text input without ``-d`` (it goes to whichever display last
|
||||
took focus, i.e. possibly another session's app), or input aimed at another
|
||||
session's virtual display.
|
||||
|
||||
Unknown commands pass: this guards against interrupting other sessions, it is
|
||||
not a sandbox.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shlex
|
||||
|
||||
from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need
|
||||
from python_pkg.phone_guard._remote import REMOTE, _display_flag
|
||||
|
||||
_SEPARATORS = {";", "&&", "||", "|", "&", "(", ")", "\n"}
|
||||
_GLOBAL_WITH_ARG = {"-s", "-t", "-H", "-P", "-L"}
|
||||
_HEREDOC = re.compile(r"<<-?\s*(['\"]?)(\w+)\1[^\n]*\n.*?\n\s*\2\s*(?=\n|$)", re.DOTALL)
|
||||
|
||||
_WHOLE_PHONE_ADB = {
|
||||
"reboot",
|
||||
"root",
|
||||
"unroot",
|
||||
"remount",
|
||||
"disable-verity",
|
||||
"enable-verity",
|
||||
"tcpip",
|
||||
"usb",
|
||||
"kill-server",
|
||||
"sideload",
|
||||
}
|
||||
_READ_ADB = {
|
||||
"devices",
|
||||
"get-serialno",
|
||||
"get-state",
|
||||
"version",
|
||||
"help",
|
||||
"logcat",
|
||||
"pull",
|
||||
"push",
|
||||
"wait-for-device",
|
||||
"forward",
|
||||
"reverse",
|
||||
"bugreport",
|
||||
"start-server",
|
||||
"sync",
|
||||
}
|
||||
_KEYS = {"text", "keyevent", "keycombination"}
|
||||
_POINTER = {"tap", "swipe", "draganddrop", "press", "roll", "motionevent", "scroll"}
|
||||
|
||||
|
||||
def _tokens(command: str) -> list[str]:
|
||||
"""Shell tokens, heredoc bodies removed (a script being WRITTEN is not run)."""
|
||||
lexer = shlex.shlex(
|
||||
_HEREDOC.sub("\n", command), posix=True, punctuation_chars=";&|()"
|
||||
)
|
||||
lexer.whitespace = " \t\r"
|
||||
lexer.whitespace_split = True
|
||||
try:
|
||||
return list(lexer)
|
||||
except ValueError:
|
||||
return []
|
||||
|
||||
|
||||
def _split(tokens: list[str]) -> list[list[str]]:
|
||||
"""Split a token stream into simple commands."""
|
||||
groups: list[list[str]] = [[]]
|
||||
for word in tokens:
|
||||
if word in _SEPARATORS or set(word) <= set(";&|"):
|
||||
groups.append([])
|
||||
else:
|
||||
groups[-1].append(word)
|
||||
return [g for g in groups if g]
|
||||
|
||||
|
||||
def adb_calls(command: str) -> list[list[str]]:
|
||||
"""Every ``adb ...`` simple command, as its argument list (no ``adb``)."""
|
||||
calls = []
|
||||
for simple in _split(_tokens(command)):
|
||||
for i, word in enumerate(simple):
|
||||
if word == "adb" or word.endswith("/adb"):
|
||||
calls.append(simple[i + 1 :])
|
||||
break
|
||||
return calls
|
||||
|
||||
|
||||
def _input(args: list[str], serial: str | None) -> Need:
|
||||
display = _display_flag(args, "-d")
|
||||
verb = next((a for a in args if a in _KEYS | _POINTER), "")
|
||||
if display is not None and display != 0:
|
||||
return Need(display=display, serial=serial)
|
||||
if display is None and verb in _KEYS:
|
||||
return Need(
|
||||
block=f"`input {verb}` without -d goes to whichever display last took "
|
||||
"focus -- possibly another session's app. Add `-d 0` for the real "
|
||||
"screen or `-d <your display>` (phone_vd.sh id).",
|
||||
serial=serial,
|
||||
)
|
||||
return Need(scope=SCREEN, serial=serial)
|
||||
|
||||
|
||||
def _remote(args: list[str], serial: str | None) -> list[Need]:
|
||||
"""Needs of one ``adb shell`` command line."""
|
||||
found = []
|
||||
for simple in _split(_tokens(" ".join(args))):
|
||||
tool, rest = simple[0], simple[1:]
|
||||
found.append(_remote_one(tool, rest, serial))
|
||||
return found
|
||||
|
||||
|
||||
def _remote_one(tool: str, rest: list[str], serial: str | None) -> Need:
|
||||
"""The need of one remote command, stamped with the call's serial."""
|
||||
if tool == "input":
|
||||
return _input(rest, serial)
|
||||
need = REMOTE.get(tool, lambda _rest: Need())(rest)
|
||||
return Need(scope=need.scope, block=need.block, display=need.display, serial=serial)
|
||||
|
||||
|
||||
def needs(command: str, apk_package: dict[str, str] | None = None) -> list[Need]:
|
||||
"""Every adb call's :class:`Need` in ``command``.
|
||||
|
||||
``apk_package`` maps an APK path to its package, resolved by the caller
|
||||
(reading an APK is I/O); an unknown APK's install needs the screen.
|
||||
"""
|
||||
known = apk_package or {}
|
||||
result: list[Need] = []
|
||||
for call in adb_calls(command):
|
||||
serial = None
|
||||
i = 0
|
||||
while i < len(call) and call[i].startswith("-"):
|
||||
if call[i] in _GLOBAL_WITH_ARG and i + 1 < len(call):
|
||||
serial = call[i + 1] if call[i] == "-s" else serial
|
||||
i += 2
|
||||
else:
|
||||
i += 1
|
||||
sub, args = (call[i], call[i + 1 :]) if i < len(call) else ("", [])
|
||||
if sub in {"shell", "exec-out"}:
|
||||
result.extend(_remote(args, serial))
|
||||
elif sub in _WHOLE_PHONE_ADB:
|
||||
result.append(Need(scope=WHOLE_PHONE, serial=serial))
|
||||
elif sub in {"install", "install-multiple", "uninstall"}:
|
||||
target = next((a for a in reversed(args) if not a.startswith("-")), "")
|
||||
package = known.get(target) or (target if sub == "uninstall" else None)
|
||||
result.append(
|
||||
Need(scope=f"pkg:{package}" if package else SCREEN, serial=serial)
|
||||
)
|
||||
elif sub in _READ_ADB or not sub:
|
||||
result.append(Need(serial=serial))
|
||||
else:
|
||||
result.append(Need(serial=serial))
|
||||
return result
|
||||
138
python_pkg/phone_guard/hook.py
Normal file
138
python_pkg/phone_guard/hook.py
Normal file
@ -0,0 +1,138 @@
|
||||
"""PreToolUse hook: lease -- or refuse -- raw ``adb`` before the Bash tool runs it.
|
||||
|
||||
``android_ui``, ``phone_deploy.sh`` and ``phone_vd.sh`` take their own leases,
|
||||
but a bare ``adb shell input tap`` in a Bash call took none, which is how one
|
||||
session's tap landed in another session's app (2026-09-25). Claude Code pipes
|
||||
the tool call here as JSON; exit 2 blocks it and shows stderr to the model.
|
||||
|
||||
Leases are taken as this session (``phone_lease.owner_id`` finds the
|
||||
``claude`` ancestor), so a later ``android_ui`` call from the same session
|
||||
simply refreshes them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from python_pkg.android_ui._a11y_helper import build_tool
|
||||
from python_pkg.android_ui._elements import UiAutomationError
|
||||
from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE
|
||||
from python_pkg.phone_guard.classify import adb_calls, needs
|
||||
from python_pkg.phone_guard.vd_state import display_owner, own_display
|
||||
from python_pkg.phone_lease import (
|
||||
NoPhoneError,
|
||||
PhoneBusyError,
|
||||
Terms,
|
||||
acquire,
|
||||
owner_id,
|
||||
resolve_serial,
|
||||
scope_of,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterable
|
||||
from typing import TextIO
|
||||
|
||||
from python_pkg.phone_guard._need import Need
|
||||
|
||||
BLOCK = 2
|
||||
# Under Claude Code's 60 s hook timeout, with room to report.
|
||||
_WAIT_SECONDS = 40.0
|
||||
|
||||
|
||||
def _apk_packages(command: str) -> dict[str, str]:
|
||||
"""Map each APK an ``adb install`` names to its package (via aapt2)."""
|
||||
found: dict[str, str] = {}
|
||||
try:
|
||||
aapt2 = str(build_tool("aapt2"))
|
||||
except UiAutomationError:
|
||||
return found # no SDK: an unknown APK's install takes the screen lease
|
||||
for call in adb_calls(command):
|
||||
for word in call:
|
||||
if not word.endswith(".apk"):
|
||||
continue
|
||||
try:
|
||||
done = subprocess.run(
|
||||
[aapt2, "dump", "packagename", word],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
except OSError, subprocess.TimeoutExpired:
|
||||
continue
|
||||
if done.returncode == 0 and done.stdout.strip():
|
||||
found[word] = done.stdout.strip()
|
||||
return found
|
||||
|
||||
|
||||
def _scope(need: Need, serial: str, owner: str) -> str:
|
||||
"""The lease scope a need maps to; raises PermissionError for a foreign display."""
|
||||
if need.display is not None:
|
||||
mine = own_display(serial, owner)
|
||||
if mine is not None and mine.display == need.display:
|
||||
return scope_of(mine.package)
|
||||
other = display_owner(serial, need.display)
|
||||
whose = f"session {other.owner_dir}'s" if other else "not your"
|
||||
msg = (
|
||||
f"display {need.display} is {whose} virtual display; drive only "
|
||||
"your own (`phone_vd.sh id`)"
|
||||
)
|
||||
raise PermissionError(msg)
|
||||
if need.scope == SCREEN:
|
||||
return scope_of()
|
||||
if need.scope == WHOLE_PHONE:
|
||||
return scope_of(whole_phone=True)
|
||||
return str(need.scope)
|
||||
|
||||
|
||||
def decide(command: str, note: str = "") -> str | None:
|
||||
"""Take every lease ``command`` needs; return why it must not run, if so."""
|
||||
found: Iterable[Need] = needs(command, _apk_packages(command))
|
||||
owner = owner_id()
|
||||
for need in found:
|
||||
if need.block:
|
||||
return need.block
|
||||
if need.scope is None and need.display is None:
|
||||
continue
|
||||
try:
|
||||
serial = resolve_serial(need.serial)
|
||||
except NoPhoneError:
|
||||
continue # no phone or several without -s: adb itself will refuse
|
||||
try:
|
||||
scope = _scope(need, serial, owner)
|
||||
acquire(
|
||||
serial,
|
||||
note or f"hook: {command[:60]}",
|
||||
scope=scope,
|
||||
terms=Terms(wait=_WAIT_SECONDS),
|
||||
owner=owner,
|
||||
)
|
||||
except PermissionError as exc:
|
||||
return str(exc)
|
||||
except PhoneBusyError as exc:
|
||||
return (
|
||||
f"{exc} -- wait for it, use your own virtual display, or ask the user"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def main(stdin: TextIO | None = None) -> int:
|
||||
"""Hook entry point: 0 lets the Bash call run, 2 blocks it."""
|
||||
try:
|
||||
payload = json.load(stdin or sys.stdin)
|
||||
except json.JSONDecodeError:
|
||||
return 0
|
||||
if payload.get("tool_name") != "Bash":
|
||||
return 0
|
||||
command = str((payload.get("tool_input") or {}).get("command") or "")
|
||||
if "adb" not in command:
|
||||
return 0
|
||||
reason = decide(command)
|
||||
if reason is None:
|
||||
return 0
|
||||
sys.stderr.write(f"phone_guard: {reason}\n")
|
||||
return BLOCK
|
||||
1
python_pkg/phone_guard/tests/__init__.py
Normal file
1
python_pkg/phone_guard/tests/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for :mod:`python_pkg.phone_guard`."""
|
||||
33
python_pkg/phone_guard/tests/conftest.py
Normal file
33
python_pkg/phone_guard/tests/conftest.py
Normal file
@ -0,0 +1,33 @@
|
||||
"""Keep every phone_guard test off the live lease and virtual-display state."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.phone_lease import lease as lease_mod
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolated_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
||||
"""Redirect ~/.cache/phone-lease and /tmp/phone_vd into tmp_path."""
|
||||
monkeypatch.setenv("PHONE_VD_STATE_DIR", str(tmp_path / "phone_vd"))
|
||||
with patch.object(lease_mod, "LEASE_DIR", tmp_path / "leases"):
|
||||
yield tmp_path
|
||||
|
||||
|
||||
def make_display(
|
||||
root: Path, serial: str, owner_dir: str, display: str, package: str
|
||||
) -> None:
|
||||
"""Write the state phone_vd.sh leaves for one running display."""
|
||||
state = root / "phone_vd" / serial / owner_dir
|
||||
state.mkdir(parents=True)
|
||||
(state / "display").write_text(display)
|
||||
(state / "sf").write_text("123")
|
||||
(state / "package").write_text(package)
|
||||
121
python_pkg/phone_guard/tests/test_classify.py
Normal file
121
python_pkg/phone_guard/tests/test_classify.py
Normal file
@ -0,0 +1,121 @@
|
||||
"""What each adb call needs, from the Bash command that runs it."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need
|
||||
from python_pkg.phone_guard.classify import adb_calls, needs
|
||||
|
||||
|
||||
def only(command: str) -> Need:
|
||||
found = needs(command)
|
||||
assert len(found) == 1, found
|
||||
return found[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("command", "scope"),
|
||||
[
|
||||
# reads and file transfer never need a lease
|
||||
("adb devices", None),
|
||||
("adb logcat -d", None),
|
||||
("adb pull /sdcard/a.png /tmp", None),
|
||||
("adb push a.jpg /sdcard/", None),
|
||||
("adb shell getprop ro.x", None),
|
||||
("adb exec-out screencap -p", None),
|
||||
("adb shell dumpsys window", None),
|
||||
("adb shell settings get global x", None),
|
||||
("adb shell cmd connectivity get-chain3-enabled", None),
|
||||
("adb shell cmd package list", None),
|
||||
("adb shell pm list packages", None),
|
||||
("adb shell am broadcast -a X", None),
|
||||
("adb shell am instrument -w a/.B", None),
|
||||
("adb shell wm size", None),
|
||||
("adb shell uiautomator dump", None),
|
||||
("adb", None),
|
||||
("adb frobnicate", None),
|
||||
# the real screen
|
||||
("adb shell input tap 1 2", SCREEN),
|
||||
("adb shell input -d 0 text hi", SCREEN),
|
||||
("adb shell input swipe 1 2 3 4", SCREEN),
|
||||
("adb shell am start -n a/.B", SCREEN),
|
||||
("adb shell monkey -p a 1", SCREEN),
|
||||
("adb shell wm size 1080x2400", SCREEN),
|
||||
("adb shell pm clear --user 0", SCREEN),
|
||||
("adb install -r unknown.apk", SCREEN),
|
||||
# one app
|
||||
("adb shell am force-stop dev.kuhy.todo", "pkg:dev.kuhy.todo"),
|
||||
("adb shell pm grant a.b android.permission.X", "pkg:a.b"),
|
||||
("adb shell pm clear --user 0 -f a.b", "pkg:a.b"),
|
||||
(
|
||||
"adb shell cmd connectivity set-package-networking-enabled false a.b",
|
||||
"pkg:a.b",
|
||||
),
|
||||
("adb uninstall a.b", "pkg:a.b"),
|
||||
# the whole phone
|
||||
("adb reboot", WHOLE_PHONE),
|
||||
("adb kill-server", WHOLE_PHONE),
|
||||
("adb shell svc wifi disable", WHOLE_PHONE),
|
||||
("adb shell settings put global airplane_mode_on 1", WHOLE_PHONE),
|
||||
("adb shell cmd connectivity airplane-mode enable", WHOLE_PHONE),
|
||||
("adb shell cmd connectivity set-chain3-enabled true", WHOLE_PHONE),
|
||||
(
|
||||
"adb shell cmd connectivity set-package-networking-enabled false",
|
||||
WHOLE_PHONE,
|
||||
),
|
||||
("adb shell dumpsys deviceidle force-idle", WHOLE_PHONE),
|
||||
("adb shell dumpsys battery unplug", WHOLE_PHONE),
|
||||
("adb shell reboot", WHOLE_PHONE),
|
||||
("adb shell setprop a b", WHOLE_PHONE),
|
||||
],
|
||||
)
|
||||
def test_scopes(command: str, scope: str | None) -> None:
|
||||
need = only(command)
|
||||
assert need.scope == scope
|
||||
assert need.block is None
|
||||
|
||||
|
||||
def test_text_without_display_is_blocked() -> None:
|
||||
for verb in ("text hi", "keyevent 4", "keycombination 113 29"):
|
||||
assert "without -d" in (only(f"adb shell input {verb}").block or "")
|
||||
|
||||
|
||||
def test_a_virtual_display_is_reported_not_leased() -> None:
|
||||
need = only("adb shell input -d 9 text hi")
|
||||
assert (need.display, need.scope) == (9, None)
|
||||
assert only("adb shell am start --display 9 -n a/.B").display == 9
|
||||
|
||||
|
||||
def test_a_display_flag_without_a_number_is_not_a_display() -> None:
|
||||
assert only("adb shell am start --display main -n a/.B").display is None
|
||||
|
||||
|
||||
def test_serial_and_global_options() -> None:
|
||||
need = only("adb -s R58M1 -t 3 shell am force-stop a.b")
|
||||
assert (need.serial, need.scope) == ("R58M1", "pkg:a.b")
|
||||
assert only("adb -e -H host shell input tap 1 1").scope == SCREEN
|
||||
|
||||
|
||||
def test_installs_use_the_resolved_package() -> None:
|
||||
found = needs("adb install -r -t app.apk", {"app.apk": "a.b"})
|
||||
assert found[0].scope == "pkg:a.b"
|
||||
|
||||
|
||||
def test_every_call_in_a_compound_command() -> None:
|
||||
found = needs(
|
||||
"adb logcat -d | grep x && "
|
||||
"/opt/sdk/adb -s S shell 'svc data disable; input tap 1 1'"
|
||||
)
|
||||
assert [n.scope for n in found] == [None, WHOLE_PHONE, SCREEN]
|
||||
|
||||
|
||||
def test_text_that_only_mentions_adb_is_ignored() -> None:
|
||||
assert needs('echo "adb shell input tap 1 1"') == []
|
||||
assert needs("cat <<'EOF' > s.sh\nadb shell input tap 1 1\nEOF\nls") == []
|
||||
assert needs("grep adbd log") == []
|
||||
|
||||
|
||||
def test_unparseable_quoting_is_left_alone() -> None:
|
||||
assert needs("adb shell 'input tap") == []
|
||||
assert adb_calls("") == []
|
||||
128
python_pkg/phone_guard/tests/test_hook.py
Normal file
128
python_pkg/phone_guard/tests/test_hook.py
Normal file
@ -0,0 +1,128 @@
|
||||
"""The PreToolUse hook: which Bash calls run, which lease first, which stop."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.android_ui._elements import UiAutomationError
|
||||
from python_pkg.phone_guard import hook
|
||||
from python_pkg.phone_guard.tests.conftest import make_display
|
||||
from python_pkg.phone_lease import NoPhoneError, PhoneBusyError
|
||||
|
||||
MOD = "python_pkg.phone_guard.hook"
|
||||
|
||||
|
||||
def run(command: str, tool: str = "Bash") -> int:
|
||||
return hook.main(
|
||||
io.StringIO(json.dumps({"tool_name": tool, "tool_input": {"command": command}}))
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def acquire() -> MagicMock:
|
||||
with (
|
||||
patch(f"{MOD}.acquire") as mock,
|
||||
patch(f"{MOD}.owner_id", return_value="claude:1"),
|
||||
patch(f"{MOD}.resolve_serial", side_effect=lambda s=None: s or "S"),
|
||||
patch(f"{MOD}.build_tool", side_effect=UiAutomationError("no sdk")),
|
||||
):
|
||||
yield mock
|
||||
|
||||
|
||||
class TestMain:
|
||||
def test_ignores_what_it_cannot_or_need_not_judge(self, acquire: MagicMock) -> None:
|
||||
assert hook.main(io.StringIO("not json")) == 0
|
||||
assert run("adb shell input tap 1 1", tool="Read") == 0
|
||||
assert run("ls -la") == 0
|
||||
assert hook.main(io.StringIO(json.dumps({"tool_name": "Bash"}))) == 0
|
||||
acquire.assert_not_called()
|
||||
|
||||
def test_reads_pass_without_a_lease(self, acquire: MagicMock) -> None:
|
||||
assert run("adb logcat -d") == 0
|
||||
acquire.assert_not_called()
|
||||
|
||||
def test_a_block_exits_2_with_the_reason(
|
||||
self, acquire: MagicMock, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
assert run("adb shell input text hi") == hook.BLOCK
|
||||
assert "without -d" in capsys.readouterr().err
|
||||
acquire.assert_not_called()
|
||||
|
||||
|
||||
class TestLeases:
|
||||
def test_the_real_screen(self, acquire: MagicMock) -> None:
|
||||
assert run("adb shell input tap 1 1") == 0
|
||||
assert acquire.call_args.kwargs["scope"] == "pkg:__screen__"
|
||||
assert acquire.call_args.kwargs["owner"] == "claude:1"
|
||||
assert acquire.call_args.args[0] == "S"
|
||||
|
||||
def test_the_whole_phone_and_an_app(self, acquire: MagicMock) -> None:
|
||||
assert run("adb -s X reboot && adb shell am force-stop a.b") == 0
|
||||
scopes = [c.kwargs["scope"] for c in acquire.call_args_list]
|
||||
assert scopes == ["whole-phone", "pkg:a.b"]
|
||||
assert acquire.call_args_list[0].args[0] == "X"
|
||||
|
||||
def test_own_display_takes_its_app_lease(
|
||||
self, acquire: MagicMock, isolated_state: Path
|
||||
) -> None:
|
||||
make_display(isolated_state, "S", "claude_1", "9", "dev.kuhy.todo")
|
||||
assert run("adb shell input -d 9 tap 5 5") == 0
|
||||
assert acquire.call_args.kwargs["scope"] == "pkg:dev.kuhy.todo"
|
||||
|
||||
def test_another_sessions_display_is_refused(
|
||||
self,
|
||||
acquire: MagicMock,
|
||||
isolated_state: Path,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
make_display(isolated_state, "S", "claude_2", "9", "x.y")
|
||||
assert run("adb shell input -d 9 tap 5 5") == hook.BLOCK
|
||||
assert "session claude_2's" in capsys.readouterr().err
|
||||
assert run("adb shell input -d 4 tap 5 5") == hook.BLOCK
|
||||
assert "not your" in capsys.readouterr().err
|
||||
acquire.assert_not_called()
|
||||
|
||||
def test_busy_phone_blocks_naming_the_holder(
|
||||
self, acquire: MagicMock, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
acquire.side_effect = PhoneBusyError("held by claude:2")
|
||||
assert run("adb shell input tap 1 1") == hook.BLOCK
|
||||
err = capsys.readouterr().err
|
||||
assert "held by claude:2" in err
|
||||
assert "own virtual display" in err
|
||||
|
||||
def test_no_resolvable_phone_is_left_to_adb(self, acquire: MagicMock) -> None:
|
||||
with patch(f"{MOD}.resolve_serial", side_effect=NoPhoneError("several")):
|
||||
assert run("adb shell input tap 1 1") == 0
|
||||
acquire.assert_not_called()
|
||||
|
||||
|
||||
class TestApkPackages:
|
||||
def test_reads_the_package_with_aapt2(self, tmp_path: Path) -> None:
|
||||
done = subprocess.CompletedProcess([], 0, "a.b\n", "")
|
||||
with (
|
||||
patch(f"{MOD}.build_tool", return_value=tmp_path / "aapt2"),
|
||||
patch(f"{MOD}.subprocess.run", return_value=done) as runner,
|
||||
):
|
||||
assert hook._apk_packages("adb install -r app.apk other.txt") == {
|
||||
"app.apk": "a.b"
|
||||
}
|
||||
assert runner.call_args.args[0][0] == str(tmp_path / "aapt2")
|
||||
|
||||
def test_failures_leave_the_apk_unknown(self, tmp_path: Path) -> None:
|
||||
with patch(f"{MOD}.build_tool", return_value=tmp_path / "aapt2"):
|
||||
with patch(f"{MOD}.subprocess.run", side_effect=OSError("gone")):
|
||||
assert hook._apk_packages("adb install a.apk") == {}
|
||||
failed = subprocess.CompletedProcess([], 1, "", "bad apk")
|
||||
with patch(f"{MOD}.subprocess.run", return_value=failed):
|
||||
assert hook._apk_packages("adb install a.apk") == {}
|
||||
|
||||
def test_no_sdk_means_no_lookup(self) -> None:
|
||||
with patch(f"{MOD}.build_tool", side_effect=UiAutomationError("no sdk")):
|
||||
assert hook._apk_packages("adb install a.apk") == {}
|
||||
17
python_pkg/phone_guard/tests/test_main_entry.py
Normal file
17
python_pkg/phone_guard/tests/test_main_entry.py
Normal file
@ -0,0 +1,17 @@
|
||||
"""``python -m python_pkg.phone_guard`` runs the hook."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import runpy
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_module_entry_exits_with_the_hook_code() -> None:
|
||||
with (
|
||||
patch("python_pkg.phone_guard.hook.main", return_value=2),
|
||||
pytest.raises(SystemExit) as done,
|
||||
):
|
||||
runpy.run_module("python_pkg.phone_guard", run_name="__main__")
|
||||
assert done.value.code == 2
|
||||
45
python_pkg/phone_guard/tests/test_vd_state.py
Normal file
45
python_pkg/phone_guard/tests/test_vd_state.py
Normal file
@ -0,0 +1,45 @@
|
||||
"""Reading phone_vd.sh's per-session state."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from python_pkg.phone_guard import vd_state
|
||||
from python_pkg.phone_guard.tests.conftest import make_display
|
||||
|
||||
|
||||
def test_default_root_is_under_the_temp_dir(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("PHONE_VD_STATE_DIR")
|
||||
assert vd_state.state_root().name == "phone_vd"
|
||||
|
||||
|
||||
def test_owner_dir_name_matches_the_script() -> None:
|
||||
assert vd_state.owner_dir_name("claude:123") == "claude_123"
|
||||
|
||||
|
||||
def test_lists_and_finds_displays(isolated_state: Path) -> None:
|
||||
make_display(isolated_state, "S", "claude_1", "9", "a.b")
|
||||
make_display(isolated_state, "S", "claude_2", "10", "c.d")
|
||||
make_display(isolated_state, "S", "claude_3", "gone", "e.f")
|
||||
assert [d.display for d in vd_state.displays("S")] == [9, 10]
|
||||
mine = vd_state.own_display("S", "claude:1")
|
||||
assert mine is not None
|
||||
assert (mine.display, mine.package, mine.sf) == (9, "a.b", "123")
|
||||
assert vd_state.own_display("S", "claude:9") is None
|
||||
other = vd_state.display_owner("S", 10)
|
||||
assert other is not None
|
||||
assert other.owner_dir == "claude_2"
|
||||
assert vd_state.display_owner("S", 11) is None
|
||||
|
||||
|
||||
def test_no_state_means_no_displays() -> None:
|
||||
assert vd_state.displays("NOPE") == []
|
||||
|
||||
|
||||
def test_unreadable_files_read_as_empty(isolated_state: Path) -> None:
|
||||
make_display(isolated_state, "S", "claude_1", "9", "a.b")
|
||||
(isolated_state / "phone_vd" / "S" / "claude_1" / "package").unlink()
|
||||
(isolated_state / "phone_vd" / "S" / "claude_1" / "package").mkdir()
|
||||
assert vd_state.displays("S")[0].package == ""
|
||||
83
python_pkg/phone_guard/vd_state.py
Normal file
83
python_pkg/phone_guard/vd_state.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""Which session owns which virtual display, read from ``phone_vd.sh`` state.
|
||||
|
||||
``phone_vd.sh`` keeps one directory per session under
|
||||
``$PHONE_VD_STATE_DIR/<serial>/<owner>/`` holding ``display`` (the logical id
|
||||
``input -d`` takes), ``sf`` (the SurfaceFlinger id ``screencap -d`` takes) and
|
||||
``package``. The owner directory name is the phone_lease owner id with
|
||||
anything outside ``[A-Za-z0-9_.-]`` replaced by ``_``, exactly as the script
|
||||
writes it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import tempfile
|
||||
|
||||
_UNSAFE = re.compile(r"[^A-Za-z0-9_.-]")
|
||||
|
||||
|
||||
def state_root() -> Path:
|
||||
"""The directory ``phone_vd.sh`` keeps its per-phone state in.
|
||||
|
||||
``$PHONE_VD_STATE_DIR``, else ``phone_vd`` under the temp dir -- the same
|
||||
``${TMPDIR:-/tmp}`` rule the script applies, so both sides agree.
|
||||
"""
|
||||
default = Path(tempfile.gettempdir()) / "phone_vd"
|
||||
return Path(os.environ.get("PHONE_VD_STATE_DIR", str(default)))
|
||||
|
||||
|
||||
def owner_dir_name(owner: str) -> str:
|
||||
"""``owner`` as ``phone_vd.sh`` names its state directory."""
|
||||
return _UNSAFE.sub("_", owner)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SessionDisplay:
|
||||
"""One running virtual display and whose it is."""
|
||||
|
||||
owner_dir: str
|
||||
display: int
|
||||
sf: str
|
||||
package: str
|
||||
|
||||
|
||||
def _read(path: Path) -> str:
|
||||
try:
|
||||
return path.read_text(encoding="utf-8").strip()
|
||||
except OSError:
|
||||
return ""
|
||||
|
||||
|
||||
def displays(serial: str) -> list[SessionDisplay]:
|
||||
"""Every session's virtual display on ``serial``."""
|
||||
root = state_root() / serial
|
||||
found: list[SessionDisplay] = []
|
||||
if not root.is_dir():
|
||||
return found
|
||||
for entry in sorted(root.iterdir()):
|
||||
display = _read(entry / "display")
|
||||
if not display.isdigit():
|
||||
continue
|
||||
found.append(
|
||||
SessionDisplay(
|
||||
owner_dir=entry.name,
|
||||
display=int(display),
|
||||
sf=_read(entry / "sf"),
|
||||
package=_read(entry / "package"),
|
||||
)
|
||||
)
|
||||
return found
|
||||
|
||||
|
||||
def own_display(serial: str, owner: str) -> SessionDisplay | None:
|
||||
"""``owner``'s virtual display on ``serial``, if it has one running."""
|
||||
mine = owner_dir_name(owner)
|
||||
return next((d for d in displays(serial) if d.owner_dir == mine), None)
|
||||
|
||||
|
||||
def display_owner(serial: str, display: int) -> SessionDisplay | None:
|
||||
"""The session whose virtual display has logical id ``display``."""
|
||||
return next((d for d in displays(serial) if d.display == display), None)
|
||||
Loading…
Reference in New Issue
Block a user