chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): productionize proven sku reveal flow

Browse Source
This commit is contained in:
QiuSW
2026-08-05 16:55:10 +08:00
parent fec21ae857
commit 60cf05b08b
5 changed files with 895 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/tests/pdd/test_sku_reveal_spike.py
+23 -13
View File
@@ -18,17 +18,19 @@ 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,
_candidate_projection,
_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,
_require_safe_reveal_path,
_revealed_unselected_projection as _candidate_projection,
)
from cmbuyer_client.pdd.sku_selection_runner import (
UiautomatorSkuPanelAdapter as _RevealEvidenceAdapter,
)
@@ -350,7 +352,7 @@ def _candidate_cross_variant_residue(variant: str, kind: str) -> str:
def _candidate_with_color_unselected() -> str:
root = ElementTree.fromstring(_candidate_unselected())
root = ElementTree.fromstring(_candidate_current_only())
target = next(
node
for node in root.iter("node")
@@ -364,7 +366,7 @@ def _candidate_with_color_unselected() -> str:
def _candidate_with_submit_risk(kind: str) -> str:
root = ElementTree.fromstring(_candidate_unselected())
root = ElementTree.fromstring(_candidate_current_only())
parents = {child: parent for parent in root.iter() for child in parent}
submit = next(
node
@@ -379,6 +381,13 @@ def _candidate_with_submit_risk(kind: str) -> str:
parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")))
elif kind == "moved_up":
submit.set("bounds", "[285,1900][795,1959]")
elif kind in {"extra_altered", "extra_moved_up"}:
extra = ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode"))
if kind == "extra_altered":
extra.set("text", "异常提交订单文案")
else:
extra.set("bounds", "[285,1900][795,1959]")
parent.append(extra)
else:
raise AssertionError(kind)
return ElementTree.tostring(root, encoding="unicode")
@@ -416,9 +425,9 @@ def _png() -> str:
class _FakeDevice:
def __init__(self) -> None:
self.hierarchy = "<hierarchy />"
self.empty_hierarchy = _EMPTY
self.color_hierarchy = _COLOR_ONLY
self.after_hierarchy = _candidate_unselected()
self.empty_hierarchy = _CURRENT_ONLY_EMPTY
self.color_hierarchy = _CURRENT_ONLY_COLOR
self.after_hierarchy = _candidate_current_only()
self.calls: list[tuple[object, ...]] = []
self.swipe_error = False
@@ -545,8 +554,9 @@ class SkuRevealSpikeTests(unittest.TestCase):
self.assertTrue((target / "after" / "hierarchy.xml").is_file())
def test_candidate_price_variants_reject_hybrid_duplicate_unknown_and_parent_drift(self) -> None:
# 两个真实嵌套 variant 各自唯一命中；任意混搭、额外子节点或父子关系漂移均拒绝。
_candidate_projection(_parse_nodes(_candidate_unselected()))
# 生产只放行 current-only；历史 dual 及任意混搭、额外节点、父链漂移均拒绝。
with self.assertRaises(SkuSelectionError):
_candidate_projection(_parse_nodes(_candidate_unselected()))
_candidate_projection(_parse_nodes(_candidate_current_only()))
for variant in ("dual", "current_only"):
kinds = ["hybrid", "duplicate", "unknown", "unknown_child", "parent_drift"]
@@ -683,7 +693,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
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"):
for kind in ("missing", "duplicate", "moved_up", "extra_altered", "extra_moved_up"):
with self.subTest(kind=kind):
self._assert_after_rejected_once(_candidate_with_submit_risk(kind))
@@ -739,7 +749,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
device.jsonrpc_call = drift # type: ignore[method-assign]
with TemporaryDirectory() as directory:
target = Path(directory) / "evidence"
with self.assertRaises(SkuRevealSpikeError) as raised:
with self.assertRaises((SkuRevealSpikeError, SkuSelectionError)) as raised:
self._capturer(device).capture(
"192.168.0.173:5555",
"937122477375",
@@ -850,7 +860,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
"not-a-bound",
),
):
with self.subTest(), self.assertRaises(SkuRevealSpikeError):
with self.subTest(), self.assertRaises((SkuRevealSpikeError, SkuSelectionError)):
_require_safe_reveal_path(_parse_nodes(hierarchy))
client/tests/pdd/test_sku_selection.py
+357 -30
View File
@@ -3,8 +3,10 @@ from __future__ import annotations
import ast
import base64
from contextlib import redirect_stderr
from functools import lru_cache
from io import BytesIO
import importlib.util
import json
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
@@ -49,6 +51,14 @@ _TASK_SIZE = "M(建议100-115)"
_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
def _revealed_current_only() -> str:
# 与 spike 测试共用由已验 M fixture 机械恢复出的 current-only 未选态；生产判据只在源码一处。
from tests.pdd.test_sku_reveal_spike import _candidate_current_only
return _candidate_current_only()
@lru_cache(maxsize=1)
def _png_base64() -> str:
image = Image.new("RGB", (1080, 2376), "white")
raw = BytesIO()
@@ -59,8 +69,9 @@ def _png_base64() -> str:
class _RawDevice:
def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
self.hierarchy = hierarchy
self.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
self.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
self.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
self.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8")
self.revealed_hierarchy = _revealed_current_only()
self.version = "8.17.0"
self.package = "com.xunmeng.pinduoduo"
self.screenshot = _png_base64() if screenshot is None else screenshot
@@ -89,6 +100,11 @@ class _RawDevice:
raise AssertionError(params)
self._apply_tap(int(params[0]), int(params[1]))
return ""
if method == "swipe":
if params != [360, 1900, 360, 1300, 30]:
raise AssertionError(params)
self.hierarchy = self.revealed_hierarchy
return ""
raise AssertionError(method)
def _apply_tap(self, x: int, y: int) -> None:
@@ -405,25 +421,223 @@ class SkuSelectionFlowTests(unittest.TestCase):
flow.open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_target_mapping_stops_at_unproven_reveal_then_s_to_m_is_exact(self) -> None:
def test_current_only_exact_transition_reveals_then_selects_m_once(self) -> None:
device = _RawDevice()
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(flow.read_sku_unit_price(), "12.88")
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_actions(device, "pressKey"), [])
restored = _RawDevice(_S_FIXTURE.read_text(encoding="utf-8"))
restored_flow = _flow(restored)
restored_flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(restored_flow.read_sku_unit_price(), "12.88")
restored_flow.exit_sku_panel_safely()
self.assertEqual(_tap_centers(restored), [(635, 1624)])
self.assertEqual(_actions(restored, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
self.assertEqual(_actions(restored, "swipe"), [])
def test_dual_color_only_is_zero_action_before_reveal(self) -> None:
for fixture in (_COLOR_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE):
with self.subTest(fixture=fixture.name):
device = _RawDevice(fixture.read_text(encoding="utf-8"))
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_actions(device, "click"), [])
self.assertEqual(_actions(device, "swipe"), [])
def test_revealed_current_only_is_a_unique_profile_but_not_a_reentry_shortcut(self) -> None:
hierarchy = _revealed_current_only()
spec = _classify_panel(_parse_nodes(hierarchy))
self.assertIs(spec.profile, _PanelProfile.SIZE_VISIBLE_UNSELECTED)
self.assertEqual(spec.price_layout.__class__.__name__, "_CurrentOnlyPriceLayout")
device = _RawDevice(hierarchy)
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_actions(device, "click"), [])
self.assertEqual(_actions(device, "swipe"), [])
def test_revealed_exact_m_contract_rejects_missing_duplicate_disabled_near_and_other_selected(self) -> None:
base = _revealed_current_only()
for kind in ("missing", "duplicate", "disabled", "near", "other_selected"):
root = ElementTree.fromstring(base)
parents = {child: parent for parent in root.iter() for child in parent}
m_action = next(
node for node in root.iter("node")
if node.get("text") == "M（建议100-115）"
)
wrapper = parents[m_action]
if kind == "missing":
parents[wrapper].remove(wrapper)
elif kind == "duplicate":
parents[wrapper].append(
ElementTree.fromstring(ElementTree.tostring(wrapper, encoding="unicode"))
)
elif kind == "disabled":
m_action.set("enabled", "false")
elif kind == "near":
m_action.set("text", "M（建议100-115）相近")
else:
s_action = next(
node for node in root.iter("node")
if node.get("text") == "S（建议80-100）"
)
s_action.set("selected", "true")
with self.subTest(kind=kind), self.assertRaises(SkuSelectionError):
_classify_panel(
_parse_nodes(ElementTree.tostring(root, encoding="unicode"))
)
def test_reveal_requires_two_consecutive_stable_frames_and_terminal_blocks_reentry(self) -> None:
class UnstableRevealDevice(_RawDevice):
def __init__(self) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.after_reveal_reads = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy" and _actions(self, "swipe"):
self.after_reveal_reads += 1
candidate = _revealed_current_only()
self.hierarchy = (
candidate
if self.after_reveal_reads % 2
else candidate.replace("请选择： 尺码", "未知摘要")
)
return super().jsonrpc_call(method, params, timeout)
device = UnstableRevealDevice()
flow = _flow(device)
with self.assertRaises(SkuSelectionError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
action_count = len(_actions(device, "click")) + len(_actions(device, "swipe"))
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387)])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(
len(_actions(device, "click")) + len(_actions(device, "swipe")),
action_count,
)
def test_reveal_timeout_is_read_only_reconciled_without_m_back_or_retry(self) -> None:
class RevealTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.delivered = delivered
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "swipe":
if self.delivered:
super().jsonrpc_call(method, params, timeout)
else:
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private reveal timeout")
return super().jsonrpc_call(method, params, timeout)
for delivered in (False, True):
with self.subTest(delivered=delivered):
device = RevealTimeoutDevice(delivered)
flow = _flow(device)
with self.assertRaises(SkuSelectionRunError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
try:
flow.reconcile_pending_action()
except SkuSelectionError:
pass
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387)])
self.assertEqual(_actions(device, "pressKey"), [])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(len(_actions(device, "swipe")), 1)
def test_m_timeout_is_read_only_reconciled_without_back_or_retry(self) -> None:
class MTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
self.delivered = delivered
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click" and params == [635, 1624]:
if self.delivered:
super().jsonrpc_call(method, params, timeout)
else:
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private M timeout")
return super().jsonrpc_call(method, params, timeout)
for delivered in (False, True):
with self.subTest(delivered=delivered):
device = MTimeoutDevice(delivered)
flow = _flow(device)
with self.assertRaises(SkuSelectionRunError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
try:
flow.reconcile_pending_action()
except SkuSelectionError:
pass
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
self.assertEqual(_actions(device, "pressKey"), [])
with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
def test_foreground_is_freshly_rechecked_before_s_reveal_and_m_mutations(self) -> None:
class ForegroundDriftDevice(_RawDevice):
def __init__(self, hierarchy: str, flip_at: int) -> None:
super().__init__(hierarchy)
self.flip_at = flip_at
self.current_reads = 0
def app_current(self) -> dict[str, str]:
current = super().app_current()
self.current_reads += 1
if self.current_reads == self.flip_at:
return {"package": "other"}
return current
cases = (
(
"s",
_S_FIXTURE.read_text(encoding="utf-8"),
2,
[],
0,
),
(
"reveal",
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
4,
[(528, 1387)],
0,
),
(
"m",
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
7,
[(528, 1387)],
1,
),
)
for name, hierarchy, flip_at, expected_taps, expected_swipes in cases:
with self.subTest(name=name):
device = ForegroundDriftDevice(hierarchy, flip_at)
with self.assertRaises(SkuSelectionError):
_flow(device).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
)
self.assertEqual(_tap_centers(device), expected_taps)
self.assertEqual(len(_actions(device, "swipe")), expected_swipes)
self.assertEqual(_actions(device, "pressKey"), [])
def test_current_only_empty_to_color_only_uses_the_matching_exact_spec(self) -> None:
device = _RawDevice()
@@ -432,15 +646,15 @@ class SkuSelectionFlowTests(unittest.TestCase):
flow = _flow(device)
flow.open_sku_panel(_TARGET_URL)
with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
self.assertEqual(len(_actions(device, "swipe")), 1)
matched = _classify_panel(_parse_nodes(device.hierarchy))
self.assertIs(matched.profile, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN)
self.assertIs(matched.profile, _PanelProfile.TARGETS_SELECTED)
self.assertEqual(
matched.price_layout.__class__.__name__,
"_CurrentOnlyPriceLayout",
"_DualPriceLayout",
)
def test_dual_and_current_only_variants_each_match_one_complete_spec(self) -> None:
@@ -585,7 +799,9 @@ class SkuSelectionFlowTests(unittest.TestCase):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertNotIn("受控显示动作尚未取证", str(raised.exception))
self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
expected = [(865, 2218), (528, 1387)] if empty_fixture is _CURRENT_ONLY_EMPTY_FIXTURE else [(865, 2218)]
self.assertEqual(_tap_centers(device), expected)
self.assertEqual(_actions(device, "swipe"), [])
self.assertEqual(_actions(device, "pressKey"), [])
def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
@@ -851,10 +1067,8 @@ class SkuSelectionFlowTests(unittest.TestCase):
flow.open_sku_panel(_TARGET_URL)
device.select_alternates()
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(
_tap_centers(device),
[(865, 2218), (635, 1624)],
)
self.assertEqual(_tap_centers(device), [(865, 2218), (635, 1624)])
self.assertEqual(_actions(device, "swipe"), [])
def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
@@ -876,7 +1090,7 @@ class SkuSelectionFlowTests(unittest.TestCase):
_flow(_RawDevice(clickable_parent)).read_sku_unit_price()
def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"}
forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click", "swipe", "scroll"}
self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
self.assertTrue(forbidden.isdisjoint(pdd.__all__))
@@ -900,6 +1114,8 @@ class SkuSelectionFlowTests(unittest.TestCase):
runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
self.assertEqual(len(click_calls), 1)
swipe_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "swipe"]
self.assertEqual(len(swipe_calls), 1)
def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
now = [0.0]
@@ -947,22 +1163,24 @@ class SkuSelectionFlowTests(unittest.TestCase):
class _CompletedFlow:
"""仅隔离 runner 文件发布测试；生产 Flow 在 reveal 取证前仍必须停止。"""
"""仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。"""
def __init__(self, device: object, *args: object, **kwargs: object) -> None:
self.device = device
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
return None
self.device.tap_sku_entry("[688,2184][1042,2253]")
def select_sku_options(self, selection: object) -> None:
return None
self.device.tap_sku_option("[372,1188][684,1587]")
self.device.reveal_size_options_once()
self.device.tap_sku_option("[439,1582][831,1667]")
def verify_target_selection_and_read_price(self, selection: object) -> str:
return "12.88"
def exit_sku_panel_safely(self) -> None:
return None
self.device.leave_sku_panel()
def reconcile_pending_action(self) -> None:
return None
@@ -989,15 +1207,55 @@ class SkuSelectionRunnerTests(unittest.TestCase):
self.assertEqual(result.unit_price, "12.88")
manifest = result.manifest_path.read_text(encoding="utf-8")
manifest_data = json.loads(manifest)
self.assertTrue(result.screenshot_path.is_file())
self.assertNotIn("device-1", manifest)
self.assertNotIn("hierarchy", manifest)
self.assertNotIn("已选", manifest)
self.assertIn('"unit_price": "12.88"', manifest)
self.assertIn('"selection_status": "restored"', manifest)
self.assertIn('"panel_status": "verified"', manifest)
self.assertIn('"safe_exit": "completed"', manifest)
self.assertIn('"panel_status": "verified_before_back"', manifest)
self.assertIn('"back_attempts": 1', manifest)
self.assertIn('"back_rpc_outcome": "completed"', manifest)
self.assertIn('"post_exit_status": "human_review_required"', manifest)
self.assertNotIn('"safe_exit"', manifest)
self.assertEqual(
manifest_data["actions"],
{
"sku_entry": {"attempts": 1, "rpc_outcome": "completed"},
"target_color": {"attempts": 1, "rpc_outcome": "completed"},
"size_reveal": {"attempts": 1, "rpc_outcome": "completed"},
"target_size": {"attempts": 1, "rpc_outcome": "completed"},
"back": {"attempts": 1, "rpc_outcome": "completed"},
},
)
self.assertFalse((target / "hierarchy.xml").exists())
self.assertEqual(len(_actions(device, "pressKey")), 1)
def test_publish_rejects_flow_stub_without_exact_action_audit_chain(self) -> None:
class IncompleteFlow(_CompletedFlow):
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
return None
def select_sku_options(self, selection: object) -> None:
return None
def exit_sku_panel_safely(self) -> None:
return None
device = _RawDevice()
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
with (
patch.object(runner_module, "SkuSelectionFlow", IncompleteFlow),
self.assertRaises(SkuSelectionRunError) as raised,
):
self._runner(_FakeAdb(), device).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
)
self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
self.assertEqual(_actions(device, "pressKey"), [])
def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None:
@@ -1223,6 +1481,34 @@ class SkuSelectionRunnerTests(unittest.TestCase):
self.assertTrue(adapter.entry_was_tapped)
self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [865, 2218], 10)])
def test_adapter_seals_every_named_mutation_before_rpc(self) -> None:
device = _RawDevice(_PRODUCT_PAGE)
adapter = UiautomatorSkuPanelAdapter(device, 10)
adapter.tap_sku_entry("[688,2184][1042,2253]")
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.tap_sku_entry("[688,2184][1042,2253]")
adapter.tap_sku_option("[372,1188][684,1587]")
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.tap_sku_option("[372,1188][684,1587]")
adapter.reveal_size_options_once()
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.reveal_size_options_once()
adapter.leave_sku_panel()
with self.assertRaises(SkuSelectionDeviceAdapterError):
adapter.leave_sku_panel()
self.assertEqual(len(_actions(device, "click")), 2)
self.assertEqual(len(_actions(device, "swipe")), 1)
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertEqual(adapter.option_attempts, 1)
self.assertEqual(adapter.entry_rpc_outcome, "completed")
self.assertEqual(adapter.reveal_rpc_outcome, "completed")
self.assertEqual(adapter.back_attempts, 1)
self.assertEqual(adapter.back_rpc_outcome, "completed")
def test_entry_stability_interruptions_never_click(self) -> None:
now = [0.0]
class SequenceDevice(_RawDevice):
@@ -1276,8 +1562,48 @@ class SkuSelectionRunnerTests(unittest.TestCase):
with self.assertRaises(SkuSelectionError): _flow(device).exit_sku_panel_safely()
self.assertEqual(_actions(device, "pressKey"), [])
def test_unproven_reveal_publishes_nothing_and_safely_exits_once(self) -> None:
def test_back_precondition_rejects_every_non_target_profile_after_screenshot_reverify(self) -> None:
drifts = (
_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
_CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8"),
_S_FIXTURE.read_text(encoding="utf-8"),
_revealed_current_only(),
)
for drift in drifts:
class DriftBeforeBackDevice(_RawDevice):
def __init__(self) -> None:
super().__init__()
self.after_screenshot_reads = 0
self.screenshot_seen = False
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "takeScreenshot":
value = super().jsonrpc_call(method, params, timeout)
self.screenshot_seen = True
return value
if method == "dumpWindowHierarchy" and self.screenshot_seen:
self.after_screenshot_reads += 1
if self.after_screenshot_reads == 2:
self.hierarchy = drift
return super().jsonrpc_call(method, params, timeout)
with self.subTest(profile=_classify_panel(_parse_nodes(drift)).profile), TemporaryDirectory() as temporary:
device = DriftBeforeBackDevice()
target = Path(temporary) / "result"
with self.assertRaises(SkuSelectionError) as raised:
self._runner(_FakeAdb(), device).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
)
self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
self.assertEqual(_actions(device, "pressKey"), [])
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
def test_dual_precondition_publishes_nothing_without_reveal_or_back(self) -> None:
device = _RawDevice()
device.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
device.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
with TemporaryDirectory() as temporary:
target = Path(temporary) / "out"
with self.assertRaises(SkuSelectionError):
@@ -1285,7 +1611,8 @@ class SkuSelectionRunnerTests(unittest.TestCase):
self.assertFalse(target.exists())
self.assertFalse((target / "manifest.json").exists())
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertEqual(_actions(device, "swipe"), [])
self.assertEqual(len(_actions(device, "pressKey")), 0)
def test_option_timeout_reconciliation_controls_back_once(self) -> None:
class OptionTimeoutDevice(_RawDevice):
@@ -1300,7 +1627,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
raise TimeoutError("uncertain option")
return super().jsonrpc_call(method, params, timeout)
for delivered, expected_back in ((False, 0), (True, 1)):
for delivered, expected_back in ((False, 0), (True, 0)):
with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
device = OptionTimeoutDevice(delivered)
adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
@@ -1336,7 +1663,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(len(_actions(device, "click")), 1)
self.assertEqual(len(_actions(device, "pressKey")), 1)
self.assertEqual(len(_actions(device, "pressKey")), 0)
class SkuSelectionCliTests(unittest.TestCase):