chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): verify same-product safe exit

Browse Source
This commit is contained in:
QiuSW
2026-08-05 17:56:16 +08:00
parent 0afe9a53b3
commit 1bb54927e1
4 changed files with 287 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/tests/pdd/test_sku_selection.py
+177 -3
View File
@@ -25,6 +25,7 @@ from cmbuyer_client.pdd.sku_selection import (
_action_bounds,
_classify_panel,
_parse_nodes,
_product_exit_projection,
resolve_task_selection,
)
from cmbuyer_client.pdd.sku_selection_runner import (
@@ -48,10 +49,12 @@ _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"
_EXIT_FIXTURE = Path(__file__).with_name("fixtures") / "product_exit_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")
_PRODUCT_EXIT = _EXIT_FIXTURE.read_text(encoding="utf-8")
def _revealed_current_only() -> str:
@@ -96,7 +99,7 @@ class _RawDevice:
if method == "takeScreenshot":
return self.screenshot
if method == "pressKey":
self.hierarchy = "<hierarchy />"
self.hierarchy = _PRODUCT_EXIT
return ""
if method == "click":
if not isinstance(params, list) or len(params) != 2:
@@ -1156,6 +1159,7 @@ class SkuSelectionFlowTests(unittest.TestCase):
_S_FIXTURE,
_M_FIXTURE,
_ENTRY_FIXTURE,
_EXIT_FIXTURE,
):
content = fixture.read_text(encoding="utf-8")
with self.subTest(fixture=fixture.name):
@@ -1164,6 +1168,170 @@ class SkuSelectionFlowTests(unittest.TestCase):
self.assertNotIn(forbidden, content)
self.assertNotIn("提交订单", content)
def test_product_exit_fixture_binds_exact_same_product_and_entry(self) -> None:
projection = _product_exit_projection(_parse_nodes(_PRODUCT_EXIT))
self.assertEqual(projection[0], "product_exit_8_17_0")
root = ElementTree.fromstring(_PRODUCT_EXIT)
title = next(
node
for node in root.iter("node")
if node.get("resource-id") == "com.xunmeng.pinduoduo:id/tv_title"
)
root.append(ElementTree.fromstring(ElementTree.tostring(title, encoding="unicode")))
duplicate = ElementTree.tostring(root, encoding="unicode")
missing = _PRODUCT_EXIT.replace(
"com.xunmeng.pinduoduo:id/tv_title",
"com.xunmeng.pinduoduo:id/other",
1,
)
drifted = _PRODUCT_EXIT.replace("[36,1571][1044,1689]", "[35,1571][1044,1689]", 1)
dangerous = _PRODUCT_EXIT.replace(
"</hierarchy>",
'<node content-desc="提交订单" package="com.xunmeng.pinduoduo" '
'class="android.view.ViewGroup" bounds="[36,1800][400,1900]" clickable="true" '
'enabled="true" visible-to-user="true" selected="false" scrollable="false" /></hierarchy>',
)
system_ui = _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo", "com.android.systemui")
non_pdd_overlay = _PRODUCT_EXIT.replace(
"</hierarchy>",
'<node content-desc="overlay" package="external.app" '
'class="android.widget.FrameLayout" bounds="[0,0][1080,2376]" clickable="true" '
'enabled="true" visible-to-user="true" selected="false" scrollable="false" /></hierarchy>',
)
lock_screen = (
'<hierarchy><node content-desc="锁屏" package="com.android.systemui" '
'class="android.widget.FrameLayout" bounds="[0,0][1080,2376]" clickable="false" '
'enabled="true" visible-to-user="true" /></hierarchy>'
)
for name, hierarchy in (
("empty", "<hierarchy />"),
("panel_still_open", _M_FIXTURE.read_text(encoding="utf-8")),
("system_ui", system_ui),
("non_pdd_overlay", non_pdd_overlay),
("lock_screen", lock_screen),
("other_pdd_page", _PRODUCT_PAGE),
("anchor_missing", missing),
("anchor_duplicate", duplicate),
("anchor_drift", drifted),
("dangerous_action", dangerous),
):
with self.subTest(name=name), self.assertRaises(SkuSelectionError):
_product_exit_projection(_parse_nodes(hierarchy))
def test_safe_exit_requires_two_consecutive_evidence_projections(self) -> None:
device = _RawDevice(_M_FIXTURE.read_text(encoding="utf-8"))
flow = _flow(device)
flow.exit_sku_panel_safely()
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertGreaterEqual(len(_actions(device, "dumpWindowHierarchy")), 3)
with self.assertRaises(SkuSelectionError):
flow.exit_sku_panel_safely()
self.assertEqual(len(_actions(device, "pressKey")), 1)
def test_safe_exit_single_hit_then_drift_resets_stability(self) -> None:
class SequenceAfterBack(_RawDevice):
def __init__(self) -> None:
super().__init__(_M_FIXTURE.read_text(encoding="utf-8"))
self.frames: list[str] = []
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "pressKey":
self.calls.append(("jsonrpc", method, params, timeout))
self.frames = [_PRODUCT_EXIT, "<hierarchy />", _PRODUCT_EXIT, _PRODUCT_EXIT]
return ""
if method == "dumpWindowHierarchy" and self.frames:
self.hierarchy = self.frames.pop(0)
return super().jsonrpc_call(method, params, timeout)
device = SequenceAfterBack()
_flow(device, 0.08).exit_sku_panel_safely()
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertEqual(device.frames, [])
def test_safe_exit_negative_pages_fail_closed_after_one_back(self) -> None:
root = ElementTree.fromstring(_PRODUCT_EXIT)
title = next(
node
for node in root.iter("node")
if node.get("resource-id") == "com.xunmeng.pinduoduo:id/tv_title"
)
root.append(ElementTree.fromstring(ElementTree.tostring(title, encoding="unicode")))
duplicate = ElementTree.tostring(root, encoding="unicode")
dangerous = _PRODUCT_EXIT.replace(
"</hierarchy>",
'<node text="立即支付" package="com.xunmeng.pinduoduo" '
'class="android.widget.TextView" bounds="[36,1800][400,1900]" clickable="true" '
'enabled="true" visible-to-user="true" selected="false" scrollable="false" /></hierarchy>',
)
class FixedPostState(_RawDevice):
def __init__(self, post: str) -> None:
super().__init__(_M_FIXTURE.read_text(encoding="utf-8"))
self.post = post
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "pressKey":
self.calls.append(("jsonrpc", method, params, timeout))
self.hierarchy = self.post
return ""
return super().jsonrpc_call(method, params, timeout)
cases = (
("empty", "<hierarchy />"),
("panel", _M_FIXTURE.read_text(encoding="utf-8")),
("system_ui", _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo", "com.android.systemui")),
(
"non_pdd_overlay",
_PRODUCT_EXIT.replace(
"</hierarchy>",
'<node content-desc="overlay" package="external.app" '
'class="android.widget.FrameLayout" bounds="[0,0][1080,2376]" clickable="true" '
'enabled="true" visible-to-user="true" selected="false" scrollable="false" /></hierarchy>',
),
),
("other_pdd", _PRODUCT_PAGE),
("missing", _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo:id/tv_title", "missing", 1)),
("duplicate", duplicate),
("drift", _PRODUCT_EXIT.replace("[36,1571][1044,1689]", "[35,1571][1044,1689]", 1)),
("danger", dangerous),
)
for name, post in cases:
with self.subTest(name=name):
device = FixedPostState(post)
with self.assertRaises(SkuSelectionError):
_flow(device).exit_sku_panel_safely()
self.assertEqual(len(_actions(device, "pressKey")), 1)
version_drift = FixedPostState(_PRODUCT_EXIT)
original_app_info = version_drift.app_info
def drifted_app_info(package_name: str) -> dict[str, str]:
value = original_app_info(package_name)
if _actions(version_drift, "pressKey"):
value["versionName"] = "8.18.0"
return value
version_drift.app_info = drifted_app_info # type: ignore[method-assign]
with self.assertRaises(SkuSelectionError):
_flow(version_drift).exit_sku_panel_safely()
self.assertEqual(len(_actions(version_drift, "pressKey")), 1)
foreground_drift = FixedPostState(_PRODUCT_EXIT)
original_app_current = foreground_drift.app_current
def drifted_app_current() -> dict[str, str]:
value = original_app_current()
if _actions(foreground_drift, "pressKey"):
value["package"] = "external.app"
return value
foreground_drift.app_current = drifted_app_current # type: ignore[method-assign]
with self.assertRaises(SkuSelectionError):
_flow(foreground_drift).exit_sku_panel_safely()
self.assertEqual(len(_actions(foreground_drift, "pressKey")), 1)
class _CompletedFlow:
"""仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。"""
@@ -1212,6 +1380,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
manifest = result.manifest_path.read_text(encoding="utf-8")
manifest_data = json.loads(manifest)
self.assertTrue(result.screenshot_path.is_file())
self.assertEqual(result.screenshot_path, target / "screenshot.png")
self.assertNotIn("device-1", manifest)
self.assertNotIn("hierarchy", manifest)
self.assertNotIn("已选", manifest)
@@ -1220,8 +1389,13 @@ class SkuSelectionRunnerTests(unittest.TestCase):
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.assertIn('"post_exit_status": "same_product_verified"', manifest)
self.assertIn('"safe_exit": "completed"', manifest)
self.assertIn('"page_identity": "same_goods_evidence_bound"', manifest)
self.assertNotIn("human_review_required", manifest)
self.assertIsNotNone(result.captured_at.tzinfo)
self.assertEqual(result.captured_at.utcoffset().total_seconds(), 0)
self.assertEqual(manifest_data["captured_at"], result.captured_at.isoformat())
self.assertEqual(
manifest_data["actions"],
{