2177 lines
100 KiB
Python
2177 lines
100 KiB
Python
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 = "<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, "<hierarchy />")
|
||
|
||
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, "<hierarchy />")
|
||
|
||
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 = "<hierarchy />"
|
||
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
|
||
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("<hierarchy />")
|
||
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("<hierarchy>private</hierarchy>")
|
||
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 <hierarchy>secret</hierarchy>")
|
||
|
||
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("<hierarchy />")
|
||
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("<hierarchy />")
|
||
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 <xml>private</xml>")
|
||
|
||
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 = "<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 <hierarchy>private</hierarchy>")
|
||
|
||
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"),
|
||
"<hierarchy />",
|
||
_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, "<hierarchy />")
|
||
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 = "<hierarchy />"
|
||
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
|
||
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_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 = "<hierarchy rotation=\"0\" />"
|
||
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 <hierarchy>page-body</hierarchy>"
|
||
|
||
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 <hierarchy>private</hierarchy>"
|
||
|
||
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()
|