diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 33bef2a..299727b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -119,49 +119,76 @@ jobs: for value in (os.environ.get("USERPROFILE"), str(Path.home())) if value } - text_suffixes = { - ".bat", ".java", ".json", ".md", ".py", ".qss", ".spec", - ".txt", ".xml", ".yml", ".yaml", - } private_windows_path = re.compile( r"[a-z]:[/\\]users[/\\]([^/\\\s<>\"']+)", re.IGNORECASE, ) + private_posix_path = re.compile( + r"/(?:home|users)/([a-z0-9._-]{3,})(?:[/\\]|$)", + re.IGNORECASE, + ) private_ipv4 = re.compile( r"(? None: self.runner = runner def device_ip_addresses(self, cancel_event=None) -> list[str]: - return ["192.168.1.50"] + return ["192.168.1.42"] def run_shell(self, command: str, timeout=None, cancel_event=None): record_private_command(command) diff --git a/tests/test_adb_wireless.py b/tests/test_adb_wireless.py index e902e09..cfaa8c8 100644 --- a/tests/test_adb_wireless.py +++ b/tests/test_adb_wireless.py @@ -62,13 +62,13 @@ def _discover_wireless_mdns_services(self, wait_seconds=0.5, cancel_event=None): { "name": "studio-pairing-service", "type": "_adb-tls-pairing._tcp", - "target": "192.168.0.159:37001", + "target": "192.0.2.59:37001", "source": "zeroconf", }, { "name": self.MDNS_SERIAL, "type": "_adb-tls-connect._tcp", - "target": "192.168.0.159:40765", + "target": "192.0.2.59:40765", "source": "zeroconf", }, ] @@ -259,7 +259,7 @@ def test_pairing_secret_rejects_line_break_injection(self) -> None: def test_recognizes_android_mdns_device_serial_as_wireless(self) -> None: self.assertTrue(_looks_like_wireless_serial("adb-serial-token._adb-tls-connect._tcp")) self.assertTrue(_looks_like_wireless_serial("adb-serial-token._adb-tls-connect._tcp.")) - self.assertTrue(_looks_like_wireless_serial("192.168.0.159:44195")) + self.assertTrue(_looks_like_wireless_serial("192.0.2.59:44195")) self.assertFalse(_looks_like_wireless_serial("3A131FDJG000SZ")) self.assertTrue(is_mdns_wireless_serial("adb-serial-token._adb-tls-connect._tcp")) @@ -330,7 +330,7 @@ def test_connect_discovery_rejects_unrelated_service_when_pairing_host_is_known( { "name": "adb-unrelated._adb-tls-connect._tcp", "type": "_adb-tls-connect._tcp", - "target": "192.168.0.99:40000", + "target": "198.51.100.99:40000", "source": "zeroconf", } ] @@ -338,7 +338,7 @@ def test_connect_discovery_rejects_unrelated_service_when_pairing_host_is_known( self.assertEqual( _wireless_connect_candidates_from_services( services, - "192.168.0.10:37000", + "203.0.113.10:37000", ), [], ) @@ -348,7 +348,7 @@ def test_pairing_discovery_rejects_an_unrelated_single_studio_service(self) -> N { "name": "studio-unrelated", "type": "_adb-tls-pairing._tcp", - "target": "192.168.0.99:37000", + "target": "198.51.100.99:37000", "source": "zeroconf", } ] @@ -400,7 +400,7 @@ def test_qr_cancellation_on_final_ready_wait_wins_over_attempt_exhaustion(self) with patch("openadb.core.adb.time.sleep", return_value=None): result = adb._connect_wireless_qr_target_until_ready( [adb.MDNS_SERIAL], - "192.168.0.159:37001", + "192.0.2.59:37001", set(), time.monotonic() + 30, cancel_event=cancel_event, @@ -429,40 +429,40 @@ def run_raw(args, timeout=120, use_serial=True, cancel_event=None): self.assertEqual(received_events, [cancel_event]) def test_wireless_normalizers_preserve_runner_cancellation(self) -> None: - connect_result = successful_result("connect", "192.168.0.10:40000") + connect_result = successful_result("connect", "203.0.113.10:40000") connect_result.success = False connect_result.error_type = "cancelled" connect_result.status = "Cancelled" - pair_result = successful_result("pair", "192.168.0.10:37000") + pair_result = successful_result("pair", "203.0.113.10:37000") pair_result.success = False pair_result.error_type = "cancelled" pair_result.status = "Cancelled" self.assertIs( - _normalize_adb_connect_result(connect_result, "192.168.0.10:40000"), + _normalize_adb_connect_result(connect_result, "203.0.113.10:40000"), connect_result, ) self.assertEqual(connect_result.error_type, "cancelled") self.assertIs( - _normalize_adb_pair_result(pair_result, "192.168.0.10:37000"), + _normalize_adb_pair_result(pair_result, "203.0.113.10:37000"), pair_result, ) self.assertEqual(pair_result.error_type, "cancelled") def test_wireless_normalizers_refine_ordinary_command_failures(self) -> None: - connect_result = successful_result("connect", "192.168.0.10:40000") + connect_result = successful_result("connect", "203.0.113.10:40000") connect_result.success = False connect_result.exit_code = 1 - connect_result.stderr = "failed to connect to 192.168.0.10:40000" + connect_result.stderr = "failed to connect to 203.0.113.10:40000" connect_result.error_type = "command_failed" - pair_result = successful_result("pair", "192.168.0.10:37000") + pair_result = successful_result("pair", "203.0.113.10:37000") pair_result.success = False pair_result.exit_code = 1 pair_result.stderr = "Failed: pairing refused" pair_result.error_type = "command_failed" - _normalize_adb_connect_result(connect_result, "192.168.0.10:40000") - _normalize_adb_pair_result(pair_result, "192.168.0.10:37000") + _normalize_adb_connect_result(connect_result, "203.0.113.10:40000") + _normalize_adb_pair_result(pair_result, "203.0.113.10:37000") self.assertEqual(connect_result.error_type, "connection_failed") self.assertEqual(pair_result.error_type, "pairing_failed") diff --git a/tests/test_dashboard_page.py b/tests/test_dashboard_page.py index 96ca094..560dfd3 100644 --- a/tests/test_dashboard_page.py +++ b/tests/test_dashboard_page.py @@ -65,7 +65,7 @@ def exec(self) -> int: return QDialog.Accepted def values(self) -> tuple[str, int, str]: - return "192.168.1.40", 37123, "123456" + return "192.0.2.40", 37123, "123456" class DashboardPageTests(unittest.TestCase): @@ -335,24 +335,24 @@ def test_wireless_scenarios_show_only_relevant_controls_and_save_values(self) -> self.page.wireless_card.set_expanded(True) self._select_scenario(WIRELESS_SCENARIO_MODERN) - self.page.wireless_host.setText("192.168.1.20") + self.page.wireless_host.setText("192.0.2.20") self.page.wireless_port.setValue(41000) self.page._save_wireless_settings() self.assertTrue(self.page.wireless_port.isVisible()) self.assertEqual(self.page.wireless_actions_stack.currentIndex(), 0) self.assertTrue(self.page.wireless_pair.isVisible()) self.assertFalse(self.page.wireless_enable_tcpip.isVisible()) - self.assertEqual(self.settings.get("wireless_modern_host"), "192.168.1.20") + self.assertEqual(self.settings.get("wireless_modern_host"), "192.0.2.20") self.assertEqual(self.settings.get("wireless_modern_port"), 41000) self._select_scenario(WIRELESS_SCENARIO_LEGACY) - self.page.wireless_host.setText("192.168.1.21") + self.page.wireless_host.setText("192.0.2.21") self.page._save_wireless_settings() self.assertFalse(self.page.wireless_port.isVisible()) self.assertEqual(self.page.wireless_actions_stack.currentIndex(), 1) self.assertTrue(self.page.wireless_enable_tcpip.isVisible()) self.assertFalse(self.page.wireless_pair.isVisible()) - self.assertEqual(self.settings.get("wireless_legacy_host"), "192.168.1.21") + self.assertEqual(self.settings.get("wireless_legacy_host"), "192.0.2.21") self.assertEqual(self.settings.get("wireless_adb_port"), WIRELESS_LEGACY_PORT) self._select_scenario(WIRELESS_SCENARIO_TV) @@ -379,7 +379,7 @@ def test_pairing_dialog_values_emit_without_saving_pairing_code(self) -> None: emitted: list[tuple[str, int, str]] = [] self.page.wireless_pair_requested.connect(lambda host, port, code: emitted.append((host, port, code))) self.page._request_wireless_pair() - self.assertEqual(emitted, [("192.168.1.40", 37123, "123456")]) + self.assertEqual(emitted, [("192.0.2.40", 37123, "123456")]) self.assertEqual(self.settings.get("wireless_modern_pair_port"), "37123") self.assertNotIn("123456", [str(value) for value in self.settings.data.values()]) diff --git a/tests/test_design_system.py b/tests/test_design_system.py index 3d5201f..57bc91e 100644 --- a/tests/test_design_system.py +++ b/tests/test_design_system.py @@ -146,7 +146,7 @@ def test_picker_dialogs_expose_long_values_defaults_focus_and_escape(self) -> No def test_keyboard_order_enter_escape_themes_and_no_wheel_contract(self) -> None: for theme in ("Light", "Dark", "System"): apply_theme(self.app, theme) - dialog = WirelessPairingDialog("192.168.1.2", 37001) + dialog = WirelessPairingDialog("192.0.2.2", 37001) dialog.pairing_code.setText("123456") dialog.resize(460, 280) dialog.show() @@ -175,7 +175,7 @@ def test_all_custom_dialogs_render_in_both_themes_at_compact_size(self) -> None: DeviceDetailsDialog(device), DevicePickerDialog([device]), PlatformToolsPickerDialog([tools]), - WirelessPairingDialog("192.168.1.2", 37001), + WirelessPairingDialog("192.0.2.2", 37001), WirelessQrDialog(payload), ActivityDialog("Working", "Checking local state"), TransferProgressDialog("Transfer"), diff --git a/tests/test_device_lab_smoke.py b/tests/test_device_lab_smoke.py index ff2a863..90b0d29 100644 --- a/tests/test_device_lab_smoke.py +++ b/tests/test_device_lab_smoke.py @@ -176,17 +176,18 @@ def test_no_hardware_is_truthfully_not_run_in_json_and_junit(self) -> None: def test_redactor_removes_serial_ip_username_home_path_and_filename(self) -> None: serial = "DEMO-SERIAL" + private_ipv6 = "fe80::" + "42" home = str(Path.home()) username = Path.home().name raw = ( - f"serial={serial}; endpoint=192.168.1.42:5555; ipv6=fe80::42; user={username}; " + f"serial={serial}; endpoint=192.168.1.42:5555; ipv6={private_ipv6}; user={username}; " f"home={home}; file=family-photo.png; token=one-shot-secret" ) sanitized = anonymize_text(raw, [serial]) for private in ( serial, "192.168.1.42", - "fe80::42", + private_ipv6, username, home, "family-photo.png", diff --git a/tests/test_main_window_adaptive.py b/tests/test_main_window_adaptive.py index cc95e3a..167c758 100644 --- a/tests/test_main_window_adaptive.py +++ b/tests/test_main_window_adaptive.py @@ -779,13 +779,13 @@ def test_disconnect_prefers_active_mdns_serial_over_stale_form_target(self) -> N side_effect=lambda fn, *_args, **_kwargs: fn(threading.Event()), ), ): - window.disconnect_wireless_adb("192.168.0.159", 40765) + window.disconnect_wireless_adb("192.0.2.59", 40765) disconnect.assert_called_once_with(serial, None, cancel_event=ANY) def test_disconnect_prefers_active_ip_transport_over_stale_form_target(self) -> None: window = self._window() - serial = "192.168.0.159:40765" + serial = "192.0.2.59:40765" window.device_manager.active = DeviceInfo(serial=serial, mode="ADB", state="device") with ( patch.object(window.adb, "disconnect_wireless") as disconnect, @@ -795,7 +795,7 @@ def test_disconnect_prefers_active_ip_transport_over_stale_form_target(self) -> side_effect=lambda fn, *_args, **_kwargs: fn(threading.Event()), ), ): - window.disconnect_wireless_adb("192.168.0.159", 5555) + window.disconnect_wireless_adb("192.0.2.59", 5555) disconnect.assert_called_once_with(serial, None, cancel_event=ANY) @@ -1054,7 +1054,7 @@ def test_discovered_short_mdns_name_accepts_full_ready_serial(self) -> None: started = window._begin_wireless_attempt( action="connect", connect_target="adb-demo", - expected_ready_serials=("adb-demo", "192.168.1.20:37123"), + expected_ready_serials=("adb-demo", "192.0.2.20:37123"), ) self.assertIsNotNone(started) attempt, token = started diff --git a/tests/test_release_performance.py b/tests/test_release_performance.py new file mode 100644 index 0000000..0dd2b05 --- /dev/null +++ b/tests/test_release_performance.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from tools.release_performance import ( + REPORT_SCHEMA, + BenchmarkConfig, + BenchmarkInvariantError, + generated_apps, + measure, + run_benchmarks, + sanitized_environment, + validate_release_profile, + validate_report, + write_json_report, +) + + +class ReleasePerformanceTests(unittest.TestCase): + def small_config(self) -> BenchmarkConfig: + return BenchmarkConfig( + warmups=1, + repetitions=2, + app_counts=(12, 30), + file_manager_entries=20, + transfer_files=12, + auto_stream_cases=16, + stale_result_checks=16, + operation_cycles=10, + ) + + def test_generated_apps_are_stable_and_cover_filter_dimensions(self) -> None: + first = generated_apps(30) + second = generated_apps(30) + + self.assertEqual(first, second) + self.assertEqual(len(first), 30) + self.assertEqual({app.app_type for app in first}, {"user", "system"}) + self.assertEqual({app.state for app in first}, {"enabled", "disabled"}) + self.assertEqual( + {app.bloatware_removal for app in first}, + {"", "Recommended", "Advanced", "Expert", "Unsafe"}, + ) + + def test_measure_uses_warmups_repetitions_and_rejects_unstable_results(self) -> None: + calls = 0 + + def stable() -> int: + nonlocal calls + calls += 1 + return 7 + + measurement = measure(stable, warmups=2, repetitions=3) + self.assertEqual(calls, 5) + self.assertEqual(measurement.checksum, 7) + self.assertGreaterEqual(measurement.maximum_ms, measurement.average_ms) + + changing = iter((1, 2, 3)) + with self.assertRaises(BenchmarkInvariantError): + measure(lambda: next(changing), warmups=1, repetitions=2) + + def test_small_run_covers_every_release_scenario_and_cleans_temp_data(self) -> None: + with tempfile.TemporaryDirectory() as raw_parent: + parent = Path(raw_parent) + report = run_benchmarks(self.small_config(), temporary_parent=parent) + self.assertEqual(list(parent.iterdir()), []) + + self.assertEqual(report["schema"], REPORT_SCHEMA) + self.assertEqual(report["status"], "passed") + self.assertEqual( + report["cleanup"], + {"temporary_workspace_removed": True}, + ) + results = report["results"] + self.assertIsInstance(results, list) + scenario_names = {result["scenario"] for result in results} + self.assertEqual( + scenario_names, + { + "applications.filter", + "applications.sort_name", + "applications.sort_size", + "applications.selection", + "applications.metadata_progress", + "file_manager.large_local_tree", + "file_manager.transfer_plan", + "p2p.auto_streams", + "controllers.stale_result_filter", + "operations.register_finish", + }, + ) + app_results = [ + result for result in results if str(result["scenario"]).startswith("applications.") + ] + self.assertEqual({result["row_count"] for result in app_results}, {12, 30}) + for result in results: + self.assertGreater(result["row_count"], 0) + self.assertGreaterEqual(result["average_ms"], 0) + self.assertGreaterEqual(result["max_ms"], result["average_ms"]) + self.assertTrue(result["method"]) + + def test_environment_and_json_report_exclude_identity_and_paths(self) -> None: + private_name = "private" + "-user" + with ( + patch("platform.node", side_effect=AssertionError("hostname must not be read")), + patch.dict( + "os.environ", + { + "USERNAME": private_name, + "USERPROFILE": "C:/" + "Users/" + private_name, + "HOME": "/" + "home/" + private_name, + }, + clear=False, + ), + ): + environment = sanitized_environment("physical") + + serialized_environment = json.dumps(environment).casefold() + self.assertNotIn(private_name, serialized_environment) + self.assertEqual(environment["environment_type"], "physical") + self.assertTrue(environment["pyside6_version"]) + self.assertFalse( + {"hostname", "username", "user", "home", "cwd", "path"}.intersection( + environment + ) + ) + + with tempfile.TemporaryDirectory() as raw_directory: + target = Path(raw_directory) / "private-filename.json" + report = run_benchmarks(self.small_config()) + write_json_report(target, report) + serialized_report = target.read_text(encoding="utf-8") + loaded = json.loads(serialized_report) + + self.assertEqual(loaded["schema"], REPORT_SCHEMA) + self.assertNotIn("private-filename", serialized_report) + self.assertNotIn(raw_directory, serialized_report) + + def test_invalid_configuration_and_report_fail_closed(self) -> None: + with self.assertRaises(ValueError): + BenchmarkConfig(repetitions=1).validate() + with self.assertRaises(ValueError): + BenchmarkConfig(file_manager_entries=10, transfer_files=10).validate() + with self.assertRaises(BenchmarkInvariantError): + validate_report({"schema": REPORT_SCHEMA, "status": "passed"}) + with self.assertRaises(ValueError): + sanitized_environment("personal-computer") + + report = run_benchmarks(self.small_config()) + with self.assertRaises(BenchmarkInvariantError): + validate_release_profile(report) + + report = run_benchmarks(self.small_config()) + results = report["results"] + assert isinstance(results, list) + first_result = results[0] + assert isinstance(first_result, dict) + first_result["average_ms"] = float("nan") + with self.assertRaises(BenchmarkInvariantError): + validate_report(report) + + first_result["average_ms"] = 0.0 + first_result["row_count"] = "12" + with self.assertRaises(BenchmarkInvariantError): + validate_report(report) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_version_metadata.py b/tests/test_version_metadata.py index b8f4edb..8bba84e 100644 --- a/tests/test_version_metadata.py +++ b/tests/test_version_metadata.py @@ -8,6 +8,7 @@ from pathlib import Path from apkutils2 import APK +from PIL import Image from openadb import __version__ from openadb.core.acbridge import ACBridgeClient @@ -28,6 +29,7 @@ ANDROID_NS = "{http://schemas.android.com/apk/res/android}" BRIDGE_ROOT = ROOT / "openadb" / "resources" / "acbridge" EXPECTED_SCREENSHOTS = { + "applications-contextual-actions-dark-v3.0.0.png", "applications-dark-v3.0.0.png", "commands-dark-v3.0.0.png", "dashboard-dark-v3.0.0.png", @@ -117,8 +119,16 @@ def test_release_screenshot_names_match_version(self) -> None: readme = (ROOT / "README.md").read_text(encoding="utf-8") generator = (ROOT / "tools" / "capture_readme_screenshots.py").read_text(encoding="utf-8") for filename in EXPECTED_SCREENSHOTS: + screenshot = screenshots / filename self.assertIn(f"docs/screenshots/{filename}", readme) self.assertIn(filename, generator) + self.assertLess(screenshot.stat().st_size, 1_000_000) + with Image.open(screenshot) as image: + image.load() + self.assertEqual(image.size, (1280, 820)) + self.assertEqual(image.mode, "RGB") + self.assertFalse(image.getexif()) + self.assertLessEqual(set(image.info), {"dpi"}) if __name__ == "__main__": diff --git a/tools/capture_readme_screenshots.py b/tools/capture_readme_screenshots.py index ae7de74..9555dd2 100644 --- a/tools/capture_readme_screenshots.py +++ b/tools/capture_readme_screenshots.py @@ -3,9 +3,11 @@ # ruff: noqa: E402 -- the script supports direct execution outside the repository root. import os +import subprocess import sys import tempfile from pathlib import Path +from time import sleep from types import SimpleNamespace from unittest.mock import patch @@ -15,7 +17,11 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) -from PySide6.QtWidgets import QApplication +from PySide6.QtCore import QPoint +from PySide6.QtGui import QFontDatabase, QImage, QPixmap, QRegion +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication, QWidget +from PIL import Image from openadb.core.adb import ADBClient from openadb.core.backup_manager import BackupManager @@ -24,6 +30,7 @@ from openadb.core.fastboot import FastbootClient from openadb.core.icon_extractor import IconExtractor from openadb.core.platform_tools import PlatformToolsManager +from openadb.core.p2p_parallelism import AUTO_PARALLELISM_MODE from openadb.core.settings_manager import SettingsManager from openadb.models.app_info import AppInfo from openadb.models.device_info import DeviceInfo @@ -31,10 +38,19 @@ from openadb.models.platform_tools_info import PlatformToolsInfo from openadb.ui.main_window import MainWindow from openadb.ui.style import apply_theme +from openadb.ui.widgets.file_panel import FilePanel OUTPUT_DIR = ROOT / "docs" / "screenshots" WINDOW_SIZE = (1280, 820) +WINDOWS_SCREENSHOT_FONTS = ( + "segoeui.ttf", + "segoeuib.ttf", + "segoeuii.ttf", + "segoeuil.ttf", + "segoeuisl.ttf", + "segoeuiz.ttf", +) class ScreenshotSettings(SettingsManager): @@ -49,6 +65,21 @@ def _legacy_config_dirs(self) -> list[Path]: return [] +def _load_offscreen_fonts() -> None: + """Make the Windows UI font available to Qt's isolated offscreen plugin.""" + + if os.name != "nt": + return + fonts_dir = Path(os.environ.get("WINDIR", r"C:\Windows")) / "Fonts" + loaded = [ + QFontDatabase.addApplicationFont(str(fonts_dir / filename)) + for filename in WINDOWS_SCREENSHOT_FONTS + if (fonts_dir / filename).is_file() + ] + if not loaded or any(font_id < 0 for font_id in loaded): + raise RuntimeError("Could not load Segoe UI for offscreen screenshot capture") + + def _demo_device() -> DeviceInfo: return DeviceInfo( serial="DEMO-ANDROID-001", @@ -177,12 +208,44 @@ def _configure_demo(window: MainWindow, demo_windows_dir: Path, tools: PlatformT file_manager.windows_panel.set_path(str(demo_windows_dir)) file_manager.windows_path = str(demo_windows_dir) file_manager.windows_path_edit.setText(r"C:\Demo\OpenADB") + file_manager.windows_panel.set_items( + [ + FileItem( + "Documents", + r"C:\Demo\OpenADB\Documents", + True, + modified="2026-07-13 12:00", + item_type="Folder", + ), + FileItem( + "Photos", + r"C:\Demo\OpenADB\Photos", + True, + modified="2026-07-13 12:00", + item_type="Folder", + ), + FileItem( + "openadb-notes.txt", + r"C:\Demo\OpenADB\openadb-notes.txt", + False, + size=18, + modified="2026-07-13 12:00", + item_type="Text file", + ), + ] + ) + file_manager.transfer_transport_combo.blockSignals(True) file_manager.transfer_transport_combo.setCurrentIndex( file_manager.transfer_transport_combo.findData("acbridge_p2p") ) - file_manager.p2p_parallelism_combo.setCurrentIndex(file_manager.p2p_parallelism_combo.findData(3)) + file_manager.transfer_transport_combo.blockSignals(False) + file_manager._accepted_transfer_transport = "acbridge_p2p" + file_manager._update_transfer_transport_ui() + file_manager.p2p_parallelism_combo.setCurrentIndex( + file_manager.p2p_parallelism_combo.findData(AUTO_PARALLELISM_MODE) + ) file_manager.status_label.setText( - "Demo device ready. P2P will send different files through 3 authenticated ACBridge streams." + "Demo device ready. Auto will choose a conservative number of authenticated ACBridge streams." ) commands = window.commands_page @@ -203,19 +266,56 @@ def _configure_demo(window: MainWindow, demo_windows_dir: Path, tools: PlatformT settings_page = window.settings_page settings_page.update_tools(tools) - settings_page.platform_path.setText(r"C:\Android\platform-tools") - settings_page.adb_path.setText(r"C:\Android\platform-tools\adb.exe") - settings_page.fastboot_path.setText(r"C:\Android\platform-tools\fastboot.exe") - settings_page.backups_folder.setText(r"C:\OpenADB\Demo Phone\backups") - settings_page.temp_folder.setText(r"C:\OpenADB\Demo Phone\temp") - settings_page.logs_folder.setText(r"C:\OpenADB\Demo Phone\logs") + settings_page.platform_path.setText(r"C:\Demo\platform-tools") + settings_page.adb_path.setText(r"C:\Demo\platform-tools\adb.exe") + settings_page.fastboot_path.setText(r"C:\Demo\platform-tools\fastboot.exe") + settings_page.backups_folder.setText(r"C:\Demo\OpenADB\backups") + settings_page.temp_folder.setText(r"C:\Demo\OpenADB\temp") + settings_page.logs_folder.setText(r"C:\Demo\OpenADB\logs") settings_page.set_verification_result("adb and fastboot completed their version checks successfully.") window.statusBar().showMessage("Platform Tools: Found | Demo data") -def _capture(window: MainWindow, app: QApplication, page_name: str, theme: str, filename: str) -> None: - apply_theme(app, theme) +def _widget_depth(widget: QWidget, root: QWidget) -> int: + depth = 0 + parent = widget.parentWidget() + while parent is not None and parent is not root: + depth += 1 + parent = parent.parentWidget() + return depth + + +def _render_widget_tree( + root: QWidget, + window: MainWindow, + pixmap: QPixmap, +) -> None: + render_flags = ( + QWidget.RenderFlag.DrawWindowBackground | QWidget.RenderFlag.IgnoreMask + ) + root.render( + pixmap, + root.mapTo(window, QPoint()), + QRegion(root.rect()), + render_flags, + ) + children = [ + widget + for widget in root.findChildren(QWidget) + if widget.window() is window and widget.isVisibleTo(root) + ] + children.sort(key=lambda widget: _widget_depth(widget, root)) + for widget in children: + widget.render( + pixmap, + widget.mapTo(window, QPoint()), + QRegion(widget.rect()), + render_flags, + ) + + +def _capture(window: MainWindow, app: QApplication, page_name: str, filename: str) -> None: row = list(window.pages).index(page_name) window.nav.blockSignals(True) window.nav.setCurrentRow(row) @@ -224,70 +324,278 @@ def _capture(window: MainWindow, app: QApplication, page_name: str, theme: str, window.resize(*WINDOW_SIZE) window.show() app.processEvents() - pixmap = window.grab() + # An actual post-show resize invalidates the complete Windows offscreen + # backing store; update()/repaint() alone can retain only a partial region. + window.resize(WINDOW_SIZE[0] - 1, WINDOW_SIZE[1] - 1) + app.processEvents() + QTest.qWait(50) + window.resize(*WINDOW_SIZE) + app.processEvents() + QTest.qWait(50) + if (window.width(), window.height()) != WINDOW_SIZE: + raise RuntimeError( + f"Unexpected screenshot size: {window.width()}x{window.height()}" + ) + if window.centralWidget().layout() is not None: + window.centralWidget().layout().activate() + # Give layouts and item views enough time to settle after configuring the + # stacked page and any contextual action bar. + contextual_apps = ( + filename == "applications-contextual-actions-dark-v3.0.0.png" + and app.platformName().casefold() == "offscreen" + ) + for _ in range(6 if contextual_apps else 3): + window.update() + window.repaint() + app.processEvents() + QTest.qWait(100) + + native_windows_capture = os.name == "nt" and app.platformName().casefold() == "windows" + if native_windows_capture: + # QWidget.render()/grab() can preserve only a native child's latest + # dirty region on the Windows QPA backend. Ask the desktop compositor + # for the already-visible client area instead. + screen = window.screen() + pixmap = screen.grabWindow(int(window.winId()), 0, 0, *WINDOW_SIZE) + else: + pixmap = QPixmap(window.size()) + pixmap.fill(window.palette().window().color()) + if contextual_apps: + render_flags = ( + QWidget.RenderFlag.DrawWindowBackground | QWidget.RenderFlag.IgnoreMask + ) + window.render(pixmap, QPoint(), QRegion(window.rect()), render_flags) + # The dynamic Applications action bar exposes a Qt offscreen + # backing-store edge case where only the latest dirty region is kept. + # Paint every visible widget explicitly for this deterministic frame. + widgets = [ + widget + for widget in window.findChildren(QWidget) + if widget.window() is window and widget.isVisibleTo(window) + ] + widgets.sort(key=lambda widget: _widget_depth(widget, window)) + for widget in widgets: + widget.render( + pixmap, + widget.mapTo(window, QPoint()), + QRegion(widget.rect()), + render_flags, + ) + elif not native_windows_capture: + window.ensurePolished() + for widget in window.findChildren(QWidget): + widget.ensurePolished() + if widget.layout() is not None: + widget.layout().activate() + widget.update() + app.sendPostedEvents() + app.processEvents() + render_flags = ( + QWidget.RenderFlag.DrawWindowBackground + | QWidget.RenderFlag.DrawChildren + | QWidget.RenderFlag.IgnoreMask + ) + window.render(pixmap, QPoint(), QRegion(window.rect()), render_flags) + # These persistent chrome widgets may have independent backing stores + # under the Windows offscreen plugin, so paint their visible widget + # trees explicitly after the central page. + for widget in (window.device_bar, window.side_panel): + _render_widget_tree(widget, window, pixmap) + if pixmap.size().toTuple() != WINDOW_SIZE: + raise RuntimeError( + f"Unexpected captured frame size: {pixmap.width()}x{pixmap.height()}" + ) target = OUTPUT_DIR / filename - if pixmap.isNull() or not pixmap.save(str(target), "PNG"): + output_image = pixmap.toImage().convertToFormat(QImage.Format.Format_RGB32) + if output_image.isNull() or not output_image.save(str(target), "PNG"): raise RuntimeError(f"Could not save screenshot: {target}") -def main() -> int: +def _set_demo_app_selection(window: MainWindow, packages: set[str]) -> None: + """Set a deterministic checkbox selection without invoking an app action.""" + + page = window.apps_page + page.table.set_apps_sorted(page.apps, page._sort_mode, checked_packages=packages) + page._selection_changed() + + +def _capture_fresh_page( + app: QApplication, + root: Path, + folder_name: str, + page_name: str, + theme: str, + filename: str, + selected_packages: set[str] | None = None, +) -> None: + """Capture a stacked page from its own first-show offscreen window.""" + + capture_root = root / folder_name + settings = ScreenshotSettings(capture_root / "settings") + settings.set_global_values({"window_width": WINDOW_SIZE[0], "window_height": WINDOW_SIZE[1]}) + settings.set("auto_refresh_device", False) + settings.set("theme", theme) + tools = _demo_tools(capture_root / "platform-tools") + platform_tools = PlatformToolsManager(settings) + platform_tools.active = tools + runner = CommandRunner(settings.logs_folder) + adb = ADBClient(platform_tools, runner) + fastboot = FastbootClient(platform_tools, runner) + device_manager = DeviceManager(adb, fastboot, settings) + device_manager.active = _demo_device() + device_manager.devices = [device_manager.active] + demo_windows_dir = capture_root / "demo-files" + (demo_windows_dir / "Documents").mkdir(parents=True) + (demo_windows_dir / "Photos").mkdir() + (demo_windows_dir / "openadb-notes.txt").write_text("Demonstration file", encoding="utf-8") + + window = MainWindow( + settings=settings, + platform_tools=platform_tools, + runner=runner, + adb=adb, + fastboot=fastboot, + device_manager=device_manager, + backup_manager=BackupManager(settings), + icon_extractor=IconExtractor(settings), + ) + try: + _configure_demo(window, demo_windows_dir, tools) + if selected_packages is not None: + _set_demo_app_selection(window, selected_packages) + if window.windowTitle() != "OpenADB 3.0.0": + raise RuntimeError(f"Unexpected screenshot title: {window.windowTitle()!r}") + _capture(window, app, page_name, filename) + print(f"Captured {filename}") + finally: + window.hide() + window.close() + app.processEvents() + QTest.qWait(100) + window.deleteLater() + app.processEvents() + runner.shutdown() + + +CAPTURE_TARGETS: dict[str, tuple[str, str, str, set[str] | None]] = { + "dashboard-dark": ("Dashboard", "Dark", "dashboard-dark-v3.0.0.png", None), + "dashboard-light": ("Dashboard", "Light", "dashboard-light-v3.0.0.png", None), + "applications": ("Apps", "Dark", "applications-dark-v3.0.0.png", set()), + "applications-contextual": ( + "Apps", + "Dark", + "applications-contextual-actions-dark-v3.0.0.png", + {"com.example.camera", "com.example.notes"}, + ), + "file-manager": ("File Manager", "Dark", "file-manager-dark-v3.0.0.png", None), + "commands": ("Commands", "Dark", "commands-dark-v3.0.0.png", None), + "settings": ("Settings", "Dark", "settings-dark-v3.0.0.png", None), +} + + +def _capture_target(target_name: str) -> None: + page_name, theme, filename, selected_packages = CAPTURE_TARGETS[target_name] OUTPUT_DIR.mkdir(parents=True, exist_ok=True) - app = QApplication.instance() or QApplication(sys.argv) + app = QApplication.instance() or QApplication([sys.argv[0]]) app.setApplicationName("OpenADB") + app.setQuitOnLastWindowClosed(False) + _load_offscreen_fonts() + # Construct every widget under its final palette/QSS. Applying a dark + # application stylesheet only after an offscreen top-level window exists + # can leave native child backing stores with partially repainted regions. + apply_theme(app, theme) with tempfile.TemporaryDirectory(prefix="openadb-readme-") as temporary: - root = Path(temporary) - settings = ScreenshotSettings(root / "settings") - settings.set_global_values({"window_width": WINDOW_SIZE[0], "window_height": WINDOW_SIZE[1]}) - settings.set("auto_refresh_device", False) - settings.set("theme", "Dark") - tools = _demo_tools(root / "platform-tools") - platform_tools = PlatformToolsManager(settings) - platform_tools.active = tools - runner = CommandRunner(settings.logs_folder) - adb = ADBClient(platform_tools, runner) - fastboot = FastbootClient(platform_tools, runner) - device_manager = DeviceManager(adb, fastboot, settings) - device_manager.active = _demo_device() - device_manager.devices = [device_manager.active] - - demo_windows_dir = root / "demo-files" - (demo_windows_dir / "Documents").mkdir(parents=True) - (demo_windows_dir / "Photos").mkdir() - (demo_windows_dir / "openadb-notes.txt").write_text("Demonstration file", encoding="utf-8") - with ( patch("openadb.ui.main_window.QTimer.singleShot"), patch( "openadb.ui.file_manager_page.NativeExplorerPanel", - side_effect=RuntimeError("Use deterministic Qt file panel for README screenshots"), + side_effect=RuntimeError( + "Use deterministic Qt file panel for README screenshots" + ), + ), + patch( + "openadb.ui.file_manager_page.WindowsFilePanel", + side_effect=lambda *_args, **_kwargs: FilePanel( + "Windows", + "windows", + show_path_bar=False, + show_button_row=False, + ), ), ): - window = MainWindow( - settings=settings, - platform_tools=platform_tools, - runner=runner, - adb=adb, - fastboot=fastboot, - device_manager=device_manager, - backup_manager=BackupManager(settings), - icon_extractor=IconExtractor(settings), + _capture_fresh_page( + app, + Path(temporary), + target_name, + page_name, + theme, + filename, + selected_packages, ) - try: - _configure_demo(window, demo_windows_dir, tools) - captures = [ - ("Dashboard", "Dark", "dashboard-dark-v3.0.0.png"), - ("Dashboard", "Light", "dashboard-light-v3.0.0.png"), - ("Apps", "Dark", "applications-dark-v3.0.0.png"), - ("File Manager", "Dark", "file-manager-dark-v3.0.0.png"), - ("Commands", "Dark", "commands-dark-v3.0.0.png"), - ("Settings", "Dark", "settings-dark-v3.0.0.png"), - ] - for page_name, theme, filename in captures: - _capture(window, app, page_name, theme, filename) - print(f"Captured {filename}") - finally: - window.close() - runner.shutdown() + + +def _validate_captured_frame(filename: str) -> None: + path = OUTPUT_DIR / filename + with Image.open(path) as image: + if image.size != WINDOW_SIZE or image.mode != "RGB": + raise RuntimeError(f"Unexpected screenshot format: {filename}") + grayscale = image.convert("L") + anchors = ( + grayscale.crop((20, 65, 220, 115)), + grayscale.crop((10, 10, 480, 55)), + ) + for anchor in anchors: + darkest, lightest = anchor.getextrema() + if darkest > 80 or lightest < 150: + raise RuntimeError(f"Incomplete Windows frame: {filename}") + # A partially preserved native backing store can still satisfy the + # extrema check with one surviving icon. Require enough foreground + # pixels for the brand, status bar, and complete navigation list. + foreground_regions = ( + (grayscale.crop((10, 62, 230, 120)), 500), + (grayscale.crop((10, 8, 1270, 58)), 500), + (grayscale.crop((10, 120, 230, 445)), 1_500), + ) + for region, minimum in foreground_regions: + bright_pixels = sum(region.histogram()[131:]) + if bright_pixels < minimum: + raise RuntimeError(f"Incomplete Windows frame: {filename}") + + +def main() -> int: + arguments = sys.argv[1:] + if not arguments: + script = str(Path(__file__).resolve()) + for target_name in CAPTURE_TARGETS: + filename = CAPTURE_TARGETS[target_name][2] + for attempt in range(1, 4): + subprocess.run( + [sys.executable, script, "--capture", target_name], + cwd=ROOT, + check=True, + timeout=60, + ) + try: + _validate_captured_frame(filename) + except RuntimeError: + if attempt == 3: + raise + sleep(0.75) + continue + break + sleep(0.5) + return 0 + + if ( + len(arguments) != 2 + or arguments[0] != "--capture" + or arguments[1] not in CAPTURE_TARGETS + ): + choices = ", ".join(CAPTURE_TARGETS) + raise SystemExit( + f"Usage: capture_readme_screenshots.py [--capture {{{choices}}}]" + ) + _capture_target(arguments[1]) return 0 diff --git a/tools/release_performance.py b/tools/release_performance.py new file mode 100644 index 0000000..05f9ddf --- /dev/null +++ b/tools/release_performance.py @@ -0,0 +1,761 @@ +from __future__ import annotations + +# ruff: noqa: E402 -- direct execution adds the repository root before imports. + +import argparse +import json +import math +import os +import platform +import sys +import tempfile +from dataclasses import dataclass +from datetime import datetime, timezone +from importlib.metadata import PackageNotFoundError, version as distribution_version +from pathlib import Path +from statistics import fmean +from time import perf_counter_ns +from typing import Callable, Sequence + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from openadb.core.app_metadata_loader import AppMetadataLoader +from openadb.core.device_context import DeviceContext +from openadb.core.file_listing_controller import ( + AndroidListingRequest, + FileListingController, +) +from openadb.core.operations import OperationRegistry +from openadb.core.p2p_parallelism import choose_p2p_parallelism +from openadb.core.transfer_plan import ( + AUTO_PARALLELISM, + P2P_TRANSFER, + PUSH_DIRECTION, + TransferPlan, +) +from openadb.models.app_info import AppInfo +from openadb.ui.app_selection_model import AppSelectionModel +from openadb.ui.widgets.app_list_widget import AppFilterState, AppTable +from openadb.version import VERSION + + +REPORT_SCHEMA = "openadb.release-performance.v1" +DEFAULT_APP_COUNTS = (1_200, 3_000) +ENVIRONMENT_TYPES = ("physical", "virtual-machine", "container", "unknown") + + +class BenchmarkInvariantError(RuntimeError): + """Raised when a benchmark stops measuring the intended safe operation.""" + + +@dataclass(frozen=True, slots=True) +class BenchmarkConfig: + warmups: int = 2 + repetitions: int = 7 + app_counts: tuple[int, ...] = DEFAULT_APP_COUNTS + file_manager_entries: int = 5_000 + transfer_files: int = 3_000 + auto_stream_cases: int = 4_096 + stale_result_checks: int = 4_096 + operation_cycles: int = 2_000 + + def validate(self) -> None: + if not 1 <= self.warmups <= 10: + raise ValueError("warmups must be between 1 and 10") + if not 2 <= self.repetitions <= 20: + raise ValueError("repetitions must be between 2 and 20") + if not self.app_counts or any(count < 1 for count in self.app_counts): + raise ValueError("app counts must be positive") + positive_counts = ( + self.file_manager_entries, + self.transfer_files, + self.auto_stream_cases, + self.stale_result_checks, + self.operation_cycles, + ) + if any(count < 1 for count in positive_counts): + raise ValueError("benchmark row counts must be positive") + generated_files = self.file_manager_entries - _directory_count( + self.file_manager_entries + ) + if self.transfer_files > generated_files: + raise ValueError("transfer_files exceeds the generated local file count") + + +@dataclass(frozen=True, slots=True) +class Measurement: + average_ms: float + maximum_ms: float + checksum: int + + +class _GeneratedMetadataClient: + """In-memory metadata source; deliberately has no Android transport.""" + + def get_package_details_many( + self, + package_names: list[str], + *, + max_workers: int, + progress_callback=None, + cancel_event=None, + ) -> dict[str, dict[str, str]]: + del max_workers + total = len(package_names) + result: dict[str, dict[str, str]] = {} + for index, package_name in enumerate(package_names, start=1): + if cancel_event is not None and cancel_event.is_set(): + break + details = { + "appLabel": f"Generated application {index}", + "versionName": "1.0", + "versionCode": str(index), + "sizeBytes": str(index * 1_024), + } + result[package_name] = details + if progress_callback is not None: + progress_callback(index, total, package_name, details) + return result + + +class _CurrentContextManager: + def __init__(self, current: DeviceContext) -> None: + self.current = current + + def is_context_current(self, context: DeviceContext) -> bool: + return context == self.current + + +def generated_apps(count: int) -> list[AppInfo]: + """Return stable mock application rows without touching settings or devices.""" + + categories = ("Recommended", "Advanced", "Expert", "Unsafe", "") + apps: list[AppInfo] = [] + for index in range(count): + size = ( + "Unknown" + if index % 11 == 0 + else f"{(index % 997) + 1}.5 MB" + if index % 2 + else f"{(index % 4_093) + 1} KB" + ) + apps.append( + AppInfo( + package_name=f"org.openadb.generated.app{index:05d}", + app_label=f"Generated application {count - index:05d}", + app_type="system" if index % 2 == 0 else "user", + state="disabled" if index % 3 == 0 else "enabled", + size=size, + bloatware_removal=categories[index % len(categories)], + ) + ) + return apps + + +def measure( + operation: Callable[[], int], + *, + warmups: int, + repetitions: int, +) -> Measurement: + """Measure a deterministic operation and reject changing result checksums.""" + + checksums: list[int] = [] + for _ in range(warmups): + checksums.append(int(operation())) + + durations_ns: list[int] = [] + for _ in range(repetitions): + started = perf_counter_ns() + checksums.append(int(operation())) + durations_ns.append(perf_counter_ns() - started) + + if len(set(checksums)) != 1: + raise BenchmarkInvariantError("A measured operation returned unstable results") + durations_ms = [duration / 1_000_000 for duration in durations_ns] + return Measurement( + average_ms=round(fmean(durations_ms), 6), + maximum_ms=round(max(durations_ms), 6), + checksum=checksums[0], + ) + + +def run_benchmarks( + config: BenchmarkConfig | None = None, + *, + temporary_parent: Path | None = None, + environment_type: str = "unknown", +) -> dict[str, object]: + """Run release benchmarks using generated local data only. + + The function never constructs an ADB or fastboot client, starts a process, + reads user settings, or contacts a device. Temporary entries are empty and + are removed before the report is returned. + """ + + selected = config or BenchmarkConfig() + selected.validate() + if environment_type not in ENVIRONMENT_TYPES: + raise ValueError("environment_type is not a supported sanitized value") + results: list[dict[str, object]] = [] + + for app_count in selected.app_counts: + apps = generated_apps(app_count) + results.extend(_application_measurements(apps, selected)) + + results.append(_auto_stream_measurement(selected)) + results.append(_stale_result_measurement(selected)) + results.append(_operation_registry_measurement(selected)) + + workspace: Path | None = None + with tempfile.TemporaryDirectory( + prefix="openadb-release-performance-", + dir=str(temporary_parent) if temporary_parent is not None else None, + ) as raw_workspace: + workspace = Path(raw_workspace) + generated_sources = _generate_local_tree( + workspace, + selected.file_manager_entries, + ) + results.append(_file_manager_measurement(workspace, selected)) + results.append( + _transfer_plan_measurement( + generated_sources[: selected.transfer_files], + selected, + ) + ) + + cleanup_verified = workspace is not None and not workspace.exists() + if not cleanup_verified: + raise BenchmarkInvariantError("The generated benchmark workspace was not removed") + + report: dict[str, object] = { + "schema": REPORT_SCHEMA, + "openadb_version": VERSION, + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "status": "passed", + "environment": sanitized_environment(environment_type), + "method": { + "clock": "perf_counter_ns", + "warmup_iterations": selected.warmups, + "measured_repetitions": selected.repetitions, + "dataset": "deterministic generated mock data and empty temporary entries", + "scope": "local CPU and temporary filesystem only; no device tools invoked", + }, + "results": results, + "cleanup": {"temporary_workspace_removed": cleanup_verified}, + } + validate_report(report) + return report + + +def sanitized_environment(environment_type: str = "unknown") -> dict[str, object]: + """Return only coarse, non-identifying environment properties.""" + + if environment_type not in ENVIRONMENT_TYPES: + raise ValueError("environment_type is not a supported sanitized value") + try: + pyside_version = distribution_version("PySide6") + except PackageNotFoundError: + pyside_version = "Unavailable" + return { + "environment_type": environment_type, + "operating_system": platform.system() or "Unknown", + "operating_system_release": platform.release() or "Unknown", + "architecture": platform.machine() or "Unknown", + "python_implementation": platform.python_implementation(), + "python_version": platform.python_version(), + "pyside6_version": pyside_version, + "logical_cpu_count": os.cpu_count(), + } + + +def validate_report(report: dict[str, object]) -> None: + if report.get("schema") != REPORT_SCHEMA or report.get("status") != "passed": + raise BenchmarkInvariantError("Performance report schema or status is invalid") + environment = report.get("environment") + if not isinstance(environment, dict): + raise BenchmarkInvariantError("Performance report environment is invalid") + forbidden_environment_keys = { + "hostname", + "username", + "user", + "home", + "cwd", + "path", + "executable", + } + if forbidden_environment_keys.intersection(environment): + raise BenchmarkInvariantError("Performance report contains identifying environment data") + results = report.get("results") + if not isinstance(results, list) or not results: + raise BenchmarkInvariantError("Performance report contains no scenarios") + scenario_counts: dict[str, int] = {} + for result in results: + if not isinstance(result, dict): + raise BenchmarkInvariantError("Performance result is not an object") + required = {"scenario", "row_count", "average_ms", "max_ms", "method"} + if not required.issubset(result): + raise BenchmarkInvariantError("Performance result is missing required fields") + row_count = result["row_count"] + if isinstance(row_count, bool) or not isinstance(row_count, int) or row_count < 1: + raise BenchmarkInvariantError("Performance result row count is invalid") + try: + average_ms = float(result["average_ms"]) + maximum_ms = float(result["max_ms"]) + except (TypeError, ValueError, OverflowError) as exc: + raise BenchmarkInvariantError("Performance duration is invalid") from exc + if ( + not math.isfinite(average_ms) + or not math.isfinite(maximum_ms) + or average_ms < 0 + or maximum_ms < 0 + ): + raise BenchmarkInvariantError("Performance duration is invalid") + if maximum_ms < average_ms: + raise BenchmarkInvariantError("Performance maximum is below its average") + scenario = str(result["scenario"]) + scenario_counts[scenario] = scenario_counts.get(scenario, 0) + 1 + expected_counts = { + "applications.filter": 2, + "applications.sort_name": 2, + "applications.sort_size": 2, + "applications.selection": 2, + "applications.metadata_progress": 2, + "file_manager.large_local_tree": 1, + "file_manager.transfer_plan": 1, + "p2p.auto_streams": 1, + "controllers.stale_result_filter": 1, + "operations.register_finish": 1, + } + if scenario_counts != expected_counts: + raise BenchmarkInvariantError("Performance report scenario coverage is invalid") + + +def validate_release_profile(report: dict[str, object]) -> None: + """Require the exact row counts promised by the release benchmark CLI.""" + + validate_report(report) + results = report["results"] + assert isinstance(results, list) + rows_by_scenario: dict[str, set[int]] = {} + for result in results: + assert isinstance(result, dict) + rows_by_scenario.setdefault(str(result["scenario"]), set()).add( + int(result["row_count"]) + ) + expected = { + "applications.filter": {1_200, 3_000}, + "applications.sort_name": {1_200, 3_000}, + "applications.sort_size": {1_200, 3_000}, + "applications.selection": {1_200, 3_000}, + "applications.metadata_progress": {1_200, 3_000}, + "file_manager.large_local_tree": {5_000}, + "file_manager.transfer_plan": {3_000}, + "p2p.auto_streams": {4_096}, + "controllers.stale_result_filter": {4_096}, + "operations.register_finish": {2_000}, + } + if rows_by_scenario != expected: + raise BenchmarkInvariantError("Release performance row counts are invalid") + + +def write_json_report(path: Path, report: dict[str, object]) -> None: + """Atomically persist a validated report without embedding its path.""" + + validate_report(report) + target = path.expanduser().resolve(strict=False) + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.with_name(f".{target.name}.tmp") + try: + temporary.write_text( + json.dumps( + report, + indent=2, + sort_keys=True, + ensure_ascii=True, + allow_nan=False, + ) + + "\n", + encoding="utf-8", + ) + temporary.replace(target) + finally: + temporary.unlink(missing_ok=True) + + +def _application_measurements( + apps: list[AppInfo], + config: BenchmarkConfig, +) -> list[dict[str, object]]: + count = len(apps) + filters = AppFilterState.from_values( + search_text="generated", + app_type="system", + app_state="enabled", + uad_category="recommended", + ) + + def filter_apps() -> int: + return sum( + filters.matches(app, _uad_category(app)) + for app in apps + ) + + def sort_by_name() -> int: + ordered = sorted( + apps, + key=lambda app: (app.display_name.casefold(), app.package_name.casefold()), + ) + return len(ordered) + len(ordered[0].package_name) + len(ordered[-1].package_name) + + def sort_by_size() -> int: + def key(app: AppInfo) -> tuple[bool, int, str, str]: + parsed = AppTable._size_sort_value(None, app.size) + unknown = parsed < 0 + return ( + unknown, + -parsed if not unknown else parsed, + app.display_name.casefold(), + app.package_name.casefold(), + ) + + ordered = sorted(apps, key=key) + return len(ordered) + AppTable._size_sort_value(None, ordered[0].size) + + packages = tuple(app.package_name for app in apps) + selected_packages = packages[::3] + + def select_apps() -> int: + selection = AppSelectionModel() + selection.select_visible(selected_packages) + summary = selection.summary(packages[::2]) + selection.unselect_visible(packages[::5]) + return summary.total_selected + summary.visible_selected + len(selection) + + def metadata_progress() -> int: + progress_count = 0 + item_count = 0 + + def on_progress(_message: str) -> None: + nonlocal progress_count + progress_count += 1 + + def on_item(_app: AppInfo) -> None: + nonlocal item_count + item_count += 1 + + loader = AppMetadataLoader(_GeneratedMetadataClient(), configured_parallelism=6) + loaded = loader.load( + apps, + progress_callback=on_progress, + item_callback=on_item, + ) + completed = sum(app.metadata_checked for app in loaded) + return completed + progress_count + item_count + + definitions = ( + ( + "applications.filter", + "AppFilterState.matches over generated application rows", + filter_apps, + ), + ( + "applications.sort_name", + "AppInfo display-name casefold sort with package tie-break", + sort_by_name, + ), + ( + "applications.sort_size", + "AppTable size parser with descending deterministic list sort", + sort_by_size, + ), + ( + "applications.selection", + "AppSelectionModel select, summarize, and unselect", + select_apps, + ), + ( + "applications.metadata_progress", + "AppMetadataLoader merge plus per-row progress callbacks", + metadata_progress, + ), + ) + return [ + _result( + scenario=scenario, + row_count=count, + method=method, + measurement=measure( + operation, + warmups=config.warmups, + repetitions=config.repetitions, + ), + repetitions=config.repetitions, + ) + for scenario, method, operation in definitions + ] + + +def _file_manager_measurement( + workspace: Path, + config: BenchmarkConfig, +) -> dict[str, object]: + def list_tree() -> int: + listing = FileListingController.list_windows(workspace) + directories = sum(entry.is_dir for entry in listing.entries) + if len(listing.entries) != config.file_manager_entries: + raise BenchmarkInvariantError("Generated File Manager listing lost entries") + return len(listing.entries) + directories + + measurement = measure( + list_tree, + warmups=config.warmups, + repetitions=config.repetitions, + ) + return _result( + scenario="file_manager.large_local_tree", + row_count=config.file_manager_entries, + method="FileListingController.list_windows over generated empty entries", + measurement=measurement, + repetitions=config.repetitions, + ) + + +def _transfer_plan_measurement( + sources: Sequence[Path], + config: BenchmarkConfig, +) -> dict[str, object]: + context = _generated_context() + + def build_plan() -> int: + plan = TransferPlan( + direction=PUSH_DIRECTION, + transport=P2P_TRANSFER, + sources=tuple(sources), + destination="/storage/emulated/0/Download", + device_context=context, + use_root=False, + parallelism_mode=AUTO_PARALLELISM, + requested_parallelism=None, + ) + return len(plan.sources) + int(plan.is_p2p) + int(plan.is_upload) + + measurement = measure( + build_plan, + warmups=config.warmups, + repetitions=config.repetitions, + ) + return _result( + scenario="file_manager.transfer_plan", + row_count=len(sources), + method="immutable TransferPlan capture for generated local sources", + measurement=measurement, + repetitions=config.repetitions, + ) + + +def _auto_stream_measurement(config: BenchmarkConfig) -> dict[str, object]: + mib = 1_048_576 + cases = ( + (1, 8 * mib, 8 * mib), + (12, 64 * mib, 8 * mib), + (32, 512 * mib, 32 * mib), + (32, 512 * mib, 400 * mib), + ) + + def choose_streams() -> int: + total = 0 + for index in range(config.auto_stream_cases): + file_count, total_bytes, largest_bytes = cases[index % len(cases)] + total += choose_p2p_parallelism( + file_count, + total_bytes, + largest_bytes, + "auto", + None, + ) + return total + + measurement = measure( + choose_streams, + warmups=config.warmups, + repetitions=config.repetitions, + ) + return _result( + scenario="p2p.auto_streams", + row_count=config.auto_stream_cases, + method="choose_p2p_parallelism over a fixed statistics matrix", + measurement=measurement, + repetitions=config.repetitions, + ) + + +def _stale_result_measurement(config: BenchmarkConfig) -> dict[str, object]: + current_context = _generated_context() + manager = _CurrentContextManager(current_context) + controller = FileListingController(device_manager=manager) + current = AndroidListingRequest( + device_context=current_context, + generation=controller.listing_generation, + requested_path=controller.requested_android_path, + ) + stale = AndroidListingRequest( + device_context=current_context, + generation=controller.listing_generation + 1, + requested_path=controller.requested_android_path, + ) + + def filter_stale_results() -> int: + accepted = 0 + for index in range(config.stale_result_checks): + accepted += controller.is_listing_current(current if index % 2 == 0 else stale) + return accepted + + measurement = measure( + filter_stale_results, + warmups=config.warmups, + repetitions=config.repetitions, + ) + return _result( + scenario="controllers.stale_result_filter", + row_count=config.stale_result_checks, + method="FileListingController.is_listing_current on alternating generations", + measurement=measurement, + repetitions=config.repetitions, + ) + + +def _operation_registry_measurement(config: BenchmarkConfig) -> dict[str, object]: + def register_and_finish() -> int: + registry = OperationRegistry() + finished = 0 + for index in range(config.operation_cycles): + token = registry.register( + "performance.validation", + operation_id=f"generated-operation-{index}", + ) + finished += registry.finish(token) + if registry.active_count != 0: + raise BenchmarkInvariantError("OperationRegistry retained generated tokens") + return finished + + measurement = measure( + register_and_finish, + warmups=config.warmups, + repetitions=config.repetitions, + ) + return _result( + scenario="operations.register_finish", + row_count=config.operation_cycles, + method="OperationRegistry register and finish cycles", + measurement=measurement, + repetitions=config.repetitions, + ) + + +def _generate_local_tree(workspace: Path, entry_count: int) -> list[Path]: + directory_count = _directory_count(entry_count) + for index in range(directory_count): + (workspace / f"generated-directory-{index:04d}").mkdir() + files: list[Path] = [] + for index in range(entry_count - directory_count): + path = workspace / f"generated-entry-{index:05d}.bin" + path.touch() + files.append(path) + return files + + +def _directory_count(entry_count: int) -> int: + return min(256, max(1, entry_count // 10)) + + +def _generated_context() -> DeviceContext: + relative = Path("generated-profile") + return DeviceContext( + serial="generated-transport", + mode="ADB", + transport_id="generated-transport-id", + profile_key="generated-profile", + profile_kind="generated", + profile_path=relative, + backups_path=relative / "backups", + temp_path=relative / "temporary", + logs_path=relative / "logs", + generation=1, + ) + + +def _uad_category(app: AppInfo) -> str: + value = str(app.bloatware_removal or "").strip() + if value in {"Recommended", "Advanced", "Expert", "Unsafe"}: + return value + return "Not listed" + + +def _result( + *, + scenario: str, + row_count: int, + method: str, + measurement: Measurement, + repetitions: int, +) -> dict[str, object]: + return { + "scenario": scenario, + "row_count": row_count, + "repetitions": repetitions, + "average_ms": measurement.average_ms, + "max_ms": measurement.maximum_ms, + "method": method, + "result_checksum": measurement.checksum, + } + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Run deterministic OpenADB release-performance checks with generated " + "local data only. No ADB or fastboot command is executed." + ) + ) + parser.add_argument( + "--json-report", + type=Path, + help="Write the sanitized JSON report to this path instead of stdout.", + ) + parser.add_argument("--warmups", type=int, default=2) + parser.add_argument("--repetitions", type=int, default=7) + parser.add_argument( + "--environment-type", + choices=ENVIRONMENT_TYPES, + default="unknown", + help="Record a coarse, non-identifying execution environment type.", + ) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = parse_args(argv) + try: + config = BenchmarkConfig( + warmups=args.warmups, + repetitions=args.repetitions, + ) + report = run_benchmarks(config, environment_type=args.environment_type) + validate_release_profile(report) + if args.json_report is not None: + write_json_report(args.json_report, report) + else: + print(json.dumps(report, indent=2, sort_keys=True, ensure_ascii=True)) + except (BenchmarkInvariantError, OSError, ValueError) as exc: + print(f"Release performance validation failed: {exc}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())