feat(cross_end): add driver-to-settlement E2E test with SQL retry

- Add DriverFreightFlowPage (Android): 28-step flow covering accept/load/unload
- Add FreightAccountingPage (Web): accounting/audit/finance-payment/shipper-payment
- Add db_helper: update_risk_control_status via pymysql
- Fix ensure_logged_in: coordinate tap on 'mine' tab + login check
- Add test_driver_to_settlement.py: full cross-end E2E with SQL retry

Co-authored-by: xst <xst@ntocc.com>

requirements.txt

@@ -5,3 +5,4 @@ pyantiword
 playwright>=1.40
 Pillow>=10.0          # 截图像素对比 (result-reporter Agent)
 # Appium-Python-Client>=3.0  # 取消注释以启用移动端自动化
+pymysql>=1.0  # 数据库操作（测试中更新风控状态等）

tests/app/_shared.py

@@ -54,11 +54,14 @@ def make_ensure_logged_in(
 ):
     """创建一个 ensure_logged_in 闭包。
 
+    策略：坐标点击底部"我的"tab（第4个图标）→ 检查"登录/注册"文本。
+    无法通过 resourceId 或 text 定位底部导航时用坐标点击兜底。
+
     Args:
         section_config: App 端配置
         config: 全局配置
         account_role: 默认账号角色名（如 "driver"、"shipper"）
-        home_indicator: 首页元素的 accessibility_id
+        home_indicator: 未使用，保留兼容
     """
     from selenium.webdriver.support.ui import WebDriverWait
     from selenium.webdriver.support import expected_conditions as EC
@@ -68,42 +71,87 @@ def make_ensure_logged_in(
         skip_login = section_config.get("skip_login", True)
         wait = WebDriverWait(driver, timeout=timeout_s)
 
-        try:
-            wait.until(
-                EC.visibility_of_element_located((AppiumBy.ACCESSIBILITY_ID, home_indicator))
-            )
-            return
-        except Exception:
-            pass
+        driver.implicitly_wait(3)
 
+        # 坐标点击底部"我的"tab（4 tab 均分宽度，第4个中心 x≈0.875*w, y≈h-50）
         try:
-            login_input = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "login_phone_input")
-            if login_input.is_displayed():
+            w = driver.get_window_size()["width"]
+            h = driver.get_window_size()["height"]
+            driver.tap([(int(w * 0.875), h - 50)])
+            driver.implicitly_wait(1)
+        except Exception as e:
+            raise RuntimeError(f"无法点击底部导航栏: {e}")
+
+        # 判断登录态：检查"登录/注册"文本
+        login_check = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
+        try:
+            el = driver.find_element(*login_check)
+            if el.is_displayed():
                 if skip_login:
                     raise RuntimeError(
-                        "APP 在登录页，未检测到登录态！请在设备上手动完成一次登录（含验证码），"
+                        "APP 未登录！请在设备上手动完成一次登录（含验证码），"
                         "之后重新运行用例即可。\n如需自动化登录，请设置 skip_login: false。"
                     )
-                account = config["accounts"].get(account_role, {})
-                login_input.send_keys(account.get("phone", ""))
-                driver.find_element(
-                    AppiumBy.ACCESSIBILITY_ID, "login_password_input"
-                ).send_keys(account.get("password", ""))
-                driver.find_element(AppiumBy.ACCESSIBILITY_ID, "login_submit_btn").click()
-                wait.until(
-                    EC.visibility_of_element_located((AppiumBy.ACCESSIBILITY_ID, home_indicator))
-                )
+                _perform_ui_login(driver, config["accounts"].get(account_role, {}), wait)
                 return
         except RuntimeError:
             raise
+        except Exception:
+            pass  # 没找到"登录/注册" → 已登录
+
+        # 回到首页
+        home_tab = (
+            AppiumBy.ANDROID_UIAUTOMATOR,
+            'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)',
+        )
+        try:
+            wait.until(EC.visibility_of_element_located(home_tab))
+            driver.find_element(*home_tab).click()
         except Exception:
             pass
 
-        raise RuntimeError("无法确定 APP 的登录状态！请确保 APP 已启动且首页或登录页可见。")
-
     return _ensure_logged_in
 
 
+def _perform_ui_login(driver, account: dict, wait) -> None:
+    """执行 UI 自动登录（skip_login: false 时使用）。"""
+    from appium.webdriver.common.appiumby import AppiumBy
+
+    phone = account.get("phone", "")
+    password = account.get("password", "")
+
+    # 点击登录入口
+    login_entry = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
+    wait.until(lambda d: d.find_element(*login_entry))
+    driver.find_element(*login_entry).click()
+    driver.implicitly_wait(1)
+
+    # 填写手机号
+    phone_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_phone")
+    wait.until(lambda d: d.find_element(*phone_input))
+    driver.find_element(*phone_input).send_keys(phone)
+
+    # 填写密码
+    password_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_pwd")
+    driver.find_element(*password_input).send_keys(password)
+
+    # 点击登录
+    login_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_login")
+    driver.find_element(*login_btn).click()
+    driver.implicitly_wait(2)
+
+    # 回到首页
+    home_tab = (
+        AppiumBy.ANDROID_UIAUTOMATOR,
+        'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)',
+    )
+    try:
+        wait.until(lambda d: d.find_element(*home_tab))
+        driver.find_element(*home_tab).click()
+    except Exception:
+        pass
+
+
 def save_failure_screenshot(driver, section_config, platform, request):
     """失败自动截图（各端 conftest teardown 中调用）。"""
     if (

tests/cross_end/README.md

@@ -0,0 +1,133 @@
+# 跨端 E2E 测试 — 货运全流程
+
+## 概述
+
+`tests/cross_end/` 目录包含跨越多个端（管理端 Web + 司机安卓 APP）的端到端自动化测试，
+验证从"发布货源"到"平台打款"的**完整业务闭环**。
+
+```
+  [管理端 Web]                    [司机安卓 APP]
+       │                                │
+  ① 发布货源 ──────────────────────────→│
+       │                          ② 接单 │
+       │                          ③ 装货 │
+       │                          ④ 卸货 │
+       │←────────────────────────────── │
+  ⑤ 核算运单                            │
+  ⑥ 核算审核                            │
+  ⑦ 货主打款                            │
+  ⑧ 平台打款给司机                      │
+```
+
+## 前置条件
+
+### 管理端
+
+```bash
+# 1. 保存管理员登录态 (仅需一次)
+python tests/web/admin/save_auth.py --role super_admin --browser chromium
+```
+
+### 司机安卓 APP
+
+| 条件 | 说明 |
+|---|---|
+| Appium Server | 启动: `appium` (默认 http://localhost:4723) |
+| Android 模拟器/真机 | 已启动并通过 `adb devices` 确认 |
+| 司机 APP | 已安装，包名 `com.arpa.ynchenggangdriver` |
+| 登录态 | 在设备上**手动登录一次** (后续 `noReset=true` 自动复用) |
+
+### Python 依赖
+
+```bash
+pip install playwright appium-python-client
+playwright install chromium
+```
+
+## 运行
+
+```bash
+# 运行跨端测试
+pytest tests/cross_end/ -m cross_end -v -s
+
+# 仅收集测试（不执行，验证 fixture 链正确）
+pytest tests/cross_end/ --collect-only
+
+# 排除跨端测试（常规运行不影响）
+pytest tests/ -m "not cross_end"
+```
+
+## 测试文件结构
+
+```
+tests/cross_end/
+├── conftest.py                           # 双 driver fixture (admin_page + app_driver)
+├── pages/
+│   ├── admin/
+│   │   ├── freight_source_page.py        # 发布货源 Page Object
+│   │   ├── waybill_management_page.py    # 运单核算 + 审核 Page Object
+│   │   └── settlement_page.py           # 货主打款 + 平台打款 Page Object
+│   └── app/
+│       ├── driver_home_page.py           # 司机首页 (复用现有)
+│       ├── driver_order_list_page.py     # 订单列表 + 接单 Page Object
+│       └── driver_transport_page.py     # 装货 + 卸货 Page Object
+├── tests/
+│   └── test_freight_full_flow.py        # 主测试: 8 步全流程
+└── README.md
+```
+
+## 选择器 `⚠️` 标记
+
+所有新 Page Object 中的定位器 (CSS 选择器 / accessibility_id) 都标有 `# ⚠️`，
+因为无法从代码静态分析中确定实际 DOM 结构或 APP 元素 ID。
+
+**替换步骤：**
+
+### Web 管理端选择器
+
+1. 启动管理端并登录
+2. 打开浏览器 DevTools (F12)
+3. 逐个导航到对应页面（货源管理、运单管理、承运结算）
+4. 检查每个表单输入框的 `placeholder`、按钮的 `text`、弹窗的 CSS class
+5. 替换 Page Object 中对应的 `# ⚠️` 选择器
+
+### APP 端 accessibility_id
+
+1. 启动 Appium Desktop 并连接到设备
+2. 使用 Inspector 功能捕获屏幕元素
+3. 找到每个按钮/输入框的 `accessibility-id` 或 `resource-id`
+4. 替换 Page Object 中对应的 `# ⚠️` 选择器
+
+## 配置
+
+所有环境配置从 `tests/config/test_config.yaml` 读取，跨端测试使用的 section:
+
+| Section | 用途 |
+|---|---|
+| `admin` | 管理端 base_url, viewport, timeout, 登录态路径 |
+| `driver_android` | Appium host, device_name, app_package, no_reset |
+| `accounts.super_admin` | 管理员手机号/密码 |
+| `accounts.driver` | 司机手机号/密码 |
+
+不需要新增配置项。
+
+## 故障排查
+
+| 现象 | 可能原因 | 解决方案 |
+|---|---|---|
+| `pytest.skip: Playwright 未安装` | 环境缺少 playwright | `pip install playwright && playwright install chromium` |
+| `pytest.skip: Appium-Python-Client 未安装` | 环境缺少 appium | `pip install appium-python-client` |
+| 登录态已过期 | storage_state 过期 | `python tests/web/admin/save_auth.py --role super_admin` |
+| APP 在登录页 | noReset 未生效或首次运行 | 在设备上手动登录一次 |
+| 无法连接到 Appium server | Appium 未启动 | 终端执行 `appium` |
+| AdbError / 设备未找到 | 模拟器未启动 | `adb devices` 确认设备在线 |
+| 订单在 APP 中找不到 | 订单同步延迟 | 在 Page Object 中增加 retry 轮询逻辑 |
+| `# ⚠️` 选择器错误 | 未更新实际选择器 | 按照上方步骤替换选择器 |
+
+## 已知限制
+
+- **仅 chromium**: 管理端不跑 browser matrix，固定使用 chromium
+- **仅 Android**: APP 端目前固定为 `driver_android`，不跑 iOS
+- **创建真实数据**: 测试会在环境中创建真实订单，需定期清理或使用测试环境
+- **无 API 清理**: 测试失败后不会自动取消/删除订单，需手动处理
+- **同时需要两个环境**: 管理端 + Appium + Android 设备必须同时可用，缺一则 skip

tests/cross_end/__init__.py

@@ -0,0 +1 @@
+# cross_end package

tests/cross_end/conftest.py

@@ -0,0 +1,197 @@
+"""跨端 E2E 测试 fixture — 同时持有 Playwright (管理端) 和 Appium (司机安卓) 两个 driver。
+
+核心挑战:
+    pytest 的单端 conftest 各自拥有 page / driver fixture，
+    跨端测试需要同时注入两者而不冲突。
+    解决方案: 使用显式命名的 fixture (admin_page / app_driver)，
+    各自独立创建/销毁生命周期。
+
+前置条件:
+    - 管理端: python tests/web/admin/save_auth.py --role super_admin --browser chromium
+    - 司机安卓: 设备上手动登录一次 (noReset=true 自动复用)
+    - Appium server: appium (默认 http://localhost:4723)
+
+运行:
+    pytest tests/cross_end/ -m cross_end -v
+"""
+
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+# ═══════════════════════════════════════════════════════════════════
+# Admin Page (Playwright)
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.fixture
+def admin_page(admin_config, request):
+    """管理端 Playwright Page — 仅 chromium，加载 super_admin 登录态。
+
+    与 tests/web/admin/conftest.py 的 page fixture 核心逻辑一致，
+    但不参数化浏览器 (跨端测试已经够慢，不必跑 browser matrix)。
+    """
+    try:
+        from playwright.sync_api import sync_playwright
+    except ImportError:
+        pytest.skip("Playwright 未安装，跳过跨端 E2E 测试。")
+
+    viewport = admin_config.get("viewport", {"width": 1920, "height": 1080})
+    timeout = admin_config.get("timeout_seconds", 120) * 1000
+    storage_state_dir = Path(admin_config.get("storage_state_dir", "tests/web/auth"))
+    browser_name = "chromium"
+
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+
+        context_kwargs = {
+            "viewport": {"width": viewport["width"], "height": viewport["height"]},
+            "locale": "zh-CN",
+        }
+
+        # 加载 super_admin 登录态
+        auth_file = storage_state_dir / "super_admin_chromium.json"
+        if auth_file.exists():
+            context_kwargs["storage_state"] = str(auth_file)
+
+        context = browser.new_context(**context_kwargs)
+        page = context.new_page()
+        page.set_default_timeout(timeout)
+
+        yield page
+
+        # 失败自动截图
+        if hasattr(request.node, "rep_call") and request.node.rep_call.failed:
+            scenario_dir = SCREENSHOTS_DIR / request.node.name
+            scenario_dir.mkdir(parents=True, exist_ok=True)
+            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+            path = str(scenario_dir / f"FAIL_admin_chromium_{ts}.png")
+            try:
+                page.screenshot(path=path, full_page=True)
+            except Exception:
+                pass
+
+        context.close()
+        browser.close()
+
+
+# ═══════════════════════════════════════════════════════════════════
+# App Driver (Appium)
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.fixture
+def app_driver(driver_android_config, request):
+    """司机安卓端 Appium driver。
+
+    复用 tests.app._shared.create_driver() 和 save_failure_screenshot()。
+    """
+    try:
+        from tests.app._shared import create_driver, save_failure_screenshot
+    except ImportError:
+        pytest.skip("Appium-Python-Client 未安装，跳过跨端 E2E 测试。")
+
+    cfg = driver_android_config
+    d = None
+    try:
+        d = create_driver(cfg, platform="android")
+        yield d
+        save_failure_screenshot(d, cfg, "android", request)
+    finally:
+        if d is not None:
+            try:
+                d.quit()
+            except Exception:
+                pass
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Login helpers
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.fixture
+def ensure_admin_login(admin_config, admin_page, config):
+    """确保管理端已登录为 super_admin。
+
+    复用 tests/web/admin/conftest.py 的登录态验证逻辑，
+    但硬编码 role="super_admin" (跨端流程始终用超管)。
+    """
+
+    def _ensure_login(role: str = "super_admin") -> None:
+        account = config["accounts"].get(role)
+        if not account:
+            raise ValueError(f"未知角色 '{role}'，可用: {list(config['accounts'].keys())}")
+
+        base_url = admin_config["base_url"]
+        skip_login = admin_config.get("skip_login", True)
+
+        if skip_login:
+            admin_page.goto(base_url)
+            admin_page.wait_for_timeout(2000)
+            if "#/login" not in admin_page.url and "/login" not in admin_page.url:
+                return
+            raise RuntimeError(
+                f"登录态已过期！请运行: python tests/web/admin/save_auth.py --role {role}"
+            )
+
+        # skip_login: false — 实时 UI 登录
+        admin_page.goto(f"{base_url}/#/login")
+        admin_page.get_by_placeholder("请输入用户名").fill(account["phone"])
+        admin_page.get_by_placeholder("请输入密码").fill(account["password"])
+        admin_page.get_by_role("button", name="登录").click()
+        admin_page.wait_for_timeout(3000)
+        if "#/login" in admin_page.url or "/login" in admin_page.url:
+            raise RuntimeError(
+                "登录失败！可能存在图形验证码。请改为手动登录: python tests/web/admin/save_auth.py"
+            )
+
+    return _ensure_login
+
+
+@pytest.fixture
+def ensure_driver_login(driver_android_config, config):
+    """确保司机安卓端已登录。
+
+    复用 tests.app._shared.make_ensure_logged_in() 工厂。
+    """
+    from tests.app._shared import make_ensure_logged_in
+
+    fn = make_ensure_logged_in(driver_android_config, config, account_role="driver")
+    return lambda d: fn(d)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Shared context
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.fixture
+def cross_end_context() -> dict:
+    """跨端流程共享上下文 — 各步骤通过此 dict 传递数据。
+
+    写入:
+        context["order_no"] = "TEST20260721..."
+        context["load_confirm"] = True
+
+    读取:
+        order_no = context["order_no"]
+    """
+    return {
+        "order_no": "",
+        "waybill_no": "",
+        "load_confirm": False,
+        "unload_confirm": False,
+        "accounting_done": False,
+        "review_done": False,
+    }
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Report hook — 让 request.node.rep_call.failed 可用于失败截图
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    rep = outcome.get_result()
+    setattr(item, f"rep_{rep.when}", rep)

tests/cross_end/pages/__init__.py

@@ -0,0 +1 @@
+# cross_end pages package

tests/cross_end/pages/admin/__init__.py

@@ -0,0 +1 @@
+# cross_end admin pages package

tests/cross_end/pages/admin/freight_accounting_page.py

@@ -0,0 +1,279 @@
+"""管理端 Page Object — 运费核算 + 运单审核 + 垫资打款 完整流程。
+
+基于 resources/selectors/正常流程-web端.md 中的元素定位。
+使用直接 hash URL 导航（比菜单点击更可靠），按钮选择器带 fallback 兜底。
+
+URL 路由:
+  运费核算:     #/tms/Carrier/freight
+  运单审核:     #/tms/Carrier/OrderDetailAudit
+  垫资-财务打款: #/tms/Carrier/dzFinancialapproval
+  垫资-货主打款: #/tms/Carrier/dzApplicationForPayment
+"""
+
+from playwright.sync_api import Page
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class FreightAccountingPage(BasePage):
+    """管理端 - 运费核算 / 运单审核 / 垫资打款。"""
+
+    def __init__(self, page: Page, admin_base_url: str = ""):
+        super().__init__(page)
+        self.admin_base_url = admin_base_url
+
+    # ════════════════════════════════════════════════════════════
+    # URL 导航
+    # ════════════════════════════════════════════════════════════
+
+    def _go(self, hash_route: str):
+        """直接 hash URL 跳转，比菜单点击稳定。"""
+        url = f"{self.admin_base_url}/#/{hash_route.lstrip('#').lstrip('/')}"
+        self.page.goto(url, wait_until="domcontentloaded")
+        self.page.wait_for_timeout(2000)
+
+    # ════════════════════════════════════════════════════════════
+    # 通用操作
+    # ════════════════════════════════════════════════════════════
+
+    def _fill_order_no(self, order_no: str):
+        """搜索栏输入运输单号（多个同名 input 时取第一个）。"""
+        input_el = self.page.locator('input[placeholder="请输入运输单号"]').first
+        self.wait_visible(input_el)
+        input_el.clear()
+        input_el.fill(order_no)
+
+    def _click_search(self):
+        """点击查询按钮。"""
+        btn = self.page.locator("button:has-text('查询')").first
+        self.wait_visible(btn)
+        btn.click()
+        self.page.wait_for_timeout(1000)
+
+    def _table_has_data(self) -> bool:
+        """检查查询结果是否有数据。"""
+        try:
+            self.page.wait_for_selector("table tbody tr", timeout=3_000)
+            return True
+        except Exception:
+            return False
+
+    def _click_table_action(self, button_index: int):
+        """点击表格固定列操作按钮（纯图标 <i> 按钮，selector 文档路径 + fallback）。
+
+        vxe-table 操作列按钮结构: td > div > button:nth-child(N) > span > i
+        """
+        # 精确路径（selector 文档原始定位）
+        sel = (
+            ".vxe-table--fixed-right-wrapper.scrolling--middle "
+            ".vxe-table--body-wrapper.fixed-right--wrapper "
+            "table > tbody > tr:nth-child(1) > td > div "
+            f"> button:nth-child({button_index}) > span > i"
+        )
+        # 宽松路径（当 DOM 结构略有变化时兜底）
+        sel_fallback = (
+            ".vxe-table--fixed-right-wrapper "
+            "table > tbody > tr:nth-child(1) "
+            f"> td:last-child > div > button:nth-child({button_index})"
+        )
+
+        for attempt in range(3):
+            try:
+                self.page.wait_for_selector("table tbody tr", timeout=5_000)
+                self.page.wait_for_timeout(300)
+                btn = self.page.locator(sel).first
+                if btn.count() == 0:
+                    btn = self.page.locator(sel_fallback).first
+                self.wait_visible(btn, timeout_ms=5_000)
+                btn.click()
+                self.page.wait_for_timeout(800)
+                return
+            except Exception:
+                if attempt == 2:
+                    raise
+                self.page.wait_for_timeout(1_000)
+
+    def _modal(self):
+        """返回当前可见弹窗内容容器（.first 避免 strict mode）。"""
+        return self.page.locator(
+            ".vxe-modal--wrapper.is--visible .vxe-modal--body .vxe-modal--content"
+        ).first
+
+    def _scroll_modal(self):
+        """滚动弹窗到底部。"""
+        modal = self._modal()
+        self.wait_visible(modal)
+        modal.evaluate("el => el.scrollTop = el.scrollHeight")
+        self.page.wait_for_timeout(500)
+
+    def _retry_click_modal_btn(self, locator, timeout_ms: int = 5_000,
+                               retries: int = 3, wait_after: int = 1500):
+        """弹窗内按钮带重试点击（每次重试重新获取元素，防 DOM 置换）。"""
+        for attempt in range(retries):
+            el = locator() if callable(locator) else locator
+            try:
+                self.wait_visible(el, timeout_ms=timeout_ms)
+                el.click()
+                self.page.wait_for_timeout(wait_after)
+                return
+            except Exception:
+                if attempt == retries - 1:
+                    raise
+                self.page.wait_for_timeout(1_000)
+
+    # ════════════════════════════════════════════════════════════
+    # 运费核算
+    # ════════════════════════════════════════════════════════════
+
+    def do_freight_accounting(self, order_no: str, retry_sql=None):
+        """运费核算：导航→搜索→（无数据时 SQL 重试）→编辑→核算。
+
+        Args:
+            order_no: 运输单号
+            retry_sql: 可调用对象，首次查询无数据时调用
+        """
+        self._go("tms/Carrier/freight")
+        self._fill_order_no(order_no)
+        self._click_search()
+
+        # 无数据 → 执行 SQL 重试
+        if not self._table_has_data():
+            if retry_sql:
+                retry_sql()
+            self._fill_order_no(order_no)
+            self._click_search()
+
+        if not self._table_has_data():
+            raise RuntimeError(f"运费核算查询无数据！单号: {order_no}")
+
+        # 点击编辑 (button_index=4, 来自 selector 文档 Step 3)
+        self._click_table_action(button_index=4)
+
+        # 核算按钮 (selector 文档 Step 4)
+        btn = self._modal().locator(
+            "form > div:nth-child(4) button:nth-child(4) span"
+        ).first
+        self._retry_click_modal_btn(btn, wait_after=1500)
+        self.screenshot("freight_accounting_done")
+
+    # ════════════════════════════════════════════════════════════
+    # 运单审核
+    # ════════════════════════════════════════════════════════════
+
+    def do_waybill_audit(self, order_no: str):
+        """运单审核：导航→搜索→审核→滚动弹窗→审核通过。"""
+        self._go("tms/Carrier/OrderDetailAudit")
+        self._fill_order_no(order_no)
+        self._click_search()
+
+        # 点击审核 (button_index=2, selector 文档 Step 10)
+        self._click_table_action(button_index=2)
+
+        # 滚动弹窗 (selector 文档 Step 11)
+        self._scroll_modal()
+
+        # 审核通过 (selector 文档 Step 12)
+        btn = self._modal().locator(
+            "._btns > button.el-button.el-button--primary.el-button--medium > span"
+        ).first
+        self._retry_click_modal_btn(btn, wait_after=1500)
+        self.screenshot("waybill_audit_done")
+
+    # ════════════════════════════════════════════════════════════
+    # 垫资-财务打款
+    # ════════════════════════════════════════════════════════════
+
+    def do_finance_payment(self, order_no: str, verify_code: str = "1234"):
+        """垫资-财务打款：导航→搜索→在线支付→滚动→勾选光大/手机→验证码→确认。"""
+        self._go("tms/Carrier/dzFinancialapproval")
+        self._fill_order_no(order_no)
+        self._click_search()
+
+        # 在线支付 (button_index=2, selector 文档 Step 18)
+        self._click_table_action(button_index=2)
+        self._scroll_modal()
+
+        # 勾选光大银行 (selector 文档 Step 20)
+        try:
+            radio = self._modal().locator(
+                "form > div:nth-child(9) > div > div:nth-child(2) "
+                "> label > span.el-radio__input > span"
+            ).first
+            self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300)
+        except Exception:
+            pass
+
+        # 勾选手机号 (selector 文档 Step 21)
+        try:
+            radio = self.page.locator(
+                "[id^='el-collapse-content-'] > div > div:nth-child(3) "
+                "> label > span.el-radio__input > span"
+            ).first
+            self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300)
+        except Exception:
+            pass
+
+        # 获取验证码 (selector 文档 Step 22)
+        try:
+            btn = self._modal().locator(
+                "form > div.el-row > div.el-col.el-col-4 button span"
+            ).first
+            self._retry_click_modal_btn(btn, timeout_ms=3_000, wait_after=500)
+        except Exception:
+            pass
+
+        # 输入验证码 (selector 文档 Step 23)
+        try:
+            inp = self._modal().locator(
+                "form > div.el-row > div.el-col.el-col-20 input"
+            ).first
+            self.wait_visible(inp)
+            inp.clear()
+            inp.fill(verify_code)
+        except Exception:
+            pass
+
+        # 确认打款 (selector 文档 Step 24)
+        btn = self._modal().locator(
+            "form > div:nth-child(12) > div > button span"
+        ).first
+        self._retry_click_modal_btn(btn, wait_after=1500)
+        self.screenshot("finance_payment_done")
+
+    # ════════════════════════════════════════════════════════════
+    # 垫资-货主打款
+    # ════════════════════════════════════════════════════════════
+
+    def do_shipper_payment(self, order_no: str):
+        """垫资-货主打款：导航→搜索→打款→勾选光大→确认打款→确认弹窗。"""
+        self._go("tms/Carrier/dzApplicationForPayment")
+        self._fill_order_no(order_no)
+        self._click_search()
+
+        # 打款 (button_index=2, selector 文档 Step 30)
+        self._click_table_action(button_index=2)
+
+        # 勾选光大 (selector 文档 Step 31)
+        try:
+            radio = self._modal().locator(
+                "form > div > div:nth-child(3) > div > div > div:nth-child(3) "
+                "> label > span.el-radio__label"
+            ).first
+            self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300)
+        except Exception:
+            pass
+
+        # 确认打款 (selector 文档 Step 31-2)
+        btn = self._modal().locator("> div > div > button > span").first
+        self._retry_click_modal_btn(btn, wait_after=1000)
+
+        # message-box 确认 (selector 文档 Step 32)
+        try:
+            msg = self.page.locator(
+                "body > div.el-message-box__wrapper .el-message-box__btns "
+                "> button.el-button.el-button--default.el-button--small.el-button--primary > span"
+            )
+            self._retry_click_modal_btn(msg, timeout_ms=5_000, wait_after=1500)
+        except Exception:
+            pass
+        self.screenshot("shipper_payment_done")

tests/cross_end/pages/admin/freight_source_page.py

@@ -0,0 +1,115 @@
+"""管理端 — 货源管理 — 发布货源 Page Object。
+
+Usage:
+    freight_page = FreightSourcePage(page)
+    freight_page.fill_freight_form(...)
+    freight_page.submit()
+    order_no = freight_page.get_published_order_no()
+"""
+
+from playwright.sync_api import Page
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class FreightSourcePage(BasePage):
+    """管理端 - 货源管理 - 发布货源。"""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+        # ⚠️ 以下定位器需要根据实际 DOM 调整
+        # 通过 Playwright Inspector 或浏览器 DevTools 获取真实选择器
+
+        # ── 货源表单字段 ──
+        self.goods_name_input = page.get_by_placeholder("请输入货物名称")           # ⚠️
+        self.weight_input = page.get_by_placeholder("请输入重量")                  # ⚠️
+        self.load_address_input = page.get_by_placeholder("请选择装货地")           # ⚠️
+        self.unload_address_input = page.get_by_placeholder("请选择卸货地")         # ⚠️
+        self.price_input = page.get_by_placeholder("请输入运费")                   # ⚠️
+
+        # ── 操作按钮 ──
+        self.submit_btn = page.get_by_role("button", name="发布")                  # ⚠️
+        self.confirm_btn = page.get_by_role("button", name="确认")                 # ⚠️
+
+        # ── 结果元素 ──
+        self.success_toast = page.locator(".el-message--success")                  # ⚠️
+        self.order_no_text = page.locator(".order-no, .freight-no")               # ⚠️ 发布成功后显示订单号的位置
+
+    def fill_freight_form(
+        self,
+        goods_name: str,
+        weight: float,
+        load_addr: str,
+        unload_addr: str,
+    ) -> None:
+        """填写货源发布表单。
+
+        Args:
+            goods_name: 货物名称
+            weight: 货物重量 (吨)
+            load_addr: 装货地址
+            unload_addr: 卸货地址
+        """
+        # ⚠️ 具体字段依赖实际表单结构，以下为通用模式
+        self.fill_field(self.goods_name_input, goods_name)
+        self.fill_field(self.weight_input, str(weight))
+        self.fill_field(self.load_address_input, load_addr)
+        self.fill_field(self.unload_address_input, unload_addr)
+        # 如有运费字段:
+        # self.fill_field(self.price_input, "1000")
+
+    def submit(self) -> None:
+        """点击发布按钮。"""
+        self.click_when_ready(self.submit_btn)
+        self.page.wait_for_timeout(1000)
+
+    def confirm_publish(self) -> None:
+        """确认发布弹窗（如果有二次确认）。"""
+        # ⚠️ 如果发布时有确认弹窗
+        try:
+            self.click_when_ready(self.confirm_btn)
+            self.page.wait_for_timeout(500)
+        except Exception:
+            pass  # 无确认弹窗则跳过
+
+    def get_published_order_no(self) -> str:
+        """发布成功后获取生成的订单号。
+
+        策略 (按优先级尝试):
+          1. 从成功提示文本中提取
+          2. 从跳转后的详情页 URL 中提取
+          3. 从列表页第一条提取
+
+        Returns:
+            订单号字符串
+
+        Raises:
+            RuntimeError: 无法获取订单号
+        """
+        # ⚠️ 策略取决于系统实际行为
+        # 策略1: 等待成功 toast
+        self.wait_visible(self.success_toast, timeout_ms=10_000)
+        toast_text = self.success_toast.inner_text()
+        self.page.wait_for_timeout(500)
+
+        # 策略2: 从当前页面提取订单号文本
+        if self.order_no_text.is_visible():
+            return self.order_no_text.inner_text().strip()
+
+        # 策略3: 从 URL 提取
+        import re
+        match = re.search(r"orderNo=(\w+)", self.page.url)
+        if match:
+            return match.group(1)
+
+        # 策略4: 从成功 toast 文本中提取数字编号
+        match = re.search(r"[A-Z]{2,6}\d{10,20}", toast_text)
+        if match:
+            return match.group(0)
+
+        raise RuntimeError(
+            f"无法获取发布的订单号。\n"
+            f"当前 URL: {self.page.url}\n"
+            f"Toast 文本: {toast_text}\n"
+            f"请更新 get_published_order_no() 中的选择器。"
+        )

tests/cross_end/pages/admin/settlement_page.py

@@ -0,0 +1,88 @@
+"""管理端 — 承运结算 — 打款 Page Object。
+
+覆盖步骤:
+    7. 货主打款 (do_shipper_payment)
+    8. 平台打款给司机 (do_platform_payment)
+
+Usage:
+    settlement_page = SettlementPage(page)
+    settlement_page.do_shipper_payment(order_no)
+    settlement_page.do_platform_payment(order_no)
+"""
+
+from playwright.sync_api import Page
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class SettlementPage(BasePage):
+    """管理端 - 承运结算 - 打款管理。"""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+        # ⚠️ 以下定位器需要根据实际 DOM 调整
+
+        # ── 搜索区 ──
+        self.search_order_input = page.get_by_placeholder("请输入订单号")            # ⚠️
+        self.search_btn = page.get_by_role("button", name="查询")                  # ⚠️
+
+        # ── 操作按钮 ──
+        self.shipper_pay_btn = page.get_by_role("button", name="货主打款")         # ⚠️
+        self.platform_pay_btn = page.get_by_role("button", name="平台打款")        # ⚠️ 或"财务打款"
+        self.confirm_btn = page.get_by_role("button", name="确定")                 # ⚠️ 确认打款弹窗
+
+        # ── 结果元素 ──
+        self.success_toast = page.locator(".el-message--success")                  # ⚠️
+        self.status_text = page.locator(".settlement-status")                     # ⚠️ 结算状态文本
+
+    def search_order(self, order_no: str) -> None:
+        """搜索运单。"""
+        self.fill_field(self.search_order_input, order_no)
+        self.click_when_ready(self.search_btn)
+        self.page.wait_for_timeout(1000)
+
+    def do_shipper_payment(self, order_no: str) -> None:
+        """货主打款流程: 搜索 → 货主打款 → 确认。
+
+        Args:
+            order_no: 订单号
+        """
+        self.search_order(order_no)
+        self.page.wait_for_timeout(500)
+
+        self.click_when_ready(self.shipper_pay_btn)
+        self.page.wait_for_timeout(500)
+
+        # 确认打款弹窗
+        self._confirm_if_present()
+
+        # 等待成功提示
+        self.wait_visible(self.success_toast, timeout_ms=10_000)
+
+    def do_platform_payment(self, order_no: str) -> None:
+        """平台打款给司机流程: 搜索 → 平台打款 → 确认。
+
+        Args:
+            order_no: 订单号
+        """
+        self.search_order(order_no)
+        self.page.wait_for_timeout(500)
+
+        self.click_when_ready(self.platform_pay_btn)
+        self.page.wait_for_timeout(500)
+
+        # 确认打款弹窗
+        self._confirm_if_present()
+
+        # 等待成功提示
+        self.wait_visible(self.success_toast, timeout_ms=10_000)
+
+    def _confirm_if_present(self) -> None:
+        """如果有确认弹窗则点击确定。"""
+        # ⚠️ 确认按钮可能以不同形式存在
+        try:
+            if self.confirm_btn.is_visible():
+                self.click_when_ready(self.confirm_btn)
+                self.page.wait_for_timeout(500)
+        except Exception:
+            pass

tests/cross_end/pages/admin/waybill_management_page.py

@@ -0,0 +1,98 @@
+"""管理端 — 运输管理 — 运单核算 Page Object。
+
+覆盖步骤:
+    5. 核算运单 (do_accounting)
+    6. 核算审核 (do_review)
+
+Usage:
+    waybill_page = WaybillManagementPage(page)
+    waybill_page.do_accounting(order_no)
+    waybill_page.do_review(order_no)
+"""
+
+from playwright.sync_api import Page
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class WaybillManagementPage(BasePage):
+    """管理端 - 运输管理 - 运单管理/核算。"""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+        # ⚠️ 以下定位器需要根据实际 DOM 调整
+
+        # ── 搜索区 ──
+        self.search_order_input = page.get_by_placeholder("请输入订单号")            # ⚠️
+        self.search_btn = page.get_by_role("button", name="查询")                  # ⚠️
+
+        # ── 操作按钮 ──
+        self.accounting_btn = page.get_by_role("button", name="核算")              # ⚠️ 核算运单
+        self.review_btn = page.get_by_role("button", name="审核")                  # ⚠️ 核算审核
+        self.confirm_btn = page.get_by_role("button", name="确定")                 # ⚠️ 确认弹窗
+
+        # ── 结果元素 ──
+        self.success_toast = page.locator(".el-message--success")                  # ⚠️
+        self.table_first_row = page.locator(".el-table__body tr:first-child")     # ⚠️ 列表第一行
+
+    def search_order(self, order_no: str) -> None:
+        """搜索运单。"""
+        self.fill_field(self.search_order_input, order_no)
+        self.click_when_ready(self.search_btn)
+        self.page.wait_for_timeout(1000)
+
+    def click_accounting(self) -> None:
+        """点击核算按钮。"""
+        self.click_when_ready(self.accounting_btn)
+        self.page.wait_for_timeout(500)
+
+    def click_review(self) -> None:
+        """点击审核按钮（核算审核）。"""
+        self.click_when_ready(self.review_btn)
+        self.page.wait_for_timeout(500)
+
+    def confirm_action(self) -> None:
+        """确认操作弹窗。"""
+        # ⚠️ 可能没有确认弹窗，try 一下
+        try:
+            if self.confirm_btn.is_visible():
+                self.click_when_ready(self.confirm_btn)
+                self.page.wait_for_timeout(500)
+        except Exception:
+            pass
+
+    def do_accounting(self, order_no: str) -> None:
+        """完整的核算运单流程: 搜索 → 核算 → 确认。
+
+        Args:
+            order_no: 订单号
+        """
+        self.search_order(order_no)
+        self.page.wait_for_timeout(500)
+        self.click_accounting()
+        self.confirm_action()
+
+        # 等待核算完成
+        self.has_success_toast()
+
+    def do_review(self, order_no: str) -> None:
+        """完整的核算审核流程: 搜索 → 审核 → 确认。
+
+        Args:
+            order_no: 订单号
+        """
+        self.search_order(order_no)
+        self.page.wait_for_timeout(500)
+        self.click_review()
+        self.confirm_action()
+
+        # 等待审核完成
+        self.has_success_toast()
+
+    def has_success_toast(self) -> bool:
+        """检查操作是否成功（有成功 toast）。"""
+        try:
+            self.wait_visible(self.success_toast, timeout_ms=5_000)
+            return True
+        except Exception:
+            return False

tests/cross_end/pages/app/__init__.py

@@ -0,0 +1 @@
+# cross_end app pages package

tests/cross_end/pages/app/driver_freight_flow_page.py

@@ -0,0 +1,212 @@
+"""司机 APP 货运流程 Page Object — 货源→接单→装货→卸货→获取单号。
+
+基于 resources/selectors/正常流程-司机.md 中的元素定位。
+使用 Appium UiAutomator2 定位策略。
+"""
+
+from appium.webdriver.common.appiumby import AppiumBy
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+from tests.app._base_page import AppBasePage
+
+
+class DriverFreightFlowPage(AppBasePage):
+    """司机端 APP - 货运完整流程（货源→接单→装货→卸货→获取单号）。"""
+
+    def __init__(self, driver, platform: str = "android"):
+        super().__init__(driver, platform)
+
+        # ── 底部导航 ──
+        # 0=首页, 1=货源, 2=消息, 3=我的
+        self.freight_source_tab = (
+            AppiumBy.ANDROID_UIAUTOMATOR,
+            'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(1)',
+        )
+
+        # ── 货源单列表 ──
+        self.freight_order_item = (
+            AppiumBy.ANDROID_UIAUTOMATOR,
+            'new UiSelector().className("android.widget.RelativeLayout").instance(3)',
+        )
+
+        # ── 数量 ──
+        self.quantity_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/et_num")
+
+        # ── 协议勾选 ──
+        self.agreement_checkbox = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/check_xieyi")
+
+        # ── 接单 ──
+        self.accept_order_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_bian")
+
+        # ── 系统弹窗取消 ──
+        self.cancel_btn = (AppiumBy.ID, "android:id/button2")
+
+        # ── 吨位 ──
+        self.weight_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/et_dun")
+
+        # ── 上传图片 ──
+        self.select_image_btn = (
+            AppiumBy.ANDROID_UIAUTOMATOR,
+            'new UiSelector().className("android.widget.ImageView").instance(2)',
+        )
+        self.album_btn = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("相册")')
+        self.image_checkbox = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/cb_check")
+        self.done_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/btn_ok")
+
+        # ── 装货/卸货确认 ──
+        self.load_unload_confirm = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_nick")
+
+        # ── 合同 ──
+        self.contract_webview = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/textHtml")
+        self.acknowledge_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_affirm")
+        self.confirm_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ok_tv3")
+
+    # ════════════════════════════════════════════════════════════
+    # 基础操作封装
+    # ════════════════════════════════════════════════════════════
+
+    def _wait_and_click(self, locator, timeout_s: float = 10.0):
+        wait = WebDriverWait(self.driver, timeout=timeout_s)
+        wait.until(EC.visibility_of_element_located(locator))
+        self.driver.find_element(*locator).click()
+
+    def _wait_and_fill(self, locator, value: str, timeout_s: float = 10.0):
+        wait = WebDriverWait(self.driver, timeout=timeout_s)
+        wait.until(EC.visibility_of_element_located(locator))
+        el = self.driver.find_element(*locator)
+        el.clear()
+        el.send_keys(value)
+
+    def _tap_by_text(self, text: str, timeout_s: float = 10.0):
+        locator = (AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().text("{text}")')
+        self._wait_and_click(locator, timeout_s)
+
+    def _safe_tap_by_text(self, text: str, timeout_s: float = 5.0):
+        """容错点击：元素不存在时静默跳过。"""
+        try:
+            self._tap_by_text(text, timeout_s)
+        except Exception:
+            pass
+
+    def _swipe_up(self, locator=None, duration_ms: int = 800):
+        """向上滑动。指定 locator 时在元素内滑动，否则全屏滑动。"""
+        if locator:
+            try:
+                wait = WebDriverWait(self.driver, timeout=5.0)
+                el = wait.until(EC.presence_of_element_located(locator))
+                size = el.size
+                loc = el.location
+                start_x = loc["x"] + size["width"] // 2
+                start_y = loc["y"] + int(size["height"] * 0.8)
+                end_y = loc["y"] + int(size["height"] * 0.2)
+                self.driver.swipe(start_x, start_y, start_x, end_y, duration_ms)
+                return
+            except Exception:
+                pass
+        # 全屏滑动
+        size = self.driver.get_window_size()
+        self.driver.swipe(
+            start_x=size["width"] // 2,
+            start_y=int(size["height"] * 0.7),
+            end_x=size["width"] // 2,
+            end_y=int(size["height"] * 0.3),
+            duration=duration_ms,
+        )
+
+    # ════════════════════════════════════════════════════════════
+    # 接单流程 (步骤 1-5)
+    # ════════════════════════════════════════════════════════════
+
+    def accept_order_flow(self, quantity: str = "1"):
+        """接单完整流程：货源 tab → 货源单 → 数量 → 协议 → 接单。"""
+        self._wait_and_click(self.freight_source_tab)     # ① 点击货源
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.freight_order_item)      # ② 点击货源单
+        self.driver.implicitly_wait(2)
+        self._wait_and_fill(self.quantity_input, quantity) # ③ 填写数量
+        self._wait_and_click(self.agreement_checkbox)      # ④ 勾选协议
+        self._wait_and_click(self.accept_order_btn)        # ⑤ 接单
+        self.driver.implicitly_wait(2)
+
+    # ════════════════════════════════════════════════════════════
+    # 装货流程 (步骤 6-17)
+    # ════════════════════════════════════════════════════════════
+
+    def load_goods_flow(self, weight: str = "10"):
+        """装货完整流程。"""
+        self._tap_by_text("装货")                           # ⑥ 点击装货
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.cancel_btn)               # ⑦ 点击取消（系统弹窗）
+        self.driver.implicitly_wait(1)
+        self._wait_and_fill(self.weight_input, weight)      # ⑧ 输入吨位
+        self._wait_and_click(self.select_image_btn)         # ⑨ 选择图片
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.album_btn)                # ⑩ 相册
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.image_checkbox)           # ⑪ 勾选图片
+        self._wait_and_click(self.done_btn)                 # ⑫ 完成
+        self.driver.implicitly_wait(2)
+        self._wait_and_click(self.agreement_checkbox)       # ⑬ 勾选协议
+        self._wait_and_click(self.load_unload_confirm)      # ⑭ 装货确认
+        self.driver.implicitly_wait(1)
+        self._swipe_up(self.contract_webview)               # ⑮ 滑动合同
+        self.driver.implicitly_wait(1)
+        self._safe_tap_by_text("已知晓", timeout_s=5.0)     # ⑯ 已知晓（容错）
+        self._safe_tap_by_text("确定", timeout_s=5.0)       # ⑰ 确定（容错）
+        self.driver.implicitly_wait(2)
+
+    # ════════════════════════════════════════════════════════════
+    # 卸货流程 (步骤 19-27)
+    # ════════════════════════════════════════════════════════════
+
+    def unload_goods_flow(self, weight: str = "10"):
+        """卸货完整流程。"""
+        self._tap_by_text("卸货")                           # ⑲ 点击卸货
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.cancel_btn)               # ⑳ 点击取消（系统弹窗）
+        self.driver.implicitly_wait(1)
+        self._wait_and_fill(self.weight_input, weight)      # ㉑ 输入吨位
+        self._wait_and_click(self.select_image_btn)         # ㉒ 选择图片
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.album_btn)                # ㉓ 相册
+        self.driver.implicitly_wait(1)
+        self._wait_and_click(self.image_checkbox)           # ㉔ 勾选图片
+        self._wait_and_click(self.done_btn)                 # ㉕ 完成
+        self.driver.implicitly_wait(2)
+        self._wait_and_click(self.load_unload_confirm)      # ㉖ 卸货确认
+        self.driver.implicitly_wait(1)
+        self._safe_tap_by_text("确定", timeout_s=5.0)       # ㉗ 确定（容错）
+        self.driver.implicitly_wait(2)
+
+    # ════════════════════════════════════════════════════════════
+    # 获取单号 (步骤 28)
+    # ════════════════════════════════════════════════════════════
+
+    def get_transport_order_no(self, timeout_s: float = 10.0) -> str:
+        """获取运输单号。从"运输单号：RHJC..."中正则提取纯单号。"""
+        import re
+
+        locator = (
+            AppiumBy.ANDROID_UIAUTOMATOR,
+            'new UiSelector().textContains("运输单号")',
+        )
+        wait = WebDriverWait(self.driver, timeout=timeout_s)
+        el = wait.until(EC.visibility_of_element_located(locator))
+        full_text = el.text
+        match = re.search(r"RHJC\S+", full_text)
+        if match:
+            return match.group()
+        return full_text.replace("运输单号：", "").replace("运输单号:", "").strip()
+
+    # ════════════════════════════════════════════════════════════
+    # 全流程
+    # ════════════════════════════════════════════════════════════
+
+    def full_driver_flow(self, quantity: str = "1", load_weight: str = "10",
+                         unload_weight: str = "10") -> str:
+        """执行完整司机端流程：接单→装货→卸货→返回运输单号。"""
+        self.accept_order_flow(quantity)
+        self.load_goods_flow(load_weight)
+        self.unload_goods_flow(unload_weight)
+        return self.get_transport_order_no()

tests/cross_end/pages/app/driver_home_page.py

@@ -0,0 +1,7 @@
+"""跨端测试 — 司机端首页 Page Object。
+
+直接复用 tests/app/driver_android/pages/driver_home_page.py。
+"""
+
+# re-export existing page object
+from tests.app.driver_android.pages.driver_home_page import DriverHomePage  # noqa: F401

tests/cross_end/pages/app/driver_order_list_page.py

@@ -0,0 +1,92 @@
+"""司机端 APP — 订单列表 + 接单 Page Object。
+
+Usage:
+    order_list = DriverOrderListPage(driver)
+    order_list.search_and_accept_order(order_no)
+"""
+
+from tests.app._base_page import AppBasePage
+
+
+class DriverOrderListPage(AppBasePage):
+    """司机端 APP - 订单列表 - 接单。"""
+
+    def __init__(self, driver, platform: str = "android"):
+        super().__init__(driver, platform)
+        # ⚠️ 以下 accessibility_id 需要根据实际 APP 元素 ID 调整
+        # 使用 Appium Desktop Inspector 或 uiautomatorviewer 获取
+
+        # ── 搜索区 ──
+        self.search_bar = "et_search"                       # ⚠️ 搜索框
+        self.search_input = "et_search_input"               # ⚠️ 搜索输入框
+        self.search_btn = "btn_search"                      # ⚠️ 搜索按钮
+
+        # ── 列表 ──
+        self.order_list_container = "rv_order_list"         # ⚠️ 订单列表容器
+        self.order_item_prefix = "order_item_"              # ⚠️ 订单项 ID 前缀
+        self.order_detail_btn = "btn_order_detail"          # ⚠️ 进入订单详情
+
+        # ── 接单 ──
+        self.accept_order_btn = "btn_accept_order"          # ⚠️ 接单按钮
+        self.confirm_accept_btn = "btn_confirm"             # ⚠️ 确认接单按钮
+        self.accept_success_text = "接单成功"               # ⚠️ 成功提示文本
+
+    def search_order(self, order_no: str) -> None:
+        """搜索指定订单。"""
+        # 如果搜索框是独立按钮需要先点击展开
+        try:
+            self.tap_by_id(self.search_bar)
+        except Exception:
+            pass
+
+        self.fill_by_id(self.search_input, order_no)
+        self.tap_by_id(self.search_btn)
+        self.driver.implicitly_wait(3)
+
+    def accept_order(self, order_no: str) -> None:
+        """接受指定订单。
+
+        Args:
+            order_no: 订单号
+        """
+        # 点击搜索结果中的订单
+        # ⚠️ 实际定位方式取决于列表项如何展示订单号
+        # 策略1: 通过订单号文本点击
+        self.tap_by_text(order_no)
+        self.driver.implicitly_wait(1)
+
+        # 策略2: 通过详情按钮进入
+        # self.tap_by_id(f"{self.order_item_prefix}{order_no}")
+        # self.tap_by_id(self.order_detail_btn)
+
+    def search_and_accept_order(self, order_no: str) -> None:
+        """搜索订单并接单 — 完整的接单流程。
+
+        Args:
+            order_no: 订单号
+        """
+        # 1. 搜索
+        self.search_order(order_no)
+
+        # 2. 进入订单详情
+        self.accept_order(order_no)
+
+        # 3. 点击接单
+        self.tap_by_id(self.accept_order_btn)
+        self.driver.implicitly_wait(1)
+
+        # 4. 确认接单 (如有弹窗)
+        try:
+            self.tap_by_id(self.confirm_accept_btn)
+        except Exception:
+            pass  # 无确认弹窗则跳过
+
+        self.driver.implicitly_wait(2)
+
+    def verify_order_accepted(self) -> bool:
+        """验证订单已被接受（成功提示出现）。"""
+        try:
+            self.wait_visible_by_text(self.accept_success_text, timeout_s=5)
+            return True
+        except Exception:
+            return False

tests/cross_end/pages/app/driver_transport_page.py

@@ -0,0 +1,132 @@
+"""司机端 APP — 运输中（装货/卸货） Page Object。
+
+Usage:
+    transport = DriverTransportPage(driver)
+    transport.load_goods(order_no)
+    transport.unload_goods(order_no)
+"""
+
+from tests.app._base_page import AppBasePage
+
+
+class DriverTransportPage(AppBasePage):
+    """司机端 APP - 运输中 - 装货/卸货。"""
+
+    def __init__(self, driver, platform: str = "android"):
+        super().__init__(driver, platform)
+        # ⚠️ 以下 accessibility_id 需要根据实际 APP 元素 ID 调整
+        # 使用 Appium Desktop Inspector 或 uiautomatorviewer 获取
+
+        # ── 导航 tab ──
+        self.my_orders_tab = "tab_orders"                   # ⚠️ 我的运单 tab
+
+        # ── 运单列表/详情 ──
+        self.order_item = "order_item"                      # ⚠️ 运单项
+        self.order_detail_btn = "btn_order_detail"          # ⚠️ 进入详情
+
+        # ── 装货 ──
+        self.load_goods_btn = "btn_load_goods"              # ⚠️ 装货按钮
+        self.confirm_load_btn = "btn_confirm"               # ⚠️ 确认装货
+        self.upload_load_photo_btn = "btn_upload_photo"     # ⚠️ 上传装货照片 (如有)
+        self.load_success_text = "装货成功"                 # ⚠️ 成功提示
+
+        # ── 卸货 ──
+        self.unload_goods_btn = "btn_unload_goods"          # ⚠️ 卸货按钮
+        self.confirm_unload_btn = "btn_confirm"             # ⚠️ 确认卸货
+        self.upload_unload_photo_btn = "btn_upload_photo"   # ⚠️ 上传卸货照片 (如有)
+        self.unload_success_text = "卸货成功"               # ⚠️ 成功提示
+
+    def go_to_my_orders(self) -> None:
+        """切换到我的运单 tab。"""
+        self.tap_by_id(self.my_orders_tab)
+        self.driver.implicitly_wait(1)
+
+    def find_and_open_order(self, order_no: str) -> None:
+        """查找并打开指定运单的详情页。
+
+        Args:
+            order_no: 订单号
+        """
+        # ⚠️ 策略取决于 APP 实际布局:
+        # 策略1: 通过订单号文本定位并点击
+        try:
+            self.tap_by_text(order_no)
+            self.driver.implicitly_wait(1)
+            return
+        except Exception:
+            pass
+
+        # 策略2: 滚动查找
+        for _ in range(5):  # 最多滚动 5 次
+            try:
+                self.tap_by_text(order_no)
+                self.driver.implicitly_wait(1)
+                return
+            except Exception:
+                self.scroll_down()
+
+        raise RuntimeError(f"在运单列表中未找到订单: {order_no}")
+
+    def load_goods(self, order_no: str) -> None:
+        """执行装货流程。
+
+        Args:
+            order_no: 订单号
+        """
+        self.go_to_my_orders()
+        self.find_and_open_order(order_no)
+
+        # 点击装货
+        self.tap_by_id(self.load_goods_btn)
+        self.driver.implicitly_wait(1)
+
+        # 上传照片 (如有)
+        try:
+            self.tap_by_id(self.upload_load_photo_btn)
+            self.driver.implicitly_wait(1)
+        except Exception:
+            pass
+
+        # 确认装货
+        self.tap_by_id(self.confirm_load_btn)
+        self.driver.implicitly_wait(2)
+
+    def unload_goods(self, order_no: str) -> None:
+        """执行卸货流程。
+
+        Args:
+            order_no: 订单号
+        """
+        self.go_to_my_orders()
+        self.find_and_open_order(order_no)
+
+        # 点击卸货
+        self.tap_by_id(self.unload_goods_btn)
+        self.driver.implicitly_wait(1)
+
+        # 上传照片 (如有)
+        try:
+            self.tap_by_id(self.upload_unload_photo_btn)
+            self.driver.implicitly_wait(1)
+        except Exception:
+            pass
+
+        # 确认卸货
+        self.tap_by_id(self.confirm_unload_btn)
+        self.driver.implicitly_wait(2)
+
+    def verify_load_complete(self) -> bool:
+        """验证装货完成。"""
+        try:
+            self.wait_visible_by_text(self.load_success_text, timeout_s=5)
+            return True
+        except Exception:
+            return False
+
+    def verify_unload_complete(self) -> bool:
+        """验证卸货完成。"""
+        try:
+            self.wait_visible_by_text(self.unload_success_text, timeout_s=5)
+            return True
+        except Exception:
+            return False

tests/cross_end/tests/__init__.py

@@ -0,0 +1 @@
+# cross_end tests package

tests/cross_end/tests/test_driver_to_settlement.py

@@ -0,0 +1,129 @@
+"""跨端 E2E 测试 — 司机接单装货卸货 + 管理端核算打款完整流程。
+
+业务流程:
+    司机 APP (Android):
+        1. 接单: 点击货源 → 选择货源单 → 填写数量 → 勾选协议 → 接单
+        2. 装货: 装货 → 取消弹窗 → 吨位 → 选图 → 相册 → 勾图 → 完成 → 勾选 → 装货确认 → 合同 → 已知晓 → 确定
+        3. 卸货: 卸货 → 取消弹窗 → 吨位 → 选图 → 相册 → 勾图 → 完成 → 卸货确认 → 确定
+        4. 获取运输单号
+
+    Web 管理端:
+        5. 运费核算: URL 跳转 → 输入单号 → 查询 →（无数据时 SQL 重试）→ 编辑 → 核算
+        6. 运单审核: URL 跳转 → 输入单号 → 查询 → 审核 → 滚动弹窗 → 审核通过
+        7. 垫资-财务打款: URL 跳转 → 输入单号 → 查询 → 在线支付 → 滚动 →
+           勾选光大 → 勾选手机号 → 获取验证码 → 输入验证码 → 确认打款
+        8. 垫资-货主打款: URL 跳转 → 输入单号 → 查询 → 打款 → 勾选光大 → 确认打款 → 确认弹窗
+
+运行:
+    pytest tests/cross_end/tests/test_driver_to_settlement.py -m cross_end -v -s
+
+前置条件:
+    - Appium server: http://localhost:4723
+    - Android 模拟器已启动，司机 APP 已安装并登录过一次 (noReset)
+    - 管理端 auth 态: python tests/web/admin/save_auth.py --role super_admin
+    - MySQL 数据库可连接 (pymysql)
+"""
+
+import pytest
+
+from tests.cross_end.pages.admin.freight_accounting_page import FreightAccountingPage
+from tests.cross_end.pages.app.driver_freight_flow_page import DriverFreightFlowPage
+from tests.utils.db_helper import update_risk_control_status
+
+pytestmark = [pytest.mark.cross_end, pytest.mark.p0, pytest.mark.smoke]
+
+
+class TestDriverToSettlement:
+    """完整业务流程: 司机 APP 接单→装货→卸货 → Web 管理端核算→打款。
+
+    测试覆盖:
+        - 司机 APP: 货源浏览、接单、装货（含拍照上传、合同确认）、卸货（含拍照上传）
+        - Web 管理端: 运费核算、运单审核、垫资-财务打款（含验证码）、垫资-货主打款
+        - 数据库: SQL 更新风控状态（运费核算查询无数据时自动执行）
+    """
+
+    def test_driver_to_settlement_full_flow(
+        self,
+        admin_page,
+        admin_config,
+        app_driver,
+        ensure_admin_login,
+        ensure_driver_login,
+        cross_end_context,
+        logger,
+    ):
+        """验证从司机 APP 接单到管理端核算打款的完整跨端业务流程。"""
+        logger.info("=" * 60)
+        logger.info("跨端 E2E: 司机APP → 管理端核算打款 完整流程开始")
+        logger.info("=" * 60)
+
+        # ── 登录确认 ──
+        ensure_admin_login("super_admin")
+        ensure_driver_login(app_driver)
+
+        # ── 初始化 Page Objects ──
+        driver_flow = DriverFreightFlowPage(app_driver)
+        admin_accounting = FreightAccountingPage(admin_page, admin_base_url=admin_config["base_url"])
+
+        # ════════════════════════════════════════════════════════════
+        # Phase 1: 司机 Android APP — 接单 + 装货 + 卸货
+        # ════════════════════════════════════════════════════════════
+
+        logger.info("Phase 1: 司机 APP — 开始接单流程")
+        driver_flow.accept_order_flow(quantity="1")
+        driver_flow.screenshot("step_accept_order")
+        logger.info("  -> 接单流程完成")
+
+        logger.info("Phase 1: 司机 APP — 开始装货流程")
+        driver_flow.load_goods_flow(weight="10")
+        driver_flow.screenshot("step_load_goods")
+        logger.info("  -> 装货流程完成")
+
+        logger.info("Phase 1: 司机 APP — 开始卸货流程")
+        driver_flow.unload_goods_flow(weight="10")
+        driver_flow.screenshot("step_unload_goods")
+        logger.info("  -> 卸货流程完成")
+
+        logger.info("Phase 1: 司机 APP — 获取运输单号")
+        transport_order_no = driver_flow.get_transport_order_no()
+        cross_end_context["order_no"] = transport_order_no
+        logger.info(f"  -> 运输单号: {transport_order_no}")
+
+        # ════════════════════════════════════════════════════════════
+        # Phase 2: Web 管理端 — 核算 + 审核 + 打款
+        # ════════════════════════════════════════════════════════════
+
+        logger.info("Phase 2: Web管理端 — 运费核算（首次查询无数据时自动 SQL 重试）")
+        admin_accounting.do_freight_accounting(
+            transport_order_no,
+            retry_sql=lambda: update_risk_control_status(transport_order_no, status=1),
+        )
+        cross_end_context["accounting_done"] = True
+        logger.info(f"  -> 运费核算完成: {transport_order_no}")
+
+        logger.info("Phase 2: Web管理端 — 运单审核")
+        admin_accounting.do_waybill_audit(transport_order_no)
+        cross_end_context["review_done"] = True
+        logger.info(f"  -> 运单审核完成: {transport_order_no}")
+
+        logger.info("Phase 2: Web管理端 — 垫资-财务打款")
+        admin_accounting.do_finance_payment(transport_order_no, verify_code="1234")
+        logger.info(f"  -> 垫资-财务打款完成: {transport_order_no}")
+
+        logger.info("Phase 2: Web管理端 — 垫资-货主打款")
+        admin_accounting.do_shipper_payment(transport_order_no)
+        logger.info(f"  -> 垫资-货主打款完成: {transport_order_no}")
+
+        # ════════════════════════════════════════════════════════════
+        # 最终验证
+        # ════════════════════════════════════════════════════════════
+        logger.info("=" * 60)
+        logger.info("跨端 E2E: 司机APP → 管理端核算打款 完整流程结束!")
+        logger.info(f"  运输单号:       {cross_end_context['order_no']}")
+        logger.info(f"  核算完成:       {cross_end_context['accounting_done']}")
+        logger.info(f"  审核完成:       {cross_end_context['review_done']}")
+        logger.info("=" * 60)
+
+        assert cross_end_context["order_no"], "运输单号不能为空"
+        assert cross_end_context["accounting_done"], "核算未完成"
+        assert cross_end_context["review_done"], "审核未完成"

tests/cross_end/tests/test_freight_full_flow.py

@@ -0,0 +1,197 @@
+"""跨端 E2E 测试 — 货运全流程。
+
+业务流程 (8 步):
+    1. Web 管理端: 发布货源
+    2. 司机安卓 APP: 接单
+    3. 司机安卓 APP: 装货
+    4. 司机安卓 APP: 卸货
+    5. Web 管理端: 核算运单
+    6. Web 管理端: 核算审核
+    7. Web 管理端: 货主打款
+    8. Web 管理端: 平台打款给司机
+
+运行:
+    pytest tests/cross_end/ -m cross_end -v -s
+
+前置条件:
+    - Appium server: appium (http://localhost:4723)
+    - Android 模拟器已启动，司机 APP 已安装并登录过一次 (noReset)
+    - 管理端 auth 态: python tests/web/admin/save_auth.py --role super_admin
+"""
+
+import pytest
+
+from tests.cross_end.pages.admin.freight_source_page import FreightSourcePage
+from tests.cross_end.pages.admin.waybill_management_page import WaybillManagementPage
+from tests.cross_end.pages.admin.settlement_page import SettlementPage
+from tests.cross_end.pages.app.driver_home_page import DriverHomePage
+from tests.cross_end.pages.app.driver_order_list_page import DriverOrderListPage
+from tests.cross_end.pages.app.driver_transport_page import DriverTransportPage
+from tests.utils.data_helper import generate_order_no
+
+pytestmark = [pytest.mark.cross_end, pytest.mark.p0, pytest.mark.smoke]
+
+
+class TestFreightFullFlow:
+    """货运全流程跨端 E2E — 管理端发布货源 → 司机 APP 接单装货卸货 → 管理端核算打款。"""
+
+    def test_full_freight_flow(
+        self,
+        admin_page,
+        app_driver,
+        ensure_admin_login,
+        ensure_driver_login,
+        cross_end_context,
+        logger,
+    ):
+        """验证从发布货源到平台打款的完整跨端流程。"""
+        logger.info("=" * 60)
+        logger.info("跨端 E2E: 货运全流程开始")
+        logger.info("=" * 60)
+
+        # ── 登录 ──
+        ensure_admin_login("super_admin")
+        ensure_driver_login(app_driver)
+
+        # 生成唯一订单号 (用于货源描述，实际订单号由系统生成)
+        from tests.web.admin.pages.components.navbar import NavBar
+        navbar = NavBar(admin_page)
+        test_order_prefix = generate_order_no()
+
+        # ════════════════════════════════════════════════════════
+        # Step 1: Web 管理端 — 发布货源
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 1/8: Web管理端 — 发布货源")
+
+        # ⚠️ 菜单名称需要根据实际管理端调整
+        navbar.go_to_submenu("运输管理", "货源管理")
+        admin_page.wait_for_timeout(1000)
+
+        freight_page = FreightSourcePage(admin_page)
+        freight_page.fill_freight_form(
+            goods_name=f"测试货物_{test_order_prefix}",
+            weight=10,
+            # ⚠️ 地址格式根据实际系统调整
+            load_addr="云南省昆明市五华区",
+            unload_addr="云南省昆明市官渡区",
+        )
+        freight_page.submit()
+        freight_page.confirm_publish()
+
+        # 截图留档
+        freight_page.screenshot("step1_freight_published")
+
+        # 获取系统生成的订单号
+        published_order_no = freight_page.get_published_order_no()
+        cross_end_context["order_no"] = published_order_no
+        logger.info(f"  -> 发布成功, 订单号: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # Step 2: 司机安卓 APP — 接单
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 2/8: 司机安卓APP — 接单")
+
+        driver_home = DriverHomePage(app_driver)
+        driver_home.go_to_order_list()
+
+        order_list_page = DriverOrderListPage(app_driver)
+        order_list_page.search_and_accept_order(published_order_no)
+
+        if not order_list_page.verify_order_accepted():
+            order_list_page.screenshot("step2_order_accept_failed")
+            raise AssertionError(f"司机端接单失败: {published_order_no}")
+
+        order_list_page.screenshot("step2_order_accepted")
+        logger.info(f"  -> 接单成功: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # Step 3: 司机安卓 APP — 装货
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 3/8: 司机安卓APP — 装货")
+
+        transport_page = DriverTransportPage(app_driver)
+        transport_page.load_goods(published_order_no)
+        cross_end_context["load_confirm"] = True
+
+        transport_page.screenshot("step3_loaded")
+        logger.info("  -> 装货完成")
+
+        # ════════════════════════════════════════════════════════
+        # Step 4: 司机安卓 APP — 卸货
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 4/8: 司机安卓APP — 卸货")
+
+        transport_page.unload_goods(published_order_no)
+        cross_end_context["unload_confirm"] = True
+
+        transport_page.screenshot("step4_unloaded")
+        logger.info("  -> 卸货完成")
+
+        # ════════════════════════════════════════════════════════
+        # Step 5: Web 管理端 — 核算运单
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 5/8: Web管理端 — 核算运单")
+
+        # ⚠️ 菜单名称需要根据实际管理端调整
+        # 运单核算可能在 "运输管理 > 运单管理" 或 "承运结算" 下
+        navbar.go_to_submenu("运输管理", "承运结算")
+        admin_page.wait_for_timeout(1000)
+
+        waybill_page = WaybillManagementPage(admin_page)
+        waybill_page.do_accounting(published_order_no)
+        cross_end_context["accounting_done"] = True
+
+        waybill_page.screenshot("step5_accounting_done")
+        logger.info(f"  -> 核算完成: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # Step 6: Web 管理端 — 核算审核
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 6/8: Web管理端 — 核算审核")
+
+        waybill_page.do_review(published_order_no)
+        cross_end_context["review_done"] = True
+
+        waybill_page.screenshot("step6_review_done")
+        logger.info(f"  -> 审核完成: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # Step 7: Web 管理端 — 货主打款
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 7/8: Web管理端 — 货主打款")
+
+        # ⚠️ 打款页面可能与运单核算在同一菜单下或不同菜单
+        settlement_page = SettlementPage(admin_page)
+        settlement_page.do_shipper_payment(published_order_no)
+
+        settlement_page.screenshot("step7_shipper_paid")
+        logger.info(f"  -> 货主打款完成: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # Step 8: Web 管理端 — 平台打款给司机
+        # ════════════════════════════════════════════════════════
+        logger.info("Step 8/8: Web管理端 — 平台打款给司机")
+
+        settlement_page.do_platform_payment(published_order_no)
+
+        settlement_page.screenshot("step8_platform_paid")
+        logger.info(f"  -> 平台打款完成: {published_order_no}")
+
+        # ════════════════════════════════════════════════════════
+        # 最终验证
+        # ════════════════════════════════════════════════════════
+        logger.info("=" * 60)
+        logger.info("跨端 E2E: 货运全流程完成!")
+        logger.info(f"  订单号:       {cross_end_context['order_no']}")
+        logger.info(f"  装货确认:     {cross_end_context['load_confirm']}")
+        logger.info(f"  卸货确认:     {cross_end_context['unload_confirm']}")
+        logger.info(f"  核算完成:     {cross_end_context['accounting_done']}")
+        logger.info(f"  审核完成:     {cross_end_context['review_done']}")
+        logger.info("=" * 60)
+
+        # 断言: 所有步骤必须完成
+        assert cross_end_context["order_no"], "订单号不能为空"
+        assert cross_end_context["load_confirm"], "装货未确认"
+        assert cross_end_context["unload_confirm"], "卸货未确认"
+        assert cross_end_context["accounting_done"], "核算未完成"
+        assert cross_end_context["review_done"], "审核未完成"

tests/pytest.ini

@@ -31,6 +31,7 @@ markers =
     shipper_ios: 货主苹果端 (Appium)
     appium: 所有 Appium 测试
     playwright: 所有 Playwright 测试
+    cross_end: 跨端 E2E 测试（管理端 + 司机安卓端联动）
 
 # 日志格式
 log_cli = true

tests/utils/db_helper.py

@@ -0,0 +1,72 @@
+"""MySQL 数据库操作工具 — 测试中的数据库校验/数据修改。
+
+Usage:
+    from tests.utils.db_helper import update_risk_control_status
+
+    affected = update_risk_control_status("RHJC040026071600021457570057", status=1)
+"""
+
+import pymysql
+
+DEFAULT_DB_CONFIG = {
+    "host": "36.137.133.24",
+    "port": 3306,
+    "user": "ntocc",
+    "password": "Rgxi3!kD9jjXpgzx",
+    "database": "ntocc",
+    "charset": "utf8mb4",
+}
+
+
+def get_connection(config: dict | None = None):
+    """获取数据库连接。"""
+    cfg = config or DEFAULT_DB_CONFIG
+    return pymysql.connect(**cfg)
+
+
+def execute_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> int:
+    """执行 SQL（INSERT/UPDATE/DELETE），返回影响行数。"""
+    conn = get_connection(config)
+    try:
+        with conn.cursor() as cursor:
+            affected = cursor.execute(sql, params)
+            conn.commit()
+            return affected
+    finally:
+        conn.close()
+
+
+def query_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> list[dict]:
+    """执行查询 SQL，返回结果列表。"""
+    conn = get_connection(config)
+    try:
+        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
+            cursor.execute(sql, params)
+            return cursor.fetchall()
+    finally:
+        conn.close()
+
+
+def update_risk_control_status(order_code: str, status: int = 1, config: dict | None = None) -> int:
+    """更新运单风控状态。
+
+    执行 SQL:
+        UPDATE tms_order_detail_ext SET risk_control_status = <status>
+        WHERE order_detail_code IN (
+            SELECT `code` FROM `tms_order_detail` WHERE order_code = '<order_code>'
+        )
+
+    Args:
+        order_code: 运输单号（对应 tms_order_detail.order_code）
+        status: 风控状态值，默认 1
+        config: 数据库连接配置
+    Returns:
+        影响行数
+    """
+    sql = (
+        "UPDATE tms_order_detail_ext SET risk_control_status = %s "
+        "WHERE order_detail_code IN ("
+        "  SELECT `code` FROM `tms_order_detail` WHERE order_code = %s"
+        ")"
+    )
+    return execute_sql(sql, (status, order_code), config)