feat(client): add guarded product link capture
This commit is contained in:
QiuSW
@@ -0,0 +1,186 @@
|
||||
"""商品打开围栏的离线测试；所有设备和命令均为 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",
|
||||
hierarchy: str = HIERARCHY,
|
||||
timeout_on_screenshot: bool = False,
|
||||
) -> None:
|
||||
self.version = version
|
||||
self.current_package = current_package
|
||||
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")
|
||||
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) -> ProductOpenCapturer:
|
||||
return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2)
|
||||
|
||||
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")
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(ProductPackageMismatchError):
|
||||
self._capturer(adb, device).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
|
||||
self.assertEqual(device.calls, ["app_info", "app_current"])
|
||||
self.assertFalse(target.exists())
|
||||
|
||||
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")
|
||||
Reference in New Issue
Block a user