From 994b6c805499f6d70d25fcf2262c32d9343af60a Mon Sep 17 00:00:00 2001
From: QiuSW <105186638@qq.com>
Date: Wed, 5 Aug 2026 18:08:11 +0800
Subject: [PATCH] feat(client): capture T-105 quantity states
---
.../scripts/capture_quantity_gate2_spike.py | 117 +++++
.../pdd/quantity_gate2_spike.py | 303 ++++++++++++
client/tests/pdd/test_quantity_gate2.py | 452 ++++++++++++++++++
3 files changed, 872 insertions(+)
create mode 100644 client/scripts/capture_quantity_gate2_spike.py
create mode 100644 client/src/cmbuyer_client/pdd/quantity_gate2_spike.py
create mode 100644 client/tests/pdd/test_quantity_gate2.py
diff --git a/client/scripts/capture_quantity_gate2_spike.py b/client/scripts/capture_quantity_gate2_spike.py
new file mode 100644
index 0000000..426f6d3
--- /dev/null
+++ b/client/scripts/capture_quantity_gate2_spike.py
@@ -0,0 +1,117 @@
+"""采集 T-105 人工准备的数量 1/2 两态证据；不执行页面操作。"""
+
+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.quantity_gate2_spike import (
+ DECLARED_QUANTITIES,
+ EXPECTED_GOODS_ID,
+ QuantityGate2EvidenceCapturer,
+ QuantityGate2EvidenceError,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+ parser = argparse.ArgumentParser(description="采集 T-105 人工准备的数量两态本机证据。")
+ parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+ parser.add_argument("--goods-id", required=True, help="T-105 已批准的目标商品标识。")
+ parser.add_argument(
+ "--state",
+ required=True,
+ choices=sorted(DECLARED_QUANTITIES),
+ help="人工声明状态：initial=数量1，target=数量2。",
+ )
+ parser.add_argument(
+ "--declared-quantity",
+ required=True,
+ type=int,
+ help="人工看到的数量；必须与状态一致。",
+ )
+ 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()
+ or arguments.serial != arguments.serial.strip()
+ ):
+ raise ValueError("必须显式提供非空 --serial。")
+ if type(arguments.goods_id) is not str or arguments.goods_id != EXPECTED_GOODS_ID:
+ raise ValueError("--goods-id 不是 T-105 已批准目标。")
+ if type(arguments.state) is not str or arguments.state not in DECLARED_QUANTITIES:
+ raise ValueError("--state 必须是批准的人工声明状态。")
+ if type(arguments.declared_quantity) is not int or (
+ arguments.declared_quantity != DECLARED_QUANTITIES[arguments.state]
+ ):
+ raise ValueError("--declared-quantity 必须与人工声明状态一致。")
+ 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 = QuantityGate2EvidenceCapturer(
+ 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.goods_id,
+ arguments.state,
+ arguments.declared_quantity,
+ arguments.output_dir,
+ )
+ except (DeviceConnectionError, QuantityGate2EvidenceError):
+ # 第三方异常可能含 serial、Activity、本机路径或页面正文，CLI 只输出固定摘要。
+ 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/quantity_gate2_spike.py b/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py
new file mode 100644
index 0000000..9576272
--- /dev/null
+++ b/client/src/cmbuyer_client/pdd/quantity_gate2_spike.py
@@ -0,0 +1,303 @@
+"""T-105 数量两态的纯只读取证；人工确认前不识别或操作数量控件。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from hashlib import sha256
+import json
+from math import isfinite
+import os
+from pathlib import Path
+import shutil
+from typing import Any, Protocol
+from uuid import uuid4
+
+from adbutils.errors import AdbTimeout
+from PIL import Image, UnidentifiedImageError
+from uiautomator2.exceptions import HTTPTimeoutError
+
+from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
+from ..device.baseline import (
+ HIERARCHY_PARAMS,
+ PDD_PACKAGE,
+ SCREENSHOT_PARAMS,
+ _save_base64_screenshot,
+ _sha256_file,
+ _validate_hierarchy,
+)
+from .product_open import EXPECTED_PDD_VERSION
+from .product_url import parse_product_url
+
+
+EXPECTED_GOODS_ID = "937122477375"
+EXPECTED_DEVICE_MODEL = "PKG110"
+EXPECTED_ANDROID_VERSION = "16"
+EXPECTED_SCREEN_SIZE = (1080, 2376)
+TARGET_SELECTION = {"color": "黑色CHA(纯棉)", "size": "M(建议100-115)"}
+DECLARED_QUANTITIES = {"initial": 1, "target": 2}
+
+
+class QuantityGate2EvidenceError(RuntimeError):
+ """T-105 两态证据未形成完整原子产物。"""
+
+
+class QuantityGate2EvidenceTimeoutError(QuantityGate2EvidenceError):
+ """只读取证设备调用超时。"""
+
+
+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: ...
+
+
+@dataclass(frozen=True)
+class QuantityGate2EvidenceResult:
+ output_directory: Path
+ manifest_path: Path
+ screenshot_path: Path
+ hierarchy_path: Path
+ app_path: Path
+
+
+class QuantityGate2EvidenceCapturer:
+ """记录人工准备的数量状态，不从页面推断声明是否正确。"""
+
+ def __init__(
+ self,
+ adb_client: AdbClient,
+ connector: Callable[[str], QuantityGate2ReadDevice],
+ 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,
+ goods_id: str,
+ human_declared_state: str,
+ human_declared_quantity: int,
+ output_directory: Path,
+ ) -> QuantityGate2EvidenceResult:
+ if self._started:
+ raise QuantityGate2EvidenceError("同一取证器不可重复调用。")
+ self._started = True
+ _validate_inputs(serial, goods_id, human_declared_state, human_declared_quantity)
+ target = Path(output_directory)
+ _validate_new_target(target)
+
+ staging: Path | None = None
+ try:
+ inspection = self._adb_client.inspect(serial)
+ _require_expected_device(inspection)
+ device = self._connector(serial)
+ initial_app = _require_read_precondition(device)
+
+ target.parent.mkdir(parents=True, exist_ok=True)
+ staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
+ staging.mkdir()
+
+ screenshot_path = staging / "screenshot.png"
+ screenshot_payload = _read_rpc(
+ device,
+ "takeScreenshot",
+ SCREENSHOT_PARAMS,
+ self._timeout_seconds,
+ )
+ if not isinstance(screenshot_payload, str):
+ raise QuantityGate2EvidenceError("数量状态截图无效，未发布证据。")
+ _save_base64_screenshot(screenshot_payload, screenshot_path)
+ _require_screenshot_size(screenshot_path)
+
+ hierarchy = _read_rpc(
+ device,
+ "dumpWindowHierarchy",
+ HIERARCHY_PARAMS,
+ self._timeout_seconds,
+ )
+ _validate_hierarchy(hierarchy)
+ hierarchy_path = staging / "hierarchy.xml"
+ hierarchy_path.write_text(hierarchy, encoding="utf-8")
+
+ final_app = _require_read_precondition(device)
+ if final_app != initial_app:
+ raise QuantityGate2EvidenceError("数量状态取证期间前台页面漂移，未发布证据。")
+ app_path = staging / "app.json"
+ app_path.write_text(
+ json.dumps(final_app, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+
+ manifest_path = staging / "manifest.json"
+ manifest_path.write_text(
+ json.dumps(
+ _manifest(
+ inspection,
+ serial,
+ goods_id,
+ human_declared_state,
+ human_declared_quantity,
+ 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, QuantityGate2EvidenceError):
+ _clean_staging(staging)
+ raise
+ except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+ _clean_staging(staging)
+ raise QuantityGate2EvidenceTimeoutError("数量两态只读取证超时，未发布证据。") from error
+ except (OSError, UnidentifiedImageError, ValueError) as error:
+ _clean_staging(staging)
+ raise QuantityGate2EvidenceError("数量两态证据无法原子发布，未发布证据。") from error
+ except Exception as error:
+ _clean_staging(staging)
+ raise QuantityGate2EvidenceError("数量两态只读取证未完成，未发布证据。") from error
+
+ return QuantityGate2EvidenceResult(
+ output_directory=target,
+ manifest_path=target / "manifest.json",
+ screenshot_path=target / "screenshot.png",
+ hierarchy_path=target / "hierarchy.xml",
+ app_path=target / "app.json",
+ )
+
+
+def _is_positive_finite(value: object) -> bool:
+ return (
+ isinstance(value, (int, float))
+ and not isinstance(value, bool)
+ and value > 0
+ and isfinite(value)
+ )
+
+
+def _validate_inputs(serial: object, goods_id: object, state: object, quantity: object) -> None:
+ if type(serial) is not str or not serial.strip() or serial != serial.strip():
+ raise QuantityGate2EvidenceError("必须显式提供非空设备通道。")
+ if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID:
+ raise QuantityGate2EvidenceError("商品不是 T-105 已批准取证目标。")
+ parse_product_url(f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}")
+ if type(state) is not str or state not in DECLARED_QUANTITIES:
+ raise QuantityGate2EvidenceError("人工声明状态无效。")
+ if type(quantity) is not int or quantity != DECLARED_QUANTITIES[state]:
+ raise QuantityGate2EvidenceError("人工声明数量与批准状态不一致。")
+
+
+def _validate_new_target(target: Path) -> None:
+ if target.exists() or not target.name:
+ raise QuantityGate2EvidenceError("输出目录必须是不存在的明确新目录。")
+
+
+def _require_expected_device(inspection: DeviceInspection) -> None:
+ if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION:
+ raise QuantityGate2EvidenceError("设备不是已批准取证组合。")
+
+
+def _require_read_precondition(device: QuantityGate2ReadDevice) -> 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")
+ if not isinstance(activity, str) or not activity.strip():
+ raise QuantityGate2EvidenceError("前台应用摘要不完整。")
+ if device.window_size() != EXPECTED_SCREEN_SIZE:
+ raise QuantityGate2EvidenceError("屏幕坐标空间不是已批准尺寸。")
+ return {"package": PDD_PACKAGE, "activity": activity, "pdd_version": EXPECTED_PDD_VERSION}
+
+
+def _read_rpc(
+ device: QuantityGate2ReadDevice,
+ method: str,
+ params: object,
+ timeout_seconds: float,
+) -> object:
+ return device.jsonrpc_call(method, params, timeout=timeout_seconds)
+
+
+def _require_screenshot_size(path: Path) -> None:
+ with Image.open(path) as image:
+ image.load()
+ if image.size != EXPECTED_SCREEN_SIZE or image.format != "PNG":
+ raise QuantityGate2EvidenceError("数量状态截图格式或尺寸无效。")
+
+
+def _clean_staging(staging: Path | None) -> None:
+ if staging is not None and staging.exists():
+ shutil.rmtree(staging)
+
+
+def _manifest(
+ inspection: DeviceInspection,
+ serial: str,
+ goods_id: str,
+ state: str,
+ quantity: int,
+ screenshot_path: Path,
+ hierarchy_path: Path,
+ app_path: Path,
+) -> dict[str, Any]:
+ return {
+ "schema_version": 1,
+ "operation": "t105-quantity-gate2-readonly-evidence",
+ "captured_at": datetime.now(UTC).isoformat(),
+ "product": {
+ "goods_id": goods_id,
+ "canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}",
+ },
+ "human_declared_state": state,
+ "human_declared_quantity": quantity,
+ "human_declared_selection": TARGET_SELECTION,
+ "review_status": "human_review_required",
+ "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,
+ },
+ "artifacts": [
+ {
+ "path": screenshot_path.name,
+ "role": "quantity_state_raw_screenshot",
+ "sha256": _sha256_file(screenshot_path),
+ },
+ {
+ "path": hierarchy_path.name,
+ "role": "quantity_state_raw_hierarchy_local_only",
+ "sha256": _sha256_file(hierarchy_path),
+ },
+ {
+ "path": app_path.name,
+ "role": "quantity_state_app_identity",
+ "sha256": _sha256_file(app_path),
+ },
+ ],
+ }
diff --git a/client/tests/pdd/test_quantity_gate2.py b/client/tests/pdd/test_quantity_gate2.py
new file mode 100644
index 0000000..0343974
--- /dev/null
+++ b/client/tests/pdd/test_quantity_gate2.py
@@ -0,0 +1,452 @@
+"""T-105 数量两态只读取证与后续生产边界的离线测试。"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import base64
+from contextlib import redirect_stderr
+from functools import lru_cache
+from importlib.util import module_from_spec, spec_from_file_location
+from io import BytesIO, StringIO
+import json
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+
+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.pdd.quantity_gate2_spike import (
+ DECLARED_QUANTITIES,
+ EXPECTED_GOODS_ID,
+ QuantityGate2EvidenceCapturer,
+ QuantityGate2EvidenceError,
+ QuantityGate2EvidenceTimeoutError,
+ QuantityGate2ReadDevice,
+)
+
+
+SERIAL = "192.168.0.173:5555"
+HIERARCHY = ""
+
+
+@lru_cache(maxsize=1)
+def _png_base64() -> str:
+ raw = BytesIO()
+ Image.new("RGB", (1080, 2376), color="white").save(raw, format="PNG")
+ return base64.b64encode(raw.getvalue()).decode("ascii")
+
+
+class FakeAdbClient:
+ def __init__(self, *, model: str = "PKG110", android_version: str = "16") -> None:
+ self.calls: list[str] = []
+ self.inspection = DeviceInspection(
+ device=AdbDevice(serial=SERIAL, state="device", model=model),
+ model=model,
+ android_version=android_version,
+ )
+
+ def inspect(self, serial: str) -> DeviceInspection:
+ self.calls.append(serial)
+ return self.inspection
+
+
+class FakeReadDevice:
+ def __init__(
+ 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]:
+ self.calls.append(("app_info", package_name))
+ self.app_info_reads += 1
+ 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
+ if self.window_reads > 1 and self.post_screen_size is not None:
+ return self.post_screen_size
+ return self.screen_size
+
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
+ self.calls.append(("jsonrpc", method, params, timeout))
+ if method == "takeScreenshot":
+ return self.screenshot
+ if method == "dumpWindowHierarchy":
+ return self.hierarchy
+ raise AssertionError(f"unexpected RPC {method}")
+
+
+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)
+ assert spec is not None and spec.loader is not None
+ module = module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+def _namespace(**changes: object) -> argparse.Namespace:
+ values: dict[str, object] = {
+ "serial": SERIAL,
+ "goods_id": EXPECTED_GOODS_ID,
+ "state": "initial",
+ "declared_quantity": 1,
+ "output_dir": Path("evidence"),
+ "timeout": 10.0,
+ "adb": "adb",
+ }
+ values.update(changes)
+ return argparse.Namespace(**values)
+
+
+class QuantityGate2EvidenceTests(unittest.TestCase):
+ def _capturer(
+ self,
+ adb: FakeAdbClient,
+ device: FakeReadDevice,
+ *,
+ connector_calls: list[str] | None = None,
+ ) -> QuantityGate2EvidenceCapturer:
+ def connect(serial: str) -> FakeReadDevice:
+ if connector_calls is not None:
+ connector_calls.append(serial)
+ return device
+
+ return QuantityGate2EvidenceCapturer(adb, connect, 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()
+ connector_calls: list[str] = []
+ target = Path(temporary) / state
+
+ result = self._capturer(adb, device, connector_calls=connector_calls).capture(
+ SERIAL,
+ EXPECTED_GOODS_ID,
+ state,
+ quantity,
+ target,
+ )
+
+ manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+ app = json.loads(result.app_path.read_text(encoding="utf-8"))
+ self.assertEqual(adb.calls, [SERIAL])
+ self.assertEqual(connector_calls, [SERIAL])
+ self.assertEqual(
+ [call[1] for call in device.calls if call[0] == "jsonrpc"],
+ ["takeScreenshot", "dumpWindowHierarchy"],
+ )
+ self.assertEqual(manifest["operation"], "t105-quantity-gate2-readonly-evidence")
+ self.assertEqual(manifest["human_declared_state"], state)
+ self.assertEqual(manifest["human_declared_quantity"], quantity)
+ self.assertEqual(
+ manifest["human_declared_selection"],
+ {"color": "黑色CHA(纯棉)", "size": "M(建议100-115)"},
+ )
+ self.assertEqual(manifest["review_status"], "human_review_required")
+ self.assertEqual(manifest["product"]["goods_id"], EXPECTED_GOODS_ID)
+ self.assertEqual(app["package"], "com.xunmeng.pinduoduo")
+ self.assertEqual(
+ {path.name for path in target.iterdir()},
+ {"screenshot.png", "hierarchy.xml", "app.json", "manifest.json"},
+ )
+ serialized = result.manifest_path.read_text(encoding="utf-8")
+ self.assertNotIn(SERIAL, serialized)
+ self.assertNotIn("NewPageActivity", serialized)
+ self.assertNotIn("local raw page", serialized)
+ for artifact in manifest["artifacts"]:
+ self.assertEqual(len(artifact["sha256"]), 64)
+
+ def test_invalid_inputs_and_existing_target_fail_before_device_access(self) -> None:
+ scenarios = (
+ ("", EXPECTED_GOODS_ID, "initial", 1),
+ (f" {SERIAL}", EXPECTED_GOODS_ID, "initial", 1),
+ (SERIAL, "958756616606", "initial", 1),
+ (SERIAL, EXPECTED_GOODS_ID, "unknown", 1),
+ (SERIAL, EXPECTED_GOODS_ID, "initial", 2),
+ (SERIAL, EXPECTED_GOODS_ID, "target", 1),
+ (SERIAL, EXPECTED_GOODS_ID, "initial", True),
+ )
+ for serial, goods_id, state, quantity in scenarios:
+ with self.subTest(state=state, quantity=quantity), TemporaryDirectory() as temporary:
+ adb = FakeAdbClient()
+ connector_calls: list[str] = []
+ with self.assertRaises(QuantityGate2EvidenceError):
+ self._capturer(adb, FakeReadDevice(), connector_calls=connector_calls).capture(
+ serial,
+ goods_id,
+ state,
+ quantity,
+ Path(temporary) / "evidence",
+ )
+ self.assertEqual(adb.calls, [])
+ self.assertEqual(connector_calls, [])
+
+ with TemporaryDirectory() as temporary:
+ target = Path(temporary) / "evidence"
+ target.mkdir()
+ sentinel = target / "sentinel.txt"
+ sentinel.write_text("keep", encoding="utf-8")
+ adb = FakeAdbClient()
+ with self.assertRaises(QuantityGate2EvidenceError):
+ self._capturer(adb, FakeReadDevice()).capture(
+ SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
+ )
+ self.assertEqual(adb.calls, [])
+ self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+
+ 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))),
+ )
+ for adb, device in scenarios:
+ with self.subTest(device=device.__dict__), TemporaryDirectory() as temporary:
+ target = Path(temporary) / "evidence"
+ with self.assertRaises(QuantityGate2EvidenceError):
+ self._capturer(adb, device).capture(
+ SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
+ )
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+ self.assertEqual(
+ [call for call in device.calls if call[0] == "jsonrpc"],
+ [],
+ )
+
+ def test_invalid_screenshot_or_hierarchy_leaves_no_partial_evidence(self) -> None:
+ scenarios = (
+ FakeReadDevice(screenshot="not base64"),
+ FakeReadDevice(screenshot=123),
+ FakeReadDevice(hierarchy=""),
+ FakeReadDevice(hierarchy=123),
+ )
+ for device in scenarios:
+ with self.subTest(value=device.screenshot), TemporaryDirectory() as temporary:
+ target = Path(temporary) / "evidence"
+ with self.assertRaises(QuantityGate2EvidenceError):
+ self._capturer(FakeAdbClient(), device).capture(
+ SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
+ )
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+ def test_timeout_is_redacted_and_cleans_staging(self) -> None:
+ class TimeoutDevice(FakeReadDevice):
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
+ self.calls.append(("jsonrpc", method, params, timeout))
+ if method == "takeScreenshot":
+ raise TimeoutError(f"secret {SERIAL} C:\\private\\evidence")
+ return super().jsonrpc_call(method, params, timeout)
+
+ with TemporaryDirectory() as temporary:
+ target = Path(temporary) / "evidence"
+ with self.assertRaises(QuantityGate2EvidenceTimeoutError) as raised:
+ self._capturer(FakeAdbClient(), TimeoutDevice()).capture(
+ SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
+ )
+ self.assertNotIn(SERIAL, str(raised.exception))
+ self.assertNotIn("private", str(raised.exception))
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+ 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)),
+ )
+ for mutate in drift_mutators:
+ with TemporaryDirectory() as temporary:
+ device = FakeReadDevice()
+ mutate(device)
+ target = Path(temporary) / "evidence"
+ with self.assertRaises(QuantityGate2EvidenceError):
+ self._capturer(FakeAdbClient(), device).capture(
+ SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
+ )
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+ def test_capturer_is_one_shot_even_after_rejection(self) -> None:
+ capturer = self._capturer(FakeAdbClient(), FakeReadDevice())
+ with TemporaryDirectory() as temporary:
+ with self.assertRaises(QuantityGate2EvidenceError):
+ capturer.capture("", EXPECTED_GOODS_ID, "initial", 1, Path(temporary) / "bad")
+ with self.assertRaises(QuantityGate2EvidenceError):
+ capturer.capture(SERIAL, EXPECTED_GOODS_ID, "initial", 1, Path(temporary) / "good")
+
+ def test_read_protocol_and_capturer_expose_no_page_actions(self) -> None:
+ forbidden = {
+ "click",
+ "swipe",
+ "press",
+ "pressKey",
+ "send_keys",
+ "set_text",
+ "open_product",
+ "open_sku_panel",
+ "set_quantity",
+ "go_to_order_confirm",
+ "submit_order_once",
+ "pay",
+ }
+ self.assertTrue(forbidden.isdisjoint(QuantityGate2ReadDevice.__dict__))
+ self.assertEqual(
+ {name for name in QuantityGate2EvidenceCapturer.__dict__ if not name.startswith("_")},
+ {"capture"},
+ )
+
+
+class QuantityGate2CliAndStaticBoundaryTests(unittest.TestCase):
+ def test_cli_validation_rejects_all_unapproved_combinations(self) -> None:
+ script = _load_script()
+ invalid = (
+ _namespace(serial=""),
+ _namespace(serial=f" {SERIAL}"),
+ _namespace(goods_id="958756616606"),
+ _namespace(state="unknown"),
+ _namespace(declared_quantity=2),
+ _namespace(state="target", declared_quantity=1),
+ _namespace(output_dir=Path("")),
+ _namespace(timeout=0),
+ _namespace(timeout=float("inf")),
+ )
+ for arguments in invalid:
+ with self.subTest(arguments=arguments), self.assertRaises(ValueError):
+ script.validate_arguments(arguments) # type: ignore[attr-defined]
+
+ def test_cli_parses_exact_initial_and_target_arguments(self) -> None:
+ script = _load_script()
+ for state, quantity in DECLARED_QUANTITIES.items():
+ arguments = script.parse_arguments( # type: ignore[attr-defined]
+ [
+ "--serial",
+ SERIAL,
+ "--goods-id",
+ EXPECTED_GOODS_ID,
+ "--state",
+ state,
+ "--declared-quantity",
+ str(quantity),
+ "--output-dir",
+ f"evidence-{state}",
+ ]
+ )
+ script.validate_arguments(arguments) # type: ignore[attr-defined]
+
+ def test_cli_runtime_failure_does_not_echo_sensitive_values(self) -> None:
+ script = _load_script()
+ error = QuantityGate2EvidenceError(f"secret {SERIAL} C:\\private\\evidence raw body")
+ stderr = StringIO()
+ fake_capturer = unittest.mock.Mock()
+ fake_capturer.capture.side_effect = error
+ with (
+ patch.object(script, "QuantityGate2EvidenceCapturer", return_value=fake_capturer),
+ patch.dict(sys.modules, {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()}),
+ redirect_stderr(stderr),
+ ):
+ result = script.main(
+ [
+ "--serial",
+ SERIAL,
+ "--goods-id",
+ EXPECTED_GOODS_ID,
+ "--state",
+ "initial",
+ "--declared-quantity",
+ "1",
+ "--output-dir",
+ "C:\\private\\evidence",
+ ]
+ )
+ self.assertEqual(result, 1)
+ self.assertNotIn(SERIAL, stderr.getvalue())
+ self.assertNotIn("private", stderr.getvalue())
+ self.assertNotIn("raw body", stderr.getvalue())
+
+ def test_stage_one_sources_contain_only_approved_rpc_method_literals(self) -> None:
+ expected_by_source = {
+ Path("src/cmbuyer_client/pdd/quantity_gate2_spike.py"): {
+ "takeScreenshot",
+ "dumpWindowHierarchy",
+ },
+ Path("scripts/capture_quantity_gate2_spike.py"): set(),
+ }
+ for relative, expected in expected_by_source.items():
+ tree = ast.parse((CLIENT_ROOT / relative).read_text(encoding="utf-8"))
+ rpc_literals = {
+ node.args[1].value
+ for node in ast.walk(tree)
+ if isinstance(node, ast.Call)
+ and isinstance(node.func, ast.Name)
+ and node.func.id == "_read_rpc"
+ and len(node.args) > 1
+ and isinstance(node.args[1], ast.Constant)
+ and isinstance(node.args[1].value, str)
+ }
+ self.assertEqual(rpc_literals, expected)
+ forbidden_attributes = {
+ node.attr
+ for node in ast.walk(tree)
+ if isinstance(node, ast.Attribute)
+ and node.attr in {"click", "swipe", "press", "pressKey", "set_text", "send_keys", "start_pdd_view_intent"}
+ }
+ self.assertEqual(forbidden_attributes, set())
+
+ def test_t103_and_t104_sources_do_not_import_quantity_gate2(self) -> None:
+ forbidden_import = "quantity_gate2"
+ sources = (
+ CLIENT_ROOT / "src/cmbuyer_client/pdd/sku_selection.py",
+ CLIENT_ROOT / "src/cmbuyer_client/pdd/sku_selection_runner.py",
+ )
+ for source in sources:
+ self.assertNotIn(forbidden_import, source.read_text(encoding="utf-8"))
+
+
+if __name__ == "__main__":
+ unittest.main()