fix(client): bind revealed current-only price layout

client/tests/pdd/test_sku_reveal_spike.py

@@ -18,6 +18,7 @@ import cmbuyer_client.pdd.sku_reveal_spike as reveal_module
 from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
 from cmbuyer_client.pdd.sku_reveal_spike import (
     _annotate_reveal_failure,
+    _candidate_projection,
     _RevealEvidenceAdapter,
     safe_reveal_failure_stage,
     SkuRevealSpikeCapturer,
@@ -40,6 +41,22 @@ _CURRENT_ONLY_COLOR = (_FIXTURES / "sku_panel_color_selected_size_hidden_current
 _M_SELECTED = (_FIXTURES / "sku_panel_size_m_restored_8_17_0.xml").read_text(encoding="utf-8")
 
 
+def _inert_node(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",
+        },
+    )
+
+
 def _candidate_unselected() -> str:
     root = ElementTree.fromstring(_M_SELECTED)
     parents = {child: parent for parent in root.iter() for child in parent}
@@ -52,6 +69,38 @@ def _candidate_unselected() -> str:
             for child in list(wrapper):
                 if child.get("class") == "android.view.View":
                     wrapper.remove(child)
+    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") == "[0,366][1080,1000]"
+    )
+    price_row = next(
+        node for node in root.iter("node")
+        if node.get("class") == "android.widget.LinearLayout"
+        and node.get("bounds") == "[396,498][895,570]"
+    )
+    summary = next(
+        node for node in root.iter("node")
+        if node.get("class") == "android.widget.TextView"
+        and node.get("bounds") == "[396,654][1053,716]"
+    )
+    parents[price_row].remove(price_row)
+    parents[summary].remove(summary)
+    content_frame = _inert_node("android.widget.FrameLayout", "[0,474][1080,863]")
+    relative = _inert_node("android.widget.RelativeLayout", "[0,474][1080,863]")
+    content = _inert_node("android.view.ViewGroup", "[0,474][1080,863]")
+    price_frame = _inert_node("android.widget.FrameLayout", "[396,498][1053,570]")
+    header.append(content_frame)
+    content_frame.append(relative)
+    relative.append(content)
+    content.append(price_frame)
+    price_frame.append(price_row)
+    content.append(summary)
+    for index in range(6):
+        content.append(_inert_node("android.view.View", f"[{index},474][{index + 1},475]"))
+    for index in range(3):
+        header.append(_inert_node("android.view.View", f"[{index},366][{index + 1},367]"))
     action_root = next(
         node
         for node in root.iter("node")
@@ -110,6 +159,196 @@ def _candidate_unselected() -> str:
     return ElementTree.tostring(root, encoding="unicode")
 
 
+def _candidate_current_only() -> str:
+    root = ElementTree.fromstring(_candidate_unselected())
+    parents = {child: parent for parent in root.iter() for child in parent}
+    dual_row = next(
+        node for node in root.iter("node")
+        if node.get("class") == "android.widget.LinearLayout"
+        and node.get("bounds") == "[396,498][895,570]"
+    )
+    price_frame = parents[dual_row]
+    price_frame.remove(dual_row)
+    row = _inert_node("android.widget.LinearLayout", "[396,498][693,570]")
+    current = _inert_node("android.widget.TextView", "[396,503][675,570]")
+    current.set("text", "限1件 ¥12.88 ")
+    price_frame.append(row)
+    row.append(current)
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _candidate_price_drift(variant: str, kind: str) -> str:
+    root = ElementTree.fromstring(
+        _candidate_unselected() if variant == "dual" else _candidate_current_only()
+    )
+    parents = {child: parent for parent in root.iter() for child in parent}
+    if variant == "dual":
+        row = next(
+            node for node in root.iter("node")
+            if node.get("class") == "android.widget.LinearLayout"
+            and node.get("bounds") == "[396,498][895,570]"
+        )
+        current = next(node for node in row if node.get("text") == "快卖完 ¥12.88")
+        original = next(node for node in row if node.get("text") == "¥29.88")
+        if kind == "hybrid":
+            current.set("text", "限1件 ¥12.88 ")
+            current.set("bounds", "[396,503][675,570]")
+            row.remove(original)
+        elif kind == "duplicate":
+            row.append(ElementTree.fromstring(ElementTree.tostring(current, encoding="unicode")))
+        elif kind == "unknown":
+            current.set("text", "未知候选价格")
+        elif kind == "unknown_child":
+            row.append(ElementTree.Element("node", dict(current.attrib) | {"text": "未知子节点"}))
+        elif kind == "parent_drift":
+            row.remove(current)
+            parents[row].append(current)
+        else:
+            raise AssertionError(kind)
+    elif variant == "current_only":
+        frame = next(
+            node for node in root.iter("node")
+            if node.get("class") == "android.widget.FrameLayout"
+            and node.get("bounds") == "[396,498][1053,570]"
+        )
+        row = next(node for node in frame if node.get("bounds") == "[396,498][693,570]")
+        current = next(node for node in row if node.get("text") == "限1件 ¥12.88 ")
+        if kind == "hybrid":
+            current.set("text", "快卖完 ¥12.88")
+            current.set("bounds", "[396,503][712,570]")
+            original = ElementTree.Element(
+                "node",
+                dict(current.attrib)
+                | {
+                    "text": "¥29.88",
+                    "bounds": "[730,503][895,570]",
+                },
+            )
+            row.append(original)
+        elif kind == "duplicate":
+            parents[frame].append(
+                ElementTree.fromstring(ElementTree.tostring(frame, encoding="unicode"))
+            )
+        elif kind == "unknown":
+            current.set("bounds", "[396,502][675,570]")
+        elif kind == "unknown_child":
+            row.append(ElementTree.Element("node", dict(current.attrib) | {"text": "未知子节点"}))
+        elif kind in {"original", "coupon_original"}:
+            original = _inert_node("android.widget.TextView", "[730,503][895,570]")
+            original.set("text", "¥29.88" if kind == "original" else "券前¥29.88")
+            # 放在候选 header 之外，证明单价格分支检查整棵树，而非只看价格容器。
+            root.append(original)
+        elif kind == "parent_drift":
+            frame.remove(row)
+            parents[frame].append(row)
+        else:
+            raise AssertionError(kind)
+    else:
+        raise AssertionError(variant)
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _candidate_chain_drift(role: str, kind: str) -> str:
+    root = ElementTree.fromstring(_candidate_current_only())
+    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") == "[0,366][1080,1000]"
+    )
+    surface = parents[header]
+    content_frame = next(
+        node for node in header
+        if node.get("class") == "android.widget.FrameLayout"
+        and node.get("bounds") == "[0,474][1080,863]"
+    )
+    relative = next(iter(content_frame))
+    content = next(iter(relative))
+    price_frame = next(
+        node for node in content
+        if node.get("class") == "android.widget.FrameLayout"
+        and node.get("bounds") == "[396,498][1053,570]"
+    )
+    row = next(iter(price_frame))
+    current = next(iter(row))
+    summary = next(
+        node for node in content
+        if node.get("class") == "android.widget.TextView"
+        and node.get("bounds") == "[396,654][1053,716]"
+    )
+    roles = {
+        "header": header,
+        "content_frame": content_frame,
+        "relative": relative,
+        "content": content,
+        "price_frame": price_frame,
+        "row": row,
+        "current": current,
+        "summary": summary,
+    }
+    target = roles[role]
+    if kind == "attrs":
+        target.set("enabled", "false")
+    elif kind == "children":
+        target.append(_inert_node("android.view.View", "[1,1][2,2]"))
+    elif kind == "parent":
+        new_parents = {
+            "header": next(
+                node for node in surface
+                if node.get("class") == "androidx.recyclerview.widget.RecyclerView"
+            ),
+            "content_frame": surface,
+            "relative": header,
+            "content": content_frame,
+            "price_frame": header,
+            "row": content,
+            "current": price_frame,
+            "summary": header,
+        }
+        parents[target].remove(target)
+        new_parents[role].append(target)
+    else:
+        raise AssertionError(kind)
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _candidate_cross_variant_residue(variant: str, kind: str) -> str:
+    root = ElementTree.fromstring(
+        _candidate_unselected() if variant == "dual" else _candidate_current_only()
+    )
+    content = next(
+        node for node in root.iter("node")
+        if node.get("class") == "android.view.ViewGroup"
+        and node.get("bounds") == "[0,474][1080,863]"
+    )
+    filler = next(node for node in content if node.get("class") == "android.view.View")
+    content.remove(filler)
+    if variant == "dual":
+        row = _inert_node("android.widget.LinearLayout", "[396,498][693,570]")
+        current = _inert_node("android.widget.TextView", "[396,503][675,570]")
+        current.set("text", "限1件 ¥12.88 ")
+        row.append(current)
+        fragments = {"row": row, "current": current}
+    elif variant == "current_only":
+        row = _inert_node("android.widget.LinearLayout", "[396,498][895,570]")
+        current = _inert_node("android.widget.TextView", "[396,503][712,570]")
+        current.set("text", "快卖完 ¥12.88")
+        original = _inert_node("android.widget.TextView", "[730,503][895,570]")
+        original.set("text", "¥29.88")
+        row.extend((current, original))
+        fragments = {"row": row, "current": current, "original": original}
+    else:
+        raise AssertionError(variant)
+    if kind == "complete":
+        frame = _inert_node("android.widget.FrameLayout", "[396,498][1053,570]")
+        frame.append(row)
+        content.append(frame)
+    else:
+        # 只插入一段精确 opposing role，且维持 content 原 child-count，防止由计数检查掩盖残片检查。
+        content.append(ElementTree.fromstring(ElementTree.tostring(fragments[kind], encoding="unicode")))
+    return ElementTree.tostring(root, encoding="unicode")
+
+
 def _candidate_with_color_unselected() -> str:
     root = ElementTree.fromstring(_candidate_unselected())
     target = next(
@@ -290,6 +529,62 @@ class SkuRevealSpikeTests(unittest.TestCase):
             self.assertNotIn("start", manifest)
             self.assertNotIn("end", manifest)
 
+    def test_current_only_candidate_publishes_without_changing_other_candidate_rules(self) -> None:
+        device = _FakeDevice()
+        device.after_hierarchy = _candidate_current_only()
+        with TemporaryDirectory() as directory:
+            target = Path(directory) / "evidence"
+            result = self._capturer(device).capture(
+                "192.168.0.173:5555",
+                "937122477375",
+                target,
+            )
+
+            self.assertEqual(result.output_directory, target)
+            self.assertEqual(len(_swipes(device)), 1)
+            self.assertTrue((target / "after" / "hierarchy.xml").is_file())
+
+    def test_candidate_price_variants_reject_hybrid_duplicate_unknown_and_parent_drift(self) -> None:
+        # 两个真实嵌套 variant 各自唯一命中；任意混搭、额外子节点或父子关系漂移均拒绝。
+        _candidate_projection(_parse_nodes(_candidate_unselected()))
+        _candidate_projection(_parse_nodes(_candidate_current_only()))
+        for variant in ("dual", "current_only"):
+            kinds = ["hybrid", "duplicate", "unknown", "unknown_child", "parent_drift"]
+            if variant == "current_only":
+                kinds.extend(("original", "coupon_original"))
+            for kind in kinds:
+                with self.subTest(variant=variant, kind=kind), self.assertRaises(SkuSelectionError):
+                    _candidate_projection(
+                        _parse_nodes(_candidate_price_drift(variant, kind))
+                    )
+
+    def test_candidate_real_container_chain_rejects_every_parent_attribute_and_child_drift(self) -> None:
+        roles = (
+            "header",
+            "content_frame",
+            "relative",
+            "content",
+            "price_frame",
+            "row",
+            "current",
+            "summary",
+        )
+        for role in roles:
+            for kind in ("parent", "attrs", "children"):
+                with self.subTest(role=role, kind=kind), self.assertRaises(SkuSelectionError):
+                    _candidate_projection(_parse_nodes(_candidate_chain_drift(role, kind)))
+
+    def test_candidate_price_variants_reject_complete_and_partial_opposing_shapes(self) -> None:
+        for variant, kinds in (
+            ("dual", ("complete", "row", "current")),
+            ("current_only", ("complete", "row", "current", "original")),
+        ):
+            for kind in kinds:
+                with self.subTest(variant=variant, kind=kind), self.assertRaises(SkuSelectionError):
+                    _candidate_projection(
+                        _parse_nodes(_candidate_cross_variant_residue(variant, kind))
+                    )
+
     def test_current_only_empty_and_color_only_are_accepted_before_reveal(self) -> None:
         device = _FakeDevice()
         device.empty_hierarchy = _CURRENT_ONLY_EMPTY
@@ -429,6 +724,32 @@ class SkuRevealSpikeTests(unittest.TestCase):
         self.assertEqual(_swipes(device), [])
         self.assertEqual(safe_reveal_failure_stage(raised.exception), "reveal_precondition")
 
+    def test_after_screenshot_current_only_to_dual_drift_is_not_published(self) -> None:
+        device = _FakeDevice()
+        current_only = _candidate_current_only()
+        device.after_hierarchy = current_only
+        original_call = device.jsonrpc_call
+
+        def drift(method: str, params: object = None, timeout: float = 10) -> str:
+            value = original_call(method, params, timeout)
+            if method == "takeScreenshot" and device.hierarchy == current_only:
+                device.hierarchy = _candidate_unselected()
+            return value
+
+        device.jsonrpc_call = drift  # type: ignore[method-assign]
+        with TemporaryDirectory() as directory:
+            target = Path(directory) / "evidence"
+            with self.assertRaises(SkuRevealSpikeError) as raised:
+                self._capturer(device).capture(
+                    "192.168.0.173:5555",
+                    "937122477375",
+                    target,
+                )
+            self.assertEqual(safe_reveal_failure_stage(raised.exception), "reveal_after")
+            self.assertEqual(len(_swipes(device)), 1)
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
     def test_reveal_failure_stages_bind_attempt_candidate_after_and_publish(self) -> None:
         device = _FakeDevice()