From ff01f950eb692ae0067dec16560011e46a0c6257 Mon Sep 17 00:00:00 2001 From: chengma Date: Mon, 10 Aug 2026 19:07:06 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E5=A2=9E=E5=8A=A0PDD=E9=80=9A=E7=94=A8?= =?UTF-8?q?=E9=93=BE=E6=8E=A5=E7=9C=9F=E6=9C=BA=E8=AF=8A=E6=96=AD=E8=84=9A?= =?UTF-8?q?=E6=9C=AC=20(#118)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/test/test_pdd_home_deeplink.py | 116 ++++++++++++++ client/tools/test_pdd_home_deeplink.py | 211 +++++++++++++++++++++++++ 2 files changed, 327 insertions(+) create mode 100644 client/test/test_pdd_home_deeplink.py create mode 100644 client/tools/test_pdd_home_deeplink.py diff --git a/client/test/test_pdd_home_deeplink.py b/client/test/test_pdd_home_deeplink.py new file mode 100644 index 0000000..c92ffcf --- /dev/null +++ b/client/test/test_pdd_home_deeplink.py @@ -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 '' + + +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 = """ + + + + + + + """ + + result = classify_result(xml_data, "com.xunmeng.pinduoduo") + + self.assertEqual(result.category, "PDD 首页") + + +if __name__ == "__main__": + unittest.main() diff --git a/client/tools/test_pdd_home_deeplink.py b/client/tools/test_pdd_home_deeplink.py new file mode 100644 index 0000000..8832a9a --- /dev/null +++ b/client/tools/test_pdd_home_deeplink.py @@ -0,0 +1,211 @@ +"""验证通用 PDD 商品链接最终进入哪个 Android 页面。 + +本工具只打开固定网址并读取当前页面，不点击购买、下单或支付控件。存在已经进入 +不可逆阶段、尚未完成核单的采购任务时，会在连接手机之前拒绝运行。 +""" + +from __future__ import annotations + +import argparse +import sqlite3 +import subprocess +import sys +import time +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Optional, Sequence + + +CLIENT_ROOT = Path(__file__).resolve().parents[1] +if str(CLIENT_ROOT) not in sys.path: + sys.path.insert(0, str(CLIENT_ROOT)) + +from src.pdd_page_classifier import ( # noqa: E402 + PAGE_EXTERNAL, + PAGE_HOME, + classify_pdd_page, +) + + +TEST_URL = "https://mobile.yangkeduo.com/goods.html" +DEFAULT_DATABASE = CLIENT_ROOT / "data" / "client.db" +PDD_PACKAGE_NAME = "com.xunmeng.pinduoduo" +_BROWSER_PACKAGE_MARKERS = ( + "browser", "chrome", "edge", "ucmobile", "qqbrowser", +) +PAGE_NAMES = { + "captcha": "安全验证页", + "goods": "商品详情页", + "loading": "加载页", + "login_required": "登录页", + "network_error": "网络错误页", + "order_confirmation": "订单确认页", + "payment": "支付页", + "risk_control": "风控页", + "unknown": "无法识别的 PDD 页面", +} + + +@dataclass(frozen=True) +class LinkDiagnostic: + """浏览器打开链接后的只读诊断结果。""" + + category: str + detail: str + current_package: str + page_kind: str + + +def find_blocking_purchase_tasks(database_path: Path) -> tuple[str, ...]: + """只读查询尚未完成核单的不可逆采购任务。""" + + path = Path(database_path).resolve() + if not path.is_file(): + raise FileNotFoundError(f"找不到 Client 数据库：{path}") + connection = sqlite3.connect( + f"file:{path.as_posix()}?mode=ro", uri=True, timeout=5.0 + ) + try: + connection.execute("PRAGMA query_only = ON") + rows = connection.execute( + "SELECT t.remote_task_id FROM pdd_tasks t" + " JOIN task_runs r ON r.id = (" + " SELECT r2.id FROM task_runs r2" + " WHERE r2.task_id = t.id ORDER BY r2.id DESC LIMIT 1" + " )" + " WHERE t.task_type = 'purchase'" + " AND r.irreversible_action_at IS NOT NULL" + " AND COALESCE(t.current_step, '') <> 'reconcile_completed'" + " ORDER BY t.id" + ).fetchall() + finally: + connection.close() + return tuple(str(row[0]) for row in rows) + + +def classify_result(xml_data: str, current_package: str) -> LinkDiagnostic: + """结合当前包名和控件树，生成容易阅读的中文结论。""" + + observation = classify_pdd_page(ET.fromstring(xml_data), current_package) + if observation.kind == PAGE_HOME: + return LinkDiagnostic( + "PDD 首页", "已识别首页底部导航及选中的首页入口", current_package, + observation.kind, + ) + if observation.kind == PAGE_EXTERNAL: + if any( + marker in current_package.casefold() + for marker in _BROWSER_PACKAGE_MARKERS + ): + return LinkDiagnostic( + "浏览器", "链接仍停留在 Android 浏览器", current_package, + observation.kind, + ) + return LinkDiagnostic( + "无法判断", + "链接停留在非 PDD 应用或系统应用选择页面", + current_package, + observation.kind, + ) + if current_package == PDD_PACKAGE_NAME or observation.pdd_hierarchy: + return LinkDiagnostic( + "PDD 其他页面", + PAGE_NAMES.get(observation.kind, observation.kind), + current_package, + observation.kind, + ) + return LinkDiagnostic( + "无法判断", "当前包名和控件树都不足以确定页面", current_package, + observation.kind, + ) + + +def run_diagnostic( + device_address: str, + database_path: Path, + wait_seconds: float, + *, + adb_runner: Callable[..., Any] = subprocess.run, + connector: Optional[Callable[[str], Any]] = None, + sleeper: Callable[[float], None] = time.sleep, +) -> LinkDiagnostic: + """先执行本地安全检查，再打开固定网址并读取一次页面。""" + + blocking_tasks = find_blocking_purchase_tasks(database_path) + if blocking_tasks: + task_ids = "、".join(blocking_tasks) + raise RuntimeError( + f"存在尚未完成核单的不可逆采购任务：{task_ids}；" + "根据安全规则，本次没有连接或操作手机" + ) + + address = str(device_address or "").strip() + if not address: + raise ValueError("必须提供 Android 设备地址") + adb_runner( + ["adb", "-s", address, "get-state"], + check=True, + capture_output=True, + text=True, + ) + adb_runner( + [ + "adb", "-s", address, "shell", "am", "start", + "-a", "android.intent.action.VIEW", "-d", TEST_URL, + ], + check=True, + capture_output=True, + text=True, + ) + sleeper(wait_seconds) + + if connector is None: + import uiautomator2 as u2 + + connector = u2.connect + device = connector(address) + current = device.app_current() + current_package = str(current.get("package") or "") + xml_data = str(device.dump_hierarchy(compressed=False)) + return classify_result(xml_data, current_package) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="测试浏览器打开 PDD 通用商品链接后进入的页面" + ) + parser.add_argument("--device", required=True, help="ADB 设备地址") + parser.add_argument( + "--wait-seconds", type=float, default=5.0, + help="打开链接后等待页面跳转的秒数，默认 5 秒", + ) + parser.add_argument( + "--database", type=Path, default=DEFAULT_DATABASE, + help="Client 数据库路径", + ) + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if not 0 <= args.wait_seconds <= 30: + parser.error("--wait-seconds 必须在 0 到 30 之间") + try: + result = run_diagnostic( + args.device, args.database, args.wait_seconds + ) + except Exception as exc: + print(f"测试未执行：{exc}", file=sys.stderr) + return 2 + + print(f"测试网址：{TEST_URL}") + print(f"当前包名：{result.current_package or '未知'}") + print(f"页面分类：{result.category}") + print(f"判断说明：{result.detail}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())