311 lines
12 KiB
Python
311 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise Qlyra's shared Flutter engine across Android launcher/VIEW tasks.
|
|
|
|
Uses only Python's standard library and an installed adb. It preserves app data
|
|
and does not force-stop the process. JSON output contains counts and identities;
|
|
screenshots, UI XML and detailed device logs are saved only after a failure.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from xml.etree import ElementTree
|
|
|
|
|
|
PACKAGE = "ru.qlyra.app"
|
|
HOST_NAMES = {"MainActivity", "IntentEntry", "MinimalIcon"}
|
|
REMOTE_UI = "/sdcard/qlyra-resume-smoke.xml"
|
|
UI_TIMEOUT = 15.0
|
|
|
|
|
|
class SmokeFailure(Exception):
|
|
pass
|
|
|
|
|
|
class Adb:
|
|
def __init__(self, executable, serial):
|
|
self.prefix = [executable] + (["-s", serial] if serial else [])
|
|
|
|
def run(self, *args, timeout=25):
|
|
try:
|
|
result = subprocess.run(
|
|
self.prefix + list(args), capture_output=True, timeout=timeout
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as error:
|
|
raise SmokeFailure("ADB command could not complete") from error
|
|
if result.returncode:
|
|
raise SmokeFailure("ADB command failed: " + " ".join(args[:3]))
|
|
return result.stdout
|
|
|
|
def text(self, *args, **kwargs):
|
|
return self.run(*args, **kwargs).decode("utf-8", errors="replace")
|
|
|
|
|
|
def parse_hosts(activity_dump):
|
|
"""Read only dumpsys activity's top TASK/ACTIVITY headers, not view details."""
|
|
task_id = None
|
|
hosts = []
|
|
for line in activity_dump.splitlines():
|
|
task = re.match(r"^\s{0,2}TASK \S+ .*?\bid=(\d+)\b", line)
|
|
if task:
|
|
task_id = int(task.group(1))
|
|
continue
|
|
activity = re.match(
|
|
r"^\s{0,4}ACTIVITY (\S+) \S+ pid=(\d+)\b", line
|
|
)
|
|
if not activity:
|
|
continue
|
|
component = activity.group(1)
|
|
if "/" not in component:
|
|
continue
|
|
package, class_name = component.split("/", 1)
|
|
if package != PACKAGE or class_name.rsplit(".", 1)[-1] not in HOST_NAMES:
|
|
continue
|
|
hosts.append(
|
|
{"task_id": task_id, "pid": int(activity.group(2)),
|
|
"component": component}
|
|
)
|
|
return hosts
|
|
|
|
|
|
def count_app_labels(ui_xml):
|
|
root = ElementTree.fromstring(ui_xml)
|
|
return sum(
|
|
1
|
|
for node in root.iter("node")
|
|
if node.get("package") == PACKAGE
|
|
and (node.get("text", "").strip() or node.get("content-desc", "").strip())
|
|
)
|
|
|
|
|
|
class ResumeSmoke:
|
|
def __init__(self, adb, output, cycles, background_seconds):
|
|
self.adb = adb
|
|
self.output = output
|
|
self.cycles = cycles
|
|
self.background_seconds = background_seconds
|
|
self.baseline = None
|
|
self.launcher_component = PACKAGE + "/.MainActivity"
|
|
self.latest_ui = ""
|
|
self.log_marker = "resume-smoke-" + uuid.uuid4().hex
|
|
self.log_file = None
|
|
self.log_process = None
|
|
self.report = {"status": "running", "cycles": cycles, "checks": []}
|
|
self.failures = []
|
|
|
|
def start_logs(self):
|
|
self.log_file = (self.output / "logcat.raw.txt").open("wb")
|
|
try:
|
|
self.log_process = subprocess.Popen(
|
|
self.adb.prefix + ["logcat", "-v", "threadtime"],
|
|
stdout=self.log_file, stderr=subprocess.STDOUT,
|
|
)
|
|
except OSError as error:
|
|
raise SmokeFailure("Could not start device log capture") from error
|
|
self.adb.run("shell", "log", "-p", "i", "-t", "QlyraResumeSmoke", self.log_marker)
|
|
|
|
def stop_logs(self):
|
|
if self.log_process is not None:
|
|
self.log_process.terminate()
|
|
try:
|
|
self.log_process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
self.log_process.kill()
|
|
self.log_process.wait(timeout=5)
|
|
if self.log_file is not None:
|
|
self.log_file.close()
|
|
|
|
def scenario_log(self):
|
|
raw_path = self.output / "logcat.raw.txt"
|
|
if not raw_path.exists():
|
|
return ""
|
|
if self.log_file is not None and not self.log_file.closed:
|
|
self.log_file.flush()
|
|
lines = raw_path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
for index, line in enumerate(lines):
|
|
if self.log_marker in line:
|
|
return "\n".join(lines[index:]) + "\n"
|
|
raise SmokeFailure("Device log marker was not captured")
|
|
|
|
def read_ui(self, timeout=12):
|
|
# Remove only this script's temporary dump so a failed dump cannot pass
|
|
# using the preceding screen's semantics.
|
|
self.adb.run("shell", "rm", "-f", REMOTE_UI)
|
|
self.adb.run("shell", "uiautomator", "dump", "--compressed", REMOTE_UI,
|
|
timeout=timeout)
|
|
self.latest_ui = self.adb.text("shell", "cat", REMOTE_UI)
|
|
return count_app_labels(self.latest_ui)
|
|
|
|
def wait_for_ui(self):
|
|
deadline = time.monotonic() + UI_TIMEOUT
|
|
self.latest_ui = ""
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
count = self.read_ui(timeout=max(1, min(12, deadline - time.monotonic())))
|
|
if count:
|
|
return count
|
|
except (SmokeFailure, ElementTree.ParseError):
|
|
pass
|
|
time.sleep(0.3)
|
|
return 0
|
|
|
|
def hosts(self):
|
|
return parse_hosts(self.adb.text("shell", "dumpsys", "activity", PACKAGE))
|
|
|
|
def checkpoint(self, name, *, baseline=False):
|
|
count = self.wait_for_ui()
|
|
hosts = self.hosts()
|
|
pid_text = self.adb.text("shell", "pidof", PACKAGE).strip()
|
|
pids = [int(value) for value in pid_text.split() if value.isdigit()]
|
|
check = {"stage": name, "host_count": len(hosts), "hosts": hosts,
|
|
"pids": pids, "app_label_count": count}
|
|
self.report["checks"].append(check)
|
|
errors = []
|
|
if len(hosts) != 1 or hosts[0].get("task_id") is None:
|
|
errors.append(name + ": expected one live Flutter host in one task")
|
|
if len(pids) != 1:
|
|
errors.append(name + ": expected one application process")
|
|
if count == 0:
|
|
errors.append(name + ": no visible application text or accessibility labels")
|
|
if baseline:
|
|
if errors:
|
|
raise SmokeFailure("; ".join(errors))
|
|
self.baseline = hosts[0]
|
|
self.report.update({"pid": self.baseline["pid"],
|
|
"task_id": self.baseline["task_id"]})
|
|
elif self.baseline:
|
|
if pids != [self.baseline["pid"]]:
|
|
errors.append(name + ": application process changed")
|
|
if len(hosts) == 1 and any(
|
|
hosts[0][key] != self.baseline[key] for key in ("pid", "task_id")
|
|
):
|
|
errors.append(name + ": original Flutter host task changed")
|
|
self.failures.extend(errors)
|
|
return errors
|
|
|
|
def home(self):
|
|
self.adb.run("shell", "input", "keyevent", "KEYCODE_HOME")
|
|
time.sleep(self.background_seconds)
|
|
|
|
def launcher(self):
|
|
self.adb.run("shell", "am", "start", "-W", "-a", "android.intent.action.MAIN",
|
|
"-c", "android.intent.category.LAUNCHER", "-f", "0x10200000",
|
|
"-n", self.launcher_component)
|
|
|
|
def run(self):
|
|
if self.adb.text("get-state").strip() != "device":
|
|
raise SmokeFailure("ADB device is not ready")
|
|
resolved = self.adb.text("shell", "cmd", "package", "resolve-activity", "--brief",
|
|
"-a", "android.intent.action.MAIN", "-c",
|
|
"android.intent.category.LAUNCHER", "-p", PACKAGE)
|
|
for line in resolved.splitlines():
|
|
component = line.strip()
|
|
if component.startswith(PACKAGE + "/") and \
|
|
component.rsplit(".", 1)[-1] in HOST_NAMES:
|
|
self.launcher_component = component
|
|
break
|
|
else:
|
|
raise SmokeFailure("Application launcher component could not be resolved")
|
|
self.start_logs()
|
|
self.launcher()
|
|
self.checkpoint("initial-launch", baseline=True)
|
|
for cycle in range(1, self.cycles + 1):
|
|
self.home()
|
|
self.adb.run("shell", "am", "start", "-W", "-a", "android.intent.action.VIEW",
|
|
"-c", "android.intent.category.BROWSABLE", "-d",
|
|
"qlyra://resume-smoke", "-f", "0x18000000",
|
|
"-n", PACKAGE + "/.IntentEntry")
|
|
# Keep going if VIEW exposed a second host, so diagnostics capture
|
|
# the original task's return rather than only the duplicate task.
|
|
self.checkpoint("cycle-%d-view" % cycle)
|
|
self.home()
|
|
self.adb.run("shell", "am", "start", "-W", "--task",
|
|
str(self.baseline["task_id"]), "-f", "0x00200000",
|
|
"-a", "android.intent.action.MAIN", "-c",
|
|
"android.intent.category.LAUNCHER", "-n",
|
|
self.launcher_component)
|
|
self.checkpoint("cycle-%d-original-task-return" % cycle)
|
|
if self.failures:
|
|
raise SmokeFailure("; ".join(self.failures))
|
|
self.home()
|
|
self.launcher()
|
|
self.checkpoint("cycle-%d-launcher-return" % cycle)
|
|
if self.failures:
|
|
raise SmokeFailure("; ".join(self.failures))
|
|
# Give the streaming log reader time to receive the last lifecycle event.
|
|
time.sleep(0.3)
|
|
self.stop_logs()
|
|
log = self.scenario_log()
|
|
warnings = [line for line in log.splitlines()
|
|
if "FlutterFragment" in line and
|
|
("evicted" in line or "called after detach" in line)]
|
|
if warnings:
|
|
raise SmokeFailure("FlutterFragment eviction/detached lifecycle warning detected")
|
|
if "FATAL EXCEPTION" in log and ("Process: " + PACKAGE) in log:
|
|
raise SmokeFailure("Application fatal exception detected")
|
|
self.report["status"] = "passed"
|
|
|
|
def capture_failure(self):
|
|
self.report["diagnostic_errors"] = []
|
|
commands = {
|
|
"failure-screen.png": ("exec-out", "screencap", "-p"),
|
|
"failure-activity.txt": ("shell", "dumpsys", "activity", PACKAGE),
|
|
"failure-activities.txt": ("shell", "dumpsys", "activity", "activities"),
|
|
}
|
|
for name, command in commands.items():
|
|
try:
|
|
(self.output / name).write_bytes(self.adb.run(*command))
|
|
except SmokeFailure:
|
|
self.report["diagnostic_errors"].append(name)
|
|
try:
|
|
self.read_ui()
|
|
except (SmokeFailure, ElementTree.ParseError):
|
|
self.report["diagnostic_errors"].append("failure-ui.xml")
|
|
if self.latest_ui:
|
|
(self.output / "failure-ui.xml").write_text(self.latest_ui, encoding="utf-8")
|
|
self.stop_logs()
|
|
try:
|
|
log = self.scenario_log()
|
|
except SmokeFailure:
|
|
log = ""
|
|
self.report["diagnostic_errors"].append("failure-logcat.txt")
|
|
(self.output / "failure-logcat.txt").write_text(log, encoding="utf-8")
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--adb", default="adb", help="Path to adb executable")
|
|
parser.add_argument("--serial", help="ADB device serial (required with multiple devices)")
|
|
parser.add_argument("--output", type=Path, default=Path("artifacts/android-resume-smoke"))
|
|
parser.add_argument("--cycles", type=int, default=3)
|
|
parser.add_argument("--background-seconds", type=float, default=1.0)
|
|
args = parser.parse_args(argv)
|
|
if args.cycles < 1 or args.background_seconds < 0:
|
|
parser.error("cycles must be positive and background-seconds cannot be negative")
|
|
args.output.mkdir(parents=True, exist_ok=True)
|
|
smoke = ResumeSmoke(Adb(args.adb, args.serial), args.output,
|
|
args.cycles, args.background_seconds)
|
|
exit_code = 0
|
|
try:
|
|
smoke.run()
|
|
except (SmokeFailure, OSError) as error:
|
|
exit_code = 1
|
|
smoke.report.update({"status": "failed", "error": str(error)})
|
|
smoke.capture_failure()
|
|
finally:
|
|
smoke.stop_logs()
|
|
(args.output / "logcat.raw.txt").unlink(missing_ok=True)
|
|
report = json.dumps(smoke.report, ensure_ascii=True, separators=(",", ":"))
|
|
(args.output / "report.json").write_text(report + "\n", encoding="utf-8")
|
|
print(report)
|
|
return exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|