feat(client): add T-106 confirmation evidence capture
This commit is contained in:
QiuSW
@@ -0,0 +1,112 @@
|
||||
"""采集 T-106 人工准备的确认页四态证据；不执行页面操作。"""
|
||||
|
||||
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.order_confirm_spike import (
|
||||
Android16ForegroundReader,
|
||||
DECLARED_STATES,
|
||||
EXPECTED_GOODS_ID,
|
||||
OrderConfirmEvidenceCapturer,
|
||||
OrderConfirmEvidenceError,
|
||||
)
|
||||
|
||||
|
||||
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="采集 T-106 人工准备的确认页四态本机证据。")
|
||||
parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
|
||||
parser.add_argument("--goods-id", required=True, help="T-106 已批准的目标商品标识。")
|
||||
parser.add_argument(
|
||||
"--state",
|
||||
required=True,
|
||||
choices=DECLARED_STATES,
|
||||
help=(
|
||||
"人工声明状态：Gate2 后导航来源、确认页 Gate3、最终控件可见、"
|
||||
"人工一次返回后的安全页。"
|
||||
),
|
||||
)
|
||||
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-106 已批准目标。")
|
||||
if type(arguments.state) is not str or arguments.state not in DECLARED_STATES:
|
||||
raise ValueError("--state 必须是批准的人工声明状态。")
|
||||
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
|
||||
|
||||
adb_runner = SubprocessAdbRunner(arguments.adb)
|
||||
capturer = OrderConfirmEvidenceCapturer(
|
||||
AdbClient(adb_runner, timeout_seconds=arguments.timeout),
|
||||
NoReconnectUiautomatorConnector(
|
||||
adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
|
||||
u2.connect,
|
||||
),
|
||||
Android16ForegroundReader(adb_runner, arguments.timeout),
|
||||
timeout_seconds=arguments.timeout,
|
||||
)
|
||||
try:
|
||||
capturer.capture(
|
||||
arguments.serial,
|
||||
arguments.goods_id,
|
||||
arguments.state,
|
||||
arguments.output_dir,
|
||||
)
|
||||
except (DeviceConnectionError, OrderConfirmEvidenceError):
|
||||
# 第三方异常可能带设备、页面正文或本机目录，命令行只输出固定摘要。
|
||||
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())
|
||||
@@ -0,0 +1,345 @@
|
||||
"""T-106 确认页四态的纯只读取证；页面对应性只由人工确认。"""
|
||||
|
||||
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 re
|
||||
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, CommandRunner, 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)
|
||||
DECLARED_STATES = (
|
||||
"gate2-navigation-source",
|
||||
"confirm-gate3",
|
||||
"submit-control-visible",
|
||||
"returned-safe-page",
|
||||
)
|
||||
|
||||
|
||||
class OrderConfirmEvidenceError(RuntimeError):
|
||||
"""T-106 四态证据未形成完整原子产物。"""
|
||||
|
||||
|
||||
class OrderConfirmEvidenceTimeoutError(OrderConfirmEvidenceError):
|
||||
"""只读取证设备调用超时。"""
|
||||
|
||||
|
||||
class OrderConfirmReadDevice(Protocol):
|
||||
"""T-106 唯一设备边界；故意只有读取能力。"""
|
||||
|
||||
def app_info(self, package_name: str) -> dict[str, Any]: ...
|
||||
|
||||
def window_size(self) -> tuple[int, int]: ...
|
||||
|
||||
def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any: ...
|
||||
|
||||
|
||||
class OrderConfirmForegroundReader(Protocol):
|
||||
"""读取 Android 前台摘要，不暴露通用命令执行。"""
|
||||
|
||||
def read(self, serial: str) -> dict[str, str]: ...
|
||||
|
||||
|
||||
class Android16ForegroundReader:
|
||||
"""读取 Android 16 的唯一 top-resumed Activity。"""
|
||||
|
||||
_TOP_RESUMED_PATTERN = re.compile(
|
||||
r"(?m)^\s*topResumedActivity=ActivityRecord\{[^\r\n}]*?\s+u\d+\s+"
|
||||
r"(?P<package>[^/\s]+)/(?P<activity>[^\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 OrderConfirmEvidenceError("必须显式提供非空设备通道。")
|
||||
result = self._runner.run(
|
||||
("-s", serial, "shell", "dumpsys", "activity", "activities"),
|
||||
self._timeout_seconds,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise OrderConfirmEvidenceError("Android 前台摘要读取失败，未发布证据。")
|
||||
matches = list(self._TOP_RESUMED_PATTERN.finditer(result.stdout))
|
||||
if len(matches) != 1:
|
||||
raise OrderConfirmEvidenceError("Android 前台摘要不唯一，未发布证据。")
|
||||
match = matches[0]
|
||||
return {"package": match.group("package"), "activity": match.group("activity")}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OrderConfirmEvidenceResult:
|
||||
output_directory: Path
|
||||
manifest_path: Path
|
||||
screenshot_path: Path
|
||||
hierarchy_path: Path
|
||||
app_path: Path
|
||||
|
||||
|
||||
class OrderConfirmEvidenceCapturer:
|
||||
"""采集人工准备的单个稳定状态，不解析页面业务字段。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
adb_client: AdbClient,
|
||||
connector: Callable[[str], OrderConfirmReadDevice],
|
||||
foreground_reader: OrderConfirmForegroundReader,
|
||||
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
|
||||
|
||||
def capture(
|
||||
self,
|
||||
serial: str,
|
||||
goods_id: str,
|
||||
human_declared_state: str,
|
||||
output_directory: Path,
|
||||
) -> OrderConfirmEvidenceResult:
|
||||
if self._started:
|
||||
raise OrderConfirmEvidenceError("同一取证器不可重复调用。")
|
||||
self._started = True
|
||||
_validate_inputs(serial, goods_id, human_declared_state)
|
||||
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,
|
||||
self._foreground_reader.read(serial),
|
||||
)
|
||||
|
||||
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 OrderConfirmEvidenceError("页面截图无效，未发布证据。")
|
||||
_save_base64_screenshot(screenshot_payload, screenshot_path)
|
||||
_require_screenshot(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,
|
||||
self._foreground_reader.read(serial),
|
||||
)
|
||||
if final_app != initial_app:
|
||||
raise OrderConfirmEvidenceError("取证期间前台页面漂移，未发布证据。")
|
||||
|
||||
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,
|
||||
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, OrderConfirmEvidenceError):
|
||||
_clean_staging(staging)
|
||||
raise
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
|
||||
_clean_staging(staging)
|
||||
raise OrderConfirmEvidenceTimeoutError("确认页四态只读取证超时，未发布证据。") from error
|
||||
except (OSError, UnidentifiedImageError, ValueError) as error:
|
||||
_clean_staging(staging)
|
||||
raise OrderConfirmEvidenceError("确认页四态证据无法原子发布，未发布证据。") from error
|
||||
except Exception as error:
|
||||
_clean_staging(staging)
|
||||
raise OrderConfirmEvidenceError("确认页四态只读取证未完成，未发布证据。") from error
|
||||
|
||||
return OrderConfirmEvidenceResult(
|
||||
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) -> None:
|
||||
if type(serial) is not str or not serial.strip() or serial != serial.strip():
|
||||
raise OrderConfirmEvidenceError("必须显式提供非空设备通道。")
|
||||
if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID:
|
||||
raise OrderConfirmEvidenceError("商品不是 T-106 已批准取证目标。")
|
||||
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_STATES:
|
||||
raise OrderConfirmEvidenceError("人工声明状态无效。")
|
||||
|
||||
|
||||
def _validate_new_target(target: Path) -> None:
|
||||
if target.exists() or not target.name:
|
||||
raise OrderConfirmEvidenceError("输出目录必须是不存在的明确新目录。")
|
||||
|
||||
|
||||
def _require_expected_device(inspection: DeviceInspection) -> None:
|
||||
if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION:
|
||||
raise OrderConfirmEvidenceError("设备不是已批准取证组合。")
|
||||
|
||||
|
||||
def _require_read_precondition(
|
||||
device: OrderConfirmReadDevice,
|
||||
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 OrderConfirmEvidenceError("拼多多版本不是已批准取证版本。")
|
||||
if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
|
||||
raise OrderConfirmEvidenceError("拼多多不在前台。")
|
||||
activity = current.get("activity")
|
||||
if not isinstance(activity, str) or not activity.strip():
|
||||
raise OrderConfirmEvidenceError("前台应用摘要不完整。")
|
||||
if device.window_size() != EXPECTED_SCREEN_SIZE:
|
||||
raise OrderConfirmEvidenceError("屏幕坐标空间不是已批准尺寸。")
|
||||
return {"package": PDD_PACKAGE, "activity": activity, "pdd_version": EXPECTED_PDD_VERSION}
|
||||
|
||||
|
||||
def _read_rpc(
|
||||
device: OrderConfirmReadDevice,
|
||||
method: str,
|
||||
params: object,
|
||||
timeout_seconds: float,
|
||||
) -> object:
|
||||
return device.jsonrpc_call(method, params, timeout=timeout_seconds)
|
||||
|
||||
|
||||
def _require_screenshot(path: Path) -> None:
|
||||
with Image.open(path) as image:
|
||||
image.load()
|
||||
if image.size != EXPECTED_SCREEN_SIZE or image.format != "PNG":
|
||||
raise OrderConfirmEvidenceError("页面截图格式或尺寸无效。")
|
||||
|
||||
|
||||
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,
|
||||
screenshot_path: Path,
|
||||
hierarchy_path: Path,
|
||||
app_path: Path,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"operation": "t106-order-confirm-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,
|
||||
"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": "human_prepared_state_raw_screenshot",
|
||||
"sha256": _sha256_file(screenshot_path),
|
||||
},
|
||||
{
|
||||
"path": hierarchy_path.name,
|
||||
"role": "human_prepared_state_raw_hierarchy_local_only",
|
||||
"sha256": _sha256_file(hierarchy_path),
|
||||
},
|
||||
{
|
||||
"path": app_path.name,
|
||||
"role": "human_prepared_state_app_identity",
|
||||
"sha256": _sha256_file(app_path),
|
||||
},
|
||||
],
|
||||
}
|
||||
@@ -0,0 +1,597 @@
|
||||
"""T-106 确认页四态纯只读取证的离线测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import base64
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
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, CommandResult, DeviceInspection
|
||||
from cmbuyer_client.pdd.order_confirm_spike import (
|
||||
Android16ForegroundReader,
|
||||
DECLARED_STATES,
|
||||
EXPECTED_GOODS_ID,
|
||||
OrderConfirmEvidenceCapturer,
|
||||
OrderConfirmEvidenceError,
|
||||
OrderConfirmEvidenceTimeoutError,
|
||||
OrderConfirmForegroundReader,
|
||||
OrderConfirmReadDevice,
|
||||
)
|
||||
|
||||
|
||||
SERIAL = "192.168.0.173:5555"
|
||||
HIERARCHY = (
|
||||
"<?xml version='1.0' encoding='UTF-8'?><hierarchy rotation='0'>"
|
||||
"<node text='local raw page' /></hierarchy>"
|
||||
)
|
||||
TOP_RESUMED = (
|
||||
" topResumedActivity=ActivityRecord{101034589 u0 "
|
||||
"com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n"
|
||||
)
|
||||
|
||||
|
||||
@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",
|
||||
screen_size: tuple[int, int] = (1080, 2376),
|
||||
screenshot: object | None = None,
|
||||
hierarchy: object = HIERARCHY,
|
||||
) -> None:
|
||||
self.version = version
|
||||
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.window_reads = 0
|
||||
self.post_version: 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 else self.version
|
||||
return {"versionName": version}
|
||||
|
||||
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("unexpected read RPC")
|
||||
|
||||
|
||||
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 else self.package
|
||||
activity = self.post_activity if len(self.calls) > 1 and self.post_activity 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:
|
||||
path = CLIENT_ROOT / "scripts" / "capture_order_confirm_spike.py"
|
||||
specification = spec_from_file_location("capture_order_confirm_spike_for_test", path)
|
||||
assert specification is not None and specification.loader is not None
|
||||
module = module_from_spec(specification)
|
||||
specification.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _namespace(**changes: object) -> argparse.Namespace:
|
||||
values: dict[str, object] = {
|
||||
"serial": SERIAL,
|
||||
"goods_id": EXPECTED_GOODS_ID,
|
||||
"state": DECLARED_STATES[0],
|
||||
"output_dir": Path("evidence"),
|
||||
"timeout": 10.0,
|
||||
"adb": "adb",
|
||||
}
|
||||
values.update(changes)
|
||||
return argparse.Namespace(**values)
|
||||
|
||||
|
||||
class Android16ForegroundReaderTests(unittest.TestCase):
|
||||
def test_reads_exact_unique_top_resumed_activity(self) -> None:
|
||||
runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED))
|
||||
reader = Android16ForegroundReader(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_duplicate_or_failed_summary_is_rejected(self) -> None:
|
||||
outputs = (
|
||||
"",
|
||||
"mResumedActivity: ActivityRecord{1 u0 com.xunmeng.pinduoduo/.Main t1}\n",
|
||||
TOP_RESUMED + TOP_RESUMED,
|
||||
"topResumedActivity=null\n",
|
||||
)
|
||||
for output in outputs:
|
||||
with self.subTest(output=output):
|
||||
reader = Android16ForegroundReader(
|
||||
FakeCommandRunner(CommandResult(stdout=output)),
|
||||
timeout_seconds=7,
|
||||
)
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
reader.read(SERIAL)
|
||||
|
||||
reader = Android16ForegroundReader(
|
||||
FakeCommandRunner(CommandResult(stdout=TOP_RESUMED, returncode=1)),
|
||||
timeout_seconds=7,
|
||||
)
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
reader.read(SERIAL)
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
reader.read(f" {SERIAL}")
|
||||
|
||||
|
||||
class OrderConfirmEvidenceTests(unittest.TestCase):
|
||||
def _capturer(
|
||||
self,
|
||||
adb: FakeAdbClient,
|
||||
device: FakeReadDevice,
|
||||
*,
|
||||
foreground: FakeForegroundReader | None = None,
|
||||
connector_calls: list[str] | None = None,
|
||||
) -> OrderConfirmEvidenceCapturer:
|
||||
def connect(serial: str) -> FakeReadDevice:
|
||||
if connector_calls is not None:
|
||||
connector_calls.append(serial)
|
||||
return device
|
||||
|
||||
return OrderConfirmEvidenceCapturer(
|
||||
adb, # type: ignore[arg-type]
|
||||
connect,
|
||||
foreground or FakeForegroundReader(),
|
||||
timeout_seconds=2,
|
||||
)
|
||||
|
||||
def test_all_four_human_states_publish_only_raw_read_evidence(self) -> None:
|
||||
self.assertEqual(len(DECLARED_STATES), 4)
|
||||
for state in DECLARED_STATES:
|
||||
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,
|
||||
foreground=foreground,
|
||||
connector_calls=connector_calls,
|
||||
).capture(SERIAL, EXPECTED_GOODS_ID, state, target)
|
||||
|
||||
manifest_text = result.manifest_path.read_text(encoding="utf-8")
|
||||
manifest = json.loads(manifest_text)
|
||||
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"],
|
||||
)
|
||||
self.assertEqual(manifest["operation"], "t106-order-confirm-readonly-evidence")
|
||||
self.assertEqual(manifest["human_declared_state"], state)
|
||||
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"},
|
||||
)
|
||||
self.assertNotIn(SERIAL, manifest_text)
|
||||
self.assertNotIn("NewPageActivity", manifest_text)
|
||||
self.assertNotIn("local raw page", manifest_text)
|
||||
for artifact in manifest["artifacts"]:
|
||||
self.assertEqual(len(artifact["sha256"]), 64)
|
||||
|
||||
def test_invalid_inputs_and_existing_target_stop_before_device_access(self) -> None:
|
||||
scenarios = (
|
||||
("", EXPECTED_GOODS_ID, DECLARED_STATES[0]),
|
||||
(f" {SERIAL}", EXPECTED_GOODS_ID, DECLARED_STATES[0]),
|
||||
(SERIAL, "958756616606", DECLARED_STATES[0]),
|
||||
(SERIAL, EXPECTED_GOODS_ID, "unknown"),
|
||||
)
|
||||
for serial, goods_id, state in scenarios:
|
||||
with self.subTest(state=state), TemporaryDirectory() as temporary:
|
||||
adb = FakeAdbClient()
|
||||
connector_calls: list[str] = []
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
self._capturer(
|
||||
adb,
|
||||
FakeReadDevice(),
|
||||
connector_calls=connector_calls,
|
||||
).capture(serial, goods_id, state, 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(OrderConfirmEvidenceError):
|
||||
self._capturer(adb, FakeReadDevice()).capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], 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(), 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, foreground in scenarios:
|
||||
with self.subTest(device=device.__dict__), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
self._capturer(adb, device, foreground=foreground).capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], 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_never_publishes(self) -> None:
|
||||
devices = (
|
||||
FakeReadDevice(screenshot="not base64"),
|
||||
FakeReadDevice(screenshot=123),
|
||||
FakeReadDevice(hierarchy="<not-hierarchy />"),
|
||||
FakeReadDevice(hierarchy=123),
|
||||
)
|
||||
for device in devices:
|
||||
with self.subTest(value=device.screenshot), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
self._capturer(FakeAdbClient(), device).capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
|
||||
)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
|
||||
|
||||
def test_timeout_and_unexpected_failure_are_redacted(self) -> None:
|
||||
class BrokenDevice(FakeReadDevice):
|
||||
def __init__(self, failure: BaseException) -> None:
|
||||
super().__init__()
|
||||
self.failure = failure
|
||||
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
|
||||
raise self.failure
|
||||
|
||||
failures = (
|
||||
(TimeoutError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceTimeoutError),
|
||||
(RuntimeError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceError),
|
||||
)
|
||||
for failure, expected in failures:
|
||||
with self.subTest(expected=expected), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(expected) as raised:
|
||||
self._capturer(FakeAdbClient(), BrokenDevice(failure)).capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], 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_identity_drift_never_publishes(self) -> None:
|
||||
mutators = (
|
||||
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 mutators:
|
||||
with TemporaryDirectory() as temporary:
|
||||
device = FakeReadDevice()
|
||||
foreground = FakeForegroundReader()
|
||||
mutate(device, foreground)
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
self._capturer(
|
||||
FakeAdbClient(), device, foreground=foreground
|
||||
).capture(SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target)
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
|
||||
|
||||
def test_atomic_publish_failure_cleans_staging(self) -> None:
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with (
|
||||
patch(
|
||||
"cmbuyer_client.pdd.order_confirm_spike.os.rename",
|
||||
side_effect=OSError(f"private {SERIAL}"),
|
||||
),
|
||||
self.assertRaises(OrderConfirmEvidenceError) as raised,
|
||||
):
|
||||
self._capturer(FakeAdbClient(), FakeReadDevice()).capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
|
||||
)
|
||||
self.assertNotIn(SERIAL, str(raised.exception))
|
||||
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(OrderConfirmEvidenceError):
|
||||
capturer.capture("", EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "bad")
|
||||
with self.assertRaises(OrderConfirmEvidenceError):
|
||||
capturer.capture(
|
||||
SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "good"
|
||||
)
|
||||
|
||||
def test_public_boundaries_expose_reading_only(self) -> None:
|
||||
forbidden = {
|
||||
"click",
|
||||
"swipe",
|
||||
"press",
|
||||
"pressKey",
|
||||
"send_keys",
|
||||
"set_text",
|
||||
"start_pdd_view_intent",
|
||||
"open_product",
|
||||
"open_sku_panel",
|
||||
"set_quantity",
|
||||
"go_to_order_confirm",
|
||||
"submit_order_once",
|
||||
"pay",
|
||||
}
|
||||
self.assertTrue(forbidden.isdisjoint(OrderConfirmReadDevice.__dict__))
|
||||
self.assertTrue(forbidden.isdisjoint(OrderConfirmForegroundReader.__dict__))
|
||||
self.assertEqual(
|
||||
{name for name in Android16ForegroundReader.__dict__ if not name.startswith("_")},
|
||||
{"read"},
|
||||
)
|
||||
self.assertEqual(
|
||||
{name for name in OrderConfirmEvidenceCapturer.__dict__ if not name.startswith("_")},
|
||||
{"capture"},
|
||||
)
|
||||
|
||||
|
||||
class OrderConfirmCliAndStaticBoundaryTests(unittest.TestCase):
|
||||
def test_cli_parses_each_approved_state(self) -> None:
|
||||
script = _load_script()
|
||||
for state in DECLARED_STATES:
|
||||
arguments = script.parse_arguments( # type: ignore[attr-defined]
|
||||
[
|
||||
"--serial",
|
||||
SERIAL,
|
||||
"--goods-id",
|
||||
EXPECTED_GOODS_ID,
|
||||
"--state",
|
||||
state,
|
||||
"--output-dir",
|
||||
f"evidence-{state}",
|
||||
]
|
||||
)
|
||||
script.validate_arguments(arguments) # type: ignore[attr-defined]
|
||||
|
||||
def test_cli_validation_rejects_unapproved_values(self) -> None:
|
||||
script = _load_script()
|
||||
invalid = (
|
||||
_namespace(serial=""),
|
||||
_namespace(serial=f" {SERIAL}"),
|
||||
_namespace(goods_id="958756616606"),
|
||||
_namespace(state="unknown"),
|
||||
_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_runtime_failure_never_echoes_sensitive_values(self) -> None:
|
||||
script = _load_script()
|
||||
fake_capturer = unittest.mock.Mock()
|
||||
fake_capturer.capture.side_effect = OrderConfirmEvidenceError(
|
||||
f"secret {SERIAL} C:\\private\\raw.xml body"
|
||||
)
|
||||
stderr = StringIO()
|
||||
with (
|
||||
patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer),
|
||||
patch.dict(
|
||||
sys.modules,
|
||||
{"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()},
|
||||
),
|
||||
redirect_stderr(stderr),
|
||||
):
|
||||
result = script.main( # type: ignore[attr-defined]
|
||||
[
|
||||
"--serial",
|
||||
SERIAL,
|
||||
"--goods-id",
|
||||
EXPECTED_GOODS_ID,
|
||||
"--state",
|
||||
DECLARED_STATES[0],
|
||||
"--output-dir",
|
||||
"C:\\private\\evidence",
|
||||
]
|
||||
)
|
||||
self.assertEqual(result, 1)
|
||||
self.assertNotIn(SERIAL, stderr.getvalue())
|
||||
self.assertNotIn("private", stderr.getvalue())
|
||||
self.assertNotIn("body", stderr.getvalue())
|
||||
|
||||
def test_cli_success_does_not_echo_local_path_or_device(self) -> None:
|
||||
script = _load_script()
|
||||
fake_capturer = unittest.mock.Mock()
|
||||
stdout = StringIO()
|
||||
with (
|
||||
patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer),
|
||||
patch.dict(
|
||||
sys.modules,
|
||||
{"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()},
|
||||
),
|
||||
redirect_stdout(stdout),
|
||||
):
|
||||
result = script.main( # type: ignore[attr-defined]
|
||||
[
|
||||
"--serial",
|
||||
SERIAL,
|
||||
"--goods-id",
|
||||
EXPECTED_GOODS_ID,
|
||||
"--state",
|
||||
DECLARED_STATES[0],
|
||||
"--output-dir",
|
||||
"C:\\private\\evidence",
|
||||
]
|
||||
)
|
||||
self.assertEqual(result, 0)
|
||||
self.assertNotIn(SERIAL, stdout.getvalue())
|
||||
self.assertNotIn("private", stdout.getvalue())
|
||||
fake_capturer.capture.assert_called_once()
|
||||
|
||||
def test_sources_have_only_approved_read_rpc_literals_and_no_ui_mutators(self) -> None:
|
||||
expected_by_source = {
|
||||
Path("src/cmbuyer_client/pdd/order_confirm_spike.py"): {
|
||||
"takeScreenshot",
|
||||
"dumpWindowHierarchy",
|
||||
},
|
||||
Path("scripts/capture_order_confirm_spike.py"): set(),
|
||||
}
|
||||
forbidden_attributes = {
|
||||
"click",
|
||||
"swipe",
|
||||
"press",
|
||||
"pressKey",
|
||||
"send_keys",
|
||||
"set_text",
|
||||
"start_pdd_view_intent",
|
||||
"open_product",
|
||||
"open_sku_panel",
|
||||
"set_quantity_and_readback",
|
||||
"go_to_order_confirm",
|
||||
"submit_order_once",
|
||||
}
|
||||
forbidden_import_fragments = {
|
||||
"quantity_gate2_runner",
|
||||
"sku_selection_runner",
|
||||
"submission",
|
||||
"payment",
|
||||
}
|
||||
for relative, expected_rpc in expected_by_source.items():
|
||||
source = (CLIENT_ROOT / relative).read_text(encoding="utf-8")
|
||||
tree = ast.parse(source)
|
||||
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_rpc)
|
||||
observed_attributes = {
|
||||
node.attr
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Attribute) and node.attr in forbidden_attributes
|
||||
}
|
||||
self.assertEqual(observed_attributes, set())
|
||||
imported_modules = {
|
||||
alias.name
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom))
|
||||
for alias in node.names
|
||||
}
|
||||
for fragment in forbidden_import_fragments:
|
||||
self.assertTrue(all(fragment not in module for module in imported_modules))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user