cmbuyer/client/tests/pdd/test_product_open.py

"""商品打开围栏的离线测试；所有设备和命令均为 fake。"""

from __future__ import annotations

import base64
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, IntentLaunchSummary
from cmbuyer_client.pdd.product_open import (
    ProductOpenCapturer,
    ProductOpenTimeoutError,
    ProductOpenUiDevice,
    ProductHierarchyCaptureError,
    ProductPackageMismatchError,
    ProductScreenshotCaptureError,
    ProductVersionMismatchError,
)
from cmbuyer_client.pdd.product_url import ProductUrl, ProductUrlError


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 rotation='0'><node /></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[tuple[str, str | None]] = []
        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(("inspect", serial))
        return self.inspection

    def start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary:
        self.calls.append(("intent", goods_id))
        return IntentLaunchSummary(status="ok", returncode=0)


class FakeUiDevice:
    def __init__(
        self,
        *,
        version: str = "8.17.0",
        current_package: str = "com.xunmeng.pinduoduo",
        current_packages: list[str] | None = None,
        hierarchy: str = HIERARCHY,
        timeout_on_screenshot: bool = False,
    ) -> None:
        self.version = version
        self.current_package = current_package
        self.current_packages = list(current_packages) if current_packages is not None else None
        self.hierarchy = hierarchy
        self.timeout_on_screenshot = timeout_on_screenshot
        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")
        if self.current_packages:
            package = self.current_packages.pop(0)
            self.current_package = package
        return {"package": self.current_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":
            if self.timeout_on_screenshot:
                raise TimeoutError("raw remote detail")
            return _png_base64()
        if method == "dumpWindowHierarchy":
            return self.hierarchy
        raise AssertionError(f"unexpected RPC {method}")


class ProductOpenTests(unittest.TestCase):
    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice, **kwargs: object) -> ProductOpenCapturer:
        return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2, **kwargs)

    def test_success_uses_canonical_url_and_redacted_atomic_manifest(self) -> None:
        adb = FakeAdbClient()
        device = FakeUiDevice()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            result = self._capturer(adb, device).open_and_capture(SERIAL, URL, target)
            manifest = result.manifest_path.read_text(encoding="utf-8")

            self.assertTrue(result.screenshot_path.exists())
            self.assertTrue(result.hierarchy_path.exists())
            self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
            self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
            self.assertIn('"goods_id": "123"', manifest)
            self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
            self.assertNotIn(SERIAL, manifest)
            self.assertNotIn("sensitive.activity.name", manifest)
            self.assertNotIn(HIERARCHY, manifest)

    def test_version_mismatch_halts_before_intent(self) -> None:
        adb = FakeAdbClient()
        for version in ("8.17.1", " 8.17.0 "):
            with self.subTest(version=version), TemporaryDirectory() as temporary:
                target = Path(temporary) / "evidence"
                with self.assertRaises(ProductVersionMismatchError):
                    self._capturer(adb, FakeUiDevice(version=version)).open_and_capture(SERIAL, URL, target)

                self.assertEqual(adb.calls[-1:], [("inspect", SERIAL)])
                self.assertFalse(target.exists())

    def test_public_entry_rejects_caller_constructed_url_value_object(self) -> None:
        adb = FakeAdbClient()
        with TemporaryDirectory() as temporary:
            with self.assertRaises(ProductUrlError):
                self._capturer(adb, FakeUiDevice()).open_and_capture(
                    SERIAL,
                    ProductUrl(goods_id="123", canonical_url="https://example.invalid/"),  # type: ignore[arg-type]
                    Path(temporary) / "evidence",
                )

            self.assertEqual(adb.calls, [])

    def test_foreground_package_mismatch_halts_before_capture(self) -> None:
        adb = FakeAdbClient()
        device = FakeUiDevice(current_package="com.example.other")
        clock = FakeClock()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with self.assertRaises(ProductPackageMismatchError):
                self._capturer(
                    adb,
                    device,
                    foreground_poll_interval_seconds=0.5,
                    monotonic_clock=clock.monotonic,
                    sleep_function=clock.sleep,
                ).open_and_capture(SERIAL, URL, target)

            self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current", "app_current"])
            self.assertEqual(clock.sleeps, [0.5, 0.5, 0.5, 0.5])
            self.assertFalse(target.exists())

    def test_foreground_package_poll_waits_for_pdd_before_reading_evidence(self) -> None:
        adb = FakeAdbClient()
        device = FakeUiDevice(current_packages=["com.example.other", "com.xunmeng.pinduoduo"])
        clock = FakeClock()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            result = self._capturer(
                adb,
                device,
                foreground_poll_interval_seconds=0.25,
                monotonic_clock=clock.monotonic,
                sleep_function=clock.sleep,
            ).open_and_capture(SERIAL, URL, target)

            self.assertTrue(result.manifest_path.exists())
            self.assertEqual(clock.sleeps, [0.25])
            self.assertEqual(
                device.calls,
                ["app_info", "app_current", "app_current", "takeScreenshot", "dumpWindowHierarchy"],
            )

    def test_foreground_package_poll_stops_at_deadline_without_evidence(self) -> None:
        adb = FakeAdbClient()
        device = FakeUiDevice(current_packages=["com.example.other", "", "com.example.other"])
        clock = FakeClock()
        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with self.assertRaises(ProductPackageMismatchError):
                self._capturer(
                    adb,
                    device,
                    foreground_poll_interval_seconds=0.8,
                    monotonic_clock=clock.monotonic,
                    sleep_function=clock.sleep,
                ).open_and_capture(SERIAL, URL, target)

            self.assertEqual(len(clock.sleeps), 3)
            for actual, expected in zip(clock.sleeps, (0.8, 0.8, 0.4), strict=True):
                self.assertAlmostEqual(actual, expected)
            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current"])
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])

    def test_foreground_poll_interval_must_be_positive_and_finite(self) -> None:
        for interval in (0, -0.1, float("inf"), float("nan"), True):
            with self.subTest(interval=interval):
                with self.assertRaises(ValueError):
                    ProductOpenCapturer(
                        FakeAdbClient(),
                        lambda serial: FakeUiDevice(),
                        timeout_seconds=2,
                        foreground_poll_interval_seconds=interval,  # type: ignore[arg-type]
                    )

    def test_timeout_and_invalid_hierarchy_leave_no_partial_evidence(self) -> None:
        scenarios = (
            (FakeUiDevice(timeout_on_screenshot=True), ProductOpenTimeoutError),
            (FakeUiDevice(hierarchy="<not-hierarchy />"), ProductHierarchyCaptureError),
        )
        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).open_and_capture(SERIAL, URL, target)

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

    def test_invalid_screenshot_is_a_distinct_redacted_failure(self) -> None:
        class InvalidScreenshotDevice(FakeUiDevice):
            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                if method == "takeScreenshot":
                    self.calls.append(method)
                    return "not valid base64!"
                return super().jsonrpc_call(method, params, timeout)

        with TemporaryDirectory() as temporary:
            target = Path(temporary) / "evidence"
            with self.assertRaises(ProductScreenshotCaptureError) as raised:
                self._capturer(FakeAdbClient(), InvalidScreenshotDevice()).open_and_capture(SERIAL, URL, target)

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

    def test_read_only_protocol_has_no_ui_operation_methods(self) -> None:
        forbidden = {"click", "swipe", "send_keys", "set_text", "press", "long_click"}

        self.assertTrue(forbidden.isdisjoint(ProductOpenUiDevice.__dict__))
        self.assertEqual(base64.b64decode(_png_base64())[:8], b"\x89PNG\r\n\x1a\n")


class FakeClock:
    def __init__(self) -> None:
        self.value = 0.0
        self.sleeps: list[float] = []

    def monotonic(self) -> float:
        return self.value

    def sleep(self, seconds: float) -> None:
        self.sleeps.append(seconds)
        self.value += seconds