from __future__ import annotations
import ast
import base64
from contextlib import redirect_stderr, redirect_stdout
from functools import lru_cache
from io import BytesIO, StringIO
import importlib.util
import json
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from xml.etree import ElementTree
from PIL import Image
import cmbuyer_client.pdd as pdd
import cmbuyer_client.pdd.sku_selection_runner as runner_module
from cmbuyer_client.device.adb import AdbDevice, DeviceConnectionError, DeviceInspection
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
from cmbuyer_client.pdd.sku_selection import (
_PanelProfile,
SkuPanelDevice,
_action_bounds,
_classify_panel,
_parse_nodes,
resolve_task_selection,
)
from cmbuyer_client.pdd.sku_selection_runner import (
SkuExitSpikeCapturer,
SkuExitSpikeError,
SkuSelectionDeviceAdapterError,
SkuSelectionRunError,
SkuSelectionScreenshotError,
UiautomatorSkuExitAdapter,
UiautomatorSkuPanelAdapter,
safe_failure_stage,
)
_FIXTURES = Path(__file__).with_name("fixtures")
_EMPTY_FIXTURE = _FIXTURES / "sku_panel_empty_8_17_0.xml"
_COLOR_FIXTURE = _FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml"
_CURRENT_ONLY_EMPTY_FIXTURE = _FIXTURES / "sku_panel_empty_current_only_8_17_0.xml"
_CURRENT_ONLY_COLOR_FIXTURE = _FIXTURES / "sku_panel_color_selected_size_hidden_current_only_8_17_0.xml"
_S_FIXTURE = _FIXTURES / "sku_panel_size_s_selected_8_17_0.xml"
_M_FIXTURE = _FIXTURES / "sku_panel_size_m_restored_8_17_0.xml"
_FIXTURE = _M_FIXTURE
_ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml"
_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
_TASK_COLOR = "黑色CHA(纯棉)"
_TASK_SIZE = "M(建议100-115)"
_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
def _revealed_current_only() -> str:
# 与 spike 测试共用由已验 M fixture 机械恢复出的 current-only 未选态；生产判据只在源码一处。
from tests.pdd.test_sku_reveal_spike import _candidate_current_only
return _candidate_current_only()
@lru_cache(maxsize=1)
def _png_base64() -> str:
image = Image.new("RGB", (1080, 2376), "white")
raw = BytesIO()
image.save(raw, format="PNG")
return base64.b64encode(raw.getvalue()).decode("ascii")
class _RawDevice:
def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
self.hierarchy = hierarchy
self.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
self.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8")
self.revealed_hierarchy = _revealed_current_only()
self.version = "8.17.0"
self.package = "com.xunmeng.pinduoduo"
self.screenshot = _png_base64() if screenshot is None else screenshot
self.calls: list[tuple[object, ...]] = []
self.fail_color_readback = False
def app_info(self, package_name: str) -> dict[str, str]:
self.calls.append(("app_info", package_name))
return {"versionName": self.version}
def app_current(self) -> dict[str, str]:
self.calls.append(("app_current",))
return {"package": self.package}
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
self.calls.append(("jsonrpc", method, params, timeout))
if method == "dumpWindowHierarchy":
return self.hierarchy
if method == "takeScreenshot":
return self.screenshot
if method == "pressKey":
self.hierarchy = ""
return ""
if method == "click":
if not isinstance(params, list) or len(params) != 2:
raise AssertionError(params)
self._apply_tap(int(params[0]), int(params[1]))
return ""
if method == "swipe":
if params != [360, 1900, 360, 1300, 30]:
raise AssertionError(params)
self.hierarchy = self.revealed_hierarchy
return ""
raise AssertionError(method)
def _apply_tap(self, x: int, y: int) -> None:
if (
(x, y) == (865, 2218)
and "快要抢光 ¥ 12.88" in self.hierarchy
and "[396,498][895,570]" not in self.hierarchy
):
self.hierarchy = self.panel_hierarchy
return
if (x, y) == (528, 1387):
self.hierarchy = (
_EMPTY_FIXTURE.read_text(encoding="utf-8")
if self.fail_color_readback
else self.color_hierarchy
)
return
if (x, y) == (635, 1624):
self.hierarchy = _M_FIXTURE.read_text(encoding="utf-8")
return
raise AssertionError((x, y))
def window_size(self) -> tuple[int, int]:
self.calls.append(("window_size",))
return 1080, 2376
def select_alternates(self) -> None:
self.panel_hierarchy = _S_FIXTURE.read_text(encoding="utf-8")
if self.hierarchy != _PRODUCT_PAGE:
self.hierarchy = self.panel_hierarchy
def _center(bounds: str) -> tuple[int, int]:
left_top, right_bottom = bounds.split("][")
left, top = (int(value) for value in left_top.removeprefix("[").split(","))
right, bottom = (int(value) for value in right_bottom.removesuffix("]").split(","))
return left + (right - left) // 2, top + (bottom - top) // 2
def _flow(device: _RawDevice, timeout_seconds: float = 0.02) -> SkuSelectionFlow:
now = [0.0]
return SkuSelectionFlow(
UiautomatorSkuPanelAdapter(device, 10),
timeout_seconds,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
def _fast_runner_flow(
device: SkuPanelDevice,
entry_wait_timeout_seconds: float = 0.02,
) -> SkuSelectionFlow:
now = [0.0]
return SkuSelectionFlow(
device,
entry_wait_timeout_seconds,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
def _mutate_unique_node(
hierarchy: str,
*,
attribute: str,
value: str | None,
text: str | None = None,
desc: str | None = None,
class_name: str | None = None,
bounds: str | None = None,
) -> str:
root = ElementTree.fromstring(hierarchy)
matches = [
node
for node in root.iter("node")
if (text is None or node.get("text") == text)
and (desc is None or node.get("content-desc") == desc)
and (class_name is None or node.get("class") == class_name)
and (bounds is None or node.get("bounds") == bounds)
]
if len(matches) != 1:
raise AssertionError(f"fixture node match count: {len(matches)}")
if value is None:
matches[0].attrib.pop(attribute, None)
else:
matches[0].set(attribute, value)
return ElementTree.tostring(root, encoding="unicode")
def _nested_dual_price_layout(
hierarchy: str,
*,
header_bounds: str,
row_bounds: str,
summary_bounds: str,
content_bounds: str,
price_frame_bounds: str,
) -> str:
root = ElementTree.fromstring(hierarchy)
parents = {child: parent for parent in root.iter() for child in parent}
header = next(
node for node in root.iter("node")
if node.get("class") == "android.widget.LinearLayout"
and node.get("bounds") == header_bounds
)
row = next(
node for node in root.iter("node")
if node.get("class") == "android.widget.LinearLayout"
and node.get("bounds") == row_bounds
)
summary = next(
node for node in root.iter("node")
if node.get("class") == "android.widget.TextView"
and node.get("bounds") == summary_bounds
)
parents[row].remove(row)
parents[summary].remove(summary)
def inert(class_name: str, bounds: str) -> ElementTree.Element:
return ElementTree.Element(
"node",
{
"package": "com.xunmeng.pinduoduo",
"class": class_name,
"bounds": bounds,
"clickable": "false",
"enabled": "true",
"visible-to-user": "true",
"selected": "false",
"scrollable": "false",
},
)
content_frame = inert("android.widget.FrameLayout", content_bounds)
relative = inert("android.widget.RelativeLayout", content_bounds)
content = inert("android.view.ViewGroup", content_bounds)
price_frame = inert("android.widget.FrameLayout", price_frame_bounds)
header.append(content_frame)
content_frame.append(relative)
relative.append(content)
content.append(price_frame)
price_frame.append(row)
content.append(summary)
return ElementTree.tostring(root, encoding="unicode")
def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]:
parents = {child: parent for parent in root.iter() for child in parent}
child = next(node for node in root.iter("node") if node.get("text") == "快要抢光 ¥ 12.88")
return [child, parents[child]]
def _entry_sibling(root: ElementTree.Element) -> ElementTree.Element:
return next(node for node in root.iter("node") if node.get("text") == "免拼购买")
def _mutate_entry(depth: int, attribute: str, value: str) -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
_entry_chain(root)[depth].set(attribute, value)
return ElementTree.tostring(root, encoding="unicode")
def _without_entry() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.remove(_entry_chain(root)[1])
return ElementTree.tostring(root, encoding="unicode")
def _duplicate_entry() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
entry_root = _entry_chain(root)[1]
root.append(ElementTree.fromstring(ElementTree.tostring(entry_root, encoding="unicode")))
return ElementTree.tostring(root, encoding="unicode")
def _extra_entry_action_ancestor() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
action = _entry_chain(root)[1]
root.append(ElementTree.Element("node", dict(action.attrib)))
return ElementTree.tostring(root, encoding="unicode")
def _entry_with_panel_price_marker() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.append(ElementTree.Element("node", {"bounds": "[396,498][895,570]"}))
return ElementTree.tostring(root, encoding="unicode")
def _dynamic_product_page(value: str, action_desc: str = "") -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.set("dynamic-page-value", value)
_entry_chain(root)[1].set("content-desc", action_desc or "快要抢光¥12.88")
return ElementTree.tostring(root, encoding="unicode")
def _with_overlapping_clickable(package: str, class_name: str, bounds: str) -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.append(
ElementTree.Element(
"node",
{
"package": package,
"class": class_name,
"bounds": bounds,
"clickable": "true",
"enabled": "true",
"visible-to-user": "true",
},
)
)
return ElementTree.tostring(root, encoding="unicode")
def _with_action_subtree_child(
text: str,
*,
bounds: str = "[500,2170][600,2200]",
) -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
action = _entry_chain(root)[1]
ElementTree.SubElement(
action,
"node",
{
"text": text,
"package": "com.xunmeng.pinduoduo",
"class": "android.widget.TextView",
"bounds": bounds,
"clickable": "false",
"enabled": "true",
"visible-to-user": "true",
},
)
return ElementTree.tostring(root, encoding="unicode")
def _mutate_entry_sibling(attribute: str, value: str) -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
_entry_sibling(root).set(attribute, value)
return ElementTree.tostring(root, encoding="unicode")
def _without_entry_sibling() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
_entry_chain(root)[1].remove(_entry_sibling(root))
return ElementTree.tostring(root, encoding="unicode")
def _entry_sibling_elsewhere() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.append(ElementTree.fromstring(ElementTree.tostring(_entry_sibling(root), encoding="unicode")))
return ElementTree.tostring(root, encoding="unicode")
def _home_with_unverified_entry_labels() -> str:
root = ElementTree.Element("hierarchy")
for index in range(2):
ElementTree.SubElement(
root,
"node",
{
"text": "快要抢光",
"package": "com.xunmeng.pinduoduo",
"class": "android.widget.TextView",
"clickable": "false",
"enabled": "true",
"visible-to-user": "true",
"bounds": f"[{index},{index}][{index + 1},{index + 1}]",
},
)
return ElementTree.tostring(root, encoding="unicode")
def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]
def _tap_centers(device: _RawDevice) -> list[tuple[int, int]]:
return [tuple(call[2]) for call in _actions(device, "click")] # type: ignore[misc]
class _FakeAdb:
def __init__(self) -> None:
self.calls: list[tuple[object, ...]] = []
self.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "PKG110", "16")
self.on_intent: callable | None = None
def inspect(self, serial: str) -> DeviceInspection:
self.calls.append(("inspect", serial))
return self.inspection
def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
self.calls.append(("intent", serial, goods_id))
if self.on_intent is not None:
self.on_intent()
return object()
class SkuSelectionFlowTests(unittest.TestCase):
def _assert_entry_rejected_without_click(self, hierarchy: str) -> None:
now = [0.0]
device = _RawDevice(hierarchy)
flow = SkuSelectionFlow(
UiautomatorSkuPanelAdapter(device, 10),
0.01,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_current_only_exact_transition_reveals_then_selects_m_once(self) -> None:
device = _RawDevice()
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(flow.read_sku_unit_price(), "12.88")
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_actions(device, "pressKey"), [])
restored = _RawDevice(_S_FIXTURE.read_text(encoding="utf-8"))
restored_flow = _flow(restored)
restored_flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(restored_flow.read_sku_unit_price(), "12.88")
self.assertEqual(_tap_centers(restored), [(635, 1624)])
self.assertEqual(_actions(restored, "swipe"), [])
def test_dual_color_only_is_zero_action_before_reveal(self) -> None:
for fixture in (_COLOR_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE):
with self.subTest(fixture=fixture.name):
device = _RawDevice(fixture.read_text(encoding="utf-8"))
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_actions(device, "click"), [])
self.assertEqual(_actions(device, "swipe"), [])
def test_revealed_current_only_is_a_unique_profile_but_not_a_reentry_shortcut(self) -> None:
hierarchy = _revealed_current_only()
spec = _classify_panel(_parse_nodes(hierarchy))
self.assertIs(spec.profile, _PanelProfile.SIZE_VISIBLE_UNSELECTED)
self.assertEqual(spec.price_layout.__class__.__name__, "_CurrentOnlyPriceLayout")
device = _RawDevice(hierarchy)
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_actions(device, "click"), [])
self.assertEqual(_actions(device, "swipe"), [])
def test_revealed_exact_m_contract_rejects_missing_duplicate_disabled_near_and_other_selected(self) -> None:
base = _revealed_current_only()
for kind in ("missing", "duplicate", "disabled", "near", "other_selected"):
root = ElementTree.fromstring(base)
parents = {child: parent for parent in root.iter() for child in parent}
m_action = next(
node for node in root.iter("node")
if node.get("text") == "M（建议100-115）"
)
wrapper = parents[m_action]
if kind == "missing":
parents[wrapper].remove(wrapper)
elif kind == "duplicate":
parents[wrapper].append(
ElementTree.fromstring(ElementTree.tostring(wrapper, encoding="unicode"))
)
elif kind == "disabled":
m_action.set("enabled", "false")
elif kind == "near":
m_action.set("text", "M（建议100-115）相近")
else:
s_action = next(
node for node in root.iter("node")
if node.get("text") == "S（建议80-100）"
)
s_action.set("selected", "true")
with self.subTest(kind=kind), self.assertRaises(SkuSelectionError):
_classify_panel(
_parse_nodes(ElementTree.tostring(root, encoding="unicode"))
)
def test_reveal_requires_two_consecutive_stable_frames_and_terminal_blocks_reentry(self) -> None:
class UnstableRevealDevice(_RawDevice):
def __init__(self) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.after_reveal_reads = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy" and _actions(self, "swipe"):
self.after_reveal_reads += 1
candidate = _revealed_current_only()
self.hierarchy = (
candidate
if self.after_reveal_reads % 2
else candidate.replace("请选择： 尺码", "未知摘要")
)
return super().jsonrpc_call(method, params, timeout)
device = UnstableRevealDevice()
flow = _flow(device)
with self.assertRaises(SkuSelectionError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
action_count = len(_actions(device, "click")) + len(_actions(device, "swipe"))
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387)])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(
len(_actions(device, "click")) + len(_actions(device, "swipe")),
action_count,
)
def test_reveal_timeout_is_read_only_reconciled_without_m_back_or_retry(self) -> None:
class RevealTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.delivered = delivered
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "swipe":
if self.delivered:
super().jsonrpc_call(method, params, timeout)
else:
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private reveal timeout")
return super().jsonrpc_call(method, params, timeout)
for delivered in (False, True):
with self.subTest(delivered=delivered):
device = RevealTimeoutDevice(delivered)
flow = _flow(device)
with self.assertRaises(SkuSelectionRunError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
try:
flow.reconcile_pending_action()
except SkuSelectionError:
pass
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387)])
self.assertEqual(_actions(device, "pressKey"), [])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(len(_actions(device, "swipe")), 1)
def test_m_timeout_is_read_only_reconciled_without_back_or_retry(self) -> None:
class MTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.delivered = delivered
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click" and params == [635, 1624]:
if self.delivered:
super().jsonrpc_call(method, params, timeout)
else:
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private M timeout")
return super().jsonrpc_call(method, params, timeout)
for delivered in (False, True):
with self.subTest(delivered=delivered):
device = MTimeoutDevice(delivered)
flow = _flow(device)
with self.assertRaises(SkuSelectionRunError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
try:
flow.reconcile_pending_action()
except SkuSelectionError:
pass
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
self.assertEqual(_actions(device, "pressKey"), [])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
def test_foreground_is_freshly_rechecked_before_s_reveal_and_m_mutations(self) -> None:
class ForegroundDriftDevice(_RawDevice):
def __init__(self, hierarchy: str, flip_at: int) -> None:
super().__init__(hierarchy)
self.flip_at = flip_at
self.current_reads = 0
def app_current(self) -> dict[str, str]:
current = super().app_current()
self.current_reads += 1
if self.current_reads == self.flip_at:
return {"package": "other"}
return current
cases = (
(
"s",
_S_FIXTURE.read_text(encoding="utf-8"),
2,
[],
0,
),
(
"reveal",
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
4,
[(528, 1387)],
0,
),
(
"m",
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
7,
[(528, 1387)],
1,
),
)
for name, hierarchy, flip_at, expected_taps, expected_swipes in cases:
with self.subTest(name=name):
device = ForegroundDriftDevice(hierarchy, flip_at)
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_tap_centers(device), expected_taps)
self.assertEqual(len(_actions(device, "swipe")), expected_swipes)
self.assertEqual(_actions(device, "pressKey"), [])
def test_current_only_empty_to_color_only_uses_the_matching_exact_spec(self) -> None:
device = _RawDevice()
device.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
device.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8")
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
self.assertEqual(len(_actions(device, "swipe")), 1)
matched = _classify_panel(_parse_nodes(device.hierarchy))
self.assertIs(matched.profile, _PanelProfile.TARGETS_SELECTED)
self.assertEqual(
matched.price_layout.__class__.__name__,
"_DualPriceLayout",
)
def test_dual_and_current_only_variants_each_match_one_complete_spec(self) -> None:
cases = (
(_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY, "_DualPriceLayout"),
(_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, "_DualPriceLayout"),
(_CURRENT_ONLY_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY, "_CurrentOnlyPriceLayout"),
(_CURRENT_ONLY_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, "_CurrentOnlyPriceLayout"),
)
for fixture, profile, layout_name in cases:
with self.subTest(fixture=fixture.name):
spec = _classify_panel(_parse_nodes(fixture.read_text(encoding="utf-8")))
self.assertIs(spec.profile, profile)
self.assertEqual(spec.price_layout.__class__.__name__, layout_name)
def test_dual_price_historical_nested_parent_chains_remain_accepted(self) -> None:
for fixture, expected_profile in (
(_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY),
(_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN),
):
with self.subTest(fixture=fixture.name):
nested = _nested_dual_price_layout(
fixture.read_text(encoding="utf-8"),
header_bounds="[0,366][1080,1077]",
row_bounds="[396,575][912,647]",
summary_bounds="[396,731][1053,793]",
content_bounds="[0,551][1080,940]",
price_frame_bounds="[396,575][1053,647]",
)
spec = _classify_panel(_parse_nodes(nested))
self.assertIs(spec.profile, expected_profile)
self.assertEqual(spec.price_layout.__class__.__name__, "_DualPriceLayout")
nested_s = _nested_dual_price_layout(
_S_FIXTURE.read_text(encoding="utf-8"),
header_bounds="[0,366][1080,1000]",
row_bounds="[396,498][895,570]",
summary_bounds="[396,654][1053,716]",
content_bounds="[0,474][1080,863]",
price_frame_bounds="[396,498][1053,570]",
)
self.assertIs(
_classify_panel(_parse_nodes(nested_s)).profile,
_PanelProfile.SIZE_VISIBLE_NON_TARGET,
)
nested_m = _nested_dual_price_layout(
_M_FIXTURE.read_text(encoding="utf-8"),
header_bounds="[0,366][1080,1000]",
row_bounds="[396,498][895,570]",
summary_bounds="[396,654][1053,716]",
content_bounds="[0,474][1080,863]",
price_frame_bounds="[396,498][1053,570]",
)
self.assertEqual(_flow(_RawDevice(nested_m)).read_sku_unit_price(), "12.88")
def test_current_only_price_variant_rejects_cross_variant_duplicate_unknown_and_parent_drift(self) -> None:
base = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
def mutate(kind: str) -> str:
root = ElementTree.fromstring(base)
parents = {child: parent for parent in root.iter() for child in parent}
frame = next(
node for node in root.iter("node")
if node.get("class") == "android.widget.FrameLayout"
and node.get("bounds") == "[396,575][1053,647]"
)
row = next(
node for node in frame
if node.get("class") == "android.widget.LinearLayout"
and node.get("bounds") == "[396,575][693,647]"
)
current = next(node for node in row if node.get("text") == "限1件 ¥12.88 ")
if kind == "cross_variant":
ElementTree.SubElement(
row,
"node",
{
"text": "券前¥29.88",
"package": "com.xunmeng.pinduoduo",
"class": "android.widget.TextView",
"bounds": "[693,580][912,647]",
"clickable": "false",
"enabled": "true",
"visible-to-user": "true",
"selected": "false",
"scrollable": "false",
},
)
elif kind == "duplicate_frame":
parents[frame].append(ElementTree.fromstring(ElementTree.tostring(frame, encoding="unicode")))
elif kind == "duplicate_current":
row.append(ElementTree.fromstring(ElementTree.tostring(current, encoding="unicode")))
elif kind == "unknown_text":
current.set("text", "未知促销文字")
elif kind == "unknown_bounds":
frame.set("bounds", "[396,574][1053,647]")
elif kind == "unknown_child":
row.append(ElementTree.Element("node", dict(current.attrib) | {"text": "未知节点"}))
elif kind == "parent_drift":
frame.remove(row)
parents[frame].append(row)
elif kind == "summary_parent_drift":
summary = next(
node for node in root.iter("node")
if node.get("bounds") == "[396,731][1053,793]"
and node.get("class") == "android.widget.TextView"
)
parents[summary].remove(summary)
parents[parents[summary]].append(summary)
else:
raise AssertionError(kind)
return ElementTree.tostring(root, encoding="unicode")
for kind in (
"cross_variant",
"duplicate_frame",
"duplicate_current",
"unknown_text",
"unknown_bounds",
"unknown_child",
"parent_drift",
"summary_parent_drift",
):
with self.subTest(kind=kind), self.assertRaises(SkuSelectionError):
_classify_panel(_parse_nodes(mutate(kind)))
def test_color_click_rejects_dual_current_only_cross_variant_in_both_directions(self) -> None:
cases = (
(_CURRENT_ONLY_EMPTY_FIXTURE, _COLOR_FIXTURE),
(_EMPTY_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE),
)
for empty_fixture, color_fixture in cases:
with self.subTest(empty=empty_fixture.name, color=color_fixture.name):
device = _RawDevice()
device.panel_hierarchy = empty_fixture.read_text(encoding="utf-8")
device.color_hierarchy = color_fixture.read_text(encoding="utf-8")
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
with self.assertRaises(SkuSelectionError) as raised:
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertNotIn("受控显示动作尚未取证", str(raised.exception))
expected = [(865, 2218), (528, 1387)] if empty_fixture is _CURRENT_ONLY_EMPTY_FIXTURE else [(865, 2218)]
self.assertEqual(_tap_centers(device), expected)
self.assertEqual(_actions(device, "swipe"), [])
self.assertEqual(_actions(device, "pressKey"), [])
def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
device = _RawDevice()
_flow(device).open_sku_panel(_TARGET_URL)
self.assertEqual(_tap_centers(device), [(865, 2218)])
self.assertNotIn((763, 2247), _tap_centers(device)) # 父容器中心不是获准目标。
self.assertNotIn((772, 2280), _tap_centers(device)) # 第二行“免拼购买”永不点击。
def test_entry_stability_uses_verified_chain_projection_not_whole_xml(self) -> None:
class DynamicProductDevice(_RawDevice):
def __init__(self) -> None:
super().__init__()
self.frame = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy" and "快要抢光" in self.hierarchy:
self.frame += 1
self.hierarchy = _dynamic_product_page(str(self.frame))
return super().jsonrpc_call(method, params, timeout)
now = [0.0]
device = DynamicProductDevice()
flow = SkuSelectionFlow(
UiautomatorSkuPanelAdapter(device, 10),
0.03,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
flow.open_sku_panel(_TARGET_URL, "")
self.assertEqual(device.frame, 2)
self.assertEqual(_tap_centers(device), [(865, 2218)])
def test_entry_action_description_must_be_stable_across_frames(self) -> None:
class ChangingActionDescriptionDevice(_RawDevice):
def __init__(self) -> None:
super().__init__()
self.frame = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy":
self.frame += 1
self.hierarchy = _dynamic_product_page(
str(self.frame),
f"活动{self.frame}快要抢光",
)
return super().jsonrpc_call(method, params, timeout)
now = [0.0]
device = ChangingActionDescriptionDevice()
flow = SkuSelectionFlow(
UiautomatorSkuPanelAdapter(device, 10),
0.02,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL, "")
self.assertEqual(_actions(device, "click"), [])
def test_unverified_home_entry_labels_do_not_trigger_old_product_rejection(self) -> None:
device = _RawDevice()
_flow(device).open_sku_panel(
_TARGET_URL,
_home_with_unverified_entry_labels(),
)
self.assertEqual(_tap_centers(device), [(865, 2218)])
def test_entry_child_and_every_ancestor_attribute_drift_never_clicks(self) -> None:
expected_clickable = ("false", "true")
for depth in range(2):
changes = {
"package": "other.package",
"class": "android.widget.Button",
"bounds": "[1,1][2,2]",
"clickable": "true" if expected_clickable[depth] == "false" else "false",
"enabled": "false",
"visible-to-user": "false",
}
for attribute, value in changes.items():
with self.subTest(depth=depth, attribute=attribute):
self._assert_entry_rejected_without_click(_mutate_entry(depth, attribute, value))
def test_entry_chain_text_and_description_drift_never_clicks(self) -> None:
cases = [
_mutate_entry(1, "text", "祖先文字漂移")
] + [
_mutate_entry(0, "content-desc", "祖先描述漂移"),
_mutate_entry(1, "content-desc", "祖先描述漂移"),
] + [
_mutate_entry(1, "content-desc", forbidden + "快要抢光")
for forbidden in (
"免拼购买", "单独购买", "直接拼成", "提交订单", "支付", "先用后付", "0元下单",
"立即购买", "确认下单", "立即付款", "订单详情",
)
]
for hierarchy in cases:
with self.subTest():
self._assert_entry_rejected_without_click(hierarchy)
def test_any_live_clickable_covering_entry_center_blocks_click(self) -> None:
cases = (
_with_overlapping_clickable(
"com.xunmeng.pinduoduo",
"android.widget.Button",
"[850,2200][900,2230]",
),
_with_overlapping_clickable(
"com.android.systemui",
"android.view.ViewGroup",
"[800,2100][1000,2300]",
),
_with_overlapping_clickable(
"com.android.systemui",
"android.view.ViewGroup",
"not-a-bound",
),
)
for hierarchy in cases:
with self.subTest():
self._assert_entry_rejected_without_click(hierarchy)
def test_pre_intent_old_entry_is_rejected_even_with_extra_action_node(self) -> None:
for old_page in (
_extra_entry_action_ancestor(),
_entry_with_panel_price_marker(),
_mutate_entry(1, "content-desc", "立即购买快要抢光"),
):
with self.subTest():
device = _RawDevice()
flow = _flow(device)
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL, old_page)
self.assertEqual(_actions(device, "click"), [])
def test_action_subtree_dangerous_or_ambiguous_children_never_click(self) -> None:
cases = (
_with_action_subtree_child("提交订单"),
_with_action_subtree_child("免拼购买"),
_with_action_subtree_child("快要抢光"),
)
for hierarchy in cases:
with self.subTest():
self._assert_entry_rejected_without_click(hierarchy)
def test_duplicate_entry_and_forbidden_sibling_entry_never_click(self) -> None:
self._assert_entry_rejected_without_click(_duplicate_entry())
self._assert_entry_rejected_without_click(_extra_entry_action_ancestor())
self._assert_entry_rejected_without_click(_without_entry())
self._assert_entry_rejected_without_click(_mutate_entry(0, "clickable", "true"))
def test_header_promotion_label_alone_never_becomes_entry(self) -> None:
self._assert_entry_rejected_without_click(_without_entry())
def test_entry_leaf_and_parent_amount_identity_drift_never_clicks(self) -> None:
for hierarchy in (
_PRODUCT_PAGE.replace('text="快要抢光 ¥ 12.88"', 'text="快要抢光 ¥ 13.88"'),
_mutate_entry(1, "content-desc", "快要抢光¥13.88"),
):
with self.subTest():
self._assert_entry_rejected_without_click(hierarchy)
def test_entry_sibling_is_exact_inert_unique_and_in_same_parent(self) -> None:
cases = (
_without_entry_sibling(),
_mutate_entry_sibling("text", "免拼购买 "),
_mutate_entry_sibling("bounds", "[688,2255][856,2305]"),
_mutate_entry_sibling("clickable", "true"),
_with_action_subtree_child("免拼购买", bounds="[500,2260][650,2300]"),
_entry_sibling_elsewhere(),
)
for hierarchy in cases:
with self.subTest():
self._assert_entry_rejected_without_click(hierarchy)
def test_unknown_task_or_ui_variants_are_rejected_without_action(self) -> None:
for color, size in (("黑色 CHA (纯棉）", _TASK_SIZE), (_TASK_COLOR, "M（建议100-115）"), ("黑色CHA（纯棉）", _TASK_SIZE)):
with self.subTest(color=color, size=size), self.assertRaises(SkuSelectionError):
resolve_task_selection(color, size)
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8"))
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE).__class__("粉红", "L（建议115-130）")
)
self.assertEqual(_actions(device, "click"), [])
def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None:
base = _FIXTURE.read_text(encoding="utf-8")
cases = (
_mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="selected", value=None),
_mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="selected", value="maybe"),
_mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="bounds", value="[1,1][20,20]"),
_mutate_unique_node(base, desc="黑色 CHA (纯棉）", class_name="android.view.ViewGroup", attribute="enabled", value="false"),
)
for hierarchy in cases:
with self.subTest(), self.assertRaises(SkuSelectionError):
_flow(_RawDevice(hierarchy)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
def test_unknown_clickable_action_ancestor_is_rejected_by_profile(self) -> None:
root = ElementTree.fromstring(_EMPTY_FIXTURE.read_text(encoding="utf-8"))
parents = {child: parent for parent in root.iter() for child in parent}
target = next(
node
for node in root.iter("node")
if node.get("content-desc") == "黑色 CHA (纯棉）"
and node.get("class") == "android.view.ViewGroup"
)
region = parents[target]
region.remove(target)
unknown = ElementTree.SubElement(
region,
"node",
{
"package": "com.xunmeng.pinduoduo",
"class": "android.view.ViewGroup",
"bounds": "[0,1188][1080,2046]",
"clickable": "true",
"enabled": "true",
"visible-to-user": "true",
"selected": "false",
"scrollable": "false",
},
)
unknown.append(target)
with self.assertRaises(SkuSelectionError):
_classify_panel(_parse_nodes(ElementTree.tostring(root, encoding="unicode")))
def test_invalid_bounds_stop_before_action(self) -> None:
for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"):
with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError):
_action_bounds(bounds)
device = _RawDevice(_PRODUCT_PAGE.replace("[688,2184][1042,2253]", "[0,0][1081,1]"))
with self.assertRaises(SkuSelectionError):
_flow(device).open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_color_readback_failure_never_attempts_second_option(self) -> None:
device = _RawDevice()
device.fail_color_readback = True
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
with self.assertRaises(SkuSelectionError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
def test_non_target_selection_restores_each_dimension_once(self) -> None:
device = _RawDevice()
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
device.select_alternates()
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(865, 2218), (635, 1624)])
self.assertEqual(_actions(device, "swipe"), [])
def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
with self.subTest(replacement=replacement):
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", replacement))
with self.assertRaises(SkuSelectionError):
_flow(device).read_sku_unit_price()
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88"))
with self.assertRaises(SkuSelectionError):
_flow(device).read_sku_unit_price()
clickable_parent = _mutate_unique_node(
_FIXTURE.read_text(encoding="utf-8"),
class_name="android.widget.LinearLayout",
bounds="[396,498][895,570]",
attribute="clickable",
value="true",
)
with self.assertRaises(SkuSelectionError):
_flow(_RawDevice(clickable_parent)).read_sku_unit_price()
def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click", "swipe", "scroll"}
self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
self.assertTrue(forbidden.isdisjoint(pdd.__all__))
def test_static_ast_boundary_limits_flow_runner_adapter_and_cli(self) -> None:
root = Path(__file__).resolve().parents[2]
files = (
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py",
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py",
root / "scripts" / "run_t103_sku_selection.py",
root / "src" / "cmbuyer_client" / "pdd" / "sku_reveal_spike.py",
root / "scripts" / "capture_sku_reveal_spike.py",
)
forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment")
for path in files:
source = path.read_text(encoding="utf-8")
with self.subTest(path=path.name):
self.assertTrue(all(token not in source.lower() for token in forbidden))
tree = ast.parse(source)
self.assertFalse(any(isinstance(node, ast.ImportFrom) and node.module in {"selenium", "requests"} for node in ast.walk(tree)))
runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
self.assertEqual(len(click_calls), 1)
swipe_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "swipe"]
self.assertEqual(len(swipe_calls), 1)
def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
now = [0.0]
device = _RawDevice()
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), 0.01, 0.01, lambda: now[0], lambda seconds: now.__setitem__(0, now[0] + seconds))
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL, _PRODUCT_PAGE)
self.assertEqual(_actions(device, "click"), [])
def test_action_postcondition_wait_never_repeats_entry_click(self) -> None:
class NoPanelAfterEntry(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.calls.append(("jsonrpc", method, params, timeout))
return ""
return super().jsonrpc_call(method, params, timeout)
device = NoPanelAfterEntry()
with self.assertRaises(SkuSelectionError):
_flow(device).open_sku_panel(_TARGET_URL)
self.assertEqual(_tap_centers(device), [(865, 2218)])
duplicate = _duplicate_entry()
device = _RawDevice(duplicate)
with self.assertRaises(SkuSelectionError):
_flow(device).open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None:
for fixture in (
_EMPTY_FIXTURE,
_COLOR_FIXTURE,
_CURRENT_ONLY_EMPTY_FIXTURE,
_CURRENT_ONLY_COLOR_FIXTURE,
_S_FIXTURE,
_M_FIXTURE,
_ENTRY_FIXTURE,
):
content = fixture.read_text(encoding="utf-8")
with self.subTest(fixture=fixture.name):
self.assertNotRegex(content, r"1[3-9]\d{9}")
for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"):
self.assertNotIn(forbidden, content)
self.assertNotIn("提交订单", content)
class _CompletedFlow:
"""仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。"""
def __init__(self, device: object, *args: object, **kwargs: object) -> None:
self.device = device
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
self.device.tap_sku_entry("[688,2184][1042,2253]")
def select_sku_options(self, selection: object) -> None:
self.device.tap_sku_option("[372,1188][684,1587]")
self.device.reveal_size_options_once()
self.device.tap_sku_option("[439,1582][831,1667]")
def verify_target_selection_and_read_price(self, selection: object) -> str:
return "12.88"
def exit_sku_panel_safely(self) -> None:
self.device.leave_sku_panel()
def reconcile_pending_action(self) -> None:
return None
class SkuSelectionRunnerTests(unittest.TestCase):
def setUp(self) -> None:
flow_patch = patch.object(runner_module, "SkuSelectionFlow", _fast_runner_flow)
flow_patch.start()
self.addCleanup(flow_patch.stop)
def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner:
device.hierarchy = ""
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("", ''))
return SkuSelectionRunner(adb, lambda serial: device, 0.03)
def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None:
adb = _FakeAdb()
device = _RawDevice()
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
with patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow):
result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual(result.unit_price, "12.88")
manifest = result.manifest_path.read_text(encoding="utf-8")
manifest_data = json.loads(manifest)
self.assertTrue(result.screenshot_path.is_file())
self.assertNotIn("device-1", manifest)
self.assertNotIn("hierarchy", manifest)
self.assertNotIn("已选", manifest)
self.assertIn('"unit_price": "12.88"', manifest)
self.assertIn('"selection_status": "restored"', manifest)
self.assertIn('"panel_status": "verified_before_back"', manifest)
self.assertIn('"back_attempts": 1', manifest)
self.assertIn('"back_rpc_outcome": "completed"', manifest)
self.assertIn('"post_exit_status": "human_review_required"', manifest)
self.assertNotIn('"safe_exit"', manifest)
self.assertEqual(
manifest_data["actions"],
{
"sku_entry": {"attempts": 1, "rpc_outcome": "completed"},
"target_color": {"attempts": 1, "rpc_outcome": "completed"},
"size_reveal": {"attempts": 1, "rpc_outcome": "completed"},
"target_size": {"attempts": 1, "rpc_outcome": "completed"},
"back": {"attempts": 1, "rpc_outcome": "completed"},
},
)
self.assertFalse((target / "hierarchy.xml").exists())
self.assertEqual(len(_actions(device, "pressKey")), 1)
def test_publish_rejects_flow_stub_without_exact_action_audit_chain(self) -> None:
class IncompleteFlow(_CompletedFlow):
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
return None
def select_sku_options(self, selection: object) -> None:
return None
def exit_sku_panel_safely(self) -> None:
return None
device = _RawDevice()
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
with (
patch.object(runner_module, "SkuSelectionFlow", IncompleteFlow),
self.assertRaises(SkuSelectionRunError) as raised,
):
self._runner(_FakeAdb(), device).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
)
self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
self.assertEqual(_actions(device, "pressKey"), [])
def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None:
class NoPanelAfterEntry(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.calls.append(("jsonrpc", method, params, timeout))
return ""
return super().jsonrpc_call(method, params, timeout)
adb = _FakeAdb()
device = NoPanelAfterEntry("")
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
SkuSelectionRunner(adb, lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_panel_verify")
forged = SkuSelectionError("private")
setattr(forged, "_cmbuyer_failure_stage", "sku_entry_click")
self.assertEqual(safe_failure_stage(forged), "unknown")
class HostileSetterError(DeviceConnectionError):
def __setattr__(self, name: str, value: object) -> None:
raise KeyboardInterrupt("SERIAL=private secret")
class FailingAdb(_FakeAdb):
def inspect(self, serial: str) -> DeviceInspection:
raise HostileSetterError("private")
with TemporaryDirectory() as temporary, self.assertRaises(HostileSetterError) as hostile:
SkuSelectionRunner(FailingAdb(), lambda serial: _RawDevice(), 10).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(hostile.exception), "unknown")
def test_failure_stage_identifies_sku_entry_pre_intent(self) -> None:
device = _RawDevice(_PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
SkuSelectionRunner(_FakeAdb(), lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_pre_intent")
self.assertEqual(_actions(device, "click"), [])
def test_failure_stage_identifies_sku_entry_discovery(self) -> None:
device = _RawDevice("")
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
SkuSelectionRunner(_FakeAdb(), lambda serial: device, 0.01).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_discovery")
self.assertEqual(_actions(device, "click"), [])
def test_failure_stage_identifies_sku_entry_click(self) -> None:
class ClickFailureDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private device detail")
return super().jsonrpc_call(method, params, timeout)
adb = _FakeAdb()
device = ClickFailureDevice("")
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError) as raised:
SkuSelectionRunner(adb, lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_click")
self.assertEqual(len(_actions(device, "click")), 1)
def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
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)
with (
patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow),
patch.object(runner_module.os, "rename", side_effect=create_target_then_rename),
self.assertRaises(SkuSelectionRunError),
):
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep")
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
with (
patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow),
self.assertRaises(SkuSelectionScreenshotError) as screenshot_failure,
):
self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual(safe_failure_stage(screenshot_failure.exception), "screenshot_capture")
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
target = Path(temporary) / "write-failure"
with (
patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow),
patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")),
):
with self.assertRaises(SkuSelectionScreenshotError):
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".write-failure.staging-*")), [])
adb = _FakeAdb()
device = _RawDevice()
target.mkdir()
sentinel = target / "keep"
sentinel.write_text("keep", encoding="utf-8")
with self.assertRaises(SkuSelectionRunError) as existing_target_failure:
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual(safe_failure_stage(existing_target_failure.exception), "precheck")
self.assertEqual(adb.calls, [])
self.assertEqual(device.calls, [])
self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
def test_device_screen_and_output_preflight_fail_before_any_click(self) -> None:
with TemporaryDirectory() as temporary:
adb = _FakeAdb()
adb.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "wrong", "16")
device = _RawDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(device.calls, [])
class WrongScreenDevice(_RawDevice):
def window_size(self) -> tuple[int, int]:
return 1080, 1920
device = WrongScreenDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "screen")
self.assertEqual(_actions(device, "click"), [])
parent_file = Path(temporary) / "not-a-directory"
parent_file.write_text("x", encoding="utf-8")
adb = _FakeAdb()
device = _RawDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, parent_file / "result")
self.assertEqual(adb.calls, [])
self.assertEqual(device.calls, [])
def test_small_but_valid_png_is_not_accepted(self) -> None:
image = Image.new("RGB", (1, 1), "white")
raw = BytesIO(); image.save(raw, format="PNG")
with (
TemporaryDirectory() as temporary,
patch.object(runner_module, "SkuSelectionFlow", _CompletedFlow),
self.assertRaises(SkuSelectionScreenshotError),
):
self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
def test_failure_after_entry_attempts_one_safe_exit_and_hides_device_detail(self) -> None:
adb = _FakeAdb()
device = _RawDevice()
device.fail_color_readback = True
device.select_alternates()
with TemporaryDirectory() as temporary:
with self.assertRaises(SkuSelectionError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(_actions(device, "pressKey"), [])
class FailingRawDevice(_RawDevice):
def app_info(self, package_name: str) -> dict[str, str]:
raise RuntimeError("device-1 private")
with self.assertRaises(SkuSelectionDeviceAdapterError) as raised:
UiautomatorSkuPanelAdapter(FailingRawDevice(), 10).app_info("com.xunmeng.pinduoduo")
self.assertNotIn("device-1", str(raised.exception))
self.assertNotIn("private", str(raised.exception))
def test_unverified_failure_never_sends_blind_back(self) -> None:
class InvalidAfterOptionDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
value = super().jsonrpc_call(method, params, timeout)
if method == "click" and "[396,498][895,570]" in self.hierarchy:
self.hierarchy = ""
return value
device = InvalidAfterOptionDevice()
device.select_alternates()
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(_actions(device, "pressKey"), [])
def test_adapter_timeout_is_mapped_without_third_party_detail(self) -> None:
class TimeoutRawDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
raise TimeoutError("device-1 private")
with self.assertRaises(SkuSelectionRunError) as raised:
UiautomatorSkuPanelAdapter(TimeoutRawDevice(), 10).dump_window_hierarchy()
self.assertNotIn("device-1", str(raised.exception))
self.assertNotIn("private", str(raised.exception))
def test_entry_attempt_is_recorded_before_unconfirmed_click_and_not_retried(self) -> None:
class TimeoutTapDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
self.calls.append(("jsonrpc", method, params, timeout))
if method == "click":
raise TimeoutError("device detail")
return super().jsonrpc_call(method, params, timeout)
adapter = UiautomatorSkuPanelAdapter(TimeoutTapDevice(), 10)
with self.assertRaises(SkuSelectionRunError):
adapter.tap_sku_entry("[688,2184][1042,2253]")
self.assertTrue(adapter.entry_was_tapped)
self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [865, 2218], 10)])
def test_adapter_seals_every_named_mutation_before_rpc(self) -> None:
device = _RawDevice(_PRODUCT_PAGE)
adapter = UiautomatorSkuPanelAdapter(device, 10)
adapter.tap_sku_entry("[688,2184][1042,2253]")
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.tap_sku_entry("[688,2184][1042,2253]")
adapter.tap_sku_option("[372,1188][684,1587]")
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.tap_sku_option("[372,1188][684,1587]")
adapter.reveal_size_options_once()
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.reveal_size_options_once()
adapter.leave_sku_panel()
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.leave_sku_panel()
self.assertEqual(len(_actions(device, "click")), 2)
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertEqual(adapter.option_attempts, 1)
self.assertEqual(adapter.entry_rpc_outcome, "completed")
self.assertEqual(adapter.reveal_rpc_outcome, "completed")
self.assertEqual(adapter.back_attempts, 1)
self.assertEqual(adapter.back_rpc_outcome, "completed")
def test_entry_stability_interruptions_never_click(self) -> None:
now = [0.0]
class SequenceDevice(_RawDevice):
def __init__(self) -> None:
super().__init__(); self.frames = [
_dynamic_product_page("first"),
"",
_dynamic_product_page("second"),
]
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy" and self.frames:
self.hierarchy = self.frames.pop(0)
return super().jsonrpc_call(method, params, timeout)
device = SequenceDevice()
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), .02, .01, lambda: now[0], lambda x: now.__setitem__(0, now[0] + x))
with self.assertRaises(SkuSelectionError): flow.open_sku_panel(_TARGET_URL, "")
self.assertEqual(_actions(device, "click"), [])
def test_screenshot_drift_and_foreground_drift_publish_nothing_and_never_back(self) -> None:
for drift in ("color", "size", "price"):
class DriftDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
value = super().jsonrpc_call(method, params, timeout)
if method == "takeScreenshot":
if drift == "price":
self.hierarchy = self.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
else:
root = ElementTree.fromstring(self.hierarchy)
if drift == "color":
for node in root.iter("node"):
if node.get("selected") is not None and ",1000]" in node.get("bounds", ""):
node.set("selected", "false")
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
else:
for node in root.iter("node"):
if node.get("selected") is not None and ",1730]" in node.get("bounds", ""):
node.set("selected", "false")
next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
self.hierarchy = ElementTree.tostring(root, encoding="unicode")
return value
with self.subTest(drift=drift), TemporaryDirectory() as temporary:
target = Path(temporary) / "out"
with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
self._runner(_FakeAdb(), DriftDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertFalse((target / "manifest.json").exists())
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
device = _RawDevice(); device.select_alternates()
device.package = "other"
with self.assertRaises(SkuSelectionError): _flow(device).exit_sku_panel_safely()
self.assertEqual(_actions(device, "pressKey"), [])
def test_back_precondition_rejects_every_non_target_profile_after_screenshot_reverify(self) -> None:
drifts = (
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
_CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8"),
_S_FIXTURE.read_text(encoding="utf-8"),
_revealed_current_only(),
)
for drift in drifts:
class DriftBeforeBackDevice(_RawDevice):
def __init__(self) -> None:
super().__init__()
self.after_screenshot_reads = 0
self.screenshot_seen = False
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "takeScreenshot":
value = super().jsonrpc_call(method, params, timeout)
self.screenshot_seen = True
return value
if method == "dumpWindowHierarchy" and self.screenshot_seen:
self.after_screenshot_reads += 1
if self.after_screenshot_reads == 2:
self.hierarchy = drift
return super().jsonrpc_call(method, params, timeout)
with self.subTest(profile=_classify_panel(_parse_nodes(drift)).profile), TemporaryDirectory() as temporary:
device = DriftBeforeBackDevice()
target = Path(temporary) / "result"
with self.assertRaises(SkuSelectionError) as raised:
self._runner(_FakeAdb(), device).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
)
self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
self.assertEqual(_actions(device, "pressKey"), [])
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
def test_dual_precondition_publishes_nothing_without_reveal_or_back(self) -> None:
device = _RawDevice()
device.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
device.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
with TemporaryDirectory() as temporary:
target = Path(temporary) / "out"
with self.assertRaises(SkuSelectionError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertFalse((target / "manifest.json").exists())
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
self.assertEqual(_actions(device, "swipe"), [])
self.assertEqual(len(_actions(device, "pressKey")), 0)
def test_option_timeout_reconciliation_controls_back_once(self) -> None:
class OptionTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(); self.delivered = delivered; self.clicks = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.clicks += 1
if self.clicks == 2:
if self.delivered: super().jsonrpc_call(method, params, timeout)
else: self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("uncertain option")
return super().jsonrpc_call(method, params, timeout)
for delivered, expected_back in ((False, 0), (True, 0)):
with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
device = OptionTimeoutDevice(delivered)
adb = _FakeAdb(); device.hierarchy = ""
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("", ''))
runner = SkuSelectionRunner(adb, lambda serial: device, .03)
with self.assertRaises(SkuSelectionRunError):
runner.run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
self.assertEqual(len(_actions(device, "click")), 2)
self.assertEqual(len(_actions(device, "pressKey")), expected_back)
def test_back_timeout_is_never_retried(self) -> None:
class BackTimeoutDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "pressKey":
super().jsonrpc_call(method, params, timeout)
raise TimeoutError("back uncertain")
return super().jsonrpc_call(method, params, timeout)
device = BackTimeoutDevice(_M_FIXTURE.read_text(encoding="utf-8"))
with self.assertRaises(SkuSelectionRunError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 0.03)).exit_sku_panel_safely()
self.assertEqual(len(_actions(device, "pressKey")), 1)
def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None:
class DeliveredThenTimeout(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click" and self.hierarchy != _FIXTURE.read_text(encoding="utf-8"):
super().jsonrpc_call(method, params, timeout)
raise TimeoutError("delivery uncertain")
return super().jsonrpc_call(method, params, timeout)
device = DeliveredThenTimeout()
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(len(_actions(device, "click")), 1)
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_evidence_bound_rolled_target_color_variant_is_exactly_accepted(self) -> None:
hierarchy = _M_FIXTURE.read_text(encoding="utf-8")
old_bounds = "[372,1000][684,1024]"
new_bounds = "[126,1000][438,1024]"
self.assertEqual(hierarchy.count(old_bounds), 4)
device = _ExitEvidenceDevice(hierarchy.replace(old_bounds, new_bounds))
with TemporaryDirectory() as temporary:
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["post_exit_status"], "human_review_required")
self.assertEqual(len(_actions(device, "pressKey")), 1)
unknown_device = _ExitEvidenceDevice(hierarchy.replace(old_bounds, "[127,1000][439,1024]"))
with TemporaryDirectory() as temporary:
target = Path(temporary) / "unknown"
with self.assertRaises(SkuSelectionError):
self._capturer(_FakeAdb(), unknown_device).capture("device-1", target)
self.assertEqual(_actions(unknown_device, "pressKey"), [])
self.assertFalse(target.exists())
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()
valid = {
"serial": "device-1",
"url": _TARGET_URL,
"color": _TASK_COLOR,
"size": _TASK_SIZE,
"output_dir": Path("evidence"),
"timeout": 10.0,
"adb": "adb",
}
script.validate_arguments(type("Arguments", (), valid)())
for field, value in (("serial", ""), ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=1"), ("color", "黑色 CHA (纯棉）"), ("size", "M（建议100-115）"), ("timeout", 0), ("timeout", float("inf"))):
with self.subTest(field=field, value=value), self.assertRaises((ValueError, SkuSelectionError)):
script.validate_arguments(type("Arguments", (), valid | {field: value})())
def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
script = _load_runner_script()
secret = "SERIAL=192.168.0.173:5555 PATH=C:/private page-body"
class HostileGetterError(SkuSelectionError):
def __getattribute__(self, name: str) -> object:
if name == "_cmbuyer_failure_stage":
raise RuntimeError(secret)
return super().__getattribute__(name)
class HostileStage(str):
def __hash__(self) -> int:
raise RuntimeError(secret)
hostile_string = SkuSelectionError(secret)
setattr(hostile_string, "_cmbuyer_failure_stage", HostileStage("sku_entry"))
for failure in (SkuSelectionError(secret), HostileGetterError(secret), hostile_string):
class FlowFailingRunner:
def __init__(self, *args: object, **kwargs: object) -> None: pass
def run(self, *args: object, **kwargs: object) -> object:
raise failure
stderr = BytesIO()
# TextIOWrapper keeps the assertion independent from host console encoding.
import io
text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
status = script.main([
"--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
"--size", _TASK_SIZE, "--output-dir", "evidence",
])
text_stderr.flush()
output = stderr.getvalue().decode("utf-8")
self.assertEqual(status, 1)
self.assertIn("stage=unknown", output)
self.assertNotIn("Traceback", output)
self.assertNotIn(secret, output)
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)
if specification is None or specification.loader is None:
raise RuntimeError("无法加载 T-103 运行脚本。")
module = importlib.util.module_from_spec(specification)
specification.loader.exec_module(module)
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()