#!/usr/bin/env python3 """Deterministic Android debug scenario runner for Legado.""" from __future__ import annotations import argparse import base64 import datetime as dt import hashlib import json from pathlib import Path import re import shutil import subprocess import sys import time from typing import Any, Callable, Sequence import xml.etree.ElementTree as ET import zipfile EXIT_ASSERTION = 1 EXIT_CRASH = 2 EXIT_BUILD = 3 EXIT_ENVIRONMENT = 4 EXIT_SCENARIO = 5 PACKAGE = "io.legato.kazusa.debug" DEBUG_ACTIVITY = "io.legado.app.debug.DebugScenarioActivity" MAIN_ACTIVITY = "io.legado.app.ui.main.MainActivity" SOURCE_EDIT_ACTIVITY = "io.legado.app.ui.book.source.edit.BookSourceEditActivity" SOURCE_MANAGE_ACTIVITY = "io.legado.app.ui.book.source.manage.BookSourceActivity" THEME_MANIFEST_PATH = "manifest.json" THEME_ASSET_REMOTE = f"/sdcard/Android/data/{PACKAGE}/files/debug-theme-assets/background.png" DEBUG_STATE_URI = f"content://{PACKAGE}.debug-state" SCRIPT_DIR = Path(__file__).resolve().parent PROJECT_ROOT = SCRIPT_DIR.parent.parent THEME_ASSET_FIXTURE = PROJECT_ROOT / "app" / "src" / "main" / "assets" / "web" / "uploadBook" / "img" / "close.png" SCENARIO_ROOT = SCRIPT_DIR / "scenarios" RUN_ROOT = PROJECT_ROOT / ".debug-runs" FIXTURE_ROOT = PROJECT_ROOT / "app" / "src" / "debug" / "assets" / "debug-fixtures" # Capability registries. New bug scenarios extend the tool by registering an # entry / action / assertion below, never by editing execute() or # validate_scenario(). See the classes near the end of this module. ENTRIES: dict[str, "Entry"] = {} ACTIONS: dict[str, "Action"] = {} ASSERTIONS: dict[str, Callable[["Context"], bool]] = {} class ScenarioError(ValueError): pass class CommandError(RuntimeError): def __init__(self, command: Sequence[str], result: subprocess.CompletedProcess[str]): self.command = list(command) self.result = result super().__init__(f"command failed ({result.returncode}): {' '.join(command)}") def load_scenario(name: str) -> dict[str, Any]: if not re.fullmatch(r"[a-z0-9][a-z0-9/_-]*", name): raise ScenarioError(f"invalid scenario name: {name}") path = (SCENARIO_ROOT / f"{name}.json").resolve() if SCENARIO_ROOT.resolve() not in path.parents or not path.is_file(): raise ScenarioError(f"scenario not found: {name}") try: scenario = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ScenarioError(f"cannot read scenario {name}: {error}") from error validate_scenario(scenario, name) return scenario def validate_scenario(scenario: Any, requested_name: str | None = None) -> None: if not isinstance(scenario, dict): raise ScenarioError("scenario root must be an object") if scenario.get("schemaVersion") != 1: raise ScenarioError("schemaVersion must be 1") name = scenario.get("name") if not isinstance(name, str) or not name: raise ScenarioError("name must be a non-empty string") if requested_name is not None and name != requested_name: raise ScenarioError(f"scenario name mismatch: expected {requested_name}, got {name}") fixture = scenario.get("fixture") if not isinstance(fixture, str) or not re.fullmatch(r"[a-z0-9][a-z0-9-]*", fixture): raise ScenarioError("fixture must be a lowercase ID") marker = scenario.get("readyMarker") if not isinstance(marker, str) or not marker.strip(): raise ScenarioError("readyMarker must be a non-empty string") timeout = scenario.get("readyTimeoutSeconds", 30) if not isinstance(timeout, int) or timeout < 1 or timeout > 300: raise ScenarioError("readyTimeoutSeconds must be between 1 and 300") change_timeout = scenario.get("contentChangeTimeoutSeconds", 5) if not isinstance(change_timeout, int) or change_timeout < 1 or change_timeout > 30: raise ScenarioError("contentChangeTimeoutSeconds must be between 1 and 30") record = scenario.get("record", False) if not isinstance(record, bool): raise ScenarioError("record must be a boolean") validate_preferences(scenario.get("preferences")) entry = scenario.get("entry") if not isinstance(entry, dict) or entry.get("type") not in ENTRIES: raise ScenarioError(f"entry.type must be one of: {', '.join(sorted(ENTRIES))}") entry_handler = ENTRIES[entry["type"]] theme_asset = scenario.get("themeAsset") if theme_asset is not None: if entry["type"] != "theme_config": raise ScenarioError("themeAsset is only supported by theme_config scenarios") if theme_asset not in {"valid-file", "missing-file", "missing-content-uri"}: raise ScenarioError("themeAsset must be valid-file, missing-file, or missing-content-uri") expect_export_failure = scenario.get("expectThemeExportFailure", False) if not isinstance(expect_export_failure, bool): raise ScenarioError("expectThemeExportFailure must be a boolean") actions = scenario.get("actions") if not isinstance(actions, list) or not actions: raise ScenarioError("actions must be a non-empty array") for index, action in enumerate(actions): if not isinstance(action, dict): raise ScenarioError(f"actions[{index}] must be an object") action_type = action.get("type") if action_type not in entry_handler.allowed_actions: raise ScenarioError(f"actions[{index}] has an unsupported type") ACTIONS[action_type].validate(action, index) assertions = scenario.get("assertions") if not isinstance(assertions, list) or not assertions: raise ScenarioError("assertions must be a non-empty array") unknown = set(assertions) - set(ASSERTIONS) if unknown: raise ScenarioError(f"unsupported assertions: {', '.join(sorted(unknown))}") def validate_preferences(preferences: Any) -> None: # Scenario-declared app settings applied before launch (theme engine, # reader toggles, ...). Restricted to String/Boolean so the debug Activity # can apply them unambiguously via AppConfigStore. if preferences is None: return if not isinstance(preferences, dict) or not preferences: raise ScenarioError("preferences must be a non-empty object when present") for key, value in preferences.items(): if not re.fullmatch(r"[a-zA-Z][a-zA-Z0-9]*", key): raise ScenarioError(f"invalid preference key: {key}") if not isinstance(value, (str, bool)): raise ScenarioError(f"preference {key} must be a string or boolean") def validate_fixture(fixture_id: str) -> dict[str, Any]: if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", fixture_id): raise ScenarioError("fixture must be a lowercase ID") path = FIXTURE_ROOT / f"{fixture_id}.json" if not path.is_file(): raise ScenarioError(f"fixture not found: {fixture_id}") try: fixture = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ScenarioError(f"cannot read fixture {fixture_id}: {error}") from error validate_fixture_data(fixture, fixture_id) return fixture def validate_fixture_data(fixture: Any, fixture_id: str) -> None: if not isinstance(fixture, dict): raise ScenarioError(f"fixture root must be an object: {fixture_id}") if fixture.get("schemaVersion") != 1 or fixture.get("id") != fixture_id: raise ScenarioError(f"invalid fixture metadata: {fixture_id}") if fixture.get("resetPolicy") != "replace": raise ScenarioError(f"unsupported fixture resetPolicy: {fixture_id}") book = fixture.get("book") if not isinstance(book, dict): raise ScenarioError(f"fixture book must be an object: {fixture_id}") required_strings = ("bookUrl", "name", "author") if any(not isinstance(book.get(key), str) or not book[key] for key in required_strings): raise ScenarioError(f"fixture book metadata is incomplete: {fixture_id}") if book.get("pageMode") != "simulation": raise ScenarioError(f"unsupported fixture pageMode: {fixture_id}") chapters = fixture.get("chapters") if not isinstance(chapters, list) or not chapters: raise ScenarioError(f"fixture must contain chapters: {fixture_id}") start_chapter = book.get("startChapter") start_position = book.get("startPosition") if not isinstance(start_chapter, int) or start_chapter not in range(len(chapters)): raise ScenarioError(f"fixture startChapter is invalid: {fixture_id}") if not isinstance(start_position, int) or start_position < 0: raise ScenarioError(f"fixture startPosition is invalid: {fixture_id}") for index, chapter in enumerate(chapters): if not isinstance(chapter, dict): raise ScenarioError(f"fixture chapters[{index}] must be an object") for key in ("title", "readyMarker", "paragraph"): if not isinstance(chapter.get(key), str) or not chapter[key]: raise ScenarioError(f"fixture chapters[{index}].{key} is required") repeat = chapter.get("repeat") if not isinstance(repeat, int) or repeat < 1 or repeat > 1000: raise ScenarioError(f"fixture chapters[{index}].repeat must be between 1 and 1000") def run_command( command: Sequence[str], *, timeout: int = 120, check: bool = True, cwd: Path = PROJECT_ROOT, ) -> subprocess.CompletedProcess[str]: result = subprocess.run( list(command), cwd=cwd, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout, ) if check and result.returncode != 0: raise CommandError(command, result) return result def adb(device: str, *arguments: str, timeout: int = 60, check: bool = True) -> subprocess.CompletedProcess[str]: return run_command(["adb", "-s", device, *arguments], timeout=timeout, check=check) def prepare_theme_asset(device: str, scenario: dict[str, Any]) -> dict[str, Any] | None: preferences = dict(scenario.get("preferences") or {}) mode = scenario.get("themeAsset") if mode is None: return preferences or None remote_folder = THEME_ASSET_REMOTE.rsplit("/", 1)[0] adb(device, "shell", "rm", "-rf", remote_folder, timeout=10, check=False) if mode == "valid-file": adb(device, "shell", "mkdir", "-p", remote_folder, timeout=10) run_command( ["adb", "-s", device, "push", str(THEME_ASSET_FIXTURE), THEME_ASSET_REMOTE], timeout=30, ) background = THEME_ASSET_REMOTE elif mode == "missing-file": background = f"{remote_folder}/missing-background.png" else: background = f"content://{PACKAGE}.debug-theme-assets/missing-background.png" preferences["backgroundImage"] = background preferences["backgroundImageNight"] = "" return preferences def arm_theme_content_failure(device: str) -> None: adb( device, "shell", "content", "call", "--uri", f"content://{PACKAGE}.debug-theme-assets", "--method", "armFailure", timeout=15, ) def select_device(requested: str | None) -> str: if shutil.which("adb") is None: raise RuntimeError("adb is not available on PATH") result = run_command(["adb", "devices"], timeout=15) devices = [] for line in result.stdout.splitlines()[1:]: fields = line.split() if len(fields) >= 2 and fields[1] == "device": devices.append(fields[0]) if requested: if requested not in devices: raise RuntimeError(f"device is not ready: {requested}") return requested if len(devices) == 1: return devices[0] if not devices: raise RuntimeError("no ready Android device found") raise RuntimeError("multiple devices found; pass --device") def create_session(scenario_name: str, fixture: str, device: str | None) -> tuple[str, Path, dict[str, Any]]: now = dt.datetime.now().astimezone() session_id = now.strftime("%Y%m%d-%H%M%S-%f")[:-3] run_dir = RUN_ROOT / session_id run_dir.mkdir(parents=True, exist_ok=False) metadata = { "schemaVersion": 1, "sessionId": session_id, "scenario": scenario_name, "fixture": fixture, "device": device, "startedAt": now.isoformat(timespec="milliseconds"), "package": PACKAGE, } write_json(run_dir / "metadata.json", metadata) return session_id, run_dir, metadata def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def find_debug_apk() -> Path: output_root = PROJECT_ROOT / "app" / "build" / "outputs" / "apk" candidates = [ path for path in output_root.rglob("*.apk") if "androidTest" not in path.parts and "debug" in path.parts ] if not candidates: raise RuntimeError("debug APK was not produced") universal = [path for path in candidates if "universal" in path.name] return max(universal or candidates, key=lambda path: path.stat().st_mtime) def build_and_install(device: str, run_dir: Path) -> tuple[dict[str, Any], Path]: started = time.monotonic() command = ["./gradlew", ":app:assembleAppDebug", "-PenableAbiSplits=false"] result = run_command(command, timeout=900, check=False) (run_dir / "build.log").write_text(result.stdout, encoding="utf-8") build = { "status": "passed" if result.returncode == 0 else "failed", "durationMs": round((time.monotonic() - started) * 1000), "artifact": "build.log", } if result.returncode != 0: raise CommandError(command, result) apk = find_debug_apk() install_mode = install_apk(device, apk, run_dir) build["apk"] = str(apk.relative_to(PROJECT_ROOT)) build["installMode"] = install_mode return build, apk def install_apk(device: str, apk: Path, run_dir: Path) -> str: attempts = [ ("no_streaming", ["adb", "-s", device, "install", "--no-streaming", "-r", "-t", str(apk)], 120), ("streaming_retry", ["adb", "-s", device, "install", "-r", "-t", str(apk)], 60), ] log_parts = [] last_error: Exception | None = None for mode, command, timeout in attempts: try: result = run_command(command, timeout=timeout, check=False) log_parts.append(f"[{mode}] exit={result.returncode}\n{result.stdout}") if result.returncode == 0 and "Success" in result.stdout: (run_dir / "install.log").write_text("\n".join(log_parts), encoding="utf-8") return mode last_error = CommandError(command, result) except subprocess.TimeoutExpired as error: log_parts.append(f"[{mode}] timeout={timeout}s\n") last_error = error cleanup_device(device) (run_dir / "install.log").write_text("\n".join(log_parts), encoding="utf-8") raise RuntimeError(f"APK install failed after {len(attempts)} attempts: {last_error}") def launch_fixture( device: str, fixture: str, entry: str, session_id: str, preferences: dict[str, Any] | None = None, ) -> None: adb(device, "shell", "am", "force-stop", PACKAGE) adb(device, "logcat", "-c") component = f"{PACKAGE}/{DEBUG_ACTIVITY}" extras = [ "--es", "fixture", fixture, "--es", "session", session_id, "--es", "entry", entry, ] if preferences: # base64 so the JSON survives `adb shell am` re-parsing on the device # (raw JSON gets quote-stripped and brace-expanded by the device shell). payload = json.dumps(preferences, ensure_ascii=False, separators=(",", ":")).encode("utf-8") extras += ["--es", "preferencesB64", base64.b64encode(payload).decode("ascii")] result = adb( device, "shell", "am", "start", "-W", "-n", component, *extras, timeout=60, check=False, ) if result.returncode != 0 or "Error:" in result.stdout: raise CommandError(["adb", "shell", "am", "start", component], result) def ui_dump(device: str) -> str: return adb(device, "exec-out", "uiautomator", "dump", "/dev/tty", timeout=20, check=False).stdout def window_state(device: str) -> str: # Full `dumpsys window windows` output. Callers substring-match against it, # so a hit means the activity is present in the window stack, not strictly # the foreground window. return adb(device, "shell", "dumpsys", "window", "windows", timeout=20, check=False).stdout def device_is_awake(device: str) -> bool: output = adb(device, "shell", "dumpsys", "power", timeout=20, check=False).stdout return "mWakefulness=Awake" in output def keyguard_is_showing(device: str) -> bool: output = adb(device, "shell", "dumpsys", "window", "policy", timeout=20, check=False).stdout match = re.search(r"^\s+showing=(true|false)\b", output, re.MULTILINE) return match is not None and match.group(1) == "true" def device_ready_for_ui(device: str) -> bool: return device_is_awake(device) and not keyguard_is_showing(device) def prepare_device(device: str) -> dict[str, Any]: state = { "wasAwake": device_is_awake(device), "wasKeyguardShowing": keyguard_is_showing(device), "stayOnWhilePluggedIn": adb( device, "shell", "settings", "get", "global", "stay_on_while_plugged_in", timeout=20, check=False, ).stdout.strip(), } adb( device, "shell", "settings", "put", "global", "stay_on_while_plugged_in", "2", timeout=20, ) if not state["wasAwake"]: adb(device, "shell", "input", "keyevent", "KEYCODE_WAKEUP", timeout=20) if keyguard_is_showing(device): adb(device, "shell", "wm", "dismiss-keyguard", timeout=20, check=False) time.sleep(0.2) if keyguard_is_showing(device): adb(device, "shell", "input", "keyevent", "82", timeout=20, check=False) time.sleep(0.5) if not device_ready_for_ui(device): restore_device(device, state) raise RuntimeError("device could not be awakened and unlocked for UI automation") return state def restore_device(device: str, state: dict[str, Any] | None) -> None: if state is None: return stay_on = state.get("stayOnWhilePluggedIn") if stay_on in {"", "null", None}: adb( device, "shell", "settings", "delete", "global", "stay_on_while_plugged_in", timeout=20, check=False, ) else: adb( device, "shell", "settings", "put", "global", "stay_on_while_plugged_in", str(stay_on), timeout=20, check=False, ) if state.get("wasKeyguardShowing"): adb(device, "shell", "input", "keyevent", "KEYCODE_SLEEP", timeout=20, check=False) if state.get("wasAwake"): adb(device, "shell", "input", "keyevent", "KEYCODE_WAKEUP", timeout=20, check=False) elif not state.get("wasAwake"): adb(device, "shell", "input", "keyevent", "KEYCODE_SLEEP", timeout=20, check=False) def parse_ui_nodes(dump: str) -> list[dict[str, str]]: start = dump.find("") if start < 0 or end < 0: return [] try: root = ET.fromstring(dump[start : end + len("")]) except ET.ParseError: return [] return [node.attrib for node in root.iter("node")] READER_PAGE_TAG = "READER_PAGE " def extract_reader_pages(logcat: str) -> list[str]: # ReadView logs "READER_PAGE " on every render (debug # build only). This is the device-independent reader signal — uiautomator # does not expose the ReadView node on all devices. pages = [] for line in logcat.splitlines(): index = line.find(READER_PAGE_TAG) if index >= 0: pages.append(line[index + len(READER_PAGE_TAG):].strip()) return pages def reader_current_page(device: str) -> str: logcat = adb(device, "logcat", "-d", "-s", "LegadoDebug:I", timeout=20, check=False).stdout pages = extract_reader_pages(logcat) return pages[-1] if pages else "" def find_node_by_content_prefix( dump: str, prefix: str, package: str = PACKAGE, ) -> dict[str, str] | None: # The bookshelf item's label lives on a node whose clickable flag is false # (the clickable modifier sits on an ancestor). Match by content-desc and # tap the label's center; the clickable ancestor is under that point. candidates = [ node for node in parse_ui_nodes(dump) if node.get("package") == package and node.get("content-desc", "").startswith(prefix) ] return candidates[0] if candidates else None def find_node_by_text( dump: str, text: str, package: str = PACKAGE, ) -> dict[str, str] | None: candidates = [ node for node in parse_ui_nodes(dump) if node.get("package") == package and node.get("text") == text ] return candidates[0] if candidates else None def debug_state(device: str, method: str) -> dict[str, Any]: result = adb( device, "shell", "content", "call", "--uri", DEBUG_STATE_URI, "--method", method, timeout=20, ) match = re.search(r"resultB64=([A-Za-z0-9+/=]+)", result.stdout) if match is None: raise RuntimeError(f"debug state response is invalid: {result.stdout.strip()}") return json.loads(base64.b64decode(match.group(1)).decode("utf-8")) def find_node_in_same_row( dump: str, anchor: dict[str, str], resource_id: str, ) -> dict[str, str] | None: anchor_bounds = parse_bounds(anchor) if anchor_bounds is None: return None _, anchor_top, _, anchor_bottom = anchor_bounds candidates = [] for node in parse_ui_nodes(dump): if node.get("package") != PACKAGE or node.get("resource-id") != resource_id: continue bounds = parse_bounds(node) if bounds is None: continue _, top, _, bottom = bounds overlap = min(anchor_bottom, bottom) - max(anchor_top, top) if overlap > 0: candidates.append((overlap, node)) return max(candidates, key=lambda item: item[0])[1] if candidates else None def parse_bounds(node: dict[str, str]) -> tuple[int, int, int, int] | None: match = re.fullmatch(r"\[(\d+),(\d+)]\[(\d+),(\d+)]", node.get("bounds", "")) return tuple(map(int, match.groups())) if match else None def node_center(node: dict[str, str]) -> tuple[int, int]: bounds = parse_bounds(node) if bounds is None: raise RuntimeError(f"invalid accessibility bounds: {node.get('bounds', '')}") left, top, right, bottom = bounds return (left + right) // 2, (top + bottom) // 2 def wait_until_node_present( device: str, content_prefix: str, timeout_seconds: int, ) -> tuple[bool, str, dict[str, str] | None]: deadline = time.monotonic() + timeout_seconds latest = "" node = None while time.monotonic() < deadline: latest = ui_dump(device) node = find_node_by_content_prefix(latest, content_prefix) if node is not None and device_ready_for_ui(device): return True, latest, node time.sleep(0.5) return False, latest, node def wait_until_text_present( device: str, text: str, timeout_seconds: int, ) -> tuple[bool, str, dict[str, str] | None]: deadline = time.monotonic() + timeout_seconds latest = "" node = None while time.monotonic() < deadline: latest = ui_dump(device) node = find_node_by_text(latest, text) if node is not None and device_ready_for_ui(device): return True, latest, node time.sleep(0.25) return False, latest, node def find_node_by_class(dump: str, class_name: str) -> dict[str, str] | None: candidates = [ node for node in parse_ui_nodes(dump) if node.get("package") == PACKAGE and node.get("class") == class_name ] return candidates[0] if candidates else None def find_node_containing_text(dump: str, text: str) -> dict[str, str] | None: candidates = [ node for node in parse_ui_nodes(dump) if text in node.get("text", "") ] return candidates[0] if candidates else None def find_node_by_content_desc_near( dump: str, content_desc: str, anchor: dict[str, str], ) -> dict[str, str] | None: anchor_x, anchor_y = node_center(anchor) candidates = [ node for node in parse_ui_nodes(dump) if node.get("package") == PACKAGE and node.get("content-desc") == content_desc and parse_bounds(node) is not None ] return min( candidates, key=lambda node: ( abs(node_center(node)[0] - anchor_x) > 360, abs(node_center(node)[1] - anchor_y), abs(node_center(node)[0] - anchor_x), ), default=None, ) def saved_theme_manifest(device: str, theme_name: str) -> tuple[str, dict[str, Any]] | None: result = adb( device, "exec-out", "run-as", PACKAGE, "find", "files/saved_themes", "-name", "manifest.json", "-type", "f", timeout=20, check=False, ) for path in result.stdout.splitlines(): manifest = adb( device, "exec-out", "run-as", PACKAGE, "cat", path.strip(), timeout=20, check=False, ).stdout try: data = json.loads(manifest) if data.get("name") == theme_name: return path.strip(), data except json.JSONDecodeError: continue return None def saved_theme_exists(device: str, theme_name: str) -> bool: return saved_theme_manifest(device, theme_name) is not None def saved_theme_has_fixture_asset(device: str, theme_name: str) -> bool: saved = saved_theme_manifest(device, theme_name) if saved is None: return False manifest_path, manifest = saved entry = (manifest.get("assets") or {}).get("background.light") if not isinstance(entry, str) or not entry: return False asset_path = f"{manifest_path.rsplit('/', 1)[0]}/{entry}" result = adb( device, "exec-out", "run-as", PACKAGE, "sha256sum", asset_path, timeout=20, check=False, ) digest = result.stdout.split() expected = hashlib.sha256(THEME_ASSET_FIXTURE.read_bytes()).hexdigest() return bool(digest) and digest[0] == expected def wait_until_reader_ready(device: str, marker: str, timeout_seconds: int) -> tuple[bool, str, str]: deadline = time.monotonic() + timeout_seconds content = "" while time.monotonic() < deadline: content = reader_current_page(device) if marker in content and PACKAGE in window_state(device) and device_ready_for_ui(device): return True, content, content time.sleep(0.5) return False, content, content def wait_until_reader_content_changed( device: str, previous_content: str, timeout_seconds: int, ) -> tuple[bool, str, str]: deadline = time.monotonic() + timeout_seconds content = "" while time.monotonic() < deadline: content = reader_current_page(device) if content and content != previous_content: return True, content, content time.sleep(0.2) return False, content, content def resumed_activity(device: str) -> str: output = adb(device, "shell", "dumpsys", "activity", "activities", timeout=20, check=False).stdout match = re.search(r"topResumedActivity=.*?\s([\w.]+/[\w.$]+)\s", output) return match.group(1) if match else "" def wait_until_activity_resumed( device: str, activity: str, timeout_seconds: int, ) -> tuple[bool, str]: deadline = time.monotonic() + timeout_seconds resumed = "" component = f"{PACKAGE}/{activity}" while time.monotonic() < deadline: resumed = resumed_activity(device) if resumed == component and device_ready_for_ui(device): return True, resumed time.sleep(0.2) return False, resumed def application_pid(device: str) -> str | None: result = adb(device, "shell", "pidof", PACKAGE, timeout=10, check=False) return result.stdout.strip().split()[0] if result.returncode == 0 and result.stdout.strip() else None def screen_size(device: str) -> tuple[int, int]: output = adb(device, "shell", "wm", "size").stdout matches = re.findall(r"(\d+)x(\d+)", output) if not matches: raise RuntimeError(f"cannot determine screen size: {output.strip()}") width, height = matches[-1] return int(width), int(height) def capture_screenshot(device: str, path: Path) -> None: with path.open("wb") as output: result = subprocess.run( ["adb", "-s", device, "exec-out", "screencap", "-p"], cwd=PROJECT_ROOT, stdout=output, stderr=subprocess.PIPE, timeout=30, ) if result.returncode != 0: path.unlink(missing_ok=True) raise RuntimeError(result.stderr.decode(errors="replace")) class ScreenRecorder: """Optional screenrecord wrapper for capturing transient UI (animations). Best-effort: any failure to start, stop, or pull leaves the run unaffected and simply produces no video artifact. """ REMOTE = "/sdcard/legado-debug-transition.mp4" def __init__(self, device: str): self.device = device self.process: subprocess.Popen[bytes] | None = None self.saved: str | None = None self._stopped = False def start(self) -> None: adb(self.device, "shell", "rm", "-f", self.REMOTE, timeout=15, check=False) self.process = subprocess.Popen( ["adb", "-s", self.device, "shell", "screenrecord", "--time-limit", "60", self.REMOTE], cwd=PROJECT_ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) time.sleep(0.5) def stop(self, run_dir: Path, name: str = "transition.mp4") -> str | None: if self.process is None or self._stopped: return self.saved self._stopped = True # SIGINT lets screenrecord finalize a playable mp4 before it exits. pids = adb(self.device, "shell", "pidof", "screenrecord", timeout=10, check=False).stdout.split() for pid in pids: adb(self.device, "shell", "kill", "-INT", pid, timeout=10, check=False) try: self.process.wait(timeout=15) except subprocess.TimeoutExpired: self.process.kill() pull = subprocess.run( ["adb", "-s", self.device, "pull", self.REMOTE, str(run_dir / name)], cwd=PROJECT_ROOT, capture_output=True, text=True, timeout=60, ) adb(self.device, "shell", "rm", "-f", self.REMOTE, timeout=15, check=False) if pull.returncode == 0 and (run_dir / name).is_file(): self.saved = name return self.saved def collect_logcat( device: str, run_dir: Path, app_pids: Sequence[str] = (), session_id: str | None = None, ) -> str: logcat = adb(device, "logcat", "-d", "-v", "threadtime", timeout=45, check=False).stdout (run_dir / "logcat.txt").write_text(logcat, encoding="utf-8") session_logcat = "" unique_pids = list(dict.fromkeys(pid for pid in app_pids if pid)) for pid in unique_pids: pid_logcat = adb( device, "logcat", "-d", f"--pid={pid}", "-v", "threadtime", timeout=45, check=False, ).stdout session_logcat += f"--- pid={pid} ---\n{pid_logcat}" if session_id: session_lines = [line for line in logcat.splitlines() if f"[session={session_id}]" in line] for line in session_lines: if line not in session_logcat: session_logcat += line + "\n" (run_dir / "session-logcat.txt").write_text(session_logcat, encoding="utf-8") crash = extract_crash(logcat, PACKAGE, unique_pids) (run_dir / "crash.txt").write_text(crash, encoding="utf-8") return logcat def extract_crash( logcat: str, package: str = PACKAGE, app_pids: Sequence[str] = (), ) -> str: lines = logcat.splitlines() chunks = [] for index, line in enumerate(lines): chunk = lines[index : index + 100] if f"ANR in {package}" in line: chunks.extend(chunk) elif "FATAL EXCEPTION:" in line and any(package in item for item in chunk[:10]): chunks.extend(chunk) elif "Fatal signal " in line and any( re.search(rf"\b{re.escape(pid)}\b", line) for pid in app_pids ): chunks.extend(chunk) if not chunks: return "" return "\n".join(chunks).rstrip() + "\n" def parse_failure(crash: str) -> dict[str, Any] | None: if not crash: return None if "ANR in" in crash: return { "category": "anr", "confidence": "high", "detectedSignals": ["anr"], "stacktraceArtifact": "crash.txt", } if "Fatal signal " in crash: return { "category": "native_crash", "confidence": "high", "detectedSignals": ["fatal_signal"], "stacktraceArtifact": "crash.txt", } exceptions = list( re.finditer(r"(Caused by: )?([\w.$]+(?:Exception|Error))(?:: ([^\n]+))?", crash) ) caused_by = [match for match in exceptions if match.group(1)] exception = caused_by[-1] if caused_by else (exceptions[0] if exceptions else None) failure: dict[str, Any] = { "category": "uncaught_exception", "confidence": "high", "detectedSignals": ["fatal_exception"], "stacktraceArtifact": "crash.txt", } if exception: failure["exceptionType"] = exception.group(2).split(".")[-1] failure["exceptionConfidence"] = "medium" if exceptions else "low" if exception.group(3): failure["message"] = exception.group(3).strip() frame = re.search( r"at (io\.legado\.app\.[\w.$]+)\.([\w$<>]+)\(([^():]+):(\d+)\)", crash, ) if frame: failure["suspectedFrame"] = { "class": frame.group(1), "method": frame.group(2), "file": frame.group(3), "line": int(frame.group(4)), "confidence": "medium", } return failure def sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest() def make_summary( *, session_id: str, scenario: dict[str, Any], fixture: str, build: dict[str, Any], ready: bool, activity_visible: bool, content_changed: bool, action_count: int, failure: dict[str, Any] | None, ) -> tuple[dict[str, Any], int]: assertion_values = { "reader_ready": ready, "reader_activity_visible": activity_visible, "reader_content_changed": content_changed, "no_fatal_exception": failure is None, } return make_summary_from_assertions( session_id=session_id, scenario=scenario, fixture=fixture, build=build, assertion_values=assertion_values, action_count=action_count, failure=failure, ) def make_summary_from_assertions( *, session_id: str, scenario: dict[str, Any], fixture: str, build: dict[str, Any], assertion_values: dict[str, bool], action_count: int, failure: dict[str, Any] | None, extra_artifacts: dict[str, str] | None = None, ) -> tuple[dict[str, Any], int]: assertions = [ {"name": name, "passed": assertion_values[name]} for name in scenario["assertions"] ] if failure is not None: status, exit_code = "failed", EXIT_CRASH elif not all(item["passed"] for item in assertions): status, exit_code = "failed", EXIT_ASSERTION else: status, exit_code = "passed", 0 summary: dict[str, Any] = { "schemaVersion": 1, "status": status, "exitCode": exit_code, "sessionId": session_id, "scenario": scenario["name"], "fixture": fixture, "build": build, "actions": {"performed": action_count}, "assertions": assertions, "artifacts": { "metadata": "metadata.json", "logcat": "logcat.txt", "sessionLogcat": "session-logcat.txt", "crash": "crash.txt", }, } if failure is not None: summary["failure"] = failure if extra_artifacts: summary["artifacts"].update(extra_artifacts) return summary, exit_code def error_summary( session_id: str, scenario_name: str, fixture: str, exit_code: int, category: str, message: str, build: dict[str, Any] | None = None, ) -> dict[str, Any]: summary: dict[str, Any] = { "schemaVersion": 1, "status": "failed", "exitCode": exit_code, "sessionId": session_id, "scenario": scenario_name, "fixture": fixture, "failure": {"category": category, "message": message}, "artifacts": {"metadata": "metadata.json"}, } if build is not None: summary["build"] = build return summary def cleanup_device(device: str) -> None: adb(device, "shell", "am", "force-stop", PACKAGE, timeout=20, check=False) # --------------------------------------------------------------------------- # # Capability registry: entries, actions, assertions. # # A scenario is fully described by data (JSON). Each domain action and each # assertion is a small handler registered here. Adding a bug scenario means # adding a handler + a JSON file, not editing execute() / validate_scenario(). # Handlers only *record* raw observations into the Context; assertions are pure # reads of those observations. # --------------------------------------------------------------------------- # class Context: """Mutable per-run state shared across entry, actions, and assertions.""" def __init__( self, device: str, run_dir: Path, scenario: dict[str, Any], fixture_data: dict[str, Any], metadata: dict[str, Any], action_repeat: int, ): self.device = device self.run_dir = run_dir self.scenario = scenario self.fixture_data = fixture_data self.metadata = metadata self.action_repeat = action_repeat self.app_pids: list[str] = [] self.records: dict[str, Any] = {} self.state: dict[str, Any] = {} self.artifacts: dict[str, str] = {} self.action_count = 0 self.assertion_values: dict[str, bool] = {} self.failure: dict[str, Any] | None = None def note_pid(self) -> None: pid = application_pid(self.device) if pid and pid not in self.app_pids: self.app_pids.append(pid) def screenshot(self, name: str, filename: str) -> None: capture_screenshot(self.device, self.run_dir / filename) self.artifacts[name] = filename class Entry: type: str = "" allowed_actions: set[str] = set() def prepare(self, ctx: Context) -> None: raise NotImplementedError def finalize(self, ctx: Context) -> None: raise NotImplementedError class Action: name: str = "" params: dict[str, Any] = {} def validate(self, action: dict[str, Any], index: int) -> None: pass def run(self, ctx: Context, action: dict[str, Any]) -> None: raise NotImplementedError def register_entry(name: str) -> Callable[[type], type]: def decorate(cls: type) -> type: instance = cls() instance.type = name ENTRIES[name] = instance return cls return decorate def register_action(name: str) -> Callable[[type], type]: def decorate(cls: type) -> type: instance = cls() instance.name = name ACTIONS[name] = instance return cls return decorate @register_entry("reader") class ReaderEntry(Entry): allowed_actions = {"turn_page"} def prepare(self, ctx: Context) -> None: ready, before_dump, before_content = wait_until_reader_ready( ctx.device, ctx.scenario["readyMarker"], ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.note_pid() ctx.screenshot("beforeScreenshot", "before.png") ctx.records["reader_ready"] = ready ctx.records["reader_content_changed"] = False ctx.records["content_change_duration_ms"] = 0 ctx.records["ui_before"] = before_dump ctx.records["ui_after"] = before_dump ctx.records["reader_content_before"] = before_content ctx.records["reader_content_after"] = before_content ctx.state["reader_content"] = before_content def finalize(self, ctx: Context) -> None: ctx.screenshot("afterScreenshot", "after.png") windows = window_state(ctx.device) # Presence of MainActivity in the window stack; the reader runs as a # route inside MainActivity in this fork. visible = PACKAGE in windows and MAIN_ACTIVITY in windows and device_ready_for_ui(ctx.device) ctx.records["reader_activity_visible"] = visible ctx.metadata.update( { "actionRepeat": ctx.action_repeat, "uiBeforeSha256": sha256_text(ctx.records["ui_before"]), "uiAfterSha256": sha256_text(ctx.records["ui_after"]), "readerContentBeforeSha256": sha256_text(ctx.records["reader_content_before"]), "readerContentAfterSha256": sha256_text(ctx.records["reader_content_after"]), "contentChangeDurationMs": ctx.records["content_change_duration_ms"], } ) @register_entry("rss_read") class RssReadEntry(Entry): allowed_actions = {"toggle_rss_day_night"} def prepare(self, ctx: Context) -> None: ready, dump, _ = wait_until_text_present( ctx.device, "DEBUG RSS 日夜切换", ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.note_pid() ctx.screenshot("beforeScreenshot", "before.png") ctx.records["rss_read_ready"] = ready ctx.records["rss_theme_changed"] = False ctx.records["rss_pid_before"] = application_pid(ctx.device) ctx.records["ui_before"] = dump def finalize(self, ctx: Context) -> None: ctx.screenshot("afterScreenshot", "after.png") ctx.metadata.update( { "rssPidBefore": ctx.records["rss_pid_before"], "rssPidAfter": application_pid(ctx.device), } ) @register_entry("bookshelf") class BookshelfEntry(Entry): allowed_actions = {"open_fixture_book", "press_back"} def prepare(self, ctx: Context) -> None: book_name = ctx.fixture_data["book"]["name"] ready, before_dump, book_node = wait_until_node_present( ctx.device, book_name, ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.note_pid() ctx.screenshot("beforeScreenshot", "before.png") ctx.records["bookshelf_ready"] = ready ctx.records["reader_opened"] = False ctx.records["returned_to_bookshelf"] = False ctx.records["ui_before"] = before_dump ctx.records["ui_after"] = before_dump ctx.records["reader_dump"] = "" ctx.records["reader_content"] = "" ctx.state["book_name"] = book_name ctx.state["book_node"] = book_node ctx.state["returned_node"] = None def finalize(self, ctx: Context) -> None: ctx.screenshot("afterScreenshot", "after.png") book_node = ctx.state.get("book_node") returned_node = ctx.state.get("returned_node") ctx.metadata.update( { "uiBeforeSha256": sha256_text(ctx.records["ui_before"]), "readerUiSha256": sha256_text(ctx.records["reader_dump"]), "uiAfterSha256": sha256_text(ctx.records["ui_after"]), "readerContentSha256": sha256_text(ctx.records["reader_content"]), "bookshelfItemBefore": book_node.get("content-desc") if book_node else None, "bookshelfItemAfter": returned_node.get("content-desc") if returned_node else None, } ) @register_entry("source_manage") class SourceManageEntry(Entry): allowed_actions = {"open_fixture_source", "predictive_back"} def prepare(self, ctx: Context) -> None: activity_ready, resumed = wait_until_activity_resumed( ctx.device, SOURCE_MANAGE_ACTIVITY, ctx.scenario.get("readyTimeoutSeconds", 30), ) source_name = "Legado Debug Fixture" dump = ui_dump(ctx.device) source_node = find_node_by_text(dump, source_name) edit_node = find_node_in_same_row( dump, source_node, f"{PACKAGE}:id/iv_edit", ) if source_node else None ctx.note_pid() ctx.screenshot("beforeScreenshot", "before.png") ctx.records["source_manage_ready"] = activity_ready and source_node is not None and edit_node is not None ctx.records["source_edit_opened"] = False ctx.records["returned_to_source_manage"] = False ctx.records["resumed_before"] = resumed ctx.records["resumed_after"] = resumed ctx.state["source_name"] = source_name ctx.state["edit_node"] = edit_node def finalize(self, ctx: Context) -> None: ctx.screenshot("afterScreenshot", "after.png") ctx.metadata.update( { "resumedBefore": ctx.records["resumed_before"], "resumedAfter": ctx.records["resumed_after"], } ) @register_entry("theme_config") class ThemeConfigEntry(Entry): allowed_actions = { "export_current_theme", "open_theme_manage", "save_current_theme", "select_day_background", "apply_saved_theme", "delete_saved_theme", } def prepare(self, ctx: Context) -> None: ready, dump, _ = wait_until_text_present( ctx.device, "主题模式", ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.note_pid() ctx.screenshot("beforeScreenshot", "before.png") ctx.records["theme_config_ready"] = ready ctx.records["theme_manage_opened"] = False ctx.records["theme_save_succeeded"] = False ctx.records["theme_export_succeeded"] = False ctx.records["theme_save_failed"] = False ctx.records["theme_export_failed"] = False ctx.records["theme_failure_feedback"] = False ctx.records["theme_save_asset_included"] = False ctx.records["theme_export_asset_included"] = False ctx.records["theme_day_background_copied"] = False ctx.records["theme_saved_theme_applied"] = False ctx.records["theme_saved_theme_deleted"] = False ctx.records["ui_before"] = dump ctx.records["ui_after"] = dump def finalize(self, ctx: Context) -> None: ctx.screenshot("afterScreenshot", "after.png") ctx.metadata.update( { "uiBeforeSha256": sha256_text(ctx.records["ui_before"]), "uiAfterSha256": sha256_text(ctx.records["ui_after"]), "savedThemeName": ctx.state.get("theme_name"), "exportedThemeName": ctx.state.get("export_name"), } ) @register_action("turn_page") class TurnPage(Action): params = {"direction": ["next", "previous"], "count": "1-100"} def validate(self, action: dict[str, Any], index: int) -> None: if action.get("direction") not in {"next", "previous"}: raise ScenarioError(f"actions[{index}].direction must be next or previous") count = action.get("count", 1) if not isinstance(count, int) or count < 1 or count > 100: raise ScenarioError(f"actions[{index}].count must be between 1 and 100") def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("reader_ready"): return width, height = screen_size(ctx.device) direction = action["direction"] count = action.get("count", 1) * ctx.action_repeat x = round(width * (0.84 if direction == "next" else 0.16)) y = round(height * 0.5) for _ in range(count): adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 time.sleep(0.4) started = time.monotonic() changed, after_dump, after_content = wait_until_reader_content_changed( ctx.device, ctx.state["reader_content"], ctx.scenario.get("contentChangeTimeoutSeconds", 5), ) ctx.records["content_change_duration_ms"] = round((time.monotonic() - started) * 1000) if changed: ctx.records["reader_content_changed"] = True ctx.state["reader_content"] = after_content ctx.records["reader_content_after"] = after_content ctx.records["ui_after"] = after_dump @register_action("toggle_rss_day_night") class ToggleRssDayNight(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("rss_read_ready"): return before_pid = application_pid(ctx.device) debug_state(ctx.device, "toggleTheme") ctx.action_count += 1 time.sleep(1.5) after_pid = application_pid(ctx.device) ctx.records["rss_theme_changed"] = before_pid is not None and before_pid == after_pid @register_action("open_fixture_book") class OpenFixtureBook(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: book_node = ctx.state.get("book_node") if not (ctx.records.get("bookshelf_ready") and book_node is not None): return x, y = node_center(book_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 opened, reader_dump, reader_content = wait_until_reader_ready( ctx.device, ctx.scenario["readyMarker"], ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.records["reader_opened"] = opened ctx.records["reader_dump"] = reader_dump ctx.records["reader_content"] = reader_content ctx.screenshot("readerScreenshot", "reader.png") @register_action("press_back") class PressBack(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("reader_opened"): return adb(ctx.device, "shell", "input", "keyevent", "BACK", timeout=10) ctx.action_count += 1 returned, after_dump, returned_node = wait_until_node_present( ctx.device, ctx.state["book_name"], ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.records["returned_to_bookshelf"] = returned ctx.records["ui_after"] = after_dump ctx.state["returned_node"] = returned_node @register_action("predictive_back") class PredictiveBack(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("source_edit_opened"): return width, height = screen_size(ctx.device) y = round(height * 0.5) end_x = round(width * 0.45) adb( ctx.device, "shell", "input", "swipe", "1", str(y), str(end_x), str(y), "300", timeout=10, ) ctx.action_count += 1 returned, resumed = wait_until_activity_resumed( ctx.device, SOURCE_MANAGE_ACTIVITY, ctx.scenario.get("readyTimeoutSeconds", 30), ) dump = ui_dump(ctx.device) source_present = find_node_by_text(dump, ctx.state["source_name"]) is not None ctx.records["returned_to_source_manage"] = returned and source_present ctx.records["resumed_after"] = resumed @register_action("open_fixture_source") class OpenFixtureSource(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: edit_node = ctx.state.get("edit_node") if not (ctx.records.get("source_manage_ready") and edit_node is not None): return x, y = node_center(edit_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 opened, _ = wait_until_activity_resumed( ctx.device, SOURCE_EDIT_ACTIVITY, ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.records["source_edit_opened"] = opened ctx.screenshot("editScreenshot", "edit.png") @register_action("open_theme_manage") class OpenThemeManage(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("theme_config_ready"): return node = None dump = "" width, height = screen_size(ctx.device) for _ in range(6): dump = ui_dump(ctx.device) node = find_node_by_text(dump, "主题管理") if node is not None: break adb( ctx.device, "shell", "input", "swipe", str(width // 2), str(round(height * 0.84)), str(width // 2), str(round(height * 0.48)), "300", timeout=10, ) time.sleep(0.4) if node is None: ctx.records["ui_after"] = dump return x, y = node_center(node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 opened, dump, _ = wait_until_text_present( ctx.device, "保存当前设置", ctx.scenario.get("readyTimeoutSeconds", 30), ) ctx.records["theme_manage_opened"] = opened ctx.records["ui_after"] = dump ctx.screenshot("manageScreenshot", "manage.png") @register_action("select_day_background") class SelectDayBackground(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("theme_config_ready"): return fixture = PROJECT_ROOT / "app" / "src" / "main" / "assets" / "web" / "uploadBook" / "img" / "close.png" remote_fixture = "/sdcard/Download/DEBUG_THEME_BG.png" remote_folder = f"/sdcard/Android/data/{PACKAGE}/files/backgroundImage" adb(ctx.device, "shell", "rm", "-rf", remote_folder, timeout=10, check=False) run_command(["adb", "-s", ctx.device, "push", str(fixture), remote_fixture], timeout=30) adb( ctx.device, "shell", "am", "broadcast", "-a", "android.intent.action.MEDIA_SCANNER_SCAN_FILE", "-d", f"file://{remote_fixture}", timeout=15, check=False, ) day_node = None dump = "" width, height = screen_size(ctx.device) for _ in range(8): dump = ui_dump(ctx.device) day_node = find_node_by_text(dump, "白天") if day_node is not None: break adb( ctx.device, "shell", "input", "swipe", str(width // 2), str(round(height * 0.84)), str(width // 2), str(round(height * 0.48)), "300", timeout=10, ) time.sleep(0.4) if day_node is None: ctx.records["ui_after"] = dump return x, y = node_center(day_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + 10 add_node = None while time.monotonic() < deadline: dump = ui_dump(ctx.device) add_node = find_node_by_content_prefix(dump, "添加") if add_node is not None: break time.sleep(0.2) if add_node is None: ctx.records["ui_after"] = dump return x, y = node_center(add_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 filename_node = None deadline = time.monotonic() + 5 while time.monotonic() < deadline: dump = ui_dump(ctx.device) filename_node = find_node_by_text(dump, "DEBUG_THEME_BG.png") if filename_node is not None: break more_node = find_node_by_content_prefix( dump, "更多", "com.google.android.photopicker", ) if more_node is not None: x, y = node_center(more_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) browse_node = None browse_deadline = time.monotonic() + 5 while time.monotonic() < browse_deadline: dump = ui_dump(ctx.device) browse_node = find_node_by_text( dump, "浏览…", "com.google.android.photopicker", ) if browse_node is not None: break time.sleep(0.2) if browse_node is not None: x, y = node_center(browse_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) break time.sleep(0.25) if filename_node is None: deadline = time.monotonic() + 15 download_opened = False while time.monotonic() < deadline: dump = ui_dump(ctx.device) filename_node = find_node_by_text( dump, "DEBUG_THEME_BG.png", "com.google.android.documentsui", ) if filename_node is not None: break download_node = find_node_by_text( dump, "下载", "com.google.android.documentsui", ) if download_node is not None and not download_opened: x, y = node_center(download_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) download_opened = True elif download_opened: adb( ctx.device, "shell", "input", "swipe", str(width // 2), str(round(height * 0.82)), str(width // 2), str(round(height * 0.40)), "250", timeout=10, ) time.sleep(0.4) if filename_node is None: ctx.records["ui_after"] = dump return x, y = node_center(filename_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 wait_until_activity_resumed( ctx.device, MAIN_ACTIVITY, ctx.scenario.get("readyTimeoutSeconds", 30), ) expected_hash = hashlib.sha256(fixture.read_bytes()).hexdigest() copied_path = "" deadline = time.monotonic() + ctx.scenario.get("readyTimeoutSeconds", 30) while time.monotonic() < deadline: result = adb( ctx.device, "shell", "find", remote_folder, "-type", "f", timeout=10, check=False, ) candidates = [line.strip() for line in result.stdout.splitlines() if line.strip()] for candidate in candidates: digest = adb( ctx.device, "shell", "sha256sum", candidate, timeout=10, check=False, ).stdout.split() if digest and digest[0] == expected_hash: copied_path = candidate break if copied_path: break time.sleep(0.25) adb(ctx.device, "shell", "rm", "-f", remote_fixture, timeout=10, check=False) ctx.records["theme_day_background_copied"] = bool(copied_path) ctx.state["day_background_path"] = copied_path ctx.records["ui_after"] = ui_dump(ctx.device) ctx.screenshot("backgroundScreenshot", "background.png") @register_action("save_current_theme") class SaveCurrentTheme(Action): params = {"namePrefix": "non-empty ASCII string"} def validate(self, action: dict[str, Any], index: int) -> None: prefix = action.get("namePrefix") if not isinstance(prefix, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", prefix): raise ScenarioError(f"actions[{index}].namePrefix must be non-empty ASCII") def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("theme_manage_opened"): return theme_name = f"{action['namePrefix']}_{ctx.metadata['sessionId']}" ctx.state["theme_name"] = theme_name ready, _, save_current = wait_until_text_present(ctx.device, "保存当前设置", 10) if not (ready and save_current): return x, y = node_center(save_current) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + 10 edit = None dump = "" while time.monotonic() < deadline: dump = ui_dump(ctx.device) edit = find_node_by_class(dump, "android.widget.EditText") if edit is not None: break time.sleep(0.2) if edit is None: ctx.records["ui_after"] = dump return x, y = node_center(edit) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) adb(ctx.device, "shell", "input", "text", theme_name, timeout=10) dump = ui_dump(ctx.device) confirm = find_node_by_text(dump, "保存") if confirm is None: ctx.records["ui_after"] = dump return if ctx.scenario.get("themeAsset") == "missing-content-uri": arm_theme_content_failure(ctx.device) x, y = node_center(confirm) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + ctx.scenario.get("readyTimeoutSeconds", 30) succeeded = False failure_feedback = "" while time.monotonic() < deadline: if saved_theme_exists(ctx.device, theme_name): succeeded = True break dump = ui_dump(ctx.device) failure_node = find_node_containing_text(dump, "主题保存失败") if failure_node is not None: failure_feedback = failure_node.get("text", "") break time.sleep(0.25) ctx.records["theme_save_succeeded"] = succeeded ctx.records["theme_save_failed"] = not succeeded and bool(failure_feedback) ctx.records["theme_failure_feedback"] = bool(failure_feedback) if failure_feedback: ctx.state["theme_failure_feedback"] = failure_feedback if succeeded and ctx.scenario.get("themeAsset") == "valid-file": ctx.records["theme_save_asset_included"] = saved_theme_has_fixture_asset( ctx.device, theme_name, ) ctx.records["ui_after"] = ui_dump(ctx.device) @register_action("apply_saved_theme") class ApplySavedTheme(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: theme_name = ctx.state.get("theme_name") if not theme_name: return width, height = screen_size(ctx.device) name_node = None dump = "" for _ in range(20): dump = ui_dump(ctx.device) name_node = find_node_by_text(dump, theme_name) if name_node is not None: break adb( ctx.device, "shell", "input", "swipe", str(width // 2), str(round(height * 0.82)), str(width // 2), str(round(height * 0.38)), "250", timeout=10, ) time.sleep(0.3) if name_node is None: ctx.records["ui_after"] = dump return x, y = node_center(name_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 dialog_ready, dump, confirm = wait_until_text_present(ctx.device, "应用", 10) if not (dialog_ready and confirm): ctx.records["ui_after"] = dump return x, y = node_center(confirm) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + 15 while time.monotonic() < deadline: dump = ui_dump(ctx.device) if find_node_by_text(dump, theme_name) is not None and find_node_by_text(dump, "应用") is None: ctx.records["theme_saved_theme_applied"] = True break time.sleep(0.25) ctx.records["ui_after"] = dump @register_action("delete_saved_theme") class DeleteSavedTheme(Action): def run(self, ctx: Context, action: dict[str, Any]) -> None: theme_name = ctx.state.get("theme_name") if not (theme_name and ctx.records.get("theme_saved_theme_applied")): return ready, dump, name_node = wait_until_text_present(ctx.device, theme_name, 15) if not (ready and name_node): ctx.records["ui_after"] = dump return delete_node = find_node_by_content_desc_near(dump, "删除", name_node) if delete_node is None: ctx.records["ui_after"] = dump return x, y = node_center(delete_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 dialog_ready, dump, confirm = wait_until_text_present(ctx.device, "删除", 10) if not (dialog_ready and confirm): ctx.records["ui_after"] = dump return x, y = node_center(confirm) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + 15 while time.monotonic() < deadline: if not saved_theme_exists(ctx.device, theme_name): ctx.records["theme_saved_theme_deleted"] = True break time.sleep(0.25) ctx.records["ui_after"] = ui_dump(ctx.device) @register_action("export_current_theme") class ExportCurrentTheme(Action): params = {"namePrefix": "non-empty ASCII string"} def validate(self, action: dict[str, Any], index: int) -> None: prefix = action.get("namePrefix") if not isinstance(prefix, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", prefix): raise ScenarioError(f"actions[{index}].namePrefix must be non-empty ASCII") def run(self, ctx: Context, action: dict[str, Any]) -> None: if not ctx.records.get("theme_manage_opened"): return export_name = f"{action['namePrefix']}_{ctx.metadata['sessionId']}.zip" ctx.state["export_name"] = export_name remote_path = f"/sdcard/Download/{export_name}" adb(ctx.device, "shell", "rm", "-f", remote_path, timeout=10, check=False) export_current = None width, height = screen_size(ctx.device) for _ in range(15): dump = ui_dump(ctx.device) export_current = find_node_by_text(dump, "导出当前主题") if export_current is not None: break adb( ctx.device, "shell", "input", "swipe", str(width // 2), str(round(height * 0.38)), str(width // 2), str(round(height * 0.82)), "250", timeout=10, ) time.sleep(0.3) if export_current is None: return x, y = node_center(export_current) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 deadline = time.monotonic() + 15 filename_node = None while time.monotonic() < deadline: dump = ui_dump(ctx.device) filename_node = next( ( node for node in parse_ui_nodes(dump) if node.get("package") == "com.google.android.documentsui" and node.get("class") == "android.widget.EditText" ), None, ) if filename_node is not None: break time.sleep(0.25) if filename_node is None: return x, y = node_center(filename_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) adb(ctx.device, "shell", "input", "keyevent", "KEYCODE_MOVE_END", timeout=10) for _ in range(80): adb(ctx.device, "shell", "input", "keyevent", "KEYCODE_DEL", timeout=10) adb(ctx.device, "shell", "input", "text", export_name, timeout=10) dump = ui_dump(ctx.device) save_node = next( ( node for node in parse_ui_nodes(dump) if node.get("package") == "com.google.android.documentsui" and node.get("resource-id") == "android:id/button1" ), None, ) if save_node is None: ctx.records["ui_after"] = dump return if ctx.scenario.get("themeAsset") == "missing-content-uri": arm_theme_content_failure(ctx.device) x, y = node_center(save_node) adb(ctx.device, "shell", "input", "tap", str(x), str(y), timeout=10) ctx.action_count += 1 local_path = ctx.run_dir / "export.zip" deadline = time.monotonic() + ctx.scenario.get("readyTimeoutSeconds", 30) pulled = False wait_until_activity_resumed( ctx.device, MAIN_ACTIVITY, ctx.scenario.get("readyTimeoutSeconds", 30), ) last_size = -1 stable_reads = 0 failure_feedback = "" while time.monotonic() < deadline: dump = ui_dump(ctx.device) failure_node = find_node_containing_text(dump, "主题导出失败") if failure_node is not None: failure_feedback = failure_node.get("text", "") size_result = adb( ctx.device, "shell", "stat", "-c", "%s", remote_path, timeout=10, check=False, ) try: size = int(size_result.stdout.strip()) except ValueError: size = 0 stable_reads = stable_reads + 1 if size > 0 and size == last_size else 0 last_size = size if stable_reads < 2: time.sleep(0.25) continue pull = run_command( ["adb", "-s", ctx.device, "pull", remote_path, str(local_path)], timeout=30, check=False, ) if pull.returncode == 0 and local_path.is_file() and local_path.stat().st_size > 0: pulled = True break time.sleep(0.5) valid = False asset_included = False if pulled: try: with zipfile.ZipFile(local_path) as archive: manifest = json.loads(archive.read(THEME_MANIFEST_PATH).decode("utf-8")) valid = manifest.get("formatVersion") == 1 and isinstance(manifest.get("config"), dict) entry = (manifest.get("assets") or {}).get("background.light") if isinstance(entry, str) and entry in archive.namelist(): expected = hashlib.sha256(THEME_ASSET_FIXTURE.read_bytes()).hexdigest() asset_included = hashlib.sha256(archive.read(entry)).hexdigest() == expected except (OSError, KeyError, json.JSONDecodeError, zipfile.BadZipFile): valid = False adb(ctx.device, "shell", "rm", "-f", remote_path, timeout=10, check=False) if valid: ctx.artifacts["exportedTheme"] = "export.zip" ctx.records["theme_export_succeeded"] = valid ctx.records["theme_export_failed"] = not valid and bool(failure_feedback) ctx.records["theme_failure_feedback"] = bool(failure_feedback) if failure_feedback: ctx.state["theme_failure_feedback"] = failure_feedback ctx.records["theme_export_asset_included"] = asset_included ctx.records["ui_after"] = ui_dump(ctx.device) ASSERTIONS.update( { "reader_ready": lambda ctx: bool(ctx.records.get("reader_ready")), "reader_activity_visible": lambda ctx: bool(ctx.records.get("reader_activity_visible")), "reader_content_changed": lambda ctx: bool(ctx.records.get("reader_content_changed")), "rss_read_ready": lambda ctx: bool(ctx.records.get("rss_read_ready")), "rss_theme_changed": lambda ctx: bool(ctx.records.get("rss_theme_changed")), "bookshelf_ready": lambda ctx: bool(ctx.records.get("bookshelf_ready")), "reader_opened": lambda ctx: bool(ctx.records.get("reader_opened")), "returned_to_bookshelf": lambda ctx: bool(ctx.records.get("returned_to_bookshelf")), "source_manage_ready": lambda ctx: bool(ctx.records.get("source_manage_ready")), "source_edit_opened": lambda ctx: bool(ctx.records.get("source_edit_opened")), "returned_to_source_manage": lambda ctx: bool(ctx.records.get("returned_to_source_manage")), "theme_config_ready": lambda ctx: bool(ctx.records.get("theme_config_ready")), "theme_manage_opened": lambda ctx: bool(ctx.records.get("theme_manage_opened")), "theme_save_succeeded": lambda ctx: bool(ctx.records.get("theme_save_succeeded")), "theme_export_succeeded": lambda ctx: bool(ctx.records.get("theme_export_succeeded")), "theme_day_background_copied": lambda ctx: bool(ctx.records.get("theme_day_background_copied")), "theme_save_asset_included": lambda ctx: bool(ctx.records.get("theme_save_asset_included")), "theme_export_asset_included": lambda ctx: bool(ctx.records.get("theme_export_asset_included")), "theme_save_failed": lambda ctx: bool(ctx.records.get("theme_save_failed")), "theme_export_failed": lambda ctx: bool(ctx.records.get("theme_export_failed")), "theme_failure_feedback": lambda ctx: bool(ctx.records.get("theme_failure_feedback")), "theme_open_input_stream_reached": lambda ctx: bool(ctx.records.get("theme_open_input_stream_reached")), "theme_content_failure_armed": lambda ctx: bool(ctx.records.get("theme_content_failure_armed")), "theme_saved_theme_applied": lambda ctx: bool(ctx.records.get("theme_saved_theme_applied")), "theme_saved_theme_deleted": lambda ctx: bool(ctx.records.get("theme_saved_theme_deleted")), "no_fatal_exception": lambda ctx: ctx.failure is None, } ) def build_capabilities() -> dict[str, Any]: return { "schemaVersion": 1, "entries": { name: {"allowedActions": sorted(handler.allowed_actions)} for name, handler in sorted(ENTRIES.items()) }, "actions": { name: {"params": handler.params} for name, handler in sorted(ACTIONS.items()) }, "assertions": sorted(ASSERTIONS), } def execute(args: argparse.Namespace) -> int: try: scenario = load_scenario(args.scenario) fixture = args.fixture or scenario["fixture"] fixture_data = validate_fixture(fixture) markers = {chapter["readyMarker"] for chapter in fixture_data["chapters"]} if scenario["readyMarker"] not in markers: raise ScenarioError("scenario readyMarker is not provided by the fixture") if args.action_repeat < 1 or args.action_repeat > 100: raise ScenarioError("--action-repeat must be between 1 and 100") if scenario["entry"]["type"] == "bookshelf" and args.action_repeat != 1: raise ScenarioError("--action-repeat is only supported by reader scenarios") except ScenarioError as error: print(json.dumps({"status": "failed", "exitCode": EXIT_SCENARIO, "error": str(error)}, ensure_ascii=False)) return EXIT_SCENARIO try: device = select_device(args.device) except (RuntimeError, CommandError, subprocess.TimeoutExpired, OSError) as error: session_id, run_dir, _ = create_session(args.scenario, fixture, args.device) summary = error_summary(session_id, args.scenario, fixture, EXIT_ENVIRONMENT, "environment", str(error)) write_json(run_dir / "summary.json", summary) print(run_dir / "summary.json") return EXIT_ENVIRONMENT session_id, run_dir, metadata = create_session(args.scenario, fixture, device) device_state: dict[str, Any] | None = None try: device_state = prepare_device(device) metadata["deviceStateBefore"] = device_state write_json(run_dir / "metadata.json", metadata) except (RuntimeError, CommandError, subprocess.TimeoutExpired, OSError) as error: summary = error_summary(session_id, args.scenario, fixture, EXIT_ENVIRONMENT, "environment", str(error)) write_json(run_dir / "summary.json", summary) print(run_dir / "summary.json") return EXIT_ENVIRONMENT build: dict[str, Any] = {"status": "not_started"} if args.skip_build: build = {"status": "skipped"} else: try: build, _ = build_and_install(device, run_dir) except (RuntimeError, CommandError, subprocess.TimeoutExpired) as error: if isinstance(error, CommandError): build = { "status": "failed", "artifact": "build.log" if (run_dir / "build.log").exists() else "install.log", } summary = error_summary(session_id, args.scenario, fixture, EXIT_BUILD, "build_or_install", str(error), build) cleanup_device(device) restore_device(device, device_state) write_json(run_dir / "summary.json", summary) print(run_dir / "summary.json") return EXIT_BUILD entry_handler = ENTRIES[scenario["entry"]["type"]] ctx = Context(device, run_dir, scenario, fixture_data, metadata, args.action_repeat) recorder = ScreenRecorder(device) if scenario.get("record") else None try: if entry_handler.type == "theme_config": # Activity result contracts can leave their external picker task # above Legado after a failed run. Close both implementations used # by this device before launching a deterministic theme scenario. for picker_package in ( "com.google.android.documentsui", "com.google.android.photopicker", ): adb( device, "shell", "am", "force-stop", picker_package, timeout=20, check=False, ) preferences = prepare_theme_asset(device, scenario) if entry_handler.type == "theme_config" else scenario.get("preferences") launch_fixture(device, fixture, entry_handler.type, session_id, preferences) ctx.note_pid() entry_handler.prepare(ctx) if recorder is not None: recorder.start() for action in scenario["actions"]: ACTIONS[action["type"]].run(ctx, action) if recorder is not None: time.sleep(0.5) # tail so the settle after the last action is captured video = recorder.stop(run_dir) if video: ctx.artifacts["transition"] = video entry_handler.finalize(ctx) ctx.note_pid() logcat = collect_logcat(device, run_dir, ctx.app_pids, session_id) ctx.failure = parse_failure(extract_crash(logcat, PACKAGE, ctx.app_pids)) toast_signal = ( f"Toast already killed. pkg={PACKAGE}" in logcat or f"pkg={PACKAGE}" in logcat and "enqueueToast" in logcat ) open_input_stream_reached = "INJECTED_THEME_ASSET_OPEN" in logcat failure_armed = "INJECTED_THEME_ASSET_FAILURE_ARMED" in logcat ctx.records["theme_open_input_stream_reached"] = open_input_stream_reached ctx.records["theme_content_failure_armed"] = failure_armed if ( scenario.get("themeAsset") in {"missing-file", "missing-content-uri"} or scenario.get("expectThemeExportFailure") ): ctx.records["theme_failure_feedback"] = bool( ctx.records.get("theme_failure_feedback") or toast_signal ) if any(action["type"] == "save_current_theme" for action in scenario["actions"]): ctx.records["theme_save_failed"] = ( not ctx.records.get("theme_save_succeeded") and ctx.records["theme_failure_feedback"] ) if any(action["type"] == "export_current_theme" for action in scenario["actions"]): ctx.records["theme_export_failed"] = ( not ctx.records.get("theme_export_succeeded") and ctx.records["theme_failure_feedback"] ) metadata["themeOpenInputStreamReached"] = open_input_stream_reached metadata["themeContentFailureArmed"] = failure_armed metadata["themeFailureFeedbackObserved"] = bool(ctx.records.get("theme_failure_feedback")) for name in scenario["assertions"]: ctx.assertion_values[name] = ASSERTIONS[name](ctx) metadata.update( { "finishedAt": dt.datetime.now().astimezone().isoformat(timespec="milliseconds"), "appPids": ctx.app_pids, } ) write_json(run_dir / "metadata.json", metadata) summary, exit_code = make_summary_from_assertions( session_id=session_id, scenario=scenario, fixture=fixture, build=build, assertion_values=ctx.assertion_values, action_count=ctx.action_count, failure=ctx.failure, extra_artifacts=ctx.artifacts, ) except (RuntimeError, CommandError, subprocess.TimeoutExpired, OSError) as error: try: ctx.note_pid() collect_logcat(device, run_dir, ctx.app_pids, session_id) except Exception: pass summary = error_summary( session_id, args.scenario, fixture, EXIT_ENVIRONMENT, "environment", str(error), build, ) exit_code = EXIT_ENVIRONMENT finally: if recorder is not None: recorder.stop(run_dir) cleanup_device(device) restore_device(device, device_state) write_json(run_dir / "summary.json", summary) print(run_dir / "summary.json") return exit_code def parse_args(argv: Sequence[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) run_parser = subparsers.add_parser("run", help="build, install, and run one scenario") run_parser.add_argument("scenario", help="scenario ID, for example reader/page-turn") run_parser.add_argument("--device", help="adb device serial; optional when exactly one is ready") run_parser.add_argument("--fixture", help="override the scenario fixture ID") run_parser.add_argument("--action-repeat", type=int, default=1, help="repeat each domain action (default: 1)") run_parser.add_argument( "--skip-build", action="store_true", help="reuse the installed debug APK; skip Gradle build and install", ) subparsers.add_parser("list", help="print supported entries, actions, and assertions") return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv if argv is not None else sys.argv[1:]) if args.command == "run": return execute(args) if args.command == "list": print(json.dumps(build_capabilities(), ensure_ascii=False, indent=2)) return 0 return EXIT_SCENARIO if __name__ == "__main__": raise SystemExit(main())