cmbuyer/client/tests/pdd/test_quantity_gate2.py

"""T-105 数量两态只读取证与后续生产边界的离线测试。"""

from __future__ import annotations

import argparse
import ast
import base64
from contextlib import redirect_stderr
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.quantity_gate2_spike import (
    Android16TopResumedForegroundReader,
    DECLARED_QUANTITIES,
    EXPECTED_GOODS_ID,
    QuantityGate2EvidenceCapturer,
    QuantityGate2EvidenceError,
    QuantityGate2EvidenceTimeoutError,
    QuantityGate2ForegroundReader,
    QuantityGate2ReadDevice,
)


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 is not None 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(f"unexpected RPC {method}")


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 is not None else self.package
        activity = self.post_activity if len(self.calls) > 1 and self.post_activity is not None 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:
    script_path = CLIENT_ROOT / "scripts" / "capture_quantity_gate2_spike.py"
    spec = spec_from_file_location("capture_quantity_gate2_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


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


class Android16TopResumedForegroundReaderTests(unittest.TestCase):
    def test_reads_exact_unique_android16_top_resumed_activity(self) -> None:
        runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED))
        reader = Android16TopResumedForegroundReader(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_or_duplicate_top_resumed_activity_is_rejected(self) -> None:
        rejected_outputs = (
            "",
            (
                "mResumedActivity: ActivityRecord{101034589 u0 "
                "com.xunmeng.pinduoduo/.activity.NewPageActivity t1816}\n"
            ),
            TOP_RESUMED + TOP_RESUMED,
            "topResumedActivity=null\n",
        )
        for output in rejected_outputs:
            with self.subTest(output=output):
                reader = Android16TopResumedForegroundReader(
                    FakeCommandRunner(CommandResult(stdout=output)),
                    timeout_seconds=7,
                )
                with self.assertRaises(QuantityGate2EvidenceError):
                    reader.read(SERIAL)

    def test_command_failure_and_invalid_serial_are_rejected(self) -> None:
        runner = FakeCommandRunner(CommandResult(stdout=TOP_RESUMED, returncode=1))
        reader = Android16TopResumedForegroundReader(runner, timeout_seconds=7)
        with self.assertRaises(QuantityGate2EvidenceError):
            reader.read(SERIAL)
        with self.assertRaises(QuantityGate2EvidenceError):
            reader.read(f" {SERIAL}")
        self.assertEqual(len(runner.calls), 1)


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

        return QuantityGate2EvidenceCapturer(
            adb,
            connect,
            foreground or FakeForegroundReader(),
            timeout_seconds=2,
        )

    def test_initial_and_target_states_publish_only_raw_read_evidence(self) -> None:
        for state, quantity in DECLARED_QUANTITIES.items():
            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,
                    quantity,
                    target,
                )

                manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
                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"], "t105-quantity-gate2-readonly-evidence")
                self.assertEqual(manifest["human_declared_state"], state)
                self.assertEqual(manifest["human_declared_quantity"], quantity)
                self.assertEqual(
                    manifest["human_declared_selection"],
                    {"color": "黑色CHA(纯棉)", "size": "M(建议100-115)"},
                )
                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"},
                )
                serialized = result.manifest_path.read_text(encoding="utf-8")
                self.assertNotIn(SERIAL, serialized)
                self.assertNotIn("NewPageActivity", serialized)
                self.assertNotIn("local raw page", serialized)
                for artifact in manifest["artifacts"]:
                    self.assertEqual(len(artifact["sha256"]), 64)

    def test_invalid_inputs_and_existing_target_fail_before_device_access(self) -> None:
        scenarios = (
            ("", EXPECTED_GOODS_ID, "initial", 1),
            (f" {SERIAL}", EXPECTED_GOODS_ID, "initial", 1),
            (SERIAL, "958756616606", "initial", 1),
            (SERIAL, EXPECTED_GOODS_ID, "unknown", 1),
            (SERIAL, EXPECTED_GOODS_ID, "initial", 2),
            (SERIAL, EXPECTED_GOODS_ID, "target", 1),
            (SERIAL, EXPECTED_GOODS_ID, "initial", True),
        )
        for serial, goods_id, state, quantity in scenarios:
            with self.subTest(state=state, quantity=quantity), TemporaryDirectory() as temporary:
                adb = FakeAdbClient()
                connector_calls: list[str] = []
                with self.assertRaises(QuantityGate2EvidenceError):
                    self._capturer(adb, FakeReadDevice(), connector_calls=connector_calls).capture(
                        serial,
                        goods_id,
                        state,
                        quantity,
                        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(QuantityGate2EvidenceError):
                self._capturer(adb, FakeReadDevice()).capture(
                    SERIAL, EXPECTED_GOODS_ID, "initial", 1, 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(QuantityGate2EvidenceError):
                    self._capturer(adb, device, foreground=foreground).capture(
                        SERIAL, EXPECTED_GOODS_ID, "initial", 1, 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_leaves_no_partial_evidence(self) -> None:
        scenarios = (
            FakeReadDevice(screenshot="not base64"),
            FakeReadDevice(screenshot=123),
            FakeReadDevice(hierarchy="<not-hierarchy />"),
            FakeReadDevice(hierarchy=123),
        )
        for device in scenarios:
            with self.subTest(value=device.screenshot), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(QuantityGate2EvidenceError):
                    self._capturer(FakeAdbClient(), device).capture(
                        SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
                    )
                self.assertFalse(target.exists())
                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_timeout_is_redacted_and_cleans_staging(self) -> None:
        class TimeoutDevice(FakeReadDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
                self.calls.append(("jsonrpc", method, params, timeout))
                if method == "takeScreenshot":
                    raise TimeoutError(f"secret {SERIAL} C:\\private\\evidence")
                return super().jsonrpc_call(method, params, timeout)

        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with self.assertRaises(QuantityGate2EvidenceTimeoutError) as raised:
                self._capturer(FakeAdbClient(), TimeoutDevice()).capture(
                    SERIAL, EXPECTED_GOODS_ID, "initial", 1, 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_app_or_screen_drift_does_not_publish(self) -> None:
        drift_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 drift_mutators:
            with TemporaryDirectory() as temporary:
                device = FakeReadDevice()
                foreground = FakeForegroundReader()
                mutate(device, foreground)
                target = Path(temporary) / "evidence"
                with self.assertRaises(QuantityGate2EvidenceError):
                    self._capturer(
                        FakeAdbClient(),
                        device,
                        foreground=foreground,
                    ).capture(
                        SERIAL, EXPECTED_GOODS_ID, "initial", 1, target
                    )
                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(QuantityGate2EvidenceError):
                capturer.capture("", EXPECTED_GOODS_ID, "initial", 1, Path(temporary) / "bad")
            with self.assertRaises(QuantityGate2EvidenceError):
                capturer.capture(SERIAL, EXPECTED_GOODS_ID, "initial", 1, Path(temporary) / "good")

    def test_read_protocol_and_capturer_expose_no_page_actions(self) -> None:
        forbidden = {
            "click",
            "swipe",
            "press",
            "pressKey",
            "send_keys",
            "set_text",
            "open_product",
            "open_sku_panel",
            "set_quantity",
            "go_to_order_confirm",
            "submit_order_once",
            "pay",
        }
        self.assertTrue(forbidden.isdisjoint(QuantityGate2ReadDevice.__dict__))
        self.assertTrue(forbidden.isdisjoint(QuantityGate2ForegroundReader.__dict__))
        self.assertEqual(
            {
                name
                for name in Android16TopResumedForegroundReader.__dict__
                if not name.startswith("_")
            },
            {"read"},
        )
        self.assertEqual(
            {name for name in QuantityGate2EvidenceCapturer.__dict__ if not name.startswith("_")},
            {"capture"},
        )


class QuantityGate2CliAndStaticBoundaryTests(unittest.TestCase):
    def test_cli_validation_rejects_all_unapproved_combinations(self) -> None:
        script = _load_script()
        invalid = (
            _namespace(serial=""),
            _namespace(serial=f" {SERIAL}"),
            _namespace(goods_id="958756616606"),
            _namespace(state="unknown"),
            _namespace(declared_quantity=2),
            _namespace(state="target", declared_quantity=1),
            _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_parses_exact_initial_and_target_arguments(self) -> None:
        script = _load_script()
        for state, quantity in DECLARED_QUANTITIES.items():
            arguments = script.parse_arguments(  # type: ignore[attr-defined]
                [
                    "--serial",
                    SERIAL,
                    "--goods-id",
                    EXPECTED_GOODS_ID,
                    "--state",
                    state,
                    "--declared-quantity",
                    str(quantity),
                    "--output-dir",
                    f"evidence-{state}",
                ]
            )
            script.validate_arguments(arguments)  # type: ignore[attr-defined]

    def test_cli_runtime_failure_does_not_echo_sensitive_values(self) -> None:
        script = _load_script()
        error = QuantityGate2EvidenceError(f"secret {SERIAL} C:\\private\\evidence raw body")
        stderr = StringIO()
        fake_capturer = unittest.mock.Mock()
        fake_capturer.capture.side_effect = error
        with (
            patch.object(script, "QuantityGate2EvidenceCapturer", return_value=fake_capturer),
            patch.dict(sys.modules, {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()}),
            redirect_stderr(stderr),
        ):
            result = script.main(
                [
                    "--serial",
                    SERIAL,
                    "--goods-id",
                    EXPECTED_GOODS_ID,
                    "--state",
                    "initial",
                    "--declared-quantity",
                    "1",
                    "--output-dir",
                    "C:\\private\\evidence",
                ]
            )
        self.assertEqual(result, 1)
        self.assertNotIn(SERIAL, stderr.getvalue())
        self.assertNotIn("private", stderr.getvalue())
        self.assertNotIn("raw body", stderr.getvalue())

    def test_stage_one_sources_contain_only_approved_rpc_method_literals(self) -> None:
        expected_by_source = {
            Path("src/cmbuyer_client/pdd/quantity_gate2_spike.py"): {
                "takeScreenshot",
                "dumpWindowHierarchy",
            },
            Path("scripts/capture_quantity_gate2_spike.py"): set(),
        }
        for relative, expected in expected_by_source.items():
            tree = ast.parse((CLIENT_ROOT / relative).read_text(encoding="utf-8"))
            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)
            forbidden_attributes = {
                node.attr
                for node in ast.walk(tree)
                if isinstance(node, ast.Attribute)
                and node.attr in {"click", "swipe", "press", "pressKey", "set_text", "send_keys", "start_pdd_view_intent"}
            }
            self.assertEqual(forbidden_attributes, set())

    def test_t103_and_t104_sources_do_not_import_quantity_gate2(self) -> None:
        forbidden_import = "quantity_gate2"
        sources = (
            CLIENT_ROOT / "src/cmbuyer_client/pdd/sku_selection.py",
            CLIENT_ROOT / "src/cmbuyer_client/pdd/sku_selection_runner.py",
        )
        for source in sources:
            self.assertNotIn(forbidden_import, source.read_text(encoding="utf-8"))


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