#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """perf_harness.py -- LibreMail on-device performance-test harness (CLI entry point). Replicates the manual on-device perf methodology (2026-07-05, Pixel 10 Pro XL) as a repeatable, cross-platform, standard-library-only tool. Run one scenario at a time: python scripts/device-testing/perf_harness.py cold-open [opts] python scripts/device-testing/perf_harness.py message-open [opts] python scripts/device-testing/perf_harness.py back-nav [opts] python scripts/device-testing/perf_harness.py prefetch-ab [opts] python scripts/device-testing/perf_harness.py cross-provider [opts] python scripts/device-testing/perf_harness.py cold-fetch-ab [opts] # needs a debug build Common options: ``--serial`` (auto-detected if exactly one device), ``--count/-n``, ``--out``, ``--package``, ``--component``, ``--adb``, and ``--dry-run`` (print the exact command plan without touching device state -- use this to review a run before it happens). Device safety is enforced by :mod:`adb` (allow-list of adb subcommands + deny-list on shell commands; the only sanctioned mutation is clearing LibreMail's own ``cache/``). See the README for the full guarantees and the required device state. """ from __future__ import annotations import argparse import datetime as _dt import os import sys import time from typing import List, Optional # Flat-layout imports: this file's directory is on sys.path[0] when run as a script, and the # tests inject it explicitly. (The dir name contains a hyphen, so it is not an importable # package -- hence flat modules rather than `python -m`.) import breadcrumbs import fetchgate import report import scenarios from adb import Adb, DEFAULT_COMPONENT, DEFAULT_PACKAGE SCENARIOS = ( "cold-open", "message-open", "back-nav", "prefetch-ab", "cross-provider", "cold-fetch-ab", ) _DEFAULT_COUNTS = { "cold-open": 5, "message-open": 6, "back-nav": 6, "prefetch-ab": 3, "cross-provider": 8, "cold-fetch-ab": 3, } def _configure_utf8_io() -> None: """Force UTF-8 for this process and its console (issue #392). Non-ASCII sender/subject text in a uiautomator dump -- and the harness's own output on a Windows cp1252 console -- otherwise mojibake or raise ``UnicodeEncodeError``. adb output is already decoded UTF-8 in :meth:`adb.Adb.run`; this covers the interpreter and console. """ os.environ.setdefault("PYTHONUTF8", "1") os.environ.setdefault("PYTHONIOENCODING", "utf-8") for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if reconfigure is not None: try: reconfigure(encoding="utf-8", errors="replace") except (ValueError, OSError): # pragma: no cover - stream already detached pass def _make_logger(log_path: Optional[str]): handle = open(log_path, "a", encoding="utf-8") if log_path else None def log(msg: str) -> None: stamp = _dt.datetime.now().strftime("%H:%M:%S") line = f"[{stamp}] {msg}" print(line, flush=True) if handle: handle.write(line + "\n") handle.flush() return log, handle def resolve_serial(adb_path: str, requested: Optional[str]) -> Optional[str]: """Return the serial to use, auto-detecting when exactly one device is attached.""" probe = Adb(serial=None, adb_path=adb_path) serials = probe.devices() if requested: if serials and requested not in serials: print( f"warning: requested serial {requested!r} not in attached devices {serials}", file=sys.stderr, ) return requested if len(serials) == 1: return serials[0] if not serials: raise SystemExit("no devices attached; connect one or pass --serial") raise SystemExit(f"multiple devices attached {serials}; pass --serial to choose one") def _run_dir(out_root: str, scenario: str) -> str: stamp = _dt.datetime.now().strftime("%Y%m%d-%H%M%S") path = os.path.join(out_root, f"{stamp}-{scenario}") os.makedirs(path, exist_ok=True) return path def _write_filtered_extract(raw_path: str, extract_path: str) -> int: """Write the ImapPerf/MailReader/Reader/MailBackfiller subset of the raw log; return count.""" if not os.path.exists(raw_path): return 0 count = 0 with open(raw_path, "r", encoding="utf-8", errors="replace") as src, open( extract_path, "w", encoding="utf-8", newline="\n" ) as dst: for line in src: if breadcrumbs.is_perf_line(line): dst.write(line if line.endswith("\n") else line + "\n") count += 1 return count def _render_sections(scenario: str, results, adb: Adb, count: int) -> List[str]: if scenario == "cold-open": return [report.render_cold_open(results)] if scenario == "message-open": return [report.render_message_open("Message open (uncached)", results)] if scenario == "back-nav": return [report.render_back_nav(results)] if scenario == "prefetch-ab": return [ report.render_message_open( "Condition A - Fetch all on Wi-Fi (prefetch ON)", results["conditionA"] ), report.render_message_open( "Condition B - Always on-demand (prefetch OFF)", results["conditionB"] ), _ab_comparison(results), ] if scenario == "cross-provider": sections = [] for ref, rows in sorted(results.items()): sections.append(report.render_message_open(f"Provider {ref}", rows)) if not sections: sections.append("## Cross-provider\n\nNo opens captured.\n") return sections if scenario == "cold-fetch-ab": return [report.render_cold_fetch_ab(results)] return [] def _ab_comparison(results: dict) -> str: def median_took(rows): vals = [r.took_ms for r in rows if not r.skipped and not r.cached and r.took_ms] return report.aggregate(vals).median a = median_took(results["conditionA"]) b = median_took(results["conditionB"]) return ( "## Prefetch A/B comparison\n\n" f"- Condition A (prefetch ON) median openMessage: {report._ms(a)} ms\n" f"- Condition B (prefetch OFF) median openMessage: {report._ms(b)} ms\n" ) def _run_scenario(scenario: str, adb: Adb, tailer, gate, args, log): package, component, count = args.package, args.component, args.count if scenario == "cold-open": return scenarios.cold_open(adb, component, count, log) if scenario == "message-open": return scenarios.message_open(adb, package, tailer, count, log) if scenario == "back-nav": return scenarios.back_nav(adb, package, count, log) if scenario == "prefetch-ab": return scenarios.prefetch_ab(adb, package, component, tailer, count, log) if scenario == "cross-provider": return scenarios.cross_provider(adb, package, tailer, count, log) if scenario == "cold-fetch-ab": return scenarios.cold_fetch_ab(adb, package, component, tailer, gate, count, log) raise SystemExit(f"unknown scenario {scenario!r}") def build_arg_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="perf_harness.py", description="LibreMail on-device performance-test harness (stdlib-only).", ) p.add_argument("scenario", choices=SCENARIOS, help="which scenario to run") p.add_argument("--serial", help="device serial (auto-detected if exactly one attached)") p.add_argument("--package", default=DEFAULT_PACKAGE, help="target app package") p.add_argument("--component", default=DEFAULT_COMPONENT, help="launcher component") p.add_argument("--adb", default="adb", dest="adb_path", help="path to the adb executable") p.add_argument( "-n", "--count", type=int, default=None, help="samples/runs (per condition for prefetch-ab); scenario-specific default", ) p.add_argument( "--out", default=os.path.join(os.getcwd(), "device-perf-runs"), help="output root; a timestamped subdir is created per run", ) p.add_argument( "--dry-run", action="store_true", help="print the command plan without changing device state", ) return p def main(argv: Optional[List[str]] = None) -> int: _configure_utf8_io() args = build_arg_parser().parse_args(argv) if args.count is None: args.count = _DEFAULT_COUNTS[args.scenario] serial = None if args.dry_run else resolve_serial(args.adb_path, args.serial) if args.dry_run and args.serial: serial = args.serial run_dir = _run_dir(args.out, args.scenario) driver_log_path = os.path.join(run_dir, "driver.log") log, log_handle = _make_logger(driver_log_path) raw_path = os.path.join(run_dir, "session-raw.log") log(f"scenario={args.scenario} serial={serial} count={args.count} dry_run={args.dry_run}") log(f"run dir: {run_dir}") adb = Adb( serial=serial, package=args.package, adb_path=args.adb_path, dry_run=args.dry_run, logger=log, ) gate = fetchgate.FetchGate(adb, log=log) logcat_proc = None raw_handle = None tailer: scenarios.LogTailer = scenarios._NullTailer() try: if not args.dry_run: adb.clear_logcat() # Binary handle: the logcat child writes raw bytes to this fd. raw_handle = open(raw_path, "wb") logcat_proc = adb.start_logcat(raw_handle) tailer = scenarios.LogTailer(raw_path) scenarios.ensure_awake(adb, log) time.sleep(1.0) results = _run_scenario(args.scenario, adb, tailer, gate, args, log) sections = _render_sections(args.scenario, results, adb, args.count) finally: Adb.stop_logcat(logcat_proc) if raw_handle: raw_handle.close() if not args.dry_run: adb.stay_on(False) # restore default screen-timeout behaviour if log_handle: log_handle.flush() metadata = { "scenario": args.scenario, "timestamp": _dt.datetime.now().isoformat(timespec="seconds"), "serial": serial or "(dry-run)", "package": args.package, "count": args.count, } document = report.build_document(metadata, sections) tables_path = os.path.join(run_dir, "timing-tables.md") with open(tables_path, "w", encoding="utf-8", newline="\n") as fh: fh.write(document) extract_count = 0 if not args.dry_run: extract_count = _write_filtered_extract( raw_path, os.path.join(run_dir, "perf-extract.log") ) log(f"wrote {tables_path} (filtered {extract_count} breadcrumb lines)") if log_handle: log_handle.close() return 0 if __name__ == "__main__": raise SystemExit(main())