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

fix(client): identify sku entry failure substage

Browse Source
This commit is contained in:
QiuSW
2026-08-05 10:32:35 +08:00
parent e45d26fe93
commit b754cc9ca9
3 changed files with 142 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/tests/pdd/test_sku_selection.py
+45 -3
View File
@@ -605,14 +605,17 @@ class SkuSelectionRunnerTests(unittest.TestCase):
return ""
return super().jsonrpc_call(method, params, timeout)
adb = _FakeAdb()
device = NoPanelAfterEntry("<hierarchy />")
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
self._runner(_FakeAdb(), NoPanelAfterEntry()).run(
SkuSelectionRunner(adb, lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry")
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_panel_verify")
forged = SkuSelectionError("<hierarchy>private</hierarchy>")
setattr(forged, "_cmbuyer_failure_stage", "<private-stage>")
setattr(forged, "_cmbuyer_failure_stage", "sku_entry_click")
self.assertEqual(safe_failure_stage(forged), "unknown")
class HostileSetterError(DeviceConnectionError):
@@ -629,6 +632,45 @@ class SkuSelectionRunnerTests(unittest.TestCase):
)
self.assertEqual(safe_failure_stage(hostile.exception), "unknown")
def test_failure_stage_identifies_sku_entry_pre_intent(self) -> None:
device = _RawDevice(_PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
SkuSelectionRunner(_FakeAdb(), lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_pre_intent")
self.assertEqual(_actions(device, "click"), [])
def test_failure_stage_identifies_sku_entry_discovery(self) -> None:
device = _RawDevice("<hierarchy />")
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
SkuSelectionRunner(_FakeAdb(), lambda serial: device, 0.01).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_discovery")
self.assertEqual(_actions(device, "click"), [])
def test_failure_stage_identifies_sku_entry_click(self) -> None:
class ClickFailureDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("private device detail")
return super().jsonrpc_call(method, params, timeout)
adb = _FakeAdb()
device = ClickFailureDevice("<hierarchy />")
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE)
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError) as raised:
SkuSelectionRunner(adb, lambda serial: device, 0.03).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
self.assertEqual(safe_failure_stage(raised.exception), "sku_entry_click")
self.assertEqual(len(_actions(device, "click")), 1)
def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"