"""T-106 合并式最终提交面板与人工返回态纯只读取证测试。"""

from __future__ import annotations

import argparse
import ast
import base64
from contextlib import redirect_stderr, redirect_stdout
from functools import lru_cache
from importlib.util import module_from_spec, spec_from_file_location
from io import BytesIO, StringIO
import json
from pathlib import Path
import sys
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch

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, CommandResult, DeviceInspection
from cmbuyer_client.pdd.order_confirm_spike import (
    Android16ForegroundReader,
    DECLARED_STATES,
    EXPECTED_GOODS_ID,
    OrderConfirmEvidenceCapturer,
    OrderConfirmEvidenceError,
    OrderConfirmEvidenceTimeoutError,
    OrderConfirmForegroundReader,
    OrderConfirmReadDevice,
)


SERIAL = "192.168.0.173:5555"
HIERARCHY = (
    "<?xml version='1.0' encoding='UTF-8'?><hierarchy rotation='0'>"
    "<node text='local raw page' /></hierarchy>"
)
TOP_RESUMED = (
    "  topResumedActivity=ActivityRecord{101034589 u0 "
    "com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n"
)


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


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

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


class FakeReadDevice:
    def __init__(
        self,
        *,
        version: str = "8.17.0",
        screen_size: tuple[int, int] = (1080, 2376),
        screenshot: object | None = None,
        hierarchy: object = HIERARCHY,
    ) -> None:
        self.version = version
        self.screen_size = screen_size
        self.screenshot = _png_base64() if screenshot is None else screenshot
        self.hierarchy = hierarchy
        self.calls: list[tuple[object, ...]] = []
        self.app_info_reads = 0
        self.window_reads = 0
        self.post_version: str | None = None
        self.post_screen_size: tuple[int, int] | None = None

    def app_info(self, package_name: str) -> dict[str, str]:
        self.calls.append(("app_info", package_name))
        self.app_info_reads += 1
        version = self.post_version if self.app_info_reads > 1 and self.post_version else self.version
        return {"versionName": version}

    def window_size(self) -> tuple[int, int]:
        self.calls.append(("window_size",))
        self.window_reads += 1
        if self.window_reads > 1 and self.post_screen_size is not None:
            return self.post_screen_size
        return self.screen_size

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


class FakeForegroundReader:
    def __init__(
        self,
        *,
        package: str = "com.xunmeng.pinduoduo",
        activity: str = ".activity.NewPageActivity",
    ) -> None:
        self.package = package
        self.activity = activity
        self.post_package: str | None = None
        self.post_activity: str | None = None
        self.calls: list[str] = []

    def read(self, serial: str) -> dict[str, str]:
        self.calls.append(serial)
        package = self.post_package if len(self.calls) > 1 and self.post_package else self.package
        activity = self.post_activity if len(self.calls) > 1 and self.post_activity else self.activity
        return {"package": package, "activity": activity}


class FakeCommandRunner:
    def __init__(self, result: CommandResult) -> None:
        self.result = result
        self.calls: list[tuple[tuple[str, ...], float]] = []

    def run(self, arguments: tuple[str, ...], timeout_seconds: float) -> CommandResult:
        self.calls.append((arguments, timeout_seconds))
        return self.result


def _load_script() -> object:
    path = CLIENT_ROOT / "scripts" / "capture_order_confirm_spike.py"
    specification = spec_from_file_location("capture_order_confirm_spike_for_test", path)
    assert specification is not None and specification.loader is not None
    module = module_from_spec(specification)
    specification.loader.exec_module(module)
    return module


def _namespace(**changes: object) -> argparse.Namespace:
    values: dict[str, object] = {
        "serial": SERIAL,
        "goods_id": EXPECTED_GOODS_ID,
        "state": DECLARED_STATES[0],
        "output_dir": Path("evidence"),
        "timeout": 10.0,
        "adb": "adb",
    }
    values.update(changes)
    return argparse.Namespace(**values)


class Android16ForegroundReaderTests(unittest.TestCase):
    def test_reads_exact_unique_top_resumed_activity(self) -> None:
        runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED))
        reader = Android16ForegroundReader(runner, timeout_seconds=7)

        self.assertEqual(
            reader.read(SERIAL),
            {
                "package": "com.xunmeng.pinduoduo",
                "activity": ".activity.NewPageActivity",
            },
        )
        self.assertEqual(
            runner.calls,
            [
                (
                    ("-s", SERIAL, "shell", "dumpsys", "activity", "activities"),
                    7,
                )
            ],
        )

    def test_missing_legacy_duplicate_or_failed_summary_is_rejected(self) -> None:
        outputs = (
            "",
            "mResumedActivity: ActivityRecord{1 u0 com.xunmeng.pinduoduo/.Main t1}\n",
            TOP_RESUMED + TOP_RESUMED,
            "topResumedActivity=null\n",
        )
        for output in outputs:
            with self.subTest(output=output):
                reader = Android16ForegroundReader(
                    FakeCommandRunner(CommandResult(stdout=output)),
                    timeout_seconds=7,
                )
                with self.assertRaises(OrderConfirmEvidenceError):
                    reader.read(SERIAL)

        reader = Android16ForegroundReader(
            FakeCommandRunner(CommandResult(stdout=TOP_RESUMED, returncode=1)),
            timeout_seconds=7,
        )
        with self.assertRaises(OrderConfirmEvidenceError):
            reader.read(SERIAL)
        with self.assertRaises(OrderConfirmEvidenceError):
            reader.read(f" {SERIAL}")


class OrderConfirmEvidenceTests(unittest.TestCase):
    def _capturer(
        self,
        adb: FakeAdbClient,
        device: FakeReadDevice,
        *,
        foreground: FakeForegroundReader | None = None,
        connector_calls: list[str] | None = None,
    ) -> OrderConfirmEvidenceCapturer:
        def connect(serial: str) -> FakeReadDevice:
            if connector_calls is not None:
                connector_calls.append(serial)
            return device

        return OrderConfirmEvidenceCapturer(
            adb,  # type: ignore[arg-type]
            connect,
            foreground or FakeForegroundReader(),
            timeout_seconds=2,
        )

    def test_only_approved_human_states_publish_raw_read_evidence(self) -> None:
        self.assertEqual(DECLARED_STATES, ("gate2-navigation-source", "returned-safe-page"))
        for state in DECLARED_STATES:
            with self.subTest(state=state), TemporaryDirectory() as temporary:
                adb = FakeAdbClient()
                device = FakeReadDevice()
                foreground = FakeForegroundReader()
                connector_calls: list[str] = []
                target = Path(temporary) / state

                result = self._capturer(
                    adb,
                    device,
                    foreground=foreground,
                    connector_calls=connector_calls,
                ).capture(SERIAL, EXPECTED_GOODS_ID, state, target)

                manifest_text = result.manifest_path.read_text(encoding="utf-8")
                manifest = json.loads(manifest_text)
                app = json.loads(result.app_path.read_text(encoding="utf-8"))
                self.assertEqual(adb.calls, [SERIAL])
                self.assertEqual(connector_calls, [SERIAL])
                self.assertEqual(foreground.calls, [SERIAL, SERIAL])
                self.assertEqual(
                    [call[1] for call in device.calls if call[0] == "jsonrpc"],
                    ["takeScreenshot", "dumpWindowHierarchy"],
                )
                self.assertEqual(manifest["operation"], "t106-order-confirm-readonly-evidence")
                self.assertEqual(manifest["human_declared_state"], state)
                self.assertEqual(manifest["review_status"], "human_review_required")
                self.assertEqual(manifest["product"]["goods_id"], EXPECTED_GOODS_ID)
                self.assertEqual(app["package"], "com.xunmeng.pinduoduo")
                self.assertEqual(
                    {path.name for path in target.iterdir()},
                    {"screenshot.png", "hierarchy.xml", "app.json", "manifest.json"},
                )
                self.assertNotIn(SERIAL, manifest_text)
                self.assertNotIn("NewPageActivity", manifest_text)
                self.assertNotIn("local raw page", manifest_text)
                for artifact in manifest["artifacts"]:
                    self.assertEqual(len(artifact["sha256"]), 64)

    def test_invalid_inputs_and_existing_target_stop_before_device_access(self) -> None:
        scenarios = (
            ("", EXPECTED_GOODS_ID, DECLARED_STATES[0]),
            (f" {SERIAL}", EXPECTED_GOODS_ID, DECLARED_STATES[0]),
            (SERIAL, "958756616606", DECLARED_STATES[0]),
            (SERIAL, EXPECTED_GOODS_ID, "confirm-gate3"),
            (SERIAL, EXPECTED_GOODS_ID, "submit-control-visible"),
            (SERIAL, EXPECTED_GOODS_ID, "unknown"),
        )
        for serial, goods_id, state in scenarios:
            with self.subTest(state=state), TemporaryDirectory() as temporary:
                adb = FakeAdbClient()
                connector_calls: list[str] = []
                with self.assertRaises(OrderConfirmEvidenceError):
                    self._capturer(
                        adb,
                        FakeReadDevice(),
                        connector_calls=connector_calls,
                    ).capture(serial, goods_id, state, Path(temporary) / "evidence")
                self.assertEqual(adb.calls, [])
                self.assertEqual(connector_calls, [])

        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            target.mkdir()
            sentinel = target / "sentinel.txt"
            sentinel.write_text("keep", encoding="utf-8")
            adb = FakeAdbClient()
            with self.assertRaises(OrderConfirmEvidenceError):
                self._capturer(adb, FakeReadDevice()).capture(
                    SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
                )
            self.assertEqual(adb.calls, [])
            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")

    def test_device_version_package_activity_and_screen_mismatch_fail_closed(self) -> None:
        scenarios = (
            (FakeAdbClient(model="OTHER"), FakeReadDevice(), FakeForegroundReader()),
            (FakeAdbClient(android_version="15"), FakeReadDevice(), FakeForegroundReader()),
            (FakeAdbClient(), FakeReadDevice(version="8.17.1"), FakeForegroundReader()),
            (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(package="com.example.other")),
            (FakeAdbClient(), FakeReadDevice(), FakeForegroundReader(activity="")),
            (FakeAdbClient(), FakeReadDevice(screen_size=(1080, 2400)), FakeForegroundReader()),
        )
        for adb, device, foreground in scenarios:
            with self.subTest(device=device.__dict__), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(OrderConfirmEvidenceError):
                    self._capturer(adb, device, foreground=foreground).capture(
                        SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
                    )
                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
                self.assertEqual([call for call in device.calls if call[0] == "jsonrpc"], [])

    def test_invalid_screenshot_or_hierarchy_never_publishes(self) -> None:
        devices = (
            FakeReadDevice(screenshot="not base64"),
            FakeReadDevice(screenshot=123),
            FakeReadDevice(hierarchy="<not-hierarchy />"),
            FakeReadDevice(hierarchy=123),
        )
        for device in devices:
            with self.subTest(value=device.screenshot), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(OrderConfirmEvidenceError):
                    self._capturer(FakeAdbClient(), device).capture(
                        SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
                    )
                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_timeout_and_unexpected_failure_are_redacted(self) -> None:
        class BrokenDevice(FakeReadDevice):
            def __init__(self, failure: BaseException) -> None:
                super().__init__()
                self.failure = failure

            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
                raise self.failure

        failures = (
            (TimeoutError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceTimeoutError),
            (RuntimeError(f"secret {SERIAL} C:\\private\\raw.xml"), OrderConfirmEvidenceError),
        )
        for failure, expected in failures:
            with self.subTest(expected=expected), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(expected) as raised:
                    self._capturer(FakeAdbClient(), BrokenDevice(failure)).capture(
                        SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
                    )
                self.assertNotIn(SERIAL, str(raised.exception))
                self.assertNotIn("private", str(raised.exception))
                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_post_capture_identity_drift_never_publishes(self) -> None:
        mutators = (
            lambda device, foreground: setattr(device, "post_version", "8.17.1"),
            lambda device, foreground: setattr(foreground, "post_package", "com.example.other"),
            lambda device, foreground: setattr(foreground, "post_activity", "OtherActivity"),
            lambda device, foreground: setattr(device, "post_screen_size", (1080, 2400)),
        )
        for mutate in mutators:
            with TemporaryDirectory() as temporary:
                device = FakeReadDevice()
                foreground = FakeForegroundReader()
                mutate(device, foreground)
                target = Path(temporary) / "evidence"
                with self.assertRaises(OrderConfirmEvidenceError):
                    self._capturer(
                        FakeAdbClient(), device, foreground=foreground
                    ).capture(SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target)
                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_atomic_publish_failure_cleans_staging(self) -> None:
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with (
                patch(
                    "cmbuyer_client.pdd.order_confirm_spike.os.rename",
                    side_effect=OSError(f"private {SERIAL}"),
                ),
                self.assertRaises(OrderConfirmEvidenceError) as raised,
            ):
                self._capturer(FakeAdbClient(), FakeReadDevice()).capture(
                    SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], target
                )
            self.assertNotIn(SERIAL, str(raised.exception))
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_capturer_is_one_shot_even_after_rejection(self) -> None:
        capturer = self._capturer(FakeAdbClient(), FakeReadDevice())
        with TemporaryDirectory() as temporary:
            with self.assertRaises(OrderConfirmEvidenceError):
                capturer.capture("", EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "bad")
            with self.assertRaises(OrderConfirmEvidenceError):
                capturer.capture(
                    SERIAL, EXPECTED_GOODS_ID, DECLARED_STATES[0], Path(temporary) / "good"
                )

    def test_public_boundaries_expose_reading_only(self) -> None:
        forbidden = {
            "click",
            "swipe",
            "press",
            "pressKey",
            "send_keys",
            "set_text",
            "start_pdd_view_intent",
            "open_product",
            "open_sku_panel",
            "set_quantity",
            "go_to_order_confirm",
            "submit_order_once",
            "pay",
        }
        self.assertTrue(forbidden.isdisjoint(OrderConfirmReadDevice.__dict__))
        self.assertTrue(forbidden.isdisjoint(OrderConfirmForegroundReader.__dict__))
        self.assertEqual(
            {name for name in Android16ForegroundReader.__dict__ if not name.startswith("_")},
            {"read"},
        )
        self.assertEqual(
            {name for name in OrderConfirmEvidenceCapturer.__dict__ if not name.startswith("_")},
            {"capture"},
        )


class OrderConfirmCliAndStaticBoundaryTests(unittest.TestCase):
    def test_cli_parses_each_approved_state(self) -> None:
        script = _load_script()
        for state in DECLARED_STATES:
            arguments = script.parse_arguments(  # type: ignore[attr-defined]
                [
                    "--serial",
                    SERIAL,
                    "--goods-id",
                    EXPECTED_GOODS_ID,
                    "--state",
                    state,
                    "--output-dir",
                    f"evidence-{state}",
                ]
            )
            script.validate_arguments(arguments)  # type: ignore[attr-defined]

    def test_cli_validation_rejects_unapproved_values(self) -> None:
        script = _load_script()
        invalid = (
            _namespace(serial=""),
            _namespace(serial=f" {SERIAL}"),
            _namespace(goods_id="958756616606"),
            _namespace(state="confirm-gate3"),
            _namespace(state="submit-control-visible"),
            _namespace(state="unknown"),
            _namespace(output_dir=Path("")),
            _namespace(timeout=0),
            _namespace(timeout=float("inf")),
        )
        for arguments in invalid:
            with self.subTest(arguments=arguments), self.assertRaises(ValueError):
                script.validate_arguments(arguments)  # type: ignore[attr-defined]

    def test_cli_runtime_failure_never_echoes_sensitive_values(self) -> None:
        script = _load_script()
        fake_capturer = unittest.mock.Mock()
        fake_capturer.capture.side_effect = OrderConfirmEvidenceError(
            f"secret {SERIAL} C:\\private\\raw.xml body"
        )
        stderr = StringIO()
        with (
            patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer),
            patch.dict(
                sys.modules,
                {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()},
            ),
            redirect_stderr(stderr),
        ):
            result = script.main(  # type: ignore[attr-defined]
                [
                    "--serial",
                    SERIAL,
                    "--goods-id",
                    EXPECTED_GOODS_ID,
                    "--state",
                    DECLARED_STATES[0],
                    "--output-dir",
                    "C:\\private\\evidence",
                ]
            )
        self.assertEqual(result, 1)
        self.assertNotIn(SERIAL, stderr.getvalue())
        self.assertNotIn("private", stderr.getvalue())
        self.assertNotIn("body", stderr.getvalue())

    def test_cli_success_does_not_echo_local_path_or_device(self) -> None:
        script = _load_script()
        fake_capturer = unittest.mock.Mock()
        stdout = StringIO()
        with (
            patch.object(script, "OrderConfirmEvidenceCapturer", return_value=fake_capturer),
            patch.dict(
                sys.modules,
                {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()},
            ),
            redirect_stdout(stdout),
        ):
            result = script.main(  # type: ignore[attr-defined]
                [
                    "--serial",
                    SERIAL,
                    "--goods-id",
                    EXPECTED_GOODS_ID,
                    "--state",
                    DECLARED_STATES[0],
                    "--output-dir",
                    "C:\\private\\evidence",
                ]
            )
        self.assertEqual(result, 0)
        self.assertNotIn(SERIAL, stdout.getvalue())
        self.assertNotIn("private", stdout.getvalue())
        fake_capturer.capture.assert_called_once()

    def test_sources_have_only_approved_read_rpc_literals_and_no_ui_mutators(self) -> None:
        expected_by_source = {
            Path("src/cmbuyer_client/pdd/order_confirm_spike.py"): {
                "takeScreenshot",
                "dumpWindowHierarchy",
            },
            Path("scripts/capture_order_confirm_spike.py"): set(),
        }
        forbidden_attributes = {
            "click",
            "swipe",
            "press",
            "pressKey",
            "send_keys",
            "set_text",
            "start_pdd_view_intent",
            "open_product",
            "open_sku_panel",
            "set_quantity_and_readback",
            "go_to_order_confirm",
            "submit_order_once",
        }
        forbidden_import_fragments = {
            "quantity_gate2_runner",
            "sku_selection_runner",
            "submission",
            "payment",
        }
        for relative, expected_rpc in expected_by_source.items():
            source = (CLIENT_ROOT / relative).read_text(encoding="utf-8")
            tree = ast.parse(source)
            rpc_literals = {
                node.args[1].value
                for node in ast.walk(tree)
                if isinstance(node, ast.Call)
                and isinstance(node.func, ast.Name)
                and node.func.id == "_read_rpc"
                and len(node.args) > 1
                and isinstance(node.args[1], ast.Constant)
                and isinstance(node.args[1].value, str)
            }
            self.assertEqual(rpc_literals, expected_rpc)
            observed_attributes = {
                node.attr
                for node in ast.walk(tree)
                if isinstance(node, ast.Attribute) and node.attr in forbidden_attributes
            }
            self.assertEqual(observed_attributes, set())
            imported_modules = {
                alias.name
                for node in ast.walk(tree)
                if isinstance(node, (ast.Import, ast.ImportFrom))
                for alias in node.names
            }
            for fragment in forbidden_import_fragments:
                self.assertTrue(all(fragment not in module for module in imported_modules))


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