diff --git a/client/scripts/capture_sku_exit_spike.py b/client/scripts/capture_sku_exit_spike.py new file mode 100644 index 0000000..4577fe6 --- /dev/null +++ b/client/scripts/capture_sku_exit_spike.py @@ -0,0 +1,85 @@ +"""采集 T-104 阶段 A 的一次 Back 后本机证据。""" + +from __future__ import annotations + +import argparse +from math import isfinite +from pathlib import Path +import sys + + +CLIENT_ROOT = Path(__file__).resolve().parents[1] +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.sku_selection import SkuSelectionError +from cmbuyer_client.pdd.sku_selection_runner import ( + SkuExitSpikeCapturer, + SkuSelectionRunError, +) + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="采集 T-104 阶段 A 的一次 Back 后本机证据。") + parser.add_argument("--serial", required=True, help="ADB device serial;禁止自动选择。") + parser.add_argument("--output-dir", required=True, type=Path, help="全新本机证据目录;不得覆盖。") + parser.add_argument("--timeout", type=float, default=10.0, help="ADB 与设备 RPC 超时(秒)。") + parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。") + return parser.parse_args(argv) + + +def validate_arguments(arguments: argparse.Namespace) -> None: + if type(arguments.serial) is not str or not arguments.serial.strip(): + raise ValueError("必须显式提供非空 --serial。") + if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name: + raise ValueError("--output-dir 必须是明确的全新目录。") + if ( + not isinstance(arguments.timeout, (int, float)) + or isinstance(arguments.timeout, bool) + or arguments.timeout <= 0 + or not isfinite(arguments.timeout) + ): + raise ValueError("--timeout 必须是大于 0 的有限数值。") + + +def main(argv: list[str] | None = None) -> int: + arguments = parse_arguments(argv) + try: + validate_arguments(arguments) + except ValueError as error: + print(f"失败:{error}", file=sys.stderr) + return 2 + + try: + import adbutils + import uiautomator2 as u2 + except ImportError: + print("失败:缺少 uiautomator2;请在采购工具虚拟环境中运行。", file=sys.stderr) + return 2 + + capturer = SkuExitSpikeCapturer( + AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout), + NoReconnectUiautomatorConnector( + adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, + u2.connect, + ), + timeout_seconds=arguments.timeout, + ) + try: + capturer.capture(arguments.serial, arguments.output_dir) + except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError): + # 不回显第三方异常、serial、页面正文或本机路径。 + print("规格安全退出取证失败:已停止,未发布本地证据目录。", file=sys.stderr) + return 1 + except OSError: + print("规格安全退出取证失败:无法发布本地证据目录。", file=sys.stderr) + return 1 + + print("规格安全退出取证完成。") + print("人工复核:请在指定目录检查退出后截图、XML、应用摘要和 manifest。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/pdd/sku_selection_runner.py b/client/src/cmbuyer_client/pdd/sku_selection_runner.py index 763d99c..365458d 100644 --- a/client/src/cmbuyer_client/pdd/sku_selection_runner.py +++ b/client/src/cmbuyer_client/pdd/sku_selection_runner.py @@ -21,7 +21,13 @@ from adbutils.errors import AdbTimeout from uiautomator2.exceptions import HTTPTimeoutError from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection -from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file +from ..device.baseline import ( + PDD_PACKAGE, + SCREENSHOT_PARAMS, + _save_base64_screenshot, + _sha256_file, + _validate_hierarchy, +) from .product_open import EXPECTED_PDD_VERSION from .product_url import ProductUrl, ProductUrlError, parse_product_url from .sku_selection import ( @@ -37,7 +43,9 @@ from .sku_selection import ( _TARGET_SIZE_BOUNDS, _action_bounds, _annotate_sku_entry_failure, + _parse_nodes, _safe_sku_entry_failure_stage, + _unit_price, resolve_task_selection, ) @@ -86,6 +94,10 @@ class SkuSelectionDeviceAdapterError(SkuSelectionRunError): """第三方设备接口失败的脱敏映射。""" +class SkuExitSpikeError(SkuSelectionRunError): + """T-104 阶段 A 未形成完整的本机退出证据。""" + + def safe_failure_stage(error: BaseException) -> str: """返回允许公开的固定阶段码,绝不回显异常正文。""" @@ -138,6 +150,14 @@ class SkuSelectionRunResult: unit_price: str +@dataclass(frozen=True) +class SkuExitSpikeResult: + """已原子发布、仍需人工判断页面身份的 T-104 证据。""" + + output_directory: Path + manifest_path: Path + + class UiautomatorSkuPanelAdapter(SkuPanelDevice): """把 uiautomator2 缩为 T-103 所需的读取与四种命名操作。 @@ -292,6 +312,210 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice): raise SkuSelectionDeviceAdapterError("规格面板设备操作失败,已停止操作。") from error +class UiautomatorSkuExitAdapter: + """T-104 阶段 A 的窄设备边界:只读能力加唯一一次命名 Back。 + + 取证脚本不能取得 T-103 的入口、规格选项或 reveal 方法。Back 的唯一机会在 + JSON-RPC 前封存,因为超时无法证明事件没有送达,任何结果不明都不得重发。 + """ + + def __init__(self, device: Any, timeout_seconds: float) -> None: + if not _is_positive_finite(timeout_seconds): + raise ValueError("timeout_seconds 必须是大于 0 的有限数值") + self._device = device + self._timeout_seconds = timeout_seconds + self._back_attempted = False + self._back_rpc_outcome = "not_attempted" + + @property + def back_attempts(self) -> int: + return int(self._back_attempted) + + @property + def back_rpc_outcome(self) -> str: + return self._back_rpc_outcome + + def app_info(self, package_name: str) -> dict[str, Any]: + value = self._call("app_info", package_name) + if not isinstance(value, dict): + raise SkuSelectionDeviceAdapterError("无法读取应用版本,已停止取证。") + return value + + def app_current(self) -> dict[str, Any]: + value = self._call("app_current") + if not isinstance(value, dict): + raise SkuSelectionDeviceAdapterError("无法读取前台应用,已停止取证。") + return value + + def dump_window_hierarchy(self) -> str: + value = self._call( + "jsonrpc_call", + "dumpWindowHierarchy", + [False, 50], + timeout=self._timeout_seconds, + ) + if not isinstance(value, str): + raise SkuSelectionDeviceAdapterError("节点树读取失败,已停止取证。") + return value + + def capture_screenshot(self) -> str: + value = self._call( + "jsonrpc_call", + "takeScreenshot", + SCREENSHOT_PARAMS, + timeout=self._timeout_seconds, + ) + if not isinstance(value, str): + raise SkuSelectionScreenshotError("退出后截图读取失败,未发布任何证据产物。") + return value + + def display_size(self) -> tuple[int, int]: + value = self._call("window_size") + if ( + not isinstance(value, tuple) + or len(value) != 2 + or any(not isinstance(item, int) for item in value) + ): + raise SkuSelectionDeviceAdapterError("无法读取屏幕坐标空间,已停止取证。") + return value + + def leave_sku_panel(self) -> None: + if self._back_attempted: + raise SkuSelectionDeviceAdapterError("本次取证已经尝试过返回,拒绝重试。") + self._back_attempted = True + self._back_rpc_outcome = "ambiguous" + self._call( + "jsonrpc_call", + "pressKey", + ["back"], + timeout=self._timeout_seconds, + ) + self._back_rpc_outcome = "completed" + + def _call(self, method: str, *args: Any, **kwargs: Any) -> Any: + try: + operation = getattr(self._device, method) + return operation(*args, **kwargs) + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + raise SkuSelectionRunTimeoutError("T-104 设备操作超时,已停止操作。") from error + except SkuSelectionRunError: + raise + except Exception as error: + raise SkuSelectionDeviceAdapterError("T-104 设备操作失败,已停止操作。") from error + + +class SkuExitSpikeCapturer: + """从人工停驻的已验证目标面板执行一次 Back,再只读采集阶段 A 证据。""" + + def __init__( + self, + adb_client: AdbClient, + connector: Callable[[str], Any], + 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._timeout_seconds = timeout_seconds + self._started = False + + def capture(self, serial: str, output_directory: Path) -> SkuExitSpikeResult: + if self._started: + raise SkuExitSpikeError("同一取证器不可重复调用。") + self._started = True + + staging: Path | None = None + adapter: UiautomatorSkuExitAdapter | None = None + try: + if type(serial) is not str or not serial.strip(): + raise SkuExitSpikeError("必须显式提供非空设备通道。") + target = Path(output_directory) + _validate_new_target(target) + staging = _prepare_staging(target) + + inspection = self._adb_client.inspect(serial) + _require_expected_device(inspection) + adapter = UiautomatorSkuExitAdapter( + self._connector(serial), + self._timeout_seconds, + ) + + # 第一次只读核验拒绝把任意页面带入 Back 边界;第二次紧邻 Back,覆盖核验期间漂移。 + _require_t104_exit_precondition(adapter) + _require_t104_exit_precondition(adapter) + + rpc_outcome = "completed" + try: + adapter.leave_sku_panel() + except SkuSelectionRunError: + if adapter.back_attempts != 1 or adapter.back_rpc_outcome != "ambiguous": + raise + # RPC 失败不能证明 Back 未送达。只读采集可供人调和,但绝不再发第二次动作。 + rpc_outcome = "ambiguous_reconciled" + + if adapter.back_attempts != 1: + raise SkuExitSpikeError("返回动作审计不完整,未发布任何证据产物。") + if rpc_outcome == "completed" and adapter.back_rpc_outcome != "completed": + raise SkuExitSpikeError("返回动作结果不完整,未发布任何证据产物。") + + # Back 后先确认仍是已取证版本的 PDD;不能先把其他前台应用写进本地证据。 + initial_app = _require_t104_post_app(adapter) + + screenshot_path = staging / "post_exit_screenshot.png" + _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path) + _require_screenshot_size(screenshot_path) + + hierarchy = adapter.dump_window_hierarchy() + _validate_hierarchy(hierarchy) + hierarchy_path = staging / "post_exit_hierarchy.xml" + hierarchy_path.write_text(hierarchy, encoding="utf-8") + + current_app = _require_t104_post_app(adapter) + if current_app != initial_app: + raise SkuExitSpikeError("退出后应用摘要在采集期间漂移,未发布任何证据产物。") + app_path = staging / "post_exit_app.json" + app_path.write_text( + json.dumps(current_app, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps( + _t104_exit_manifest( + inspection, + serial, + rpc_outcome, + screenshot_path, + hierarchy_path, + app_path, + ), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + os.rename(staging, target) + staging = None + except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError): + _clean_staging(staging) + raise + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + _clean_staging(staging) + raise SkuExitSpikeError("T-104 取证超时,未发布任何证据产物。") from error + except OSError as error: + _clean_staging(staging) + raise SkuExitSpikeError("T-104 证据目录无法发布,未发布任何证据产物。") from error + except Exception as error: + _clean_staging(staging) + raise SkuExitSpikeError("T-104 取证未完成,未发布任何证据产物。") from error + + return SkuExitSpikeResult(target, target / "manifest.json") + + class SkuSelectionRunner: """只运行 T-103 目标规格恢复、价格确认、原始截图和一次安全退出。""" @@ -470,6 +694,87 @@ def _require_expected_device(inspection: DeviceInspection) -> None: raise SkuSelectionRunError("设备型号或 Android 版本不是已取证组合,已停止操作。") +def _require_t104_exit_precondition(adapter: UiautomatorSkuExitAdapter) -> None: + """fresh 核验 T-103 已人验的唯一 Back 前置,不产生任何页面动作。""" + + _require_expected_version(adapter.app_info(PDD_PACKAGE)) + current = adapter.app_current() + if current.get("package") != PDD_PACKAGE: + raise SkuExitSpikeError("Back 前拼多多不在前台,已停止取证。") + if adapter.display_size() != EXPECTED_SCREEN_SIZE: + raise SkuExitSpikeError("屏幕坐标空间不是已取证尺寸,已停止取证。") + if _unit_price(_parse_nodes(adapter.dump_window_hierarchy())) != EXPECTED_UNIT_PRICE: + raise SkuExitSpikeError("Back 前目标规格或当前价不匹配,已停止取证。") + + +def _post_exit_app_evidence(value: object) -> dict[str, str]: + """只保留页面身份复核需要的应用字段,不把第三方返回整体写盘。""" + + if not isinstance(value, dict): + raise SkuExitSpikeError("退出后应用摘要无效,未发布任何证据产物。") + package = value.get("package") + activity = value.get("activity") + if not isinstance(package, str) or not package.strip(): + raise SkuExitSpikeError("退出后应用包名缺失,未发布任何证据产物。") + if not isinstance(activity, str) or not activity.strip(): + raise SkuExitSpikeError("退出后 Activity 缺失,未发布任何证据产物。") + return {"package": package, "activity": activity} + + +def _require_t104_post_app(adapter: UiautomatorSkuExitAdapter) -> dict[str, str]: + current = _post_exit_app_evidence(adapter.app_current()) + if current["package"] != PDD_PACKAGE: + raise SkuExitSpikeError("Back 后拼多多不在前台,未采集页面证据。") + _require_expected_version(adapter.app_info(PDD_PACKAGE)) + return current + + +def _t104_exit_manifest( + inspection: DeviceInspection, + serial: str, + rpc_outcome: str, + screenshot_path: Path, + hierarchy_path: Path, + app_path: Path, +) -> dict[str, Any]: + """阶段 A 只写事实与哈希;不声明退出成功,也不授予价格证据角色。""" + + return { + "schema_version": 1, + "captured_at": datetime.now(UTC).isoformat(), + "operation": "t104-sku-exit-evidence", + "product": {"goods_id": EXPECTED_GOODS_ID}, + "channel": "wifi" if ":" in serial else "usb", + "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(), + "device": { + "model": inspection.model, + "android_version": inspection.android_version, + "pdd_package": PDD_PACKAGE, + "pdd_version": EXPECTED_PDD_VERSION, + }, + "back_attempts": 1, + "rpc_outcome": rpc_outcome, + "post_exit_status": "human_review_required", + "artifacts": [ + { + "path": screenshot_path.name, + "role": "post_exit_human_review_only", + "sha256": _sha256_file(screenshot_path), + }, + { + "path": hierarchy_path.name, + "role": "post_exit_raw_hierarchy_local_only", + "sha256": _sha256_file(hierarchy_path), + }, + { + "path": app_path.name, + "role": "post_exit_app_identity_human_review_only", + "sha256": _sha256_file(app_path), + }, + ], + } + + def _require_screenshot_size(screenshot_path: Path) -> None: try: with Image.open(screenshot_path) as image: diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py index e718ca3..ad5faba 100644 --- a/client/tests/pdd/test_sku_selection.py +++ b/client/tests/pdd/test_sku_selection.py @@ -2,9 +2,9 @@ from __future__ import annotations import ast import base64 -from contextlib import redirect_stderr +from contextlib import redirect_stderr, redirect_stdout from functools import lru_cache -from io import BytesIO +from io import BytesIO, StringIO import importlib.util import json from pathlib import Path @@ -28,9 +28,12 @@ from cmbuyer_client.pdd.sku_selection import ( resolve_task_selection, ) from cmbuyer_client.pdd.sku_selection_runner import ( + SkuExitSpikeCapturer, + SkuExitSpikeError, SkuSelectionDeviceAdapterError, SkuSelectionRunError, SkuSelectionScreenshotError, + UiautomatorSkuExitAdapter, UiautomatorSkuPanelAdapter, safe_failure_stage, ) @@ -1666,6 +1669,338 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertEqual(len(_actions(device, "pressKey")), 0) +class _ExitEvidenceDevice(_RawDevice): + def __init__(self, hierarchy: str | None = None) -> None: + super().__init__(hierarchy or _M_FIXTURE.read_text(encoding="utf-8")) + self.activity = "com.xunmeng.pinduoduo.activity.NewPageActivity" + + def app_current(self) -> dict[str, str]: + self.calls.append(("app_current",)) + return {"package": self.package, "activity": self.activity} + + +class SkuExitSpikeCapturerTests(unittest.TestCase): + def _capturer(self, adb: _FakeAdb, device: _ExitEvidenceDevice) -> SkuExitSpikeCapturer: + return SkuExitSpikeCapturer(adb, lambda serial: device, 0.03) + + def test_completed_back_atomically_publishes_human_review_only_evidence(self) -> None: + adb = _FakeAdb() + device = _ExitEvidenceDevice() + with TemporaryDirectory() as temporary: + target = Path(temporary) / "exit-evidence" + result = self._capturer(adb, device).capture("192.168.0.173:5555", target) + manifest_text = result.manifest_path.read_text(encoding="utf-8") + manifest = json.loads(manifest_text) + + self.assertEqual(result.output_directory, target) + self.assertEqual(manifest["product"], {"goods_id": "937122477375"}) + self.assertEqual(manifest["channel"], "wifi") + self.assertEqual(manifest["back_attempts"], 1) + self.assertEqual(manifest["rpc_outcome"], "completed") + self.assertEqual(manifest["post_exit_status"], "human_review_required") + self.assertNotIn("safe_exit", manifest_text) + self.assertNotIn("SKU_PANEL_GATE_1", manifest_text) + self.assertNotIn("192.168.0.173:5555", manifest_text) + self.assertEqual( + [item["path"] for item in manifest["artifacts"]], + [ + "post_exit_screenshot.png", + "post_exit_hierarchy.xml", + "post_exit_app.json", + ], + ) + self.assertEqual( + manifest["artifacts"][0]["role"], + "post_exit_human_review_only", + ) + for artifact in manifest["artifacts"]: + self.assertTrue((target / artifact["path"]).is_file()) + self.assertRegex(artifact["sha256"], r"^[0-9a-f]{64}$") + self.assertEqual( + json.loads((target / "post_exit_app.json").read_text(encoding="utf-8")), + { + "activity": "com.xunmeng.pinduoduo.activity.NewPageActivity", + "package": "com.xunmeng.pinduoduo", + }, + ) + + self.assertEqual(adb.calls, [("inspect", "192.168.0.173:5555")]) + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(len(_actions(device, "takeScreenshot")), 1) + self.assertEqual(_actions(device, "click"), []) + self.assertEqual(_actions(device, "swipe"), []) + + def test_precondition_mismatch_or_drift_never_sends_back_or_leaves_artifacts(self) -> None: + class WrongScreen(_ExitEvidenceDevice): + def window_size(self) -> tuple[int, int]: + self.calls.append(("window_size",)) + return 1080, 1920 + + class HierarchyDrift(_ExitEvidenceDevice): + def __init__(self) -> None: + super().__init__() + self.reads = 0 + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "dumpWindowHierarchy": + self.reads += 1 + if self.reads == 2: + self.hierarchy = _S_FIXTURE.read_text(encoding="utf-8") + return super().jsonrpc_call(method, params, timeout) + + cases: list[tuple[str, _ExitEvidenceDevice]] = [] + wrong_version = _ExitEvidenceDevice(); wrong_version.version = "8.18.0" + wrong_foreground = _ExitEvidenceDevice(); wrong_foreground.package = "other.package" + wrong_price = _ExitEvidenceDevice(); wrong_price.hierarchy = wrong_price.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88") + cases.extend( + ( + ("version", wrong_version), + ("foreground", wrong_foreground), + ("screen", WrongScreen()), + ("profile", _ExitEvidenceDevice(_S_FIXTURE.read_text(encoding="utf-8"))), + ("price", wrong_price), + ("drift", HierarchyDrift()), + ) + ) + + for name, device in cases: + with self.subTest(name=name), TemporaryDirectory() as temporary: + target = Path(temporary) / "exit-evidence" + with self.assertRaises((SkuSelectionError, SkuSelectionRunError)): + self._capturer(_FakeAdb(), device).capture("device-1", target) + self.assertEqual(_actions(device, "pressKey"), []) + self.assertEqual(_actions(device, "takeScreenshot"), []) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), []) + + def test_ambiguous_back_delivered_or_undelivered_publishes_once_for_human_review(self) -> None: + class AmbiguousBack(_ExitEvidenceDevice): + def __init__(self, delivered: bool) -> None: + super().__init__() + self.delivered = delivered + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "pressKey": + self.calls.append(("jsonrpc", method, params, timeout)) + if self.delivered: + self.hierarchy = "" + raise TimeoutError("private RPC detail") + return super().jsonrpc_call(method, params, timeout) + + for delivered in (True, False): + with self.subTest(delivered=delivered), TemporaryDirectory() as temporary: + device = AmbiguousBack(delivered) + target = Path(temporary) / "exit-evidence" + result = self._capturer(_FakeAdb(), device).capture("device-1", target) + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + self.assertEqual(manifest["back_attempts"], 1) + self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled") + self.assertEqual(manifest["post_exit_status"], "human_review_required") + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(len(_actions(device, "takeScreenshot")), 1) + + def test_non_pdd_post_app_is_rejected_before_screenshot_for_completed_and_ambiguous_back(self) -> None: + class PostAppDrift(_ExitEvidenceDevice): + def __init__(self, ambiguous: bool) -> None: + super().__init__() + self.ambiguous = ambiguous + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "pressKey": + self.calls.append(("jsonrpc", method, params, timeout)) + self.package = "external.app" + if self.ambiguous: + raise TimeoutError("private RPC detail") + return "" + return super().jsonrpc_call(method, params, timeout) + + for ambiguous in (False, True): + with self.subTest(ambiguous=ambiguous), TemporaryDirectory() as temporary: + device = PostAppDrift(ambiguous) + target = Path(temporary) / "exit-evidence" + with self.assertRaises(SkuExitSpikeError): + self._capturer(_FakeAdb(), device).capture("device-1", target) + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(_actions(device, "takeScreenshot"), []) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), []) + + class PostVersionDrift(_ExitEvidenceDevice): + def __init__(self) -> None: + super().__init__() + self.version_reads = 0 + + def app_info(self, package_name: str) -> dict[str, str]: + self.version_reads += 1 + if self.version_reads == 3: + return {"versionName": "8.18.0"} + return super().app_info(package_name) + + class MissingPostActivity(_ExitEvidenceDevice): + def app_current(self) -> dict[str, str]: + value = super().app_current() + if _actions(self, "pressKey"): + value.pop("activity") + return value + + for name, device in (("version", PostVersionDrift()), ("app_summary", MissingPostActivity())): + with self.subTest(name=name), TemporaryDirectory() as temporary: + target = Path(temporary) / "exit-evidence" + with self.assertRaises(SkuSelectionRunError): + self._capturer(_FakeAdb(), device).capture("device-1", target) + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(_actions(device, "takeScreenshot"), []) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), []) + + def test_post_capture_app_drift_or_read_failures_clean_every_artifact_without_retry(self) -> None: + class AppDrift(_ExitEvidenceDevice): + def __init__(self) -> None: + super().__init__() + self.post_reads = 0 + + def app_current(self) -> dict[str, str]: + value = super().app_current() + if _actions(self, "pressKey"): + self.post_reads += 1 + if self.post_reads == 2: + value["activity"] = "com.xunmeng.pinduoduo.activity.OtherActivity" + return value + + class BadXml(_ExitEvidenceDevice): + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "dumpWindowHierarchy" and _actions(self, "pressKey"): + self.calls.append(("jsonrpc", method, params, timeout)) + return "not-xml" + return super().jsonrpc_call(method, params, timeout) + + class ScreenshotTimeout(_ExitEvidenceDevice): + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "takeScreenshot": + self.calls.append(("jsonrpc", method, params, timeout)) + raise TimeoutError("private RPC detail") + return super().jsonrpc_call(method, params, timeout) + + bad_png = _ExitEvidenceDevice(); bad_png.screenshot = "not-image" + for name, device in ( + ("app_drift", AppDrift()), + ("bad_xml", BadXml()), + ("bad_png", bad_png), + ("screenshot_timeout", ScreenshotTimeout()), + ): + with self.subTest(name=name), TemporaryDirectory() as temporary: + target = Path(temporary) / "exit-evidence" + with self.assertRaises(SkuSelectionRunError): + self._capturer(_FakeAdb(), device).capture("device-1", target) + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), []) + + def test_existing_target_publish_race_preseal_failure_and_repeat_are_fail_closed(self) -> None: + with TemporaryDirectory() as temporary: + target = Path(temporary) / "existing" + target.mkdir() + sentinel = target / "keep" + sentinel.write_text("keep", encoding="utf-8") + adb = _FakeAdb(); device = _ExitEvidenceDevice() + with self.assertRaises(SkuSelectionRunError): + self._capturer(adb, device).capture("device-1", target) + self.assertEqual(adb.calls, []) + self.assertEqual(device.calls, []) + self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep") + + race_target = Path(temporary) / "race" + original_rename = runner_module.os.rename + + def create_target_then_rename(source: str | Path, destination: str | Path) -> None: + Path(destination).mkdir() + (Path(destination) / "sentinel").write_text("keep", encoding="utf-8") + original_rename(source, destination) + + race_device = _ExitEvidenceDevice() + with ( + patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), + self.assertRaises(SkuSelectionRunError), + ): + self._capturer(_FakeAdb(), race_device).capture("device-1", race_target) + self.assertEqual((race_target / "sentinel").read_text(encoding="utf-8"), "keep") + self.assertEqual(list(Path(temporary).glob(".race.staging-*")), []) + self.assertEqual(len(_actions(race_device, "pressKey")), 1) + + preseal_target = Path(temporary) / "preseal" + preseal_device = _ExitEvidenceDevice() + with ( + patch.object(UiautomatorSkuExitAdapter, "leave_sku_panel", side_effect=SkuSelectionRunError("private")), + self.assertRaises(SkuSelectionRunError), + ): + self._capturer(_FakeAdb(), preseal_device).capture("device-1", preseal_target) + self.assertEqual(_actions(preseal_device, "pressKey"), []) + self.assertFalse(preseal_target.exists()) + self.assertEqual(list(Path(temporary).glob(".preseal.staging-*")), []) + + repeated_device = _ExitEvidenceDevice() + capturer = self._capturer(_FakeAdb(), repeated_device) + capturer.capture("device-1", Path(temporary) / "once") + with self.assertRaises(SkuExitSpikeError): + capturer.capture("device-1", Path(temporary) / "twice") + self.assertEqual(len(_actions(repeated_device, "pressKey")), 1) + + def test_static_t104_boundary_has_only_named_back_and_read_calls(self) -> None: + root = Path(__file__).resolve().parents[2] + runner_path = root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py" + script_path = root / "scripts" / "capture_sku_exit_spike.py" + runner_tree = ast.parse(runner_path.read_text(encoding="utf-8")) + script_source = script_path.read_text(encoding="utf-8") + script_tree = ast.parse(script_source) + classes = { + node.name: node + for node in runner_tree.body + if isinstance(node, ast.ClassDef) + } + selected = (classes["UiautomatorSkuExitAdapter"], classes["SkuExitSpikeCapturer"]) + called_attributes = { + node.func.attr + for selected_class in selected + for node in ast.walk(selected_class) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + forbidden_calls = { + "start_pdd_view_intent", + "tap_sku_entry", + "tap_sku_option", + "reveal_size_options_once", + "set_quantity_and_readback", + "go_to_order_confirm", + "create_submission_fence", + "submit_order_once", + } + self.assertTrue(forbidden_calls.isdisjoint(called_attributes)) + self.assertTrue(forbidden_calls.isdisjoint( + node.attr for node in ast.walk(script_tree) if isinstance(node, ast.Attribute) + )) + self.assertTrue( + {"click", "swipe", "scroll", "intent", "quantity", "confirm", "fence", "submit", "payment"}.isdisjoint( + script_source.lower().split() + ) + ) + public_adapter_api = { + name for name in UiautomatorSkuExitAdapter.__dict__ if not name.startswith("_") + } + self.assertEqual( + public_adapter_api, + { + "app_info", + "app_current", + "dump_window_hierarchy", + "capture_screenshot", + "display_size", + "leave_sku_panel", + "back_attempts", + "back_rpc_outcome", + }, + ) + + class SkuSelectionCliTests(unittest.TestCase): def test_cli_accepts_only_target_url_and_task_values(self) -> None: script = _load_runner_script() @@ -1725,6 +2060,98 @@ class SkuSelectionCliTests(unittest.TestCase): self.assertNotIn("page-body", output) +class SkuExitSpikeCliTests(unittest.TestCase): + def test_cli_accepts_only_explicit_device_output_timeout_and_adb(self) -> None: + script = _load_exit_script() + arguments = script.parse_arguments( + [ + "--serial", "device-1", + "--output-dir", "fresh-evidence", + "--timeout", "10", + "--adb", "adb.exe", + ] + ) + script.validate_arguments(arguments) + self.assertEqual( + set(vars(arguments)), + {"serial", "output_dir", "timeout", "adb"}, + ) + for field, value in (("serial", ""), ("timeout", 0), ("timeout", float("inf"))): + with self.subTest(field=field), self.assertRaises(ValueError): + script.validate_arguments( + type( + "Arguments", + (), + { + "serial": "device-1", + "output_dir": Path("fresh-evidence"), + "timeout": 10.0, + "adb": "adb", + field: value, + }, + )() + ) + with redirect_stderr(StringIO()), self.assertRaises(SystemExit): + script.parse_arguments( + [ + "--serial", "device-1", + "--output-dir", "fresh-evidence", + "--goods-id", "937122477375", + ] + ) + + def test_cli_never_echoes_serial_page_text_or_sensitive_path(self) -> None: + script = _load_exit_script() + secret = "SERIAL=192.168.0.173:5555 PATH=C:/Users/private private" + + class FailingCapturer: + def __init__(self, *args: object, **kwargs: object) -> None: + pass + + def capture(self, *args: object, **kwargs: object) -> object: + raise SkuExitSpikeError(secret) + + stderr = BytesIO() + import io + text_stderr = io.TextIOWrapper(stderr, encoding="utf-8") + with patch.object(script, "SkuExitSpikeCapturer", FailingCapturer), redirect_stderr(text_stderr): + status = script.main( + [ + "--serial", "192.168.0.173:5555", + "--output-dir", "C:/Users/private/evidence", + ] + ) + text_stderr.flush() + output = stderr.getvalue().decode("utf-8") + self.assertEqual(status, 1) + self.assertNotIn(secret, output) + self.assertNotIn("192.168.0.173:5555", output) + self.assertNotIn("C:/Users/private", output) + self.assertNotIn("Traceback", output) + + class SuccessfulCapturer: + def __init__(self, *args: object, **kwargs: object) -> None: + pass + + def capture(self, *args: object, **kwargs: object) -> object: + return object() + + stdout = BytesIO() + text_stdout = io.TextIOWrapper(stdout, encoding="utf-8") + with patch.object(script, "SkuExitSpikeCapturer", SuccessfulCapturer), redirect_stdout(text_stdout): + status = script.main( + [ + "--serial", "192.168.0.173:5555", + "--output-dir", "C:/Users/private/evidence", + ] + ) + text_stdout.flush() + output = stdout.getvalue().decode("utf-8") + self.assertEqual(status, 0) + self.assertNotIn("192.168.0.173:5555", output) + self.assertNotIn("C:/Users/private", output) + + def _load_runner_script() -> object: path = Path(__file__).resolve().parents[2] / "scripts" / "run_t103_sku_selection.py" specification = importlib.util.spec_from_file_location("run_t103_sku_selection_test", path) @@ -1735,5 +2162,15 @@ def _load_runner_script() -> object: return module +def _load_exit_script() -> object: + path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_exit_spike.py" + specification = importlib.util.spec_from_file_location("capture_sku_exit_spike_test", path) + if specification is None or specification.loader is None: + raise RuntimeError("无法加载 T-104 阶段 A 脚本。") + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + if __name__ == "__main__": unittest.main()