684 lines
28 KiB
Python
684 lines
28 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
|
||
|
||
import cmbuyer_client.pdd.sku_reveal_spike as reveal_module
|
||
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
|
||
from cmbuyer_client.pdd.sku_reveal_spike import (
|
||
_annotate_reveal_failure,
|
||
_RevealEvidenceAdapter,
|
||
safe_reveal_failure_stage,
|
||
SkuRevealSpikeCapturer,
|
||
SkuRevealSpikeError,
|
||
_require_safe_reveal_path,
|
||
)
|
||
from cmbuyer_client.pdd.sku_selection import (
|
||
SkuSelectionError,
|
||
_annotate_sku_entry_failure,
|
||
_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")
|
||
_CURRENT_ONLY_EMPTY = (_FIXTURES / "sku_panel_empty_current_only_8_17_0.xml").read_text(encoding="utf-8")
|
||
_CURRENT_ONLY_COLOR = (_FIXTURES / "sku_panel_color_selected_size_hidden_current_only_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 _candidate_with_color_unselected() -> str:
|
||
root = ElementTree.fromstring(_candidate_unselected())
|
||
target = next(
|
||
node
|
||
for node in root.iter("node")
|
||
if node.get("class") == "android.view.ViewGroup"
|
||
and node.get("content-desc") == "黑色 CHA (纯棉）"
|
||
and node.get("bounds") == "[372,1000][684,1024]"
|
||
)
|
||
for node in target.iter("node"):
|
||
node.set("selected", "false")
|
||
return ElementTree.tostring(root, encoding="unicode")
|
||
|
||
|
||
def _candidate_with_submit_risk(kind: str) -> str:
|
||
root = ElementTree.fromstring(_candidate_unselected())
|
||
parents = {child: parent for parent in root.iter() for child in parent}
|
||
submit = next(
|
||
node
|
||
for node in root.iter("node")
|
||
if "提交订单" in node.get("text", "")
|
||
and node.get("class") == "android.widget.TextView"
|
||
)
|
||
parent = parents[submit]
|
||
if kind == "missing":
|
||
parent.remove(submit)
|
||
elif kind == "duplicate":
|
||
parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")))
|
||
elif kind == "moved_up":
|
||
submit.set("bounds", "[285,1900][795,1959]")
|
||
else:
|
||
raise AssertionError(kind)
|
||
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.empty_hierarchy = _EMPTY
|
||
self.color_hierarchy = _COLOR_ONLY
|
||
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 = self.empty_hierarchy
|
||
return ""
|
||
if params == [528, 1387]:
|
||
self.hierarchy = self.color_hierarchy
|
||
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 _assert_after_rejected_once(self, after_hierarchy: str, *, ambiguous: bool = False) -> None:
|
||
device = _FakeDevice()
|
||
device.after_hierarchy = after_hierarchy
|
||
device.swipe_error = ambiguous
|
||
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-*")), [])
|
||
|
||
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_current_only_empty_and_color_only_are_accepted_before_reveal(self) -> None:
|
||
device = _FakeDevice()
|
||
device.empty_hierarchy = _CURRENT_ONLY_EMPTY
|
||
device.color_hierarchy = _CURRENT_ONLY_COLOR
|
||
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.assertEqual(
|
||
(target / "before" / "hierarchy.xml").read_text(encoding="utf-8"),
|
||
_CURRENT_ONLY_COLOR,
|
||
)
|
||
|
||
def test_cross_price_variant_after_color_click_is_precondition_failure_without_reveal(self) -> None:
|
||
for empty_hierarchy, color_hierarchy in (
|
||
(_CURRENT_ONLY_EMPTY, _COLOR_ONLY),
|
||
(_EMPTY, _CURRENT_ONLY_COLOR),
|
||
):
|
||
with self.subTest():
|
||
device = _FakeDevice()
|
||
device.empty_hierarchy = empty_hierarchy
|
||
device.color_hierarchy = color_hierarchy
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "evidence"
|
||
with self.assertRaises(SkuSelectionError) as raised:
|
||
self._capturer(device).capture(
|
||
"192.168.0.173:5555",
|
||
"937122477375",
|
||
target,
|
||
)
|
||
self.assertEqual(
|
||
safe_reveal_failure_stage(raised.exception),
|
||
"reveal_precondition",
|
||
)
|
||
self.assertEqual(_swipes(device), [])
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
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_hidden_sizes_after_swipe_are_not_published_or_retried(self) -> None:
|
||
self._assert_after_rejected_once(_COLOR_ONLY)
|
||
|
||
def test_post_reveal_color_selected_drift_is_not_published_or_retried(self) -> None:
|
||
self._assert_after_rejected_once(_candidate_with_color_unselected())
|
||
|
||
def test_submit_hard_reject_zone_risk_is_not_published_or_retried(self) -> None:
|
||
for kind in ("missing", "duplicate", "moved_up"):
|
||
with self.subTest(kind=kind):
|
||
self._assert_after_rejected_once(_candidate_with_submit_risk(kind))
|
||
|
||
def test_ambiguous_rpc_with_unchanged_page_is_not_published_or_retried(self) -> None:
|
||
self._assert_after_rejected_once(_COLOR_ONLY, ambiguous=True)
|
||
|
||
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) as raised:
|
||
self._capturer(device).capture(
|
||
"192.168.0.173:5555",
|
||
"937122477375",
|
||
target,
|
||
)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(_swipes(device), [])
|
||
self.assertEqual(safe_reveal_failure_stage(raised.exception), "reveal_precondition")
|
||
|
||
def test_reveal_failure_stages_bind_attempt_candidate_after_and_publish(self) -> None:
|
||
device = _FakeDevice()
|
||
|
||
def fail_before_seal(adapter: _RevealEvidenceAdapter) -> None:
|
||
raise SkuRevealSpikeError("private pre-seal detail")
|
||
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "pre-seal"
|
||
with (
|
||
patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_before_seal),
|
||
self.assertRaises(SkuRevealSpikeError) as pre_seal,
|
||
):
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(safe_reveal_failure_stage(pre_seal.exception), "reveal_precondition")
|
||
self.assertEqual(_swipes(device), [])
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
# attempted 只可能在 adapter 已封存并发送唯一手势后报告。
|
||
device = _FakeDevice()
|
||
original_reveal = _RevealEvidenceAdapter.reveal_size_options_once
|
||
|
||
def fail_after_attempt(adapter: _RevealEvidenceAdapter) -> None:
|
||
original_reveal(adapter)
|
||
raise SkuRevealSpikeError("private attempted detail")
|
||
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "attempted"
|
||
with (
|
||
patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_after_attempt),
|
||
self.assertRaises(SkuRevealSpikeError) as attempted,
|
||
):
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(safe_reveal_failure_stage(attempted.exception), "reveal_attempted")
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
device = _FakeDevice()
|
||
device.after_hierarchy = _COLOR_ONLY
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "candidate"
|
||
with self.assertRaises(SkuRevealSpikeError) as candidate:
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(safe_reveal_failure_stage(candidate.exception), "reveal_candidate")
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
device = _FakeDevice()
|
||
original_capture = reveal_module._capture_frame
|
||
|
||
def fail_after_capture(adapter: object, directory: Path, hierarchy: str) -> None:
|
||
if directory.name == "after":
|
||
raise OSError("private after path")
|
||
original_capture(adapter, directory, hierarchy)
|
||
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "after"
|
||
with (
|
||
patch.object(reveal_module, "_capture_frame", fail_after_capture),
|
||
self.assertRaises(SkuRevealSpikeError) as after,
|
||
):
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(safe_reveal_failure_stage(after.exception), "reveal_after")
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
device = _FakeDevice()
|
||
with TemporaryDirectory() as directory:
|
||
target = Path(directory) / "publish"
|
||
with (
|
||
patch.object(reveal_module.os, "rename", side_effect=OSError("private publish path")),
|
||
self.assertRaises(SkuRevealSpikeError) as publish,
|
||
):
|
||
self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
|
||
self.assertEqual(safe_reveal_failure_stage(publish.exception), "reveal_publish")
|
||
self.assertEqual(len(_swipes(device)), 1)
|
||
self.assertFalse(target.exists())
|
||
self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
|
||
|
||
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.assertIn("stage=unknown", output)
|
||
self.assertNotIn("Traceback", output)
|
||
self.assertNotIn("192.168.0.173:5555", output)
|
||
self.assertNotIn("private", output)
|
||
|
||
def test_cli_reports_only_formally_annotated_entry_stage(self) -> None:
|
||
script = _load_reveal_script()
|
||
secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"
|
||
|
||
def run_with(error: BaseException) -> str:
|
||
class FailingCapturer:
|
||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||
return None
|
||
|
||
def capture(self, *args: object, **kwargs: object) -> object:
|
||
raise error
|
||
|
||
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",
|
||
]
|
||
)
|
||
self.assertEqual(status, 1)
|
||
output = stderr.getvalue()
|
||
self.assertNotIn("Traceback", output)
|
||
self.assertNotIn("192.168.0.173:5555", output)
|
||
self.assertNotIn("private", output)
|
||
return output
|
||
|
||
annotated = SkuSelectionError(secret)
|
||
_annotate_sku_entry_failure(annotated, "sku_entry_panel_verify")
|
||
_annotate_reveal_failure(annotated, "reveal_precondition")
|
||
self.assertIn("stage=sku_entry_panel_verify", run_with(annotated))
|
||
|
||
spoofed = SkuSelectionError(secret)
|
||
setattr(spoofed, "_cmbuyer_failure_stage", "sku_entry_panel_verify")
|
||
self.assertIn("stage=unknown", run_with(spoofed))
|
||
|
||
def test_cli_reports_only_formally_annotated_reveal_stage(self) -> None:
|
||
script = _load_reveal_script()
|
||
secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"
|
||
|
||
def run_with(error: BaseException) -> str:
|
||
class FailingCapturer:
|
||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||
return None
|
||
|
||
def capture(self, *args: object, **kwargs: object) -> object:
|
||
raise error
|
||
|
||
stderr = io.StringIO()
|
||
with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
|
||
self.assertEqual(
|
||
script.main(
|
||
[
|
||
"--serial", "192.168.0.173:5555",
|
||
"--goods-id", "937122477375",
|
||
"--output-dir", "evidence",
|
||
]
|
||
),
|
||
1,
|
||
)
|
||
output = stderr.getvalue()
|
||
self.assertNotIn("192.168.0.173:5555", output)
|
||
self.assertNotIn("private", output)
|
||
return output
|
||
|
||
annotated = SkuRevealSpikeError(secret)
|
||
_annotate_reveal_failure(annotated, "reveal_candidate")
|
||
self.assertIn("stage=reveal_candidate", run_with(annotated))
|
||
|
||
spoofed = SkuRevealSpikeError(secret)
|
||
setattr(spoofed, "_cmbuyer_reveal_failure_stage", "reveal_candidate")
|
||
self.assertIn("stage=unknown", run_with(spoofed))
|
||
|
||
class HostileGetterError(SkuRevealSpikeError):
|
||
def __getattribute__(self, name: str) -> object:
|
||
if name.startswith("_cmbuyer_reveal_"):
|
||
raise RuntimeError(secret)
|
||
return super().__getattribute__(name)
|
||
|
||
self.assertIn("stage=unknown", run_with(HostileGetterError(secret)))
|
||
|
||
|
||
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()
|