"""人工声明规格面板状态的离线只读取证测试。"""

from __future__ import annotations

import argparse
import base64
from importlib.util import module_from_spec, spec_from_file_location
from io import BytesIO
from pathlib import Path
import sys
from tempfile import TemporaryDirectory
import unittest

from PIL import Image


CLIENT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(CLIENT_ROOT / "src"))

from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
from cmbuyer_client.pdd.product_url import ProductUrlError
from cmbuyer_client.pdd.sku_panel_spike import (
    HUMAN_DECLARED_STATES,
    SkuPanelDeclaredStateError,
    SkuPanelEvidenceCapturer,
    SkuPanelEvidenceError,
    SkuPanelEvidenceTimeoutError,
    SkuPanelHierarchyError,
    SkuPanelPackageMismatchError,
    SkuPanelScreenshotError,
    SkuPanelUiDevice,
    SkuPanelVersionMismatchError,
)


SERIAL = "192.168.0.173:5555"
URL = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
HIERARCHY = "<?xml version='1.0' encoding='UTF-8'?><hierarchy><node text='sensitive page text' /></hierarchy>"


def _png_base64() -> str:
    image_data = BytesIO()
    Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG")
    return base64.b64encode(image_data.getvalue()).decode("ascii")


class FakeAdbClient:
    def __init__(self) -> None:
        self.calls: list[str] = []
        self.inspection = DeviceInspection(
            device=AdbDevice(serial=SERIAL, state="device", model="PKG110"),
            model="PKG110",
            android_version="16",
        )

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


class FakeUiDevice:
    def __init__(
        self,
        *,
        version: str = "8.17.0",
        package: str = "com.xunmeng.pinduoduo",
        screenshot: str | None = None,
        hierarchy: str = HIERARCHY,
    ) -> None:
        self.version = version
        self.package = package
        self.screenshot = screenshot if screenshot is not None else _png_base64()
        self.hierarchy = hierarchy
        self.calls: list[str] = []

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

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

    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
        self.calls.append(method)
        if method == "takeScreenshot":
            return self.screenshot
        if method == "dumpWindowHierarchy":
            return self.hierarchy
        raise AssertionError(f"unexpected RPC {method}")


def _load_spike_script() -> object:
    script_path = CLIENT_ROOT / "scripts" / "capture_sku_panel_spike.py"
    spec = spec_from_file_location("capture_sku_panel_spike_for_test", script_path)
    assert spec is not None and spec.loader is not None
    module = module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class SkuPanelEvidenceTests(unittest.TestCase):
    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice) -> SkuPanelEvidenceCapturer:
        return SkuPanelEvidenceCapturer(adb, lambda serial: device, timeout_seconds=2)

    def test_all_human_declared_states_publish_redacted_manifest(self) -> None:
        for state in sorted(HUMAN_DECLARED_STATES):
            with self.subTest(state=state), TemporaryDirectory() as temporary:
                adb = FakeAdbClient()
                device = FakeUiDevice()
                target = Path(temporary) / "evidence"
                result = self._capturer(adb, device).capture(SERIAL, URL, state, target)
                manifest = result.manifest_path.read_text(encoding="utf-8")

                self.assertEqual(adb.calls, [SERIAL])
                self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
                self.assertIn(f'"human_declared_state": "{state}"', manifest)
                self.assertIn('"goods_id": "123"', manifest)
                self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
                self.assertNotIn("detected_state", manifest)
                self.assertNotIn(SERIAL, manifest)
                self.assertNotIn("sensitive.activity.name", manifest)
                self.assertNotIn(HIERARCHY, manifest)

    def test_invalid_state_and_url_fail_before_device_access(self) -> None:
        adb = FakeAdbClient()
        rejected_states = ("initial", "one-dimension-selected", "all-dimensions-selected", "guessed")
        with TemporaryDirectory() as temporary:
            for state in rejected_states:
                with self.subTest(state=state), self.assertRaises(SkuPanelDeclaredStateError):
                    self._capturer(adb, FakeUiDevice()).capture(SERIAL, URL, state, Path(temporary) / "state")
            with self.assertRaises(ProductUrlError):
                self._capturer(adb, FakeUiDevice()).capture(
                    SERIAL,
                    "https://mobile.yangkeduo.com/goods.html?goods_id=12x",
                    "panel-opened-target-preselected",
                    Path(temporary) / "url",
                )

        self.assertEqual(adb.calls, [])

    def test_version_or_foreground_package_mismatch_stops_before_artifacts(self) -> None:
        scenarios = (
            (FakeUiDevice(version="8.17.1"), SkuPanelVersionMismatchError, ["app_info"]),
            (FakeUiDevice(package="com.example.other"), SkuPanelPackageMismatchError, ["app_info", "app_current"]),
        )
        for device, error_type, expected_calls in scenarios:
            with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(error_type):
                    self._capturer(FakeAdbClient(), device).capture(SERIAL, URL, "panel-opened-target-preselected", target)

                self.assertEqual(device.calls, expected_calls)
                self.assertFalse(target.exists())

    def test_screenshot_and_xml_failure_leave_no_partial_evidence(self) -> None:
        scenarios = (
            (FakeUiDevice(screenshot="not valid base64!"), SkuPanelScreenshotError),
            (FakeUiDevice(hierarchy="<not-hierarchy />"), SkuPanelHierarchyError),
        )
        for device, error_type in scenarios:
            with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(error_type):
                    self._capturer(FakeAdbClient(), device).capture(SERIAL, URL, "panel-opened-target-preselected", target)

                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_screenshot_timeout_is_redacted_and_leaves_no_partial_evidence(self) -> None:
        class TimeoutScreenshotDevice(FakeUiDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "takeScreenshot":
                    self.calls.append(method)
                    raise TimeoutError("adb 192.168.0.173:5555 raw detail")
                return super().jsonrpc_call(method, params, timeout)

        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with self.assertRaises(SkuPanelEvidenceTimeoutError) as raised:
                self._capturer(FakeAdbClient(), TimeoutScreenshotDevice()).capture(SERIAL, URL, "panel-opened-target-preselected", target)

            self.assertNotIn(SERIAL, str(raised.exception))
            self.assertNotIn("adb", str(raised.exception).lower())
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_existing_output_directory_is_not_overwritten_or_connected(self) -> None:
        adb = FakeAdbClient()
        device = FakeUiDevice()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            target.mkdir()
            sentinel = target / "sentinel.txt"
            sentinel.write_text("keep", encoding="utf-8")

            with self.assertRaises(SkuPanelEvidenceError):
                self._capturer(adb, device).capture(SERIAL, URL, "panel-opened-target-preselected", target)

            self.assertEqual(adb.calls, [])
            self.assertEqual(device.calls, [])
            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")

    def test_protocol_has_no_ui_or_purchase_operation_methods(self) -> None:
        forbidden = {
            "click",
            "swipe",
            "send_keys",
            "set_text",
            "press",
            "open_product",
            "open_sku_panel",
            "set_quantity",
            "go_to_order_confirm",
            "submit_order",
            "pay",
        }
        self.assertTrue(forbidden.isdisjoint(SkuPanelUiDevice.__dict__))
        self.assertEqual({name for name in SkuPanelEvidenceCapturer.__dict__ if not name.startswith("_")}, {"capture"})


class SkuPanelSpikeCliTests(unittest.TestCase):
    def test_validate_arguments_rejects_invalid_serial_timeout_state_and_url(self) -> None:
        script = _load_spike_script()
        valid = {
            "serial": SERIAL,
            "url": URL,
            "goods_id": None,
            "state": "panel-opened-target-preselected",
            "output_dir": Path("evidence"),
            "timeout": 10.0,
            "adb": "adb",
        }
        invalid_values = (
            ("serial", ""),
            ("timeout", 0),
            ("timeout", float("inf")),
            ("state", "not-declared"),
            ("state", "initial"),
            ("state", "one-dimension-selected"),
            ("state", "all-dimensions-selected"),
            ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=bad"),
        )
        for field, value in invalid_values:
            with self.subTest(field=field, value=value):
                arguments = argparse.Namespace(**(valid | {field: value}))
                with self.assertRaises((ValueError, ProductUrlError)):
                    script.validate_arguments(arguments)  # type: ignore[attr-defined]

    def test_goods_id_is_rebuilt_as_canonical_url(self) -> None:
        script = _load_spike_script()
        arguments = argparse.Namespace(
            serial=SERIAL,
            url=None,
            goods_id="00123",
            state="target-selection-restored",
            output_dir=Path("evidence"),
            timeout=10.0,
            adb="adb",
        )

        link = script.validate_arguments(arguments)  # type: ignore[attr-defined]
        self.assertEqual(link.canonical_url, "https://mobile.yangkeduo.com/goods.html?goods_id=00123")