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,
    _RevealEvidenceAdapter,
    safe_reveal_failure_stage,
    SkuRevealSpikeCapturer,
    SkuRevealSpikeError,
    _require_safe_reveal_path,
)
from cmbuyer_client.pdd.sku_selection import (
    SkuSelectionError,
    _annotate_sku_entry_failure,
    _parse_nodes,
)


_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 _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)
    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_with_color_unselected() -> str:
    root = ElementTree.fromstring(_candidate_unselected())
    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_unselected())
    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]")
    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 = _EMPTY
        self.color_hierarchy = _COLOR_ONLY
        self.after_hierarchy = _candidate_unselected()
        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_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"):
            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_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):
                _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()