test: 增加PDD通用链接真机诊断脚本 (#118)
This commit is contained in:
@@ -0,0 +1,116 @@
|
||||
"""PDD 通用商品链接诊断脚本测试;不连接真实手机。"""
|
||||
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from tools.test_pdd_home_deeplink import (
|
||||
TEST_URL,
|
||||
classify_result,
|
||||
find_blocking_purchase_tasks,
|
||||
run_diagnostic,
|
||||
)
|
||||
|
||||
|
||||
class FakeDevice:
|
||||
def app_current(self):
|
||||
return {"package": "com.android.chrome"}
|
||||
|
||||
def dump_hierarchy(self, compressed=False):
|
||||
del compressed
|
||||
return '<hierarchy><node package="com.android.chrome" text="网页"/></hierarchy>'
|
||||
|
||||
|
||||
def create_database(path: Path, *, step: str = "", irreversible: bool = False) -> None:
|
||||
connection = sqlite3.connect(path)
|
||||
try:
|
||||
connection.executescript(
|
||||
"CREATE TABLE pdd_tasks ("
|
||||
" id INTEGER PRIMARY KEY, remote_task_id TEXT,"
|
||||
" task_type TEXT, current_step TEXT);"
|
||||
"CREATE TABLE task_runs ("
|
||||
" id INTEGER PRIMARY KEY, task_id INTEGER,"
|
||||
" irreversible_action_at TEXT);"
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO pdd_tasks VALUES (1, 'PUR-TEST', 'purchase', ?)",
|
||||
(step,),
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO task_runs VALUES (1, 1, ?)",
|
||||
("2026-08-10T10:00:00Z" if irreversible else None,),
|
||||
)
|
||||
connection.commit()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
class PddHomeDeeplinkToolTest(unittest.TestCase):
|
||||
def test_unreconciled_irreversible_task_is_blocked_before_adb(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
database = Path(directory) / "client.db"
|
||||
create_database(
|
||||
database, step="reconcile_purchase", irreversible=True
|
||||
)
|
||||
adb_calls = []
|
||||
|
||||
with self.assertRaises(RuntimeError) as raised:
|
||||
run_diagnostic(
|
||||
"192.168.0.173:5555",
|
||||
database,
|
||||
0,
|
||||
adb_runner=lambda *args, **kwargs: adb_calls.append(
|
||||
(args, kwargs)
|
||||
),
|
||||
)
|
||||
|
||||
self.assertIn("PUR-TEST", str(raised.exception))
|
||||
self.assertEqual(adb_calls, [])
|
||||
|
||||
def test_reconciled_task_no_longer_blocks_diagnostic(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
database = Path(directory) / "client.db"
|
||||
create_database(
|
||||
database, step="reconcile_completed", irreversible=True
|
||||
)
|
||||
|
||||
self.assertEqual(find_blocking_purchase_tasks(database), ())
|
||||
|
||||
def test_safe_run_opens_only_fixed_url_and_reports_browser(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
database = Path(directory) / "client.db"
|
||||
create_database(database)
|
||||
adb_calls = []
|
||||
|
||||
result = run_diagnostic(
|
||||
"USB-001",
|
||||
database,
|
||||
0,
|
||||
adb_runner=lambda command, **_kwargs: adb_calls.append(command),
|
||||
connector=lambda _address: FakeDevice(),
|
||||
sleeper=lambda _seconds: None,
|
||||
)
|
||||
|
||||
self.assertEqual(result.category, "浏览器")
|
||||
self.assertEqual(len(adb_calls), 2)
|
||||
self.assertEqual(adb_calls[1][-1], TEST_URL)
|
||||
|
||||
def test_home_tree_is_reported_as_pdd_home(self):
|
||||
xml_data = """<hierarchy>
|
||||
<node package="com.xunmeng.pinduoduo"/>
|
||||
<node package="com.xunmeng.pinduoduo"/>
|
||||
<node package="com.xunmeng.pinduoduo"/>
|
||||
<node package="com.xunmeng.pinduoduo" content-desc="首页"
|
||||
text="首页" selected="true" clickable="true"/>
|
||||
<node package="com.xunmeng.pinduoduo" text="聊天"/>
|
||||
<node package="com.xunmeng.pinduoduo" text="个人中心"/>
|
||||
</hierarchy>"""
|
||||
|
||||
result = classify_result(xml_data, "com.xunmeng.pinduoduo")
|
||||
|
||||
self.assertEqual(result.category, "PDD 首页")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user