397 lines
15 KiB
Python
397 lines
15 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
from contextlib import redirect_stderr
|
||
from io import BytesIO
|
||
import importlib.util
|
||
import io
|
||
import json
|
||
from pathlib import Path
|
||
from tempfile import TemporaryDirectory
|
||
import unittest
|
||
from unittest.mock import patch
|
||
from xml.etree import ElementTree
|
||
|
||
from PIL import Image
|
||
|
||
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
|
||
from cmbuyer_client.pdd.sku_reveal_spike import (
|
||
SkuRevealSpikeCapturer,
|
||
SkuRevealSpikeError,
|
||
_require_safe_reveal_path,
|
||
)
|
||
from cmbuyer_client.pdd.sku_selection import SkuSelectionError, _parse_nodes
|
||
|
||
|
||
_FIXTURES = Path(__file__).with_name("fixtures")
|
||
_PRODUCT = (_FIXTURES / "product_entry_8_17_0.xml").read_text(encoding="utf-8")
|
||
_EMPTY = (_FIXTURES / "sku_panel_empty_8_17_0.xml").read_text(encoding="utf-8")
|
||
_COLOR_ONLY = (_FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml").read_text(encoding="utf-8")
|
||
_M_SELECTED = (_FIXTURES / "sku_panel_size_m_restored_8_17_0.xml").read_text(encoding="utf-8")
|
||
|
||
|
||
def _candidate_unselected() -> str:
|
||
root = ElementTree.fromstring(_M_SELECTED)
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
for node in root.iter("node"):
|
||
if node.get("text") == "已选： 黑色 CHA (纯棉） M（建议100-115）":
|
||
node.set("text", "请选择： 尺码")
|
||
if node.get("text") in {"S（建议80-100）", "M（建议100-115）"}:
|
||
node.set("selected", "false")
|
||
wrapper = parents[node]
|
||
for child in list(wrapper):
|
||
if child.get("class") == "android.view.View":
|
||
wrapper.remove(child)
|
||
action_root = next(
|
||
node
|
||
for node in root.iter("node")
|
||
if node.get("class") == "android.view.ViewGroup"
|
||
and node.get("clickable") == "true"
|
||
and node.get("bounds") == "[0,366][1080,2328]"
|
||
)
|
||
button_frame = ElementTree.SubElement(
|
||
action_root,
|
||
"node",
|
||
{
|
||
"class": "android.widget.FrameLayout",
|
||
"package": "com.xunmeng.pinduoduo",
|
||
"text": "",
|
||
"content-desc": "",
|
||
"clickable": "true",
|
||
"enabled": "true",
|
||
"visible-to-user": "true",
|
||
"selected": "false",
|
||
"scrollable": "false",
|
||
"bounds": "[0,2181][1080,2328]",
|
||
},
|
||
)
|
||
button_content = ElementTree.SubElement(
|
||
button_frame,
|
||
"node",
|
||
{
|
||
"class": "android.widget.LinearLayout",
|
||
"package": "com.xunmeng.pinduoduo",
|
||
"text": "",
|
||
"content-desc": "",
|
||
"clickable": "false",
|
||
"enabled": "true",
|
||
"visible-to-user": "true",
|
||
"selected": "false",
|
||
"scrollable": "false",
|
||
"bounds": "[273,2181][807,2328]",
|
||
},
|
||
)
|
||
ElementTree.SubElement(
|
||
button_content,
|
||
"node",
|
||
{
|
||
"class": "android.widget.TextView",
|
||
"package": "com.xunmeng.pinduoduo",
|
||
"text": "选择尺码后，提交订单",
|
||
"content-desc": "",
|
||
"clickable": "false",
|
||
"enabled": "true",
|
||
"visible-to-user": "true",
|
||
"selected": "false",
|
||
"scrollable": "false",
|
||
"bounds": "[285,2225][795,2284]",
|
||
},
|
||
)
|
||
return ElementTree.tostring(root, encoding="unicode")
|
||
|
||
|
||
def _with_clickable_overlay(package: str, class_name: str, bounds: str) -> str:
|
||
root = ElementTree.fromstring(_COLOR_ONLY)
|
||
container = next(root.iter("node"))
|
||
ElementTree.SubElement(
|
||
container,
|
||
"node",
|
||
{
|
||
"package": package,
|
||
"class": class_name,
|
||
"text": "",
|
||
"content-desc": "",
|
||
"clickable": "true",
|
||
"enabled": "true",
|
||
"visible-to-user": "true",
|
||
"selected": "false",
|
||
"scrollable": "false",
|
||
"bounds": bounds,
|
||
},
|
||
)
|
||
return ElementTree.tostring(root, encoding="unicode")
|
||
|
||
|
||
def _png() -> str:
|
||
image = Image.new("RGB", (1080, 2376), "white")
|
||
raw = BytesIO()
|
||
image.save(raw, format="PNG")
|
||
return base64.b64encode(raw.getvalue()).decode("ascii")
|
||
|
||
|
||
class _FakeDevice:
|
||
def __init__(self) -> None:
|
||
self.hierarchy = "<hierarchy />"
|
||
self.after_hierarchy = _candidate_unselected()
|
||
self.calls: list[tuple[object, ...]] = []
|
||
self.swipe_error = False
|
||
|
||
def app_info(self, package_name: str) -> dict[str, str]:
|
||
self.calls.append(("app_info", package_name))
|
||
return {"versionName": "8.17.0"}
|
||
|
||
def app_current(self) -> dict[str, str]:
|
||
self.calls.append(("app_current",))
|
||
return {"package": "com.xunmeng.pinduoduo"}
|
||
|
||
def window_size(self) -> tuple[int, int]:
|
||
self.calls.append(("window_size",))
|
||
return 1080, 2376
|
||
|
||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||
self.calls.append(("jsonrpc", method, params, timeout))
|
||
if method == "dumpWindowHierarchy":
|
||
return self.hierarchy
|
||
if method == "takeScreenshot":
|
||
return _png()
|
||
if method == "click":
|
||
if params == [865, 2218]:
|
||
self.hierarchy = _EMPTY
|
||
return ""
|
||
if params == [528, 1387]:
|
||
self.hierarchy = _COLOR_ONLY
|
||
return ""
|
||
raise AssertionError(params)
|
||
if method == "swipe":
|
||
self.hierarchy = self.after_hierarchy
|
||
if self.swipe_error:
|
||
raise TimeoutError("unknown outcome")
|
||
return ""
|
||
raise AssertionError(method)
|
||
|
||
|
||
class _FakeAdb:
|
||
def __init__(self, device: _FakeDevice) -> None:
|
||
self.device = device
|
||
self.calls: list[tuple[object, ...]] = []
|
||
|
||
def inspect(self, serial: str) -> DeviceInspection:
|
||
self.calls.append(("inspect", serial))
|
||
return DeviceInspection(AdbDevice(serial=serial, state="device"), "PKG110", "16")
|
||
|
||
def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
|
||
self.calls.append(("intent", serial, goods_id))
|
||
self.device.hierarchy = _PRODUCT
|
||
return object()
|
||
|
||
|
||
def _swipes(device: _FakeDevice) -> list[tuple[object, ...]]:
|
||
return [call for call in device.calls if call[:2] == ("jsonrpc", "swipe")]
|
||
|
||
|
||
class SkuRevealSpikeTests(unittest.TestCase):
|
||
def _capturer(self, device: _FakeDevice) -> SkuRevealSpikeCapturer:
|
||
now = [0.0]
|
||
return SkuRevealSpikeCapturer(
|
||
_FakeAdb(device),
|
||
lambda serial: device,
|
||
30,
|
||
monotonic_clock=lambda: now[0],
|
||
sleep_function=lambda seconds: now.__setitem__(0, now[0] + seconds),
|
||
)
|
||
|
||
def test_success_publishes_before_after_and_exactly_one_reveal(self) -> None:
|
||
device = _FakeDevice()
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "evidence"
|
||
result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
|
||
self.assertEqual(result.output_directory, target)
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(any(call[1] == "pressKey" for call in device.calls if call[0] == "jsonrpc"))
|
||
for relative in (
|
||
"before/screenshot.png",
|
||
"before/hierarchy.xml",
|
||
"after/screenshot.png",
|
||
"after/hierarchy.xml",
|
||
"manifest.json",
|
||
):
|
||
self.assertTrue((target / relative).is_file(), relative)
|
||
manifest_text = result.manifest_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(manifest_text)
|
||
self.assertEqual(manifest["reveal_attempts"], 1)
|
||
self.assertEqual(manifest["rpc_outcome"], "completed")
|
||
self.assertNotIn("192.168.0.173:5555", manifest_text)
|
||
self.assertNotIn("gesture", manifest)
|
||
self.assertNotIn("coordinates", manifest)
|
||
self.assertNotIn("start", manifest)
|
||
self.assertNotIn("end", manifest)
|
||
|
||
def test_ambiguous_rpc_is_read_only_reconciled_without_retry(self) -> None:
|
||
device = _FakeDevice()
|
||
device.swipe_error = True
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "evidence"
|
||
result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled")
|
||
|
||
def test_selected_m_or_precondition_drift_never_publishes_or_retries(self) -> None:
|
||
for after in (
|
||
_M_SELECTED,
|
||
_candidate_unselected().replace("请选择： 尺码", "已选： 尺码"),
|
||
):
|
||
with self.subTest():
|
||
device = _FakeDevice()
|
||
device.after_hierarchy = after
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "evidence"
|
||
with self.assertRaises(SkuRevealSpikeError):
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
device = _FakeDevice()
|
||
device.hierarchy = "<hierarchy />"
|
||
invalid_color = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
|
||
original_call = device.jsonrpc_call
|
||
|
||
def drift(method: str, params: object = None, timeout: float = 10) -> str:
|
||
value = original_call(method, params, timeout)
|
||
if method == "click" and params == [528, 1387]:
|
||
device.hierarchy = invalid_color
|
||
return value
|
||
|
||
device.jsonrpc_call = drift # type: ignore[method-assign]
|
||
with TemporaryDirectory() as directory:
|
||
with self.assertRaises((SkuSelectionError, SkuRevealSpikeError)):
|
||
self._capturer(device).capture(
|
||
"192.168.0.173:5555",
|
||
"937122477375",
|
||
Path(directory) / "evidence",
|
||
)
|
||
self.assertEqual(_swipes(device), [])
|
||
|
||
def test_invalid_goods_and_existing_target_are_zero_action(self) -> None:
|
||
device = _FakeDevice()
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "existing"
|
||
target.mkdir()
|
||
with self.assertRaises(SkuRevealSpikeError):
|
||
self._capturer(device).capture("wifi", "1", Path(directory) / "new")
|
||
with self.assertRaises(SkuRevealSpikeError):
|
||
self._capturer(device).capture("wifi", "937122477375", target)
|
||
self.assertEqual(device.calls, [])
|
||
|
||
def test_before_screenshot_drift_is_rechecked_before_zero_swipe(self) -> None:
|
||
device = _FakeDevice()
|
||
original_call = device.jsonrpc_call
|
||
|
||
def drift(method: str, params: object = None, timeout: float = 10) -> str:
|
||
value = original_call(method, params, timeout)
|
||
if method == "takeScreenshot":
|
||
device.hierarchy = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
|
||
return value
|
||
|
||
device.jsonrpc_call = drift # type: ignore[method-assign]
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "evidence"
|
||
with self.assertRaises(SkuSelectionError):
|
||
self._capturer(device).capture(
|
||
"192.168.0.173:5555",
|
||
"937122477375",
|
||
target,
|
||
)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(_swipes(device), [])
|
||
|
||
def test_complete_reveal_segment_rejects_narrow_impostor_and_invalid_bounds(self) -> None:
|
||
for hierarchy in (
|
||
_with_clickable_overlay(
|
||
"com.xunmeng.pinduoduo",
|
||
"android.view.ViewGroup",
|
||
"[350,1450][370,1460]",
|
||
),
|
||
_with_clickable_overlay(
|
||
"com.android.systemui",
|
||
"android.view.ViewGroup",
|
||
"[0,366][1080,2328]",
|
||
),
|
||
_with_clickable_overlay(
|
||
"com.xunmeng.pinduoduo",
|
||
"android.view.ViewGroup",
|
||
"not-a-bound",
|
||
),
|
||
):
|
||
with self.subTest(), self.assertRaises(SkuRevealSpikeError):
|
||
_require_safe_reveal_path(_parse_nodes(hierarchy))
|
||
|
||
|
||
class SkuRevealSpikeCliTests(unittest.TestCase):
|
||
def test_cli_has_no_gesture_or_task_specification_parameters(self) -> None:
|
||
script = _load_reveal_script()
|
||
arguments = script.parse_arguments(
|
||
[
|
||
"--serial", "device-1",
|
||
"--goods-id", "937122477375",
|
||
"--output-dir", "evidence",
|
||
]
|
||
)
|
||
self.assertEqual(
|
||
set(vars(arguments)),
|
||
{"serial", "goods_id", "output_dir", "timeout", "adb"},
|
||
)
|
||
script.validate_arguments(arguments)
|
||
for field, value in (
|
||
("serial", ""),
|
||
("goods_id", "1"),
|
||
("timeout", 0),
|
||
("timeout", float("inf")),
|
||
):
|
||
with self.subTest(field=field), self.assertRaises(ValueError):
|
||
script.validate_arguments(
|
||
type("Arguments", (), vars(arguments) | {field: value})()
|
||
)
|
||
|
||
def test_cli_failure_is_redacted(self) -> None:
|
||
script = _load_reveal_script()
|
||
secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"
|
||
|
||
class FailingCapturer:
|
||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||
return None
|
||
|
||
def capture(self, *args: object, **kwargs: object) -> object:
|
||
raise SkuRevealSpikeError(secret)
|
||
|
||
stderr = io.StringIO()
|
||
with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
|
||
status = script.main(
|
||
[
|
||
"--serial", "192.168.0.173:5555",
|
||
"--goods-id", "937122477375",
|
||
"--output-dir", "evidence",
|
||
]
|
||
)
|
||
output = stderr.getvalue()
|
||
self.assertEqual(status, 1)
|
||
self.assertNotIn("Traceback", output)
|
||
self.assertNotIn("192.168.0.173:5555", output)
|
||
self.assertNotIn("private", output)
|
||
|
||
|
||
def _load_reveal_script() -> object:
|
||
path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_reveal_spike.py"
|
||
specification = importlib.util.spec_from_file_location("capture_sku_reveal_spike_test", path)
|
||
if specification is None or specification.loader is None:
|
||
raise RuntimeError("无法加载 T-103 reveal 取证脚本。")
|
||
module = importlib.util.module_from_spec(specification)
|
||
specification.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|