|
|
@@ -1,2279 +0,0 @@
|
|
|
-#!/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("<?xml")
|
|
|
- end = dump.rfind("</hierarchy>")
|
|
|
- if start < 0 or end < 0:
|
|
|
- return []
|
|
|
- try:
|
|
|
- root = ET.fromstring(dump[start : end + len("</hierarchy>")])
|
|
|
- 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 <one-line page text>" 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, "toggleReaderDayNight")
|
|
|
- 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())
|
5151561