diff --git a/client/scripts/capture_quantity_gate2_spike.py b/client/scripts/capture_quantity_gate2_spike.py index 426f6d3..c2184ed 100644 --- a/client/scripts/capture_quantity_gate2_spike.py +++ b/client/scripts/capture_quantity_gate2_spike.py @@ -14,6 +14,7 @@ sys.path.insert(0, str(CLIENT_ROOT / "src")) from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector from cmbuyer_client.pdd.quantity_gate2_spike import ( + Android16TopResumedForegroundReader, DECLARED_QUANTITIES, EXPECTED_GOODS_ID, QuantityGate2EvidenceCapturer, @@ -84,12 +85,14 @@ def main(argv: list[str] | None = None) -> int: print("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr) return 2 + adb_runner = SubprocessAdbRunner(arguments.adb) capturer = QuantityGate2EvidenceCapturer( - AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout), + AdbClient(adb_runner, timeout_seconds=arguments.timeout), NoReconnectUiautomatorConnector( adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, u2.connect, ), + Android16TopResumedForegroundReader(adb_runner, arguments.timeout), timeout_seconds=arguments.timeout, ) try: diff --git a/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py b/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py index 9576272..d99264d 100644 --- a/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py +++ b/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py @@ -10,6 +10,7 @@ import json from math import isfinite import os from pathlib import Path +import re import shutil from typing import Any, Protocol from uuid import uuid4 @@ -18,7 +19,7 @@ from adbutils.errors import AdbTimeout from PIL import Image, UnidentifiedImageError from uiautomator2.exceptions import HTTPTimeoutError -from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection +from ..device.adb import AdbClient, CommandRunner, DeviceConnectionError, DeviceInspection from ..device.baseline import ( HIERARCHY_PARAMS, PDD_PACKAGE, @@ -52,13 +53,50 @@ class QuantityGate2ReadDevice(Protocol): def app_info(self, package_name: str) -> dict[str, Any]: ... - def app_current(self) -> dict[str, Any]: ... - def window_size(self) -> tuple[int, int]: ... def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any: ... +class QuantityGate2ForegroundReader(Protocol): + """只读取 Android 16 的唯一 resumed activity，不暴露通用 shell。""" + + def read(self, serial: str) -> dict[str, str]: ... + + +class Android16TopResumedForegroundReader: + """绕开 adbutils 2.12.0 对 Android 16 ``topResumedActivity`` 的误解析。""" + + _TOP_RESUMED_PATTERN = re.compile( + r"(?m)^\s*topResumedActivity=ActivityRecord\{[^\r\n}]*?\s+u\d+\s+" + r"(?P[^/\s]+)/(?P[^\s}]+)\s+t\d+\}\s*$" + ) + + def __init__(self, runner: CommandRunner, timeout_seconds: float) -> None: + if not _is_positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + self._runner = runner + self._timeout_seconds = timeout_seconds + + def read(self, serial: str) -> dict[str, str]: + if type(serial) is not str or not serial.strip() or serial != serial.strip(): + raise QuantityGate2EvidenceError("必须显式提供非空设备通道。") + result = self._runner.run( + ("-s", serial, "shell", "dumpsys", "activity", "activities"), + self._timeout_seconds, + ) + if result.returncode != 0: + raise QuantityGate2EvidenceError("Android 前台摘要读取失败，未发布证据。") + matches = list(self._TOP_RESUMED_PATTERN.finditer(result.stdout)) + if len(matches) != 1: + raise QuantityGate2EvidenceError("Android 前台摘要不唯一，未发布证据。") + match = matches[0] + return { + "package": match.group("package"), + "activity": match.group("activity"), + } + + @dataclass(frozen=True) class QuantityGate2EvidenceResult: output_directory: Path @@ -75,12 +113,14 @@ class QuantityGate2EvidenceCapturer: self, adb_client: AdbClient, connector: Callable[[str], QuantityGate2ReadDevice], + foreground_reader: QuantityGate2ForegroundReader, timeout_seconds: float, ) -> None: if not _is_positive_finite(timeout_seconds): raise ValueError("timeout_seconds 必须是大于 0 的有限数值") self._adb_client = adb_client self._connector = connector + self._foreground_reader = foreground_reader self._timeout_seconds = timeout_seconds self._started = False @@ -104,7 +144,10 @@ class QuantityGate2EvidenceCapturer: inspection = self._adb_client.inspect(serial) _require_expected_device(inspection) device = self._connector(serial) - initial_app = _require_read_precondition(device) + initial_app = _require_read_precondition( + device, + self._foreground_reader.read(serial), + ) target.parent.mkdir(parents=True, exist_ok=True) staging = target.parent / f".{target.name}.staging-{uuid4().hex}" @@ -132,7 +175,10 @@ class QuantityGate2EvidenceCapturer: hierarchy_path = staging / "hierarchy.xml" hierarchy_path.write_text(hierarchy, encoding="utf-8") - final_app = _require_read_precondition(device) + final_app = _require_read_precondition( + device, + self._foreground_reader.read(serial), + ) if final_app != initial_app: raise QuantityGate2EvidenceError("数量状态取证期间前台页面漂移，未发布证据。") app_path = staging / "app.json" @@ -216,12 +262,14 @@ def _require_expected_device(inspection: DeviceInspection) -> None: raise QuantityGate2EvidenceError("设备不是已批准取证组合。") -def _require_read_precondition(device: QuantityGate2ReadDevice) -> dict[str, str]: +def _require_read_precondition( + device: QuantityGate2ReadDevice, + current: object, +) -> dict[str, str]: info = device.app_info(PDD_PACKAGE) version = (info.get("versionName") or info.get("version_name")) if isinstance(info, dict) else None if version != EXPECTED_PDD_VERSION: raise QuantityGate2EvidenceError("拼多多版本不是已批准取证版本。") - current = device.app_current() if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE: raise QuantityGate2EvidenceError("拼多多不在前台。") activity = current.get("activity") diff --git a/client/tests/pdd/test_quantity_gate2.py b/client/tests/pdd/test_quantity_gate2.py index 0343974..c1367cf 100644 --- a/client/tests/pdd/test_quantity_gate2.py +++ b/client/tests/pdd/test_quantity_gate2.py @@ -22,19 +22,25 @@ from PIL import Image CLIENT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(CLIENT_ROOT / "src")) -from cmbuyer_client.device.adb import AdbDevice, DeviceInspection +from cmbuyer_client.device.adb import AdbDevice, CommandResult, DeviceInspection from cmbuyer_client.pdd.quantity_gate2_spike import ( + Android16TopResumedForegroundReader, DECLARED_QUANTITIES, EXPECTED_GOODS_ID, QuantityGate2EvidenceCapturer, QuantityGate2EvidenceError, QuantityGate2EvidenceTimeoutError, + QuantityGate2ForegroundReader, QuantityGate2ReadDevice, ) SERIAL = "192.168.0.173:5555" HIERARCHY = "" +TOP_RESUMED = ( + " topResumedActivity=ActivityRecord{101034589 u0 " + "com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n" +) @lru_cache(maxsize=1) @@ -63,25 +69,18 @@ class FakeReadDevice: self, *, version: str = "8.17.0", - package: str = "com.xunmeng.pinduoduo", - activity: str = "com.xunmeng.pinduoduo.activity.NewPageActivity", screen_size: tuple[int, int] = (1080, 2376), screenshot: object | None = None, hierarchy: object = HIERARCHY, ) -> None: self.version = version - self.package = package - self.activity = activity self.screen_size = screen_size self.screenshot = _png_base64() if screenshot is None else screenshot self.hierarchy = hierarchy self.calls: list[tuple[object, ...]] = [] self.app_info_reads = 0 - self.current_reads = 0 self.window_reads = 0 self.post_version: str | None = None - self.post_package: str | None = None - self.post_activity: str | None = None self.post_screen_size: tuple[int, int] | None = None def app_info(self, package_name: str) -> dict[str, str]: @@ -90,13 +89,6 @@ class FakeReadDevice: version = self.post_version if self.app_info_reads > 1 and self.post_version is not None else self.version return {"versionName": version} - def app_current(self) -> dict[str, str]: - self.calls.append(("app_current",)) - self.current_reads += 1 - package = self.post_package if self.current_reads > 1 and self.post_package is not None else self.package - activity = self.post_activity if self.current_reads > 1 and self.post_activity is not None else self.activity - return {"package": package, "activity": activity} - def window_size(self) -> tuple[int, int]: self.calls.append(("window_size",)) self.window_reads += 1 @@ -113,6 +105,36 @@ class FakeReadDevice: raise AssertionError(f"unexpected RPC {method}") +class FakeForegroundReader: + def __init__( + self, + *, + package: str = "com.xunmeng.pinduoduo", + activity: str = ".activity.NewPageActivity", + ) -> None: + self.package = package + self.activity = activity + self.post_package: str | None = None + self.post_activity: str | None = None + self.calls: list[str] = [] + + def read(self, serial: str) -> dict[str, str]: + self.calls.append(serial) + package = self.post_package if len(self.calls) > 1 and self.post_package is not None else self.package + activity = self.post_activity if len(self.calls) > 1 and self.post_activity is not None else self.activity + return {"package": package, "activity": activity} + + +class FakeCommandRunner: + def __init__(self, result: CommandResult) -> None: + self.result = result + self.calls: list[tuple[tuple[str, ...], float]] = [] + + def run(self, arguments: tuple[str, ...], timeout_seconds: float) -> CommandResult: + self.calls.append((arguments, timeout_seconds)) + return self.result + + def _load_script() -> object: script_path = CLIENT_ROOT / "scripts" / "capture_quantity_gate2_spike.py" spec = spec_from_file_location("capture_quantity_gate2_spike_for_test", script_path) @@ -136,12 +158,71 @@ def _namespace(**changes: object) -> argparse.Namespace: return argparse.Namespace(**values) +class Android16TopResumedForegroundReaderTests(unittest.TestCase): + def test_reads_exact_unique_android16_top_resumed_activity(self) -> None: + runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED)) + reader = Android16TopResumedForegroundReader(runner, timeout_seconds=7) + + self.assertEqual( + reader.read(SERIAL), + { + "package": "com.xunmeng.pinduoduo", + "activity": ".activity.NewPageActivity", + }, + ) + self.assertEqual( + runner.calls, + [ + ( + ( + "-s", + SERIAL, + "shell", + "dumpsys", + "activity", + "activities", + ), + 7, + ) + ], + ) + + def test_missing_legacy_or_duplicate_top_resumed_activity_is_rejected(self) -> None: + rejected_outputs = ( + "", + ( + "mResumedActivity: ActivityRecord{101034589 u0 " + "com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n" + ), + TOP_RESUMED + TOP_RESUMED, + "topResumedActivity=null\n", + ) + for output in rejected_outputs: + with self.subTest(output=output): + reader = Android16TopResumedForegroundReader( + FakeCommandRunner(CommandResult(stdout=output)), + timeout_seconds=7, + ) + with self.assertRaises(QuantityGate2EvidenceError): + reader.read(SERIAL) + + def test_command_failure_and_invalid_serial_are_rejected(self) -> None: + runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED, returncode=1)) + reader = Android16TopResumedForegroundReader(runner, timeout_seconds=7) + with self.assertRaises(QuantityGate2EvidenceError): + reader.read(SERIAL) + with self.assertRaises(QuantityGate2EvidenceError): + reader.read(f" {SERIAL}") + self.assertEqual(len(runner.calls), 1) + + class QuantityGate2EvidenceTests(unittest.TestCase): def _capturer( self, adb: FakeAdbClient, device: FakeReadDevice, *, + foreground: FakeForegroundReader | None = None, connector_calls: list[str] | None = None, ) -> QuantityGate2EvidenceCapturer: def connect(serial: str) -> FakeReadDevice: @@ -149,17 +230,28 @@ class QuantityGate2EvidenceTests(unittest.TestCase): connector_calls.append(serial) return device - return QuantityGate2EvidenceCapturer(adb, connect, timeout_seconds=2) + return QuantityGate2EvidenceCapturer( + adb, + connect, + foreground or FakeForegroundReader(), + timeout_seconds=2, + ) def test_initial_and_target_states_publish_only_raw_read_evidence(self) -> None: for state, quantity in DECLARED_QUANTITIES.items(): with self.subTest(state=state), TemporaryDirectory() as temporary: adb = FakeAdbClient() device = FakeReadDevice() + foreground = FakeForegroundReader() connector_calls: list[str] = [] target = Path(temporary) / state - result = self._capturer(adb, device, connector_calls=connector_calls).capture( + result = self._capturer( + adb, + device, + foreground=foreground, + connector_calls=connector_calls, + ).capture( SERIAL, EXPECTED_GOODS_ID, state, @@ -171,6 +263,7 @@ class QuantityGate2EvidenceTests(unittest.TestCase): app = json.loads(result.app_path.read_text(encoding="utf-8")) self.assertEqual(adb.calls, [SERIAL]) self.assertEqual(connector_calls, [SERIAL]) + self.assertEqual(foreground.calls, [SERIAL, SERIAL]) self.assertEqual( [call[1] for call in device.calls if call[0] == "jsonrpc"], ["takeScreenshot", "dumpWindowHierarchy"], @@ -236,18 +329,18 @@ class QuantityGate2EvidenceTests(unittest.TestCase): def test_device_version_package_activity_and_screen_mismatch_fail_closed(self) -> None: scenarios = ( - (FakeAdbClient(model="OTHER"), FakeReadDevice()), - (FakeAdbClient(android_version="15"), FakeReadDevice()), - (FakeAdbClient(), FakeReadDevice(version="8.17.1")), - (FakeAdbClient(), FakeReadDevice(package="com.example.other")), - (FakeAdbClient(), FakeReadDevice(activity="")), - (FakeAdbClient(), FakeReadDevice(screen_size=(1080, 2400))), + (FakeAdbClient(model="OTHER"), FakeReadDevice(), FakeForegroundReader()), + (FakeAdbClient(android_version="15"), FakeReadDevice(), FakeForegroundReader()), + (FakeAdbClient(), FakeReadDevice(version="8.17.1"), FakeForegroundReader()), + (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(package="com.example.other")), + (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(activity="")), + (FakeAdbClient(), FakeReadDevice(screen_size=(1080, 2400)), FakeForegroundReader()), ) - for adb, device in scenarios: + for adb, device, foreground in scenarios: with self.subTest(device=device.__dict__), TemporaryDirectory() as temporary: target = Path(temporary) / "evidence" with self.assertRaises(QuantityGate2EvidenceError): - self._capturer(adb, device).capture( + self._capturer(adb, device, foreground=foreground).capture( SERIAL, EXPECTED_GOODS_ID, "initial", 1, target ) self.assertFalse(target.exists()) @@ -295,18 +388,23 @@ class QuantityGate2EvidenceTests(unittest.TestCase): def test_post_capture_app_or_screen_drift_does_not_publish(self) -> None: drift_mutators = ( - lambda device: setattr(device, "post_version", "8.17.1"), - lambda device: setattr(device, "post_package", "com.example.other"), - lambda device: setattr(device, "post_activity", "OtherActivity"), - lambda device: setattr(device, "post_screen_size", (1080, 2400)), + lambda device, foreground: setattr(device, "post_version", "8.17.1"), + lambda device, foreground: setattr(foreground, "post_package", "com.example.other"), + lambda device, foreground: setattr(foreground, "post_activity", "OtherActivity"), + lambda device, foreground: setattr(device, "post_screen_size", (1080, 2400)), ) for mutate in drift_mutators: with TemporaryDirectory() as temporary: device = FakeReadDevice() - mutate(device) + foreground = FakeForegroundReader() + mutate(device, foreground) target = Path(temporary) / "evidence" with self.assertRaises(QuantityGate2EvidenceError): - self._capturer(FakeAdbClient(), device).capture( + self._capturer( + FakeAdbClient(), + device, + foreground=foreground, + ).capture( SERIAL, EXPECTED_GOODS_ID, "initial", 1, target ) self.assertFalse(target.exists()) @@ -336,6 +434,15 @@ class QuantityGate2EvidenceTests(unittest.TestCase): "pay", } self.assertTrue(forbidden.isdisjoint(QuantityGate2ReadDevice.__dict__)) + self.assertTrue(forbidden.isdisjoint(QuantityGate2ForegroundReader.__dict__)) + self.assertEqual( + { + name + for name in Android16TopResumedForegroundReader.__dict__ + if not name.startswith("_") + }, + {"read"}, + ) self.assertEqual( {name for name in QuantityGate2EvidenceCapturer.__dict__ if not name.startswith("_")}, {"capture"},