cmbuyer/client/tests/pdd/test_sku_reveal_spike.py

from __future__ import annotations

import base64
from contextlib import redirect_stderr
from io import BytesIO
import importlib.util
import io
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.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,
    safe_reveal_failure_stage,
    SkuRevealSpikeCapturer,
    SkuRevealSpikeError,
)
from cmbuyer_client.pdd.sku_selection import (
    SkuSelectionError,
    _annotate_sku_entry_failure,
    _parse_nodes,
    _require_safe_reveal_path,
    _revealed_unselected_projection as _candidate_projection,
)
from cmbuyer_client.pdd.sku_selection_runner import (
    UiautomatorSkuPanelAdapter as _RevealEvidenceAdapter,
)


_FIXTURES = Path(__file__).with_name("fixtures")
_PRODUCT = (_FIXTURES / "product_entry_8_17_0.xml").read_text(encoding="utf-8")
_EMPTY = (_FIXTURES / "sku_panel_empty_8_17_0.xml").read_text(encoding="utf-8")
_COLOR_ONLY = (_FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml").read_text(encoding="utf-8")
_CURRENT_ONLY_EMPTY = (_FIXTURES / "sku_panel_empty_current_only_8_17_0.xml").read_text(encoding="utf-8")
_CURRENT_ONLY_COLOR = (_FIXTURES / "sku_panel_color_selected_size_hidden_current_only_8_17_0.xml").read_text(encoding="utf-8")
_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}
    for node in root.iter("node"):
        if node.get("text") == "已选： 黑色 CHA (纯棉） M（建议100-115）":
            node.set("text", "请选择： 尺码")
        if node.get("text") in {"S（建议80-100）", "M（建议100-115）"}:
            node.set("selected", "false")
            wrapper = parents[node]
            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")
        if node.get("class") == "android.view.ViewGroup"
        and node.get("clickable") == "true"
        and node.get("bounds") == "[0,366][1080,2328]"
    )
    button_frame = ElementTree.SubElement(
        action_root,
        "node",
        {
            "class": "android.widget.FrameLayout",
            "package": "com.xunmeng.pinduoduo",
            "text": "",
            "content-desc": "",
            "clickable": "true",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[0,2181][1080,2328]",
        },
    )
    button_content = ElementTree.SubElement(
        button_frame,
        "node",
        {
            "class": "android.widget.LinearLayout",
            "package": "com.xunmeng.pinduoduo",
            "text": "",
            "content-desc": "",
            "clickable": "false",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[273,2181][807,2328]",
        },
    )
    ElementTree.SubElement(
        button_content,
        "node",
        {
            "class": "android.widget.TextView",
            "package": "com.xunmeng.pinduoduo",
            "text": "选择尺码后，提交订单",
            "content-desc": "",
            "clickable": "false",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[285,2225][795,2284]",
        },
    )
    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_current_only())
    target = next(
        node
        for node in root.iter("node")
        if node.get("class") == "android.view.ViewGroup"
        and node.get("content-desc") == "黑色 CHA (纯棉）"
        and node.get("bounds") == "[372,1000][684,1024]"
    )
    for node in target.iter("node"):
        node.set("selected", "false")
    return ElementTree.tostring(root, encoding="unicode")


def _candidate_with_submit_risk(kind: str) -> str:
    root = ElementTree.fromstring(_candidate_current_only())
    parents = {child: parent for parent in root.iter() for child in parent}
    submit = next(
        node
        for node in root.iter("node")
        if "提交订单" in node.get("text", "")
        and node.get("class") == "android.widget.TextView"
    )
    parent = parents[submit]
    if kind == "missing":
        parent.remove(submit)
    elif kind == "duplicate":
        parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")))
    elif kind == "moved_up":
        submit.set("bounds", "[285,1900][795,1959]")
    elif kind in {"extra_altered", "extra_moved_up"}:
        extra = ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode"))
        if kind == "extra_altered":
            extra.set("text", "异常提交订单文案")
        else:
            extra.set("bounds", "[285,1900][795,1959]")
        parent.append(extra)
    else:
        raise AssertionError(kind)
    return ElementTree.tostring(root, encoding="unicode")


def _with_clickable_overlay(package: str, class_name: str, bounds: str) -> str:
    root = ElementTree.fromstring(_COLOR_ONLY)
    container = next(root.iter("node"))
    ElementTree.SubElement(
        container,
        "node",
        {
            "package": package,
            "class": class_name,
            "text": "",
            "content-desc": "",
            "clickable": "true",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": bounds,
        },
    )
    return ElementTree.tostring(root, encoding="unicode")


def _png() -> str:
    image = Image.new("RGB", (1080, 2376), "white")
    raw = BytesIO()
    image.save(raw, format="PNG")
    return base64.b64encode(raw.getvalue()).decode("ascii")


class _FakeDevice:
    def __init__(self) -> None:
        self.hierarchy = "<hierarchy />"
        self.empty_hierarchy = _CURRENT_ONLY_EMPTY
        self.color_hierarchy = _CURRENT_ONLY_COLOR
        self.after_hierarchy = _candidate_current_only()
        self.calls: list[tuple[object, ...]] = []
        self.swipe_error = False

    def app_info(self, package_name: str) -> dict[str, str]:
        self.calls.append(("app_info", package_name))
        return {"versionName": "8.17.0"}

    def app_current(self) -> dict[str, str]:
        self.calls.append(("app_current",))
        return {"package": "com.xunmeng.pinduoduo"}

    def window_size(self) -> tuple[int, int]:
        self.calls.append(("window_size",))
        return 1080, 2376

    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 _png()
        if method == "click":
            if params == [865, 2218]:
                self.hierarchy = self.empty_hierarchy
                return ""
            if params == [528, 1387]:
                self.hierarchy = self.color_hierarchy
                return ""
            raise AssertionError(params)
        if method == "swipe":
            self.hierarchy = self.after_hierarchy
            if self.swipe_error:
                raise TimeoutError("unknown outcome")
            return ""
        raise AssertionError(method)


class _FakeAdb:
    def __init__(self, device: _FakeDevice) -> None:
        self.device = device
        self.calls: list[tuple[object, ...]] = []

    def inspect(self, serial: str) -> DeviceInspection:
        self.calls.append(("inspect", serial))
        return DeviceInspection(AdbDevice(serial=serial, state="device"), "PKG110", "16")

    def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
        self.calls.append(("intent", serial, goods_id))
        self.device.hierarchy = _PRODUCT
        return object()


def _swipes(device: _FakeDevice) -> list[tuple[object, ...]]:
    return [call for call in device.calls if call[:2] == ("jsonrpc", "swipe")]


class SkuRevealSpikeTests(unittest.TestCase):
    def _capturer(self, device: _FakeDevice) -> SkuRevealSpikeCapturer:
        now = [0.0]
        return SkuRevealSpikeCapturer(
            _FakeAdb(device),
            lambda serial: device,
            30,
            monotonic_clock=lambda: now[0],
            sleep_function=lambda seconds: now.__setitem__(0, now[0] + seconds),
        )

    def _assert_after_rejected_once(self, after_hierarchy: str, *, ambiguous: bool = False) -> None:
        device = _FakeDevice()
        device.after_hierarchy = after_hierarchy
        device.swipe_error = ambiguous
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    target,
                )
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

    def test_success_publishes_before_after_and_exactly_one_reveal(self) -> None:
        device = _FakeDevice()
        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.assertFalse(any(call[1] == "pressKey" for call in device.calls if call[0] == "jsonrpc"))
            for relative in (
                "before/screenshot.png",
                "before/hierarchy.xml",
                "after/screenshot.png",
                "after/hierarchy.xml",
                "manifest.json",
            ):
                self.assertTrue((target / relative).is_file(), relative)
            manifest_text = result.manifest_path.read_text(encoding="utf-8")
            manifest = json.loads(manifest_text)
            self.assertEqual(manifest["reveal_attempts"], 1)
            self.assertEqual(manifest["rpc_outcome"], "completed")
            self.assertNotIn("192.168.0.173:5555", manifest_text)
            self.assertNotIn("gesture", manifest)
            self.assertNotIn("coordinates", manifest)
            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:
        # 生产只放行 current-only；历史 dual 及任意混搭、额外节点、父链漂移均拒绝。
        with self.assertRaises(SkuSelectionError):
            _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
        device.color_hierarchy = _CURRENT_ONLY_COLOR
        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.assertEqual(
                (target / "before" / "hierarchy.xml").read_text(encoding="utf-8"),
                _CURRENT_ONLY_COLOR,
            )

    def test_cross_price_variant_after_color_click_is_precondition_failure_without_reveal(self) -> None:
        for empty_hierarchy, color_hierarchy in (
            (_CURRENT_ONLY_EMPTY, _COLOR_ONLY),
            (_EMPTY, _CURRENT_ONLY_COLOR),
        ):
            with self.subTest():
                device = _FakeDevice()
                device.empty_hierarchy = empty_hierarchy
                device.color_hierarchy = color_hierarchy
                with TemporaryDirectory() as directory:
                    target = Path(directory) / "evidence"
                    with self.assertRaises(SkuSelectionError) as raised:
                        self._capturer(device).capture(
                            "192.168.0.173:5555",
                            "937122477375",
                            target,
                        )
                    self.assertEqual(
                        safe_reveal_failure_stage(raised.exception),
                        "reveal_precondition",
                    )
                    self.assertEqual(_swipes(device), [])
                    self.assertFalse(target.exists())
                    self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

    def test_ambiguous_rpc_is_read_only_reconciled_without_retry(self) -> None:
        device = _FakeDevice()
        device.swipe_error = True
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(len(_swipes(device)), 1)
            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
            self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled")

    def test_selected_m_or_precondition_drift_never_publishes_or_retries(self) -> None:
        for after in (
            _M_SELECTED,
            _candidate_unselected().replace("请选择： 尺码", "已选： 尺码"),
        ):
            with self.subTest():
                device = _FakeDevice()
                device.after_hierarchy = after
                with TemporaryDirectory() as directory:
                    target = Path(directory) / "evidence"
                    with self.assertRaises(SkuRevealSpikeError):
                        self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
                    self.assertEqual(len(_swipes(device)), 1)
                    self.assertFalse(target.exists())
                    self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        device = _FakeDevice()
        device.hierarchy = "<hierarchy />"
        invalid_color = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
        original_call = device.jsonrpc_call

        def drift(method: str, params: object = None, timeout: float = 10) -> str:
            value = original_call(method, params, timeout)
            if method == "click" and params == [528, 1387]:
                device.hierarchy = invalid_color
            return value

        device.jsonrpc_call = drift  # type: ignore[method-assign]
        with TemporaryDirectory() as directory:
            with self.assertRaises((SkuSelectionError, SkuRevealSpikeError)):
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    Path(directory) / "evidence",
                )
        self.assertEqual(_swipes(device), [])

    def test_hidden_sizes_after_swipe_are_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_COLOR_ONLY)

    def test_post_reveal_color_selected_drift_is_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_candidate_with_color_unselected())

    def test_submit_hard_reject_zone_risk_is_not_published_or_retried(self) -> None:
        for kind in ("missing", "duplicate", "moved_up", "extra_altered", "extra_moved_up"):
            with self.subTest(kind=kind):
                self._assert_after_rejected_once(_candidate_with_submit_risk(kind))

    def test_ambiguous_rpc_with_unchanged_page_is_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_COLOR_ONLY, ambiguous=True)

    def test_invalid_goods_and_existing_target_are_zero_action(self) -> None:
        device = _FakeDevice()
        with TemporaryDirectory() as directory:
            target = Path(directory) / "existing"
            target.mkdir()
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture("wifi", "1", Path(directory) / "new")
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture("wifi", "937122477375", target)
        self.assertEqual(device.calls, [])

    def test_before_screenshot_drift_is_rechecked_before_zero_swipe(self) -> None:
        device = _FakeDevice()
        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":
                device.hierarchy = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
            return value

        device.jsonrpc_call = drift  # type: ignore[method-assign]
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            with self.assertRaises(SkuSelectionError) as raised:
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    target,
                )
            self.assertFalse(target.exists())
        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, SkuSelectionError)) 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()

        def fail_before_seal(adapter: _RevealEvidenceAdapter) -> None:
            raise SkuRevealSpikeError("private pre-seal detail")

        with TemporaryDirectory() as directory:
            target = Path(directory) / "pre-seal"
            with (
                patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_before_seal),
                self.assertRaises(SkuRevealSpikeError) as pre_seal,
            ):
                self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(safe_reveal_failure_stage(pre_seal.exception), "reveal_precondition")
            self.assertEqual(_swipes(device), [])
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        # attempted 只可能在 adapter 已封存并发送唯一手势后报告。
        device = _FakeDevice()
        original_reveal = _RevealEvidenceAdapter.reveal_size_options_once

        def fail_after_attempt(adapter: _RevealEvidenceAdapter) -> None:
            original_reveal(adapter)
            raise SkuRevealSpikeError("private attempted detail")

        with TemporaryDirectory() as directory:
            target = Path(directory) / "attempted"
            with (
                patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_after_attempt),
                self.assertRaises(SkuRevealSpikeError) as attempted,
            ):
                self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(safe_reveal_failure_stage(attempted.exception), "reveal_attempted")
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        device = _FakeDevice()
        device.after_hierarchy = _COLOR_ONLY
        with TemporaryDirectory() as directory:
            target = Path(directory) / "candidate"
            with self.assertRaises(SkuRevealSpikeError) as candidate:
                self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(safe_reveal_failure_stage(candidate.exception), "reveal_candidate")
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        device = _FakeDevice()
        original_capture = reveal_module._capture_frame

        def fail_after_capture(adapter: object, directory: Path, hierarchy: str) -> None:
            if directory.name == "after":
                raise OSError("private after path")
            original_capture(adapter, directory, hierarchy)

        with TemporaryDirectory() as directory:
            target = Path(directory) / "after"
            with (
                patch.object(reveal_module, "_capture_frame", fail_after_capture),
                self.assertRaises(SkuRevealSpikeError) as after,
            ):
                self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(safe_reveal_failure_stage(after.exception), "reveal_after")
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        device = _FakeDevice()
        with TemporaryDirectory() as directory:
            target = Path(directory) / "publish"
            with (
                patch.object(reveal_module.os, "rename", side_effect=OSError("private publish path")),
                self.assertRaises(SkuRevealSpikeError) as publish,
            ):
                self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(safe_reveal_failure_stage(publish.exception), "reveal_publish")
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

    def test_complete_reveal_segment_rejects_narrow_impostor_and_invalid_bounds(self) -> None:
        for hierarchy in (
            _with_clickable_overlay(
                "com.xunmeng.pinduoduo",
                "android.view.ViewGroup",
                "[350,1450][370,1460]",
            ),
            _with_clickable_overlay(
                "com.android.systemui",
                "android.view.ViewGroup",
                "[0,366][1080,2328]",
            ),
            _with_clickable_overlay(
                "com.xunmeng.pinduoduo",
                "android.view.ViewGroup",
                "not-a-bound",
            ),
        ):
            with self.subTest(), self.assertRaises((SkuRevealSpikeError, SkuSelectionError)):
                _require_safe_reveal_path(_parse_nodes(hierarchy))


class SkuRevealSpikeCliTests(unittest.TestCase):
    def test_cli_has_no_gesture_or_task_specification_parameters(self) -> None:
        script = _load_reveal_script()
        arguments = script.parse_arguments(
            [
                "--serial", "device-1",
                "--goods-id", "937122477375",
                "--output-dir", "evidence",
            ]
        )
        self.assertEqual(
            set(vars(arguments)),
            {"serial", "goods_id", "output_dir", "timeout", "adb"},
        )
        script.validate_arguments(arguments)
        for field, value in (
            ("serial", ""),
            ("goods_id", "1"),
            ("timeout", 0),
            ("timeout", float("inf")),
        ):
            with self.subTest(field=field), self.assertRaises(ValueError):
                script.validate_arguments(
                    type("Arguments", (), vars(arguments) | {field: value})()
                )

    def test_cli_failure_is_redacted(self) -> None:
        script = _load_reveal_script()
        secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"

        class FailingCapturer:
            def __init__(self, *args: object, **kwargs: object) -> None:
                return None

            def capture(self, *args: object, **kwargs: object) -> object:
                raise SkuRevealSpikeError(secret)

        stderr = io.StringIO()
        with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
            status = script.main(
                [
                    "--serial", "192.168.0.173:5555",
                    "--goods-id", "937122477375",
                    "--output-dir", "evidence",
                ]
            )
        output = stderr.getvalue()
        self.assertEqual(status, 1)
        self.assertIn("stage=unknown", output)
        self.assertNotIn("Traceback", output)
        self.assertNotIn("192.168.0.173:5555", output)
        self.assertNotIn("private", output)

    def test_cli_reports_only_formally_annotated_entry_stage(self) -> None:
        script = _load_reveal_script()
        secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"

        def run_with(error: BaseException) -> str:
            class FailingCapturer:
                def __init__(self, *args: object, **kwargs: object) -> None:
                    return None

                def capture(self, *args: object, **kwargs: object) -> object:
                    raise error

            stderr = io.StringIO()
            with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
                status = script.main(
                    [
                        "--serial", "192.168.0.173:5555",
                        "--goods-id", "937122477375",
                        "--output-dir", "evidence",
                    ]
                )
            self.assertEqual(status, 1)
            output = stderr.getvalue()
            self.assertNotIn("Traceback", output)
            self.assertNotIn("192.168.0.173:5555", output)
            self.assertNotIn("private", output)
            return output

        annotated = SkuSelectionError(secret)
        _annotate_sku_entry_failure(annotated, "sku_entry_panel_verify")
        _annotate_reveal_failure(annotated, "reveal_precondition")
        self.assertIn("stage=sku_entry_panel_verify", run_with(annotated))

        spoofed = SkuSelectionError(secret)
        setattr(spoofed, "_cmbuyer_failure_stage", "sku_entry_panel_verify")
        self.assertIn("stage=unknown", run_with(spoofed))

    def test_cli_reports_only_formally_annotated_reveal_stage(self) -> None:
        script = _load_reveal_script()
        secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"

        def run_with(error: BaseException) -> str:
            class FailingCapturer:
                def __init__(self, *args: object, **kwargs: object) -> None:
                    return None

                def capture(self, *args: object, **kwargs: object) -> object:
                    raise error

            stderr = io.StringIO()
            with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
                self.assertEqual(
                    script.main(
                        [
                            "--serial", "192.168.0.173:5555",
                            "--goods-id", "937122477375",
                            "--output-dir", "evidence",
                        ]
                    ),
                    1,
                )
            output = stderr.getvalue()
            self.assertNotIn("192.168.0.173:5555", output)
            self.assertNotIn("private", output)
            return output

        annotated = SkuRevealSpikeError(secret)
        _annotate_reveal_failure(annotated, "reveal_candidate")
        self.assertIn("stage=reveal_candidate", run_with(annotated))

        spoofed = SkuRevealSpikeError(secret)
        setattr(spoofed, "_cmbuyer_reveal_failure_stage", "reveal_candidate")
        self.assertIn("stage=unknown", run_with(spoofed))

        class HostileGetterError(SkuRevealSpikeError):
            def __getattribute__(self, name: str) -> object:
                if name.startswith("_cmbuyer_reveal_"):
                    raise RuntimeError(secret)
                return super().__getattribute__(name)

        self.assertIn("stage=unknown", run_with(HostileGetterError(secret)))


def _load_reveal_script() -> object:
    path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_reveal_spike.py"
    specification = importlib.util.spec_from_file_location("capture_sku_reveal_spike_test", path)
    if specification is None or specification.loader is None:
        raise RuntimeError("无法加载 T-103 reveal 取证脚本。")
    module = importlib.util.module_from_spec(specification)
    specification.loader.exec_module(module)
    return module


if __name__ == "__main__":
    unittest.main()