from __future__ import annotations

import ast
import base64
from contextlib import redirect_stderr
from io import BytesIO
import importlib.util
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from xml.etree import ElementTree

from PIL import Image

import cmbuyer_client.pdd as pdd
import cmbuyer_client.pdd.sku_selection_runner as runner_module
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
from cmbuyer_client.pdd.sku_selection_runner import (
    SkuSelectionDeviceAdapterError,
    SkuSelectionRunError,
    SkuSelectionScreenshotError,
    UiautomatorSkuPanelAdapter,
)


_FIXTURE = Path(__file__).with_name("fixtures") / "sku_panel_8_17_0.xml"
_ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml"
_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
_TASK_COLOR = "黑色CHA(纯棉)"
_TASK_SIZE = "M(建议100-115)"
_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")


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


class _RawDevice:
    def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
        self.hierarchy = hierarchy
        self.panel_hierarchy = _FIXTURE.read_text(encoding="utf-8")
        self.version = "8.17.0"
        self.package = "com.xunmeng.pinduoduo"
        self.screenshot = _png_base64() if screenshot is None else screenshot
        self.calls: list[tuple[object, ...]] = []
        self.fail_color_readback = False

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

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

    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
        self.calls.append(("jsonrpc", method, params, timeout))
        if method == "dumpWindowHierarchy":
            return self.hierarchy
        if method == "takeScreenshot":
            return self.screenshot
        if method == "pressKey":
            self.hierarchy = "<hierarchy />"
            return ""
        if method == "click":
            if not isinstance(params, list) or len(params) != 2:
                raise AssertionError(params)
            self._apply_tap(int(params[0]), int(params[1]))
            return ""
        raise AssertionError(method)

    def _apply_tap(self, x: int, y: int) -> None:
        if "快要抢光" in self.hierarchy and "[396,498][895,570]" not in self.hierarchy:
            self.hierarchy = self.panel_hierarchy
            return
        root = ElementTree.fromstring(self.hierarchy)
        target = next(node for node in root.iter("node") if _center(node.get("bounds", "")) == (x, y))
        color = target.get("bounds", "").endswith("][438,1172]")
        for node in root.iter("node"):
            if node.get("selected") is not None and ((color and ",1000]" in node.get("bounds", "")) or (not color and ",1730]" in node.get("bounds", ""))):
                node.set("selected", "false")
        if color and self.fail_color_readback:
            next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
        else:
            target.set("selected", "true")
        self.hierarchy = ElementTree.tostring(root, encoding="unicode")

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

    def select_alternates(self) -> None:
        root = ElementTree.fromstring(self.panel_hierarchy)
        for node in root.iter("node"):
            if node.get("selected") is not None:
                node.set("selected", "false")
        next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
        next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
        self.panel_hierarchy = ElementTree.tostring(root, encoding="unicode")
        if self.hierarchy != _PRODUCT_PAGE:
            self.hierarchy = self.panel_hierarchy


def _center(bounds: str) -> tuple[int, int]:
    left_top, right_bottom = bounds.split("][")
    left, top = (int(value) for value in left_top.removeprefix("[").split(","))
    right, bottom = (int(value) for value in right_bottom.removesuffix("]").split(","))
    return left + (right - left) // 2, top + (bottom - top) // 2


def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]:
    parents = {child: parent for parent in root.iter() for child in parent}
    child = next(node for node in root.iter("node") if node.get("text") == "快要抢光")
    chain = [child]
    for _ in range(4):
        chain.append(parents[chain[-1]])
    return chain


def _mutate_entry(depth: int, attribute: str, value: str) -> str:
    root = ElementTree.fromstring(_PRODUCT_PAGE)
    _entry_chain(root)[depth].set(attribute, value)
    return ElementTree.tostring(root, encoding="unicode")


def _without_entry() -> str:
    root = ElementTree.fromstring(_PRODUCT_PAGE)
    root.remove(_entry_chain(root)[4])
    return ElementTree.tostring(root, encoding="unicode")


def _duplicate_entry() -> str:
    root = ElementTree.fromstring(_PRODUCT_PAGE)
    entry_root = _entry_chain(root)[4]
    root.append(ElementTree.fromstring(ElementTree.tostring(entry_root, encoding="unicode")))
    return ElementTree.tostring(root, encoding="unicode")


def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
    return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]


def _tap_centers(device: _RawDevice) -> list[tuple[int, int]]:
    return [tuple(call[2]) for call in _actions(device, "click")]  # type: ignore[misc]


class _FakeAdb:
    def __init__(self) -> None:
        self.calls: list[tuple[object, ...]] = []
        self.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "PKG110", "16")
        self.on_intent: callable | None = None

    def inspect(self, serial: str) -> DeviceInspection:
        self.calls.append(("inspect", serial))
        return self.inspection

    def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
        self.calls.append(("intent", serial, goods_id))
        if self.on_intent is not None:
            self.on_intent()
        return object()


class SkuSelectionFlowTests(unittest.TestCase):
    def _assert_entry_rejected_without_click(self, hierarchy: str) -> None:
        now = [0.0]
        device = _RawDevice(hierarchy)
        flow = SkuSelectionFlow(
            UiautomatorSkuPanelAdapter(device, 10),
            0.01,
            0.01,
            lambda: now[0],
            lambda seconds: now.__setitem__(0, now[0] + seconds),
        )
        with self.assertRaises(SkuSelectionError):
            flow.open_sku_panel(_TARGET_URL)
        self.assertEqual(_actions(device, "click"), [])

    def test_target_mapping_is_exact_and_success_path_restores_target(self) -> None:
        device = _RawDevice()
        adapter = UiautomatorSkuPanelAdapter(device, 10)
        flow = SkuSelectionFlow(adapter)

        flow.open_sku_panel(_TARGET_URL)
        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
        self.assertEqual(flow.read_sku_unit_price(), "12.88")
        flow.exit_sku_panel_safely()

        self.assertEqual(_tap_centers(device), [(978, 1333)])
        self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])

    def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
        device = _RawDevice()
        SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)

        self.assertEqual(_tap_centers(device), [(978, 1333)])

    def test_entry_child_and_every_ancestor_attribute_drift_never_clicks(self) -> None:
        expected_clickable = ("false", "false", "false", "false", "true")
        for depth in range(5):
            changes = {
                "package": "other.package",
                "class": "android.widget.Button",
                "bounds": "[1,1][2,2]",
                "clickable": "true" if expected_clickable[depth] == "false" else "false",
                "enabled": "false",
                "visible-to-user": "false",
            }
            for attribute, value in changes.items():
                with self.subTest(depth=depth, attribute=attribute):
                    self._assert_entry_rejected_without_click(_mutate_entry(depth, attribute, value))

    def test_duplicate_entry_and_forbidden_sibling_entry_never_click(self) -> None:
        self._assert_entry_rejected_without_click(_duplicate_entry())
        self._assert_entry_rejected_without_click(_without_entry())
        self._assert_entry_rejected_without_click(_mutate_entry(0, "clickable", "true"))

    def test_unknown_task_or_ui_variants_are_rejected_without_action(self) -> None:
        for color, size in (("黑色 CHA (纯棉）", _TASK_SIZE), (_TASK_COLOR, "M（建议100-115）"), ("黑色CHA（纯棉）", _TASK_SIZE)):
            with self.subTest(color=color, size=size), self.assertRaises(SkuSelectionError):
                resolve_task_selection(color, size)

        device = _RawDevice(_FIXTURE.read_text(encoding="utf-8"))
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).select_sku_options(
                resolve_task_selection(_TASK_COLOR, _TASK_SIZE).__class__("粉红", "L（建议115-130）")
            )
        self.assertEqual(_actions(device, "click"), [])

    def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None:
        base = _FIXTURE.read_text(encoding="utf-8")
        cases = (
            base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
            base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'selected="maybe" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
            base.replace('bounds="[126,1000][438,1172]"', 'bounds="[1,1][20,20]"'),
            base.replace('enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'enabled="false" visible-to-user="true" bounds="[126,1000][438,1172]"'),
        )
        for hierarchy in cases:
            with self.subTest(), self.assertRaises(SkuSelectionError):
                SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(hierarchy), 10)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))

    def test_invalid_bounds_stop_before_action(self) -> None:
        for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"):
            with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError):
                _action_bounds(bounds)

        device = _RawDevice(_PRODUCT_PAGE.replace("[900,1312][1056,1355]", "[0,0][1081,1]"))
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
        self.assertEqual(_actions(device, "click"), [])

    def test_color_readback_failure_never_attempts_second_option(self) -> None:
        device = _RawDevice()
        device.fail_color_readback = True
        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
        flow.open_sku_panel(_TARGET_URL)
        device.select_alternates()
        with self.assertRaises(SkuSelectionError):
            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
        self.assertEqual(_tap_centers(device), [(978, 1333), (282, 1086)])

    def test_non_target_selection_restores_each_dimension_once(self) -> None:
        device = _RawDevice()
        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
        flow.open_sku_panel(_TARGET_URL)
        device.select_alternates()
        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
        self.assertEqual(
            _tap_centers(device),
            [(978, 1333), (282, 1086), (635, 1772)],
        )

    def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
        for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
            with self.subTest(replacement=replacement):
                device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", replacement))
                with self.assertRaises(SkuSelectionError):
                    SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
        device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88"))
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
        clickable_parent = _FIXTURE.read_text(encoding="utf-8").replace(
            '<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">',
            '<node package="" class="android.view.ViewGroup" clickable="true" bounds="[396,498][895,570]">',
        )
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(clickable_parent), 10)).read_sku_unit_price()

    def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
        forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"}
        self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
        self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
        self.assertTrue(forbidden.isdisjoint(pdd.__all__))

    def test_static_ast_boundary_limits_flow_runner_adapter_and_cli(self) -> None:
        root = Path(__file__).resolve().parents[2]
        files = (
            root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py",
            root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py",
            root / "scripts" / "run_t103_sku_selection.py",
        )
        forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment")
        for path in files:
            source = path.read_text(encoding="utf-8")
            with self.subTest(path=path.name):
                self.assertTrue(all(token not in source.lower() for token in forbidden))
                tree = ast.parse(source)
                self.assertFalse(any(isinstance(node, ast.ImportFrom) and node.module in {"selenium", "requests"} for node in ast.walk(tree)))
        runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
        click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
        self.assertEqual(len(click_calls), 1)

    def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
        now = [0.0]
        device = _RawDevice()
        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), 0.01, 0.01, lambda: now[0], lambda seconds: now.__setitem__(0, now[0] + seconds))
        with self.assertRaises(SkuSelectionError):
            flow.open_sku_panel(_TARGET_URL, _PRODUCT_PAGE)
        self.assertEqual(_actions(device, "click"), [])

    def test_action_postcondition_wait_never_repeats_entry_click(self) -> None:
        class NoPanelAfterEntry(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "click":
                    self.calls.append(("jsonrpc", method, params, timeout))
                    return ""
                return super().jsonrpc_call(method, params, timeout)

        device = NoPanelAfterEntry()
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
        self.assertEqual(_tap_centers(device), [(978, 1333)])

        duplicate = _duplicate_entry()
        device = _RawDevice(duplicate)
        with self.assertRaises(SkuSelectionError):
            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
        self.assertEqual(_actions(device, "click"), [])

    def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None:
        for fixture in (_FIXTURE, _ENTRY_FIXTURE):
            content = fixture.read_text(encoding="utf-8")
            with self.subTest(fixture=fixture.name):
                self.assertNotRegex(content, r"1[3-9]\d{9}")
                for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"):
                    self.assertNotIn(forbidden, content)
        content = _FIXTURE.read_text(encoding="utf-8")
        root = ElementTree.fromstring(content)
        leaf = next(node for node in root.iter("node") if node.get("text") == "提交订单 ¥12.88")
        self.assertEqual(leaf.get("clickable"), "false")
        self.assertEqual(leaf.get("bounds"), "[369,2225][710,2284]")


class SkuSelectionRunnerTests(unittest.TestCase):
    def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner:
        device.hierarchy = "<hierarchy />"
        adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
        return SkuSelectionRunner(adb, lambda serial: device, 10)

    def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None:
        adb = _FakeAdb()
        device = _RawDevice()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "result"
            result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)

            self.assertEqual(result.unit_price, "12.88")
            manifest = result.manifest_path.read_text(encoding="utf-8")
            self.assertTrue(result.screenshot_path.is_file())
            self.assertNotIn("device-1", manifest)
            self.assertNotIn("hierarchy", manifest)
            self.assertNotIn("已选", manifest)
            self.assertIn('"unit_price": "12.88"', manifest)
            self.assertIn('"selection_status": "restored"', manifest)
            self.assertIn('"panel_status": "verified"', manifest)
            self.assertIn('"safe_exit": "completed"', manifest)
            self.assertFalse((target / "hierarchy.xml").exists())
            self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])

    def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "result"
            original_rename = runner_module.os.rename

            def create_target_then_rename(source: str | Path, destination: str | Path) -> None:
                Path(destination).mkdir()
                (Path(destination) / "sentinel").write_text("keep", encoding="utf-8")
                original_rename(source, destination)

            with patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), self.assertRaises(SkuSelectionRunError):
                self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
            self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep")
            self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])

    def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "result"
            with self.assertRaises(SkuSelectionScreenshotError):
                self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])

            target = Path(temporary) / "write-failure"
            with patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")):
                with self.assertRaises(SkuSelectionScreenshotError):
                    self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(temporary).glob(".write-failure.staging-*")), [])

            adb = _FakeAdb()
            device = _RawDevice()
            target.mkdir()
            sentinel = target / "keep"
            sentinel.write_text("keep", encoding="utf-8")
            with self.assertRaises(SkuSelectionRunError):
                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
            self.assertEqual(adb.calls, [])
            self.assertEqual(device.calls, [])
            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")

    def test_device_screen_and_output_preflight_fail_before_any_click(self) -> None:
        with TemporaryDirectory() as temporary:
            adb = _FakeAdb()
            adb.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "wrong", "16")
            device = _RawDevice()
            with self.assertRaises(SkuSelectionRunError):
                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
            self.assertEqual(device.calls, [])

            class WrongScreenDevice(_RawDevice):
                def window_size(self) -> tuple[int, int]:
                    return 1080, 1920

            device = WrongScreenDevice()
            with self.assertRaises(SkuSelectionRunError):
                self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "screen")
            self.assertEqual(_actions(device, "click"), [])

            parent_file = Path(temporary) / "not-a-directory"
            parent_file.write_text("x", encoding="utf-8")
            adb = _FakeAdb()
            device = _RawDevice()
            with self.assertRaises(SkuSelectionRunError):
                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, parent_file / "result")
            self.assertEqual(adb.calls, [])
            self.assertEqual(device.calls, [])

    def test_small_but_valid_png_is_not_accepted(self) -> None:
        image = Image.new("RGB", (1, 1), "white")
        raw = BytesIO(); image.save(raw, format="PNG")
        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionScreenshotError):
            self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run(
                "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
            )

    def test_failure_after_entry_attempts_one_safe_exit_and_hides_device_detail(self) -> None:
        adb = _FakeAdb()
        device = _RawDevice()
        device.fail_color_readback = True
        device.select_alternates()
        with TemporaryDirectory() as temporary:
            with self.assertRaises(SkuSelectionError):
                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
        self.assertEqual(_actions(device, "pressKey"), [])

        class FailingRawDevice(_RawDevice):
            def app_info(self, package_name: str) -> dict[str, str]:
                raise RuntimeError("device-1 <xml>private</xml>")

        with self.assertRaises(SkuSelectionDeviceAdapterError) as raised:
            UiautomatorSkuPanelAdapter(FailingRawDevice(), 10).app_info("com.xunmeng.pinduoduo")
        self.assertNotIn("device-1", str(raised.exception))
        self.assertNotIn("private", str(raised.exception))

    def test_unverified_failure_never_sends_blind_back(self) -> None:
        class InvalidAfterOptionDevice(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                value = super().jsonrpc_call(method, params, timeout)
                if method == "click" and "[396,498][895,570]" in self.hierarchy:
                    self.hierarchy = "<hierarchy />"
                return value

        device = InvalidAfterOptionDevice()
        device.select_alternates()
        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError):
            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
        self.assertEqual(_actions(device, "pressKey"), [])

    def test_adapter_timeout_is_mapped_without_third_party_detail(self) -> None:
        class TimeoutRawDevice(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                raise TimeoutError("device-1 <hierarchy>private</hierarchy>")

        with self.assertRaises(SkuSelectionRunError) as raised:
            UiautomatorSkuPanelAdapter(TimeoutRawDevice(), 10).dump_window_hierarchy()
        self.assertNotIn("device-1", str(raised.exception))
        self.assertNotIn("private", str(raised.exception))

    def test_entry_attempt_is_recorded_before_unconfirmed_click_and_not_retried(self) -> None:
        class TimeoutTapDevice(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                self.calls.append(("jsonrpc", method, params, timeout))
                if method == "click":
                    raise TimeoutError("device detail")
                return super().jsonrpc_call(method, params, timeout)

        adapter = UiautomatorSkuPanelAdapter(TimeoutTapDevice(), 10)
        with self.assertRaises(SkuSelectionRunError):
            adapter.tap_sku_entry("[900,1312][1056,1355]")
        self.assertTrue(adapter.entry_was_tapped)
        self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [978, 1333], 10)])

    def test_entry_stability_interruptions_never_click(self) -> None:
        now = [0.0]
        class SequenceDevice(_RawDevice):
            def __init__(self) -> None:
                super().__init__(); self.frames = [_PRODUCT_PAGE, "<hierarchy />", _PRODUCT_PAGE]
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "dumpWindowHierarchy" and self.frames:
                    self.hierarchy = self.frames.pop(0)
                return super().jsonrpc_call(method, params, timeout)
        device = SequenceDevice()
        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), .02, .01, lambda: now[0], lambda x: now.__setitem__(0, now[0] + x))
        with self.assertRaises(SkuSelectionError): flow.open_sku_panel(_TARGET_URL, "<hierarchy />")
        self.assertEqual(_actions(device, "click"), [])

    def test_screenshot_drift_and_foreground_drift_publish_nothing_and_never_back(self) -> None:
        for drift in ("color", "size", "price"):
            class DriftDevice(_RawDevice):
                def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                    value = super().jsonrpc_call(method, params, timeout)
                    if method == "takeScreenshot":
                        if drift == "price":
                            self.hierarchy = self.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
                        else:
                            root = ElementTree.fromstring(self.hierarchy)
                            if drift == "color":
                                for node in root.iter("node"):
                                    if node.get("selected") is not None and ",1000]" in node.get("bounds", ""):
                                        node.set("selected", "false")
                                next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
                            else:
                                for node in root.iter("node"):
                                    if node.get("selected") is not None and ",1730]" in node.get("bounds", ""):
                                        node.set("selected", "false")
                                next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
                            self.hierarchy = ElementTree.tostring(root, encoding="unicode")
                    return value
            with self.subTest(drift=drift), TemporaryDirectory() as temporary:
                target = Path(temporary) / "out"
                with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
                    self._runner(_FakeAdb(), DriftDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
                self.assertFalse(target.exists())
                self.assertFalse((target / "manifest.json").exists())
                self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])

        device = _RawDevice(); device.select_alternates()
        device.package = "other"
        with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).exit_sku_panel_safely()
        self.assertEqual(_actions(device, "pressKey"), [])

    def test_screenshot_then_foreground_drift_publishes_nothing_and_never_back(self) -> None:
        class ForegroundDriftDevice(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                value = super().jsonrpc_call(method, params, timeout)
                if method == "takeScreenshot": self.package = "other"
                return value

        device = ForegroundDriftDevice()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "out"
            with self.assertRaises(SkuSelectionError):
                self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
            self.assertFalse(target.exists())
            self.assertFalse((target / "manifest.json").exists())
            self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
        self.assertEqual(_actions(device, "pressKey"), [])

    def test_option_timeout_reconciliation_controls_back_once(self) -> None:
        class OptionTimeoutDevice(_RawDevice):
            def __init__(self, delivered: bool) -> None:
                super().__init__(); self.delivered = delivered; self.clicks = 0
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "click":
                    self.clicks += 1
                    if self.clicks == 2:
                        if self.delivered: super().jsonrpc_call(method, params, timeout)
                        else: self.calls.append(("jsonrpc", method, params, timeout))
                        raise TimeoutError("uncertain option")
                return super().jsonrpc_call(method, params, timeout)

        for delivered, expected_back in ((False, 0), (True, 1)):
            with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
                device = OptionTimeoutDevice(delivered); device.select_alternates()
                adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
                adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
                runner = SkuSelectionRunner(adb, lambda serial: device, .03)
                with self.assertRaises(SkuSelectionRunError):
                    runner.run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
                self.assertEqual(len(_actions(device, "click")), 2)
                self.assertEqual(len(_actions(device, "pressKey")), expected_back)

    def test_back_timeout_is_never_retried(self) -> None:
        class BackTimeoutDevice(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "pressKey":
                    super().jsonrpc_call(method, params, timeout)
                    raise TimeoutError("back uncertain")
                return super().jsonrpc_call(method, params, timeout)

        device = BackTimeoutDevice()
        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
        self.assertEqual(len(_actions(device, "pressKey")), 1)

    def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None:
        class DeliveredThenTimeout(_RawDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "click" and self.hierarchy != _FIXTURE.read_text(encoding="utf-8"):
                    super().jsonrpc_call(method, params, timeout)
                    raise TimeoutError("delivery uncertain")
                return super().jsonrpc_call(method, params, timeout)

        device = DeliveredThenTimeout()
        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
        self.assertEqual(len(_actions(device, "click")), 1)
        self.assertEqual(len(_actions(device, "pressKey")), 1)


class SkuSelectionCliTests(unittest.TestCase):
    def test_cli_accepts_only_target_url_and_task_values(self) -> None:
        script = _load_runner_script()
        valid = {
            "serial": "device-1",
            "url": _TARGET_URL,
            "color": _TASK_COLOR,
            "size": _TASK_SIZE,
            "output_dir": Path("evidence"),
            "timeout": 10.0,
            "adb": "adb",
        }
        script.validate_arguments(type("Arguments", (), valid)())
        for field, value in (("serial", ""), ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=1"), ("color", "黑色 CHA (纯棉）"), ("size", "M（建议100-115）"), ("timeout", 0), ("timeout", float("inf"))):
            with self.subTest(field=field, value=value), self.assertRaises((ValueError, SkuSelectionError)):
                script.validate_arguments(type("Arguments", (), valid | {field: value})())

    def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
        script = _load_runner_script()

        class FlowFailingRunner:
            def __init__(self, *args: object, **kwargs: object) -> None: pass
            def run(self, *args: object, **kwargs: object) -> object:
                raise SkuSelectionError("<hierarchy>page-body</hierarchy>")

        stderr = BytesIO()
        # TextIOWrapper keeps the assertion independent from host console encoding.
        import io
        text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
        with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
            status = script.main([
                "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
                "--size", _TASK_SIZE, "--output-dir", "evidence",
            ])
        text_stderr.flush()
        output = stderr.getvalue().decode("utf-8")
        self.assertEqual(status, 1)
        self.assertNotIn("Traceback", output)
        self.assertNotIn("page-body", output)


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


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