feat(cross_end): add cross-platform E2E test support for freight full flow

Add tests/cross_end/ directory with dual-driver architecture:
- conftest.py: Playwright admin_page + Appium app_driver fixtures
- Admin Page Objects: FreightSourcePage, WaybillManagementPage, SettlementPage
- App Page Objects: DriverOrderListPage, DriverTransportPage, DriverHomePage (re-export)
- test_freight_full_flow.py: 8-step end-to-end test (publish → accept → load → unload → accounting → review → shipper pay → platform pay)
- Shared context dict for cross-step data passing
- Graceful skip when either Playwright or Appium is unavailable
- All selectors marked with ⚠️ placeholders for discovery against real environments

Add cross_end marker to pytest.ini for selective execution.

docs/superpowers/specs/2026-07-15-test-framework-scaffold-design.md

@@ -0,0 +1,65 @@
+# 测试框架搭建 — 设计说明
+
+> 2026-07-15 | 状态: 已批准
+
+## 目标
+
+基于 `ui_automation_best_practices.md` v1.0 的最佳实践，搭建可落地的 Web + App UI 自动化测试框架骨架。后续所有测试用例都留存在 `tests/` 目录下。
+
+## 交付物
+
+### 文件清单
+
+```
+tests/
+├── conftest.py                     # 根级 fixture
+├── pytest.ini                      # pytest 配置
+├── config/
+│   ├── test_config.yaml            # 环境配置
+│   └── test_config.py              # 配置加载器
+├── web/                            # Web UI (Playwright)
+│   ├── conftest.py                 # browser/page fixture
+│   ├── pages/
+│   │   ├── base_page.py            # Page Object 基类
+│   │   ├── login_page.py           # 登录页
+│   │   └── components/
+│   │       └── navbar.py           # 导航栏组件
+│   └── test_login/
+│       └── test_login.py           # 登录用例（含参数化）
+├── app/                            # App UI (Appium)
+│   ├── conftest.py                 # driver fixture
+│   ├── pages/
+│   │   ├── base_page.py            # Page Object 基类
+│   │   └── driver_home_page.py     # 司机端首页
+│   └── test_login/
+│       └── test_login.py           # 登录用例
+└── utils/
+    ├── __init__.py
+    ├── logger.py                   # 日志工具
+    └── data_helper.py              # 测试数据生成/清理
+```
+
+### 与最佳实践的对应
+
+| 最佳实践章节 | 框架落地位置 |
+| :--- | :--- |
+| §3.3 Page Object 模式 | `web/pages/base_page.py` + `login_page.py` |
+| §3.1 定位策略 | `base_page.py` 中的封装方法优先使用 getByTestId |
+| §3.4 / §4.5 截图策略 | `conftest.py` 中 fixture 的失败自动截图 |
+| §5.1 用例独立性 | `conftest.py` 中 setup/teardown fixture |
+| §5.2 数据驱动 | `test_login.py` 中的 `@pytest.mark.parametrize` |
+| §5.3 环境配置管理 | `config/test_config.yaml` + `test_config.py` |
+| §5.4 报告与日志 | `utils/logger.py` |
+| §3.6 多浏览器 | `web/conftest.py` 中 `browser_name` fixture 参数化 |
+
+## 配置数据来源
+
+从 `project_profile.md` 提取：
+- BASE_URL: `https://ybxcx.ynyun8.com:8000/admin`
+- 测试账号: super_admin / 13113113113 / 15188888888
+- 注意：密码已在 profile 中声明，config 中以占位符标注
+
+## 范围边界
+
+- ✅ 包含: 框架骨架、基类、conftest、配置、示例用例
+- ❌ 不包含: 运八完整业务用例（后续按需求逐步添加）

tests/app/_base_page.py

@@ -0,0 +1,73 @@
+"""通用 App Page Object 基类。"""
+
+from datetime import datetime
+from pathlib import Path
+
+from appium.webdriver.common.appiumby import AppiumBy
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+
+class AppBasePage:
+    """App Page Object 基类 — 各端通用。"""
+
+    def __init__(self, driver, platform: str, timeout_ms: int = 30_000):
+        self.driver = driver
+        self.platform = platform  # "android" | "ios"
+        self.timeout = timeout_ms / 1000
+
+    def wait_visible_by_id(self, accessibility_id: str, timeout_s: float | None = None) -> None:
+        wait = WebDriverWait(self.driver, timeout=timeout_s or self.timeout)
+        wait.until(
+            EC.visibility_of_element_located((AppiumBy.ACCESSIBILITY_ID, accessibility_id))
+        )
+
+    def wait_visible_by_text(self, text: str, timeout_s: float | None = None) -> None:
+        wait = WebDriverWait(self.driver, timeout=timeout_s or self.timeout)
+        locator = (
+            (AppiumBy.XPATH, f"//*[@text='{text}']")
+            if self.platform == "android"
+            else (AppiumBy.XPATH, f"//*[@label='{text}']")
+        )
+        wait.until(EC.visibility_of_element_located(locator))
+
+    def tap_by_id(self, accessibility_id: str) -> None:
+        self.wait_visible_by_id(accessibility_id)
+        self.driver.find_element(AppiumBy.ACCESSIBILITY_ID, accessibility_id).click()
+
+    def tap_by_text(self, text: str) -> None:
+        strategy = AppiumBy.XPATH
+        locator = (
+            f"//*[@text='{text}']"
+            if self.platform == "android"
+            else f"//*[@label='{text}']"
+        )
+        wait = WebDriverWait(self.driver, timeout=self.timeout)
+        wait.until(EC.visibility_of_element_located((strategy, locator)))
+        self.driver.find_element(strategy, locator).click()
+
+    def fill_by_id(self, accessibility_id: str, value: str) -> None:
+        self.wait_visible_by_id(accessibility_id)
+        element = self.driver.find_element(AppiumBy.ACCESSIBILITY_ID, accessibility_id)
+        element.clear()
+        element.send_keys(value)
+
+    def scroll_down(self) -> None:
+        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=500,
+        )
+
+    def screenshot(self, name: str) -> str:
+        scenario_dir = SCREENSHOTS_DIR / self.__class__.__name__
+        scenario_dir.mkdir(parents=True, exist_ok=True)
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        path = str(scenario_dir / f"{name}_{self.platform}_{ts}.png")
+        self.driver.save_screenshot(path)
+        return path

tests/app/_shared.py

@@ -0,0 +1,125 @@
+"""App 端公共模块 — driver 创建 + ensure_logged_in 工厂。
+
+所有 App 端 conftest 调用 share_* 函数即可，避免重复代码。
+"""
+
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+
+def create_driver(section_config: dict, platform: str):
+    """创建 Appium driver。
+
+    Args:
+        section_config: 配置中对应端的 section（如 config["driver_android"]）
+        platform: "android" | "ios"
+    """
+    try:
+        from appium import webdriver
+        from appium.options.android import UiAutomator2Options
+        from appium.options.ios import XCUITestOptions
+    except ImportError:
+        pytest.skip("Appium-Python-Client 未安装，跳过 App 测试。")
+
+    appium_host = section_config.get("appium_host", "http://localhost:4723")
+
+    if platform == "android":
+        options = UiAutomator2Options()
+        options.device_name = section_config["device_name"]
+        options.app_package = section_config["app_package"]
+        options.app_activity = section_config["app_activity"]
+        options.no_reset = section_config.get("no_reset", True)
+        options.new_command_timeout = section_config.get("new_command_timeout", 120)
+        options.automation_name = "UiAutomator2"
+    else:
+        options = XCUITestOptions()
+        options.device_name = section_config["device_name"]
+        options.bundle_id = section_config["bundle_id"]
+        options.no_reset = section_config.get("no_reset", True)
+        options.new_command_timeout = section_config.get("new_command_timeout", 120)
+        options.automation_name = "XCUITest"
+
+    return webdriver.Remote(appium_host, options=options)
+
+
+def make_ensure_logged_in(
+    section_config: dict,
+    config: dict,
+    account_role: str,
+    home_indicator: str = "tab_home",
+):
+    """创建一个 ensure_logged_in 闭包。
+
+    Args:
+        section_config: App 端配置
+        config: 全局配置
+        account_role: 默认账号角色名（如 "driver"、"shipper"）
+        home_indicator: 首页元素的 accessibility_id
+    """
+    from selenium.webdriver.support.ui import WebDriverWait
+    from selenium.webdriver.support import expected_conditions as EC
+    from appium.webdriver.common.appiumby import AppiumBy
+
+    def _ensure_logged_in(driver, timeout_s: float = 10.0) -> None:
+        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
+
+        try:
+            login_input = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "login_phone_input")
+            if login_input.is_displayed():
+                if skip_login:
+                    raise RuntimeError(
+                        "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))
+                )
+                return
+        except RuntimeError:
+            raise
+        except Exception:
+            pass
+
+        raise RuntimeError("无法确定 APP 的登录状态！请确保 APP 已启动且首页或登录页可见。")
+
+    return _ensure_logged_in
+
+
+def save_failure_screenshot(driver, section_config, platform, request):
+    """失败自动截图（各端 conftest teardown 中调用）。"""
+    if (
+        section_config.get("screenshot_on_failure", True)
+        and 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_{platform}_{ts}.png")
+        driver.save_screenshot(path)
+
+
+@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/app/driver_android/conftest.py

@@ -0,0 +1,27 @@
+"""司机安卓端 fixture — Appium + 登录态复用。"""
+
+import pytest
+from tests.app._shared import create_driver, make_ensure_logged_in, save_failure_screenshot
+
+from tests.app._shared import pytest_runtest_makereport  # noqa: F401 — report hook
+
+
+@pytest.fixture
+def driver(driver_android_config, request):
+    """司机安卓端 driver (noReset=true)。"""
+    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:
+            d.quit()
+
+
+@pytest.fixture
+def ensure_logged_in(driver_android_config, config):
+    """确保已登录态（司机角色）。"""
+    fn = make_ensure_logged_in(driver_android_config, config, account_role="driver")
+    return lambda d: fn(d)

tests/app/driver_android/pages/driver_home_page.py

@@ -0,0 +1,26 @@
+"""司机安卓端 Page Object — 首页。"""
+
+from tests.app._base_page import AppBasePage
+
+
+class DriverHomePage(AppBasePage):
+    """司机端 APP 首页。"""
+
+    def __init__(self, driver, platform: str = "android"):
+        super().__init__(driver, platform)
+        self.order_tab = "tab_orders"
+        self.mine_tab = "tab_mine"
+        self.home_tab = "tab_home"
+
+    def is_displayed(self) -> bool:
+        try:
+            self.wait_visible_by_id(self.home_tab, timeout_s=10)
+            return True
+        except Exception:
+            return False
+
+    def go_to_order_list(self) -> None:
+        self.tap_by_id(self.order_tab)
+
+    def go_to_mine(self) -> None:
+        self.tap_by_id(self.mine_tab)

tests/app/driver_android/tests/test_smoke.py

@@ -0,0 +1,18 @@
+"""司机安卓端 — 冒烟测试。"""
+
+import pytest
+
+pytestmark = [pytest.mark.driver_android]
+
+
+class TestDriverAndroidSmoke:
+    """司机安卓端基础验证。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_home_page_loaded(self, driver, ensure_logged_in):
+        """验证登录态有效，首页正常。"""
+        ensure_logged_in(driver)
+        from appium.webdriver.common.appiumby import AppiumBy
+        home = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "tab_home")
+        assert home.is_displayed(), "首页 tab 应可见"

tests/app/driver_ios/conftest.py

@@ -0,0 +1,27 @@
+"""司机苹果端 fixture — Appium + 登录态复用。"""
+
+import pytest
+from tests.app._shared import create_driver, make_ensure_logged_in, save_failure_screenshot
+
+from tests.app._shared import pytest_runtest_makereport  # noqa: F401
+
+
+@pytest.fixture
+def driver(driver_ios_config, request):
+    """司机苹果端 driver。"""
+    cfg = driver_ios_config
+    d = None
+    try:
+        d = create_driver(cfg, platform="ios")
+        yield d
+        save_failure_screenshot(d, cfg, "ios", request)
+    finally:
+        if d is not None:
+            d.quit()
+
+
+@pytest.fixture
+def ensure_logged_in(driver_ios_config, config):
+    """确保已登录态（司机角色）。"""
+    fn = make_ensure_logged_in(driver_ios_config, config, account_role="driver")
+    return lambda d: fn(d)

tests/app/driver_ios/tests/test_smoke.py

@@ -0,0 +1,18 @@
+"""司机苹果端 — 冒烟测试。"""
+
+import pytest
+
+pytestmark = [pytest.mark.driver_ios]
+
+
+class TestDriverIosSmoke:
+    """司机苹果端基础验证。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_home_page_loaded(self, driver, ensure_logged_in):
+        """验证登录态有效，首页正常。"""
+        ensure_logged_in(driver)
+        from appium.webdriver.common.appiumby import AppiumBy
+        home = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "tab_home")
+        assert home.is_displayed(), "首页 tab 应可见"

tests/app/shipper_android/conftest.py

@@ -0,0 +1,27 @@
+"""货主安卓端 fixture — Appium + 登录态复用。"""
+
+import pytest
+from tests.app._shared import create_driver, make_ensure_logged_in, save_failure_screenshot
+
+from tests.app._shared import pytest_runtest_makereport  # noqa: F401
+
+
+@pytest.fixture
+def driver(shipper_android_config, request):
+    """货主安卓端 driver。"""
+    cfg = shipper_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:
+            d.quit()
+
+
+@pytest.fixture
+def ensure_logged_in(shipper_android_config, config):
+    """确保已登录态（货主角色）。"""
+    fn = make_ensure_logged_in(shipper_android_config, config, account_role="shipper")
+    return lambda d: fn(d)

tests/app/shipper_android/tests/test_smoke.py

@@ -0,0 +1,18 @@
+"""货主安卓端 — 冒烟测试。"""
+
+import pytest
+
+pytestmark = [pytest.mark.shipper_android]
+
+
+class TestShipperAndroidSmoke:
+    """货主安卓端基础验证。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_home_page_loaded(self, driver, ensure_logged_in):
+        """验证登录态有效，首页正常。"""
+        ensure_logged_in(driver)
+        from appium.webdriver.common.appiumby import AppiumBy
+        home = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "tab_home")
+        assert home.is_displayed(), "首页 tab 应可见"

tests/app/shipper_ios/conftest.py

@@ -0,0 +1,27 @@
+"""货主苹果端 fixture — Appium + 登录态复用。"""
+
+import pytest
+from tests.app._shared import create_driver, make_ensure_logged_in, save_failure_screenshot
+
+from tests.app._shared import pytest_runtest_makereport  # noqa: F401
+
+
+@pytest.fixture
+def driver(shipper_ios_config, request):
+    """货主苹果端 driver。"""
+    cfg = shipper_ios_config
+    d = None
+    try:
+        d = create_driver(cfg, platform="ios")
+        yield d
+        save_failure_screenshot(d, cfg, "ios", request)
+    finally:
+        if d is not None:
+            d.quit()
+
+
+@pytest.fixture
+def ensure_logged_in(shipper_ios_config, config):
+    """确保已登录态（货主角色）。"""
+    fn = make_ensure_logged_in(shipper_ios_config, config, account_role="shipper")
+    return lambda d: fn(d)

tests/app/shipper_ios/tests/test_smoke.py

@@ -0,0 +1,18 @@
+"""货主苹果端 — 冒烟测试。"""
+
+import pytest
+
+pytestmark = [pytest.mark.shipper_ios]
+
+
+class TestShipperIosSmoke:
+    """货主苹果端基础验证。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_home_page_loaded(self, driver, ensure_logged_in):
+        """验证登录态有效，首页正常。"""
+        ensure_logged_in(driver)
+        from appium.webdriver.common.appiumby import AppiumBy
+        home = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "tab_home")
+        assert home.is_displayed(), "首页 tab 应可见"

tests/config/test_config.py

@@ -0,0 +1,46 @@
+"""配置加载器。
+
+从 tests/config/test_config.yaml 读取配置，提供类型安全的访问接口。
+
+Usage:
+    from tests.config.test_config import load_config
+
+    config = load_config()
+    admin_url = config["admin"]["base_url"]
+    admin_phone = config["accounts"]["super_admin"]["phone"]
+
+各端通过 fixture 直接引用对应 section:
+    def test_xxx(admin_config):
+        page.goto(admin_config["base_url"])
+"""
+
+from pathlib import Path
+from typing import Any, Dict
+
+import yaml
+
+_CONFIG_PATH = Path(__file__).parent / "test_config.yaml"
+_cache: Dict[str, Any] | None = None
+
+
+def load_config(force_reload: bool = False) -> Dict[str, Any]:
+    """加载测试配置（带缓存，force_reload=True 可强制重载）。"""
+    global _cache
+    if _cache is not None and not force_reload:
+        return _cache
+
+    if not _CONFIG_PATH.exists():
+        raise FileNotFoundError(f"配置文件不存在: {_CONFIG_PATH}")
+
+    with open(_CONFIG_PATH, encoding="utf-8") as f:
+        _cache = yaml.safe_load(f)
+
+    return _cache
+
+
+def get_config_section(section: str) -> Dict[str, Any]:
+    """获取配置中指定 section 的字典。"""
+    config = load_config()
+    if section not in config:
+        raise KeyError(f"配置中不存在 section '{section}'，可用: {list(config.keys())}")
+    return config[section]

tests/config/test_config.yaml

@@ -0,0 +1,144 @@
+# 运八网络货运平台 — 测试环境配置
+# 数据来源: knowledge_base/00_project/project_profile.md
+# 使用方式: from tests.config.test_config import load_config
+#
+# 覆盖端:
+#   Web:  管理端 admin / 司机端小程序 mini_program
+#   App:  司机安卓 driver_android / 货主安卓 shipper_android
+#         司机苹果 driver_ios / 货主苹果 shipper_ios
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 管理端 (Web/Playwright) ──
+# ═══════════════════════════════════════════════════════════════════
+admin:
+  base_url: "https://ybxcx.ynyun8.com:8000/admin"
+  api_base_url: "https://ybxcx.ynyun8.com:8000/"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+  screenshot_on_step: false
+  browsers:
+    - chromium
+    - firefox
+  viewport:
+    width: 1920
+    height: 1080
+
+  # 登录态复用: python tests/web/admin/save_auth.py 生成后自动跳过验证码
+  skip_login: true
+  storage_state_dir: "tests/web/auth"
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 司机端小程序 (miniprogram-automator / Node.js) ──
+# ═══════════════════════════════════════════════════════════════════
+# 原理: Python pytest → subprocess → node runner.js → miniprogram-automator
+#       操控微信开发者工具中的小程序，返回 JSON 结果
+#
+# 前置条件:
+#   1. 安装微信开发者工具 + 开启服务端口（设置 → 安全 → 服务端口）
+#   2. uni-app 编译为微信小程序: npm run build:mp-weixin
+#   3. 安装依赖: cd tests/web/mini_program/automator && npm install
+mini_program:
+  # uni-app 编译出的小程序项目路径（.js/.json/.wxml 所在目录）
+  project_path: "E:/test/QaAutomationHub/mini_program/dist"  # ⚠️ 请确认实际路径
+  # 微信开发者工具 CLI 路径（Windows 默认如下，macOS 为 /Applications/...）
+  dev_tool_path: "C:/Program Files (x86)/Tencent/微信web开发者工具/cli.bat"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 司机安卓端 (App/Appium) ──
+# ═══════════════════════════════════════════════════════════════════
+driver_android:
+  appium_host: "http://localhost:4723"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+  platform: android
+  skip_login: true          # 手动登录一次后 noReset 自动复用
+
+  device_name: "Android Emulator"
+  app_package: "com.arpa.ynchenggangdriver"         # ⚠️ 请确认实际包名
+  app_activity: "com.arpa.ntocc.MainActivity"       # ⚠️ 请确认实际启动 Activity
+  no_reset: true
+  new_command_timeout: 120
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 货主安卓端 (App/Appium) ──
+# ═══════════════════════════════════════════════════════════════════
+shipper_android:
+  appium_host: "http://localhost:4723"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+  platform: android
+  skip_login: true
+
+  device_name: "Android Emulator"
+  app_package: "com.arpa.ynChengGangShipper"                   # ⚠️ 请确认实际包名
+  app_activity: "com.arpa.wuche_shipper.MainActivity"                      # ⚠️ 请确认实际启动 Activity
+  no_reset: true
+  new_command_timeout: 120
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 司机苹果端 (App/Appium，需 macOS) ──
+# ═══════════════════════════════════════════════════════════════════
+driver_ios:
+  appium_host: "http://localhost:4723"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+  platform: ios
+  skip_login: true
+
+  device_name: "iPhone 15"
+  bundle_id: "com.arpa.ynchenggangdriver"            # ⚠️ 请确认实际 Bundle ID
+  no_reset: true
+  new_command_timeout: 120
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 货主苹果端 (App/Appium，需 macOS) ──
+# ═══════════════════════════════════════════════════════════════════
+shipper_ios:
+  appium_host: "http://localhost:4723"
+  timeout_seconds: 120
+  screenshot_on_failure: true
+  platform: ios
+  skip_login: true
+
+  device_name: "iPhone 15"
+  bundle_id: "com.yunba.shipper"                     # ⚠️ 请确认实际 Bundle ID
+  no_reset: true
+  new_command_timeout: 120
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 测试账号 ──
+# ═══════════════════════════════════════════════════════════════════
+accounts:
+  super_admin:
+    phone: "super_admin"
+    password: "951260684NiAn.."
+    role: "平台运营人员"
+    use_on: [admin]
+
+  team_leader:
+    phone: "13113113113"
+    password: "88888888"
+    role: "车队长"
+    use_on: [admin, mini_program]
+
+  driver:
+    phone: "15188888888"
+    password: "88888888"
+    role: "司机"
+    use_on: [mini_program, driver_android, driver_ios]
+
+  shipper:
+    phone: "13800000000"                              # ⚠️ 请确认实际货主账号
+    password: "88888888"
+    role: "货主"
+    use_on: [shipper_android, shipper_ios]
+
+# ═══════════════════════════════════════════════════════════════════
+# ── 报告 ──
+# ═══════════════════════════════════════════════════════════════════
+report:
+  logs_dir: "output/logs"
+  screenshots_dir: "output/screenshots"
+  junit_xml: "output/reports/junit.xml"

tests/conftest.py

@@ -0,0 +1,63 @@
+"""根级 conftest — 全局 fixture：配置加载、日志。
+
+Usage:
+    def test_xxx(admin_config):    # 管理端配置
+    def test_xxx(driver_android_config):  # 司机安卓配置
+"""
+
+from pathlib import Path
+import sys
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+
+@pytest.fixture(scope="session")
+def config():
+    """全局配置（session 级，只加载一次）。"""
+    from tests.config.test_config import load_config
+    return load_config()
+
+
+@pytest.fixture(scope="function")
+def logger(request):
+    """每个测试函数的 logger。"""
+    from tests.utils.logger import get_logger
+    module_name = request.module.__name__.replace("tests.", "")
+    return get_logger(module_name)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 各端配置 fixture（从 config 中摘取对应 section）
+# ═══════════════════════════════════════════════════════════════════
+
+@pytest.fixture(scope="session")
+def admin_config(config):
+    """管理端 (Playwright) 配置。"""
+    return config["admin"]
+
+@pytest.fixture(scope="session")
+def mini_program_config(config):
+    """司机端小程序 (Playwright) 配置。"""
+    return config["mini_program"]
+
+@pytest.fixture(scope="session")
+def driver_android_config(config):
+    """司机安卓端 (Appium) 配置。"""
+    return config["driver_android"]
+
+@pytest.fixture(scope="session")
+def shipper_android_config(config):
+    """货主安卓端 (Appium) 配置。"""
+    return config["shipper_android"]
+
+@pytest.fixture(scope="session")
+def driver_ios_config(config):
+    """司机苹果端 (Appium) 配置。"""
+    return config["driver_ios"]
+
+@pytest.fixture(scope="session")
+def shipper_ios_config(config):
+    """货主苹果端 (Appium) 配置。"""
+    return config["shipper_ios"]

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_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_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_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

@@ -0,0 +1,40 @@
+[pytest]
+# pytest 配置文件
+
+# 测试发现路径
+testpaths = tests
+
+# 仅执行以 test_ 开头的文件
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+
+# 默认命令行参数
+addopts =
+    -v
+    --tb=short
+    --strict-markers
+    --disable-warnings
+    -p no:cacheprovider
+
+# 自定义 markers
+markers =
+    p0: P0 核心流程（冒烟测试）
+    p1: P1 主要功能
+    p2: P2 次要/异常流程
+    smoke: 冒烟测试
+    admin: 管理端 (Playwright)
+    mini_program: 司机端小程序 (Playwright)
+    driver_android: 司机安卓端 (Appium)
+    shipper_android: 货主安卓端 (Appium)
+    driver_ios: 司机苹果端 (Appium)
+    shipper_ios: 货主苹果端 (Appium)
+    appium: 所有 Appium 测试
+    playwright: 所有 Playwright 测试
+    cross_end: 跨端 E2E 测试（管理端 + 司机安卓端联动）
+
+# 日志格式
+log_cli = true
+log_cli_level = INFO
+log_cli_format = %(asctime)s [%(levelname)s] %(name)s — %(message)s
+log_cli_date_format = %H:%M:%S

tests/utils/data_helper.py

@@ -0,0 +1,41 @@
+"""测试数据生成与清理辅助工具。
+
+提供常用方法：生成唯一标识、构建测试数据字典、API 层面清理数据。
+
+Usage:
+    from tests.utils.data_helper import generate_order_no, generate_phone
+
+    order_no = generate_order_no()
+    phone = generate_phone()
+"""
+
+import random
+import string
+from datetime import datetime
+
+
+def generate_order_no(prefix: str = "TEST") -> str:
+    """生成唯一订单号。"""
+    ts = datetime.now().strftime("%Y%m%d%H%M%S")
+    rand = "".join(random.choices(string.digits, k=4))
+    return f"{prefix}{ts}{rand}"
+
+
+def generate_phone() -> str:
+    """生成合法格式的测试手机号（138 开头）。"""
+    suffix = "".join(random.choices(string.digits, k=8))
+    return f"138{suffix}"
+
+
+def generate_shipment_no(prefix: str = "YB") -> str:
+    """生成运单号（格式: YB + 日期 + 随机数）。"""
+    ts = datetime.now().strftime("%Y%m%d")
+    rand = "".join(random.choices(string.digits, k=6))
+    return f"{prefix}{ts}{rand}"
+
+
+def generate_unique_id(prefix: str = "id") -> str:
+    """生成通用唯一 ID。"""
+    ts = datetime.now().strftime("%Y%m%d%H%M%S%f")
+    rand = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
+    return f"{prefix}_{ts}_{rand}"

tests/utils/logger.py

@@ -0,0 +1,47 @@
+"""统一日志工具。
+
+Usage:
+    from tests.utils.logger import get_logger
+
+    logger = get_logger(__name__)
+    logger.info("开始执行用例")
+    logger.error(f"用例失败: {e}")
+"""
+
+import logging
+import sys
+from datetime import datetime
+from pathlib import Path
+
+# 日志目录
+LOGS_DIR = Path("output/logs")
+LOGS_DIR.mkdir(parents=True, exist_ok=True)
+
+# 日志格式
+_FORMAT = "%(asctime)s [%(levelname)s] %(name)s — %(message)s"
+_DATE_FMT = "%Y-%m-%d %H:%M:%S"
+
+# 根 logger 配置（模块级，只初始化一次）
+_ROOT_LOGGER = logging.getLogger("tests")
+_ROOT_LOGGER.setLevel(logging.DEBUG)
+
+# 控制台 Handler
+_console_handler = logging.StreamHandler(sys.stdout)
+_console_handler.setLevel(logging.INFO)
+_console_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATE_FMT))
+
+# 文件 Handler（每次运行一个日志文件）
+_log_file = LOGS_DIR / f"test_run_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
+_file_handler = logging.FileHandler(_log_file, encoding="utf-8")
+_file_handler.setLevel(logging.DEBUG)
+_file_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATE_FMT))
+
+_ROOT_LOGGER.addHandler(_console_handler)
+_ROOT_LOGGER.addHandler(_file_handler)
+
+
+def get_logger(name: str) -> logging.Logger:
+    """获取以 `tests.` 为前缀的子 logger。"""
+    if not name.startswith("tests."):
+        name = f"tests.{name}"
+    return logging.getLogger(name)

tests/web/admin/conftest.py

@@ -0,0 +1,126 @@
+"""管理端 fixture — Playwright 浏览器实例 + 登录态复用。
+
+核心机制:
+    skip_login: true → 加载 storage_state，跳过验证码
+    skip_login: false → 执行完整 UI 登录
+
+初始化:
+    python tests/web/admin/save_auth.py
+"""
+
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+from playwright.sync_api import sync_playwright, Page
+
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+
+@pytest.fixture(params=["chromium", "firefox"])
+def browser_name(admin_config, request):
+    """每个浏览器执行一轮。"""
+    browsers = admin_config.get("browsers", ["chromium", "firefox"])
+    request.param = browsers[len(getattr(request, "_index", 0))]
+    return request.param
+
+
+@pytest.fixture
+def page(admin_config, browser_name, request) -> Page:
+    """Playwright Page，加载登录态，失败自动截图。"""
+    viewport = admin_config.get("viewport", {"width": 1920, "height": 1080})
+    timeout = admin_config.get("timeout_seconds", 120) * 1000
+    skip_login = admin_config.get("skip_login", True)
+    storage_state_dir = Path(admin_config.get("storage_state_dir", "tests/web/auth"))
+
+    with sync_playwright() as p:
+        browser_launcher = getattr(p, browser_name)
+        browser = browser_launcher.launch(headless=True)
+
+        context_kwargs = {
+            "viewport": {"width": viewport["width"], "height": viewport["height"]},
+            "locale": "zh-CN",
+        }
+
+        loaded_role = None
+        if skip_login:
+            loaded_role = _load_auth_state(context_kwargs, storage_state_dir, browser_name)
+
+        context = browser.new_context(**context_kwargs)
+        page = context.new_page()
+        page.set_default_timeout(timeout)
+        page._auth_role = loaded_role
+
+        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_{browser_name}_{ts}.png")
+            page.screenshot(path=path, full_page=True)
+
+        context.close()
+        browser.close()
+
+
+@pytest.fixture
+def ensure_login(admin_config, page, config):
+    """确保管理端已登录。"""
+
+    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:
+            auth_role = getattr(page, "_auth_role", None)
+            if auth_role == role:
+                page.goto(base_url)
+                page.wait_for_timeout(2000)
+                if "#/login" not in page.url and "/login" not in page.url:
+                    return
+                raise RuntimeError(
+                    f"登录态已过期！请运行: python tests/web/admin/save_auth.py --role {role}"
+                )
+            if auth_role and auth_role != role:
+                raise RuntimeError(
+                    f"当前 storage_state 角色为 '{auth_role}'，请求 '{role}'。\n"
+                    f"请运行: python tests/web/admin/save_auth.py --role {role}"
+                )
+            raise RuntimeError(
+                f"未找到角色 '{role}' 的登录态文件！请运行:\n"
+                f"  python tests/web/admin/save_auth.py --role {role}"
+            )
+
+        # skip_login: false
+        page.goto(f"{base_url}/#/login")
+        page.get_by_placeholder("请输入用户名").fill(account["phone"])
+        page.get_by_placeholder("请输入密码").fill(account["password"])
+        page.get_by_role("button", name="登录").click()
+        page.wait_for_timeout(3000)
+        if "#/login" in page.url or "/login" in page.url:
+            raise RuntimeError(
+                "登录失败！可能存在图形验证码。请改为手动登录: python tests/web/admin/save_auth.py"
+            )
+
+    return _ensure_login
+
+
+def _load_auth_state(context_kwargs: dict, storage_dir: Path, browser: str) -> str | None:
+    auth_files = list(storage_dir.glob(f"*_{browser}.json")) if storage_dir.exists() else []
+    if not auth_files:
+        return None
+    auth_file = auth_files[0]
+    context_kwargs["storage_state"] = str(auth_file)
+    return auth_file.stem.replace(f"_{browser}", "")
+
+
+@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/web/admin/pages/base_page.py

@@ -0,0 +1,76 @@
+"""Page Object 基类 — 封装所有页面共享的通用操作。
+
+每个具体页面应继承 BasePage，在 __init__ 中声明所有定位器。
+
+Usage:
+    class LoginPage(BasePage):
+        def __init__(self, page):
+            super().__init__(page)
+            self.phone_input = page.get_by_placeholder("请输入用户名")
+            self.login_btn = page.get_by_role("button", name="登录")
+"""
+
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+from playwright.sync_api import Page, Locator, expect
+
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+
+class BasePage:
+    """Page Object 基类。"""
+
+    def __init__(self, page: Page, timeout_ms: int = 30_000):
+        self.page = page
+        self.timeout = timeout_ms
+
+    # ── 通用等待 ──
+
+    def wait_visible(self, locator: Locator, timeout_ms: Optional[int] = None) -> None:
+        """等待元素可见。"""
+        expect(locator).to_be_visible(timeout=timeout_ms or self.timeout)
+
+    def wait_hidden(self, locator: Locator, timeout_ms: Optional[int] = None) -> None:
+        """等待元素消失（如 Loading 遮罩）。"""
+        expect(locator).to_be_hidden(timeout=timeout_ms or self.timeout)
+
+    def wait_for_text(self, text: str, timeout_ms: Optional[int] = None) -> None:
+        """等待页面上出现指定文本。"""
+        expect(self.page.get_by_text(text)).to_be_visible(timeout=timeout_ms or self.timeout)
+
+    # ── 通用操作 ──
+
+    def fill_field(self, locator: Locator, value: str) -> None:
+        """安全填写输入框（先等待可见，再清空后填写）。"""
+        self.wait_visible(locator)
+        locator.clear()
+        locator.fill(value)
+
+    def click_when_ready(self, locator: Locator) -> None:
+        """安全点击（等待可见 + 可操作）。"""
+        self.wait_visible(locator)
+        locator.click()
+
+    def get_text(self, locator: Locator) -> str:
+        """获取元素的 inner text。"""
+        self.wait_visible(locator)
+        return locator.inner_text()
+
+    # ── 截图 ──
+
+    def screenshot(self, name: str, full_page: bool = False) -> str:
+        """关键节点截图，返回文件路径。"""
+        scenario_dir = SCREENSHOTS_DIR / self.__class__.__name__
+        scenario_dir.mkdir(parents=True, exist_ok=True)
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        path = str(scenario_dir / f"{name}_{ts}.png")
+        self.page.screenshot(path=path, full_page=full_page)
+        return path
+
+    # ── 导航 ──
+
+    def navigate(self, url: str) -> None:
+        """导航到指定 URL 并等待页面加载完成。"""
+        self.page.goto(url, wait_until="domcontentloaded")

tests/web/admin/pages/components/navbar.py

@@ -0,0 +1,53 @@
+"""导航栏公共组件。
+
+跨页面复用的顶部/侧边导航操作。
+
+Usage:
+    from tests.web.admin.pages.components.navbar import NavBar
+
+    navbar = NavBar(page)
+    navbar.go_to("运输管理")
+"""
+
+from playwright.sync_api import Page, expect
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class NavBar(BasePage):
+    """侧边导航栏组件。"""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+        # ⚠️ 以下定位器需要根据实际 DOM 调整
+        self.sidebar = page.locator(".ant-layout-sider")
+        self.logout_btn = page.get_by_text("退出登录")
+
+    def go_to(self, menu_name: str) -> None:
+        """点击侧边栏菜单项展开/跳转。
+
+        Args:
+            menu_name: 菜单名称文本（如 "运输管理"、"基础信息"）
+        """
+        menu_item = self.page.get_by_text(menu_name).first
+        self.click_when_ready(menu_item)
+
+    def go_to_submenu(self, parent_menu: str, sub_menu: str) -> None:
+        """点击父级菜单后，点击子菜单项。
+
+        Args:
+            parent_menu: 父级菜单文本（如 "整车运输"）
+            sub_menu: 子菜单文本（如 "运单管理"）
+        """
+        self.go_to(parent_menu)
+        sub_item = self.page.get_by_text(sub_menu).first
+        self.click_when_ready(sub_item)
+
+    def logout(self) -> None:
+        """退出登录。"""
+        self.click_when_ready(self.logout_btn)
+        expect(self.page).to_have_url("#/login", timeout=10_000)
+
+    def is_expanded(self) -> bool:
+        """检查侧边栏是否展开。"""
+        return self.sidebar.is_visible()

tests/web/admin/pages/login_page.py

@@ -0,0 +1,54 @@
+"""登录页面 — Page Object。
+
+Usage:
+    from tests.web.admin.pages.login_page import LoginPage
+
+    login_page = LoginPage(page)
+    login_page.login(phone="super_admin", password="951260684NiAn..")
+"""
+
+from playwright.sync_api import Page, expect
+
+from tests.web.admin.pages.base_page import BasePage
+
+
+class LoginPage(BasePage):
+    """运八管理端登录页。"""
+
+    def __init__(self, page: Page):
+        super().__init__(page)
+        # ── 定位器声明 ──
+        self.username_input = page.get_by_placeholder("请输入用户名")
+        self.password_input = page.get_by_placeholder("请输入密码")
+        self.login_btn = page.get_by_role("button", name="登录")
+        self.error_tip = page.locator(".ant-form-item-explain-error")
+
+    def login(self, phone: str, password: str, expect_success: bool = True) -> None:
+        """执行登录流程。
+
+        Args:
+            phone: 用户名/手机号
+            password: 密码
+            expect_success: True=预期成功跳转，False=预期留在登录页并有错误提示
+        """
+        self.fill_field(self.username_input, phone)
+        self.fill_field(self.password_input, password)
+        self.click_when_ready(self.login_btn)
+
+        if expect_success:
+            # 等待登录成功后页面跳转（hash 路由模式）
+            self.page.wait_for_timeout(3000)
+            assert "/#/login" not in self.page.url and "/login" not in self.page.url, \
+                "登录失败，仍在登录页"
+        else:
+            # 预期登录失败，停留在当前页
+            expect(self.page).to_have_url("**/login**", timeout=5_000)
+
+    def get_error_message(self) -> str:
+        """获取登录失败时的错误提示文本。"""
+        self.wait_visible(self.error_tip)
+        return self.error_tip.inner_text()
+
+    def is_on_login_page(self) -> bool:
+        """检查是否仍在登录页（兼容 hash 路由）。"""
+        return "/login" in self.page.url or "/#/login" in self.page.url

tests/web/admin/save_auth.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""手动登录并保存浏览器登录态（cookies + localStorage）。
+
+目的: 绕过图形验证码——人工完成一次验证码登录后，
+      cookies + localStorage 序列化到 JSON 文件，后续用例自动复用。
+
+用法:
+    python tests/web/admin/save_auth.py
+
+流程:
+    1. 脚本打开浏览器到登录页
+    2. 你在浏览器中手动输入账号、密码、完成验证码并点击登录
+    3. 登录成功后按 Enter 键，脚本自动保存登录态
+    4. 后续 pytest 用例自动复用
+
+生成的 auth 文件:
+    tests/web/auth/super_admin_chromium.json
+"""
+
+import argparse
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from tests.config.test_config import load_config
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="手动登录并保存浏览器登录态（绕过图形验证码）"
+    )
+    parser.add_argument(
+        "--role",
+        default="super_admin",
+        help="登录角色（默认: super_admin），会体现在文件名中",
+    )
+    parser.add_argument(
+        "--browser",
+        default="chromium",
+        choices=["chromium", "firefox", "webkit"],
+        help="浏览器类型（默认: chromium）",
+    )
+    args = parser.parse_args()
+
+    config = load_config()
+    web_config = config["admin"]
+    base_url = web_config["base_url"]
+    login_url = f"{base_url}/#/login"
+
+    from playwright.sync_api import sync_playwright
+
+    with sync_playwright() as p:
+        browser_launcher = getattr(p, args.browser)
+        browser = browser_launcher.launch(headless=False)
+        context = browser.new_context(
+            viewport={"width": 1920, "height": 1080},
+            locale="zh-CN",
+        )
+        page = context.new_page()
+
+        print(f"\n🔗 打开登录页: {login_url}")
+        page.goto(login_url)
+
+        print(f"\n{'='*60}")
+        print(f"👤 角色: {args.role}")
+        print(f"   请在浏览器中手动完成登录（输入账号密码 + 验证码）")
+        print(f"   登录成功后回到此处按 Enter 键保存登录态...")
+        print(f"{'='*60}\n")
+
+        input(">>> 按 Enter 保存登录态...")
+
+        # 检查是否真的登录成功了
+        current = page.url
+        if "/login" in current or "/#/login" in current:
+            print(f"\n⚠️ 当前仍在登录页: {current}")
+            print("   要保存吗？(y/n): ", end="")
+            choice = input().strip().lower()
+            if choice != "y":
+                print("已取消。")
+                browser.close()
+                sys.exit(0)
+
+        print(f"\n✅ 当前页面: {current}")
+
+        auth_dir = Path(web_config.get("storage_state_dir", "tests/web/auth"))
+        auth_dir.mkdir(parents=True, exist_ok=True)
+        auth_file = auth_dir / f"{args.role}_{args.browser}.json"
+
+        context.storage_state(path=str(auth_file))
+        print(f"💾 登录态已保存到: {auth_file}")
+
+        browser.close()
+
+    print(f"\n🎉 现在可以跑了: pytest tests/web/admin/ -m p0")
+
+
+if __name__ == "__main__":
+    main()

tests/web/admin/test_login/test_login.py

@@ -0,0 +1,74 @@
+"""管理端登录测试用例。
+
+标记:
+    pytest -m admin            # 仅管理端用例
+    pytest -m p0               # P0 冒烟
+    pytest -m smoke            # 冒烟
+"""
+
+import pytest
+from playwright.sync_api import Page
+
+pytestmark = [pytest.mark.admin]
+
+
+class TestAdminSmoke:
+    """管理端冒烟 — 验证预登录态。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_admin_access_with_saved_auth(self, page: Page, admin_config, ensure_login):
+        """验证预登录态可直接访问管理后台。"""
+        ensure_login("super_admin")
+        page.goto(admin_config["base_url"])
+        page.wait_for_timeout(2000)
+        assert "#/login" not in page.url and "/login" not in page.url, (
+            "预登录态失效！请运行: python tests/web/admin/save_auth.py"
+        )
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_team_leader_access(self, page: Page, admin_config, ensure_login):
+        """验证车队长预登录态有效。"""
+        ensure_login("team_leader")
+        page.goto(admin_config["base_url"])
+        page.wait_for_timeout(2000)
+        assert "#/login" not in page.url and "/login" not in page.url
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_dashboard_loads(self, page: Page, admin_config, ensure_login):
+        """验证首页可正常加载。"""
+        ensure_login("super_admin")
+        page.goto(admin_config["base_url"])
+        assert page.title(), "首页应显示页面标题"
+
+
+class TestLoginPage:
+    """登录页交互验证。"""
+
+    @pytest.mark.p1
+    def test_login_page_accessible(self, page: Page, admin_config):
+        """验证登录页可正常访问。"""
+        page.goto(f"{admin_config['base_url']}/#/login")
+        assert page.get_by_placeholder("请输入用户名").is_visible()
+
+    @pytest.mark.p1
+    def test_login_page_elements(self, page: Page, admin_config):
+        """验证登录页元素完整。"""
+        page.goto(f"{admin_config['base_url']}/#/login")
+        assert page.get_by_placeholder("请输入用户名").is_visible()
+        assert page.get_by_placeholder("请输入密码").is_visible()
+        assert page.get_by_role("button", name="登录").is_visible()
+
+    @pytest.mark.p1
+    def test_session_redirect(self, page: Page, admin_config, ensure_login):
+        """验证已登录用户访问登录页被重定向。"""
+        ensure_login("super_admin")
+        page.goto(admin_config["base_url"])
+        page.wait_for_timeout(2000)
+        assert "#/login" not in page.url
+
+        page.goto(f"{admin_config['base_url']}/#/login")
+        page.wait_for_timeout(2000)
+        assert "#/login" not in page.url and "/login" not in page.url

tests/web/mini_program/README.md

@@ -0,0 +1,68 @@
+# 司机端小程序 — miniprogram-automator
+
+## 环境搭建
+
+### 1. 安装微信开发者工具
+
+- [下载地址](https://developers.weixin.qq.com/miniprogram/dev/devtools/download.html)
+- 安装后打开**设置 → 安全 → 安全**，开启**服务端口**
+
+### 2. 安装 Node.js 依赖
+
+```bash
+cd tests/web/mini_program/automator
+npm install
+```
+
+### 3. 编译 uni-app 为微信小程序
+
+```bash
+# 回到项目根目录（即 uni-app 源码所在目录）
+npm run build:mp-weixin
+```
+
+### 4. 配置 test_config.yaml
+
+```yaml
+mini_program:
+  project_path: "E:/path/to/your/uniapp/dist/dev/mp-weixin"  # uni-app 编译产物目录
+  dev_tool_path: "C:/Program Files (x86)/Tencent/微信web开发者工具/cli.bat"
+```
+
+## 运行
+
+```bash
+pytest tests/web/mini_program/ -v -m p0
+```
+
+## 原理
+
+```
+pytest 用例
+  ↓ conftest.py: subprocess
+  ↓
+node runner.js --action smoke --project-path ...
+  ↓ miniprogram-automator API
+  ↓
+微信开发者工具（开启服务端口）
+  ↓
+小程序运行
+  ↓
+返回 JSON { ok, message, screenshot }
+  ↓
+Python 断言
+```
+
+## 可用 actions
+
+| action | 参数 | 说明 |
+| :--- | :--- | :--- |
+| `smoke` | — | 启动小程序并检查首页 |
+| `navigate` | `page=/pages/xxx/xxx` | 导航到指定页面 |
+| `element` | `element-action=tap/exists` `selector=.class` | 操作/检查元素 |
+
+## 注意事项
+
+- **登录**：首次在开发者工具中手动登录，后续 `miniprogram-automator` 启动默认复用登录态
+- **截图**：输出到 `output/screenshots/mini_program/`
+- **CI**：微信开发者工具必须安装在 CI 机器上，且需要 GUI 环境（不能纯 headless）

tests/web/mini_program/automator/package.json

@@ -0,0 +1,11 @@
+{
+  "name": "yunba-miniprogram-automator",
+  "description": "运八司机端小程序自动化测试 — miniprogram-automator 封装",
+  "private": true,
+  "scripts": {
+    "test": "node runner.js"
+  },
+  "dependencies": {
+    "miniprogram-automator": "^0.10.0"
+  }
+}

tests/web/mini_program/automator/pages/home_page.js

@@ -0,0 +1,30 @@
+/** 小程序 Page Object（Node 版）— 首页 */
+
+class HomePage {
+  constructor(miniProgram) {
+    this.mp = miniProgram;
+  }
+
+  async goToOrderList() {
+    const page = await this.mp.currentPage();
+    const el = await page.$(".order-entry, .order-item");
+    if (el) await el.tap();
+    await this.mp.waitFor(1000);
+  }
+
+  async goToMine() {
+    const page = await this.mp.currentPage();
+    const el = await page.$(".mine-tab, .my-entry");
+    if (el) await el.tap();
+    await this.mp.waitFor(1000);
+  }
+
+  async screenshot(name) {
+    const path = require("path");
+    const file = path.resolve(__dirname, `../../../../output/screenshots/mini_program/${name}_${Date.now()}.png`);
+    await this.mp.screenshot({ path: file });
+    return file;
+  }
+}
+
+module.exports = { HomePage };

tests/web/mini_program/automator/runner.js

@@ -0,0 +1,164 @@
+/**
+ * 运八司机端小程序 — miniprogram-automator 测试执行器。
+ *
+ * 由 Python conftest.py 通过 subprocess 驱动，接收 JSON 指令，返回 JSON 结果。
+ *
+ * 用法:
+ *   node runner.js --action smoke --project-path "E:/xxx/dist"
+ *
+ * 返回 JSON:
+ *   { "ok": true/false, "message": "...", "screenshot": "path/to.png" }
+ */
+
+const automator = require("miniprogram-automator");
+const path = require("path");
+const fs = require("fs");
+const { execSync } = require("child_process");
+
+// ═══════════════════════════════════════════════════════════════
+// 参数解析
+// ═══════════════════════════════════════════════════════════════
+const args = process.argv.slice(2);
+function getArg(name) {
+  const idx = args.indexOf(`--${name}`);
+  return idx >= 0 ? args[idx + 1] : null;
+}
+
+const ACTION = getArg("action") || "smoke";
+const PROJECT_PATH = getArg("project-path") || ".";
+const DEV_TOOL_PATH =
+  getArg("dev-tool-path") || "C:/Program Files (x86)/Tencent/微信web开发者工具/cli.bat";
+const TIMEOUT = parseInt(getArg("timeout")) || 30000;
+
+const SCREENSHOTS_DIR = path.resolve(__dirname, "../../../../output/screenshots/mini_program");
+fs.mkdirSync(SCREENSHOTS_DIR, { recursive: true });
+
+// ═══════════════════════════════════════════════════════════════
+// 辅助
+// ═══════════════════════════════════════════════════════════════
+function ok(msg, screenshot) {
+  console.log(JSON.stringify({ ok: true, message: msg, screenshot: screenshot || null }));
+  process.exit(0);
+}
+
+function fail(msg, screenshot) {
+  console.error(JSON.stringify({ ok: false, message: msg, screenshot: screenshot || null }));
+  process.exit(1);
+}
+
+async function screenshot(miniProgram, name) {
+  const file = path.join(SCREENSHOTS_DIR, `${name}_${Date.now()}.png`);
+  await miniProgram.screenshot({ path: file });
+  return file;
+}
+
+// ═══════════════════════════════════════════════════════════════
+// Actions
+// ═══════════════════════════════════════════════════════════════
+
+async function actionSmoke() {
+  let miniProgram;
+  try {
+    miniProgram = await automator.launch({
+      projectPath: PROJECT_PATH,
+      cliPath: DEV_TOOL_PATH,
+    });
+
+    // 等小程序启动
+    await miniProgram.waitFor(5000);
+
+    // 检查是否启动成功
+    const currentPage = await miniProgram.currentPage();
+    if (!currentPage) {
+      return fail("小程序启动后未获取到页面", null);
+    }
+
+    // 验证首页元素
+    const elements = await currentPage.$$("view");
+    const ss = await screenshot(miniProgram, "smoke_home");
+
+    return ok(`首页加载成功，检测到 ${elements.length} 个 view 元素`, ss);
+  } catch (e) {
+    return fail(`小程序启动失败: ${e.message}`, null);
+  } finally {
+    if (miniProgram) await miniProgram.close();
+  }
+}
+
+async function actionNavigate(pagePath) {
+  let miniProgram;
+  try {
+    miniProgram = await automator.launch({
+      projectPath: PROJECT_PATH,
+      cliPath: DEV_TOOL_PATH,
+    });
+    await miniProgram.waitFor(3000);
+
+    const currentPage = await miniProgram.currentPage();
+    await currentPage.callMethod("navigateTo", { url: pagePath });
+    await miniProgram.waitFor(2000);
+
+    const newPage = await miniProgram.currentPage();
+    const ss = await screenshot(miniProgram, `nav_${pagePath.replace(/\//g, "_")}`);
+    const currentPath = await newPage.path();
+
+    return ok(`导航到 ${pagePath}，当前路径: ${currentPath}`, ss);
+  } catch (e) {
+    return fail(`页面导航失败 ${pagePath}: ${e.message}`, null);
+  } finally {
+    if (miniProgram) await miniProgram.close();
+  }
+}
+
+async function actionElement(actionType, selector) {
+  let miniProgram;
+  try {
+    miniProgram = await automator.launch({
+      projectPath: PROJECT_PATH,
+      cliPath: DEV_TOOL_PATH,
+    });
+    await miniProgram.waitFor(3000);
+
+    const currentPage = await miniProgram.currentPage();
+
+    if (actionType === "exists") {
+      const el = await currentPage.$(selector);
+      const exists = el !== null;
+      const ss = exists ? null : await screenshot(miniProgram, `missing_${selector}`);
+      return ok(exists ? `元素 "${selector}" 存在` : `元素 "${selector}" 不存在`, ss);
+    }
+
+    if (actionType === "tap") {
+      const el = await currentPage.$(selector);
+      if (!el) {
+        return fail(`元素 "${selector}" 不存在，无法点击`, null);
+      }
+      await el.tap();
+      await miniProgram.waitFor(1000);
+      const ss = await screenshot(miniProgram, `tap_${selector}`);
+      return ok(`点击 "${selector}" 成功`, ss);
+    }
+
+    return fail(`未知子操作: ${actionType}`, null);
+  } catch (e) {
+    return fail(`操作失败 ${actionType}/${selector}: ${e.message}`, null);
+  } finally {
+    if (miniProgram) await miniProgram.close();
+  }
+}
+
+// ═══════════════════════════════════════════════════════════════
+// 入口
+// ═══════════════════════════════════════════════════════════════
+(async () => {
+  switch (ACTION) {
+    case "smoke":
+      return actionSmoke();
+    case "navigate":
+      return actionNavigate(getArg("page") || "/pages/index/index");
+    case "element":
+      return actionElement(getArg("element-action") || "exists", getArg("selector") || "view");
+    default:
+      return fail(`未知 action: ${ACTION}`, null);
+  }
+})();

tests/web/mini_program/conftest.py

@@ -0,0 +1,90 @@
+"""司机端小程序 fixture — 通过 subprocess 驱动 miniprogram-automator (Node.js)。
+
+原理:
+    Python pytest → subprocess → node runner.js → miniprogram-automator
+                                           → 微信开发者工具 → 小程序
+    返回 JSON: {"ok": true/false, "message": "...", "screenshot": "path"}
+
+前置条件:
+    1. 微信开发者工具已安装 + 开启服务端口
+    2. npm install（在 tests/web/mini_program/automator/ 下）
+    3. uni-app 已编译为微信小程序
+"""
+
+import json
+import subprocess
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
+RUNNER = Path(__file__).parent / "automator" / "runner.js"
+SCREENSHOTS_DIR = Path("output/screenshots")
+
+
+def _run_runner(project_path: str, action: str, **kwargs) -> dict:
+    """调用 node runner.js，返回解析后的 JSON。"""
+    cmd = [
+        "node", str(RUNNER),
+        "--action", action,
+        "--project-path", project_path,
+        "--timeout", str(kwargs.get("timeout", 30000)),
+    ]
+    for key, value in kwargs.items():
+        if value is not None and key not in ("timeout",):
+            cmd.extend([f"--{key.replace('_', '-')}", str(value)])
+
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, cwd=str(PROJECT_ROOT))
+
+    try:
+        data = json.loads(result.stdout.strip())
+    except json.JSONDecodeError:
+        data = {"ok": False, "message": result.stderr or result.stdout, "screenshot": None}
+
+    data["_raw_stdout"] = result.stdout
+    data["_raw_stderr"] = result.stderr
+    return data
+
+
+@pytest.fixture
+def mini_program_config(config):
+    """小程序配置。"""
+    return config["mini_program"]
+
+
+@pytest.fixture
+def run_mini_action(mini_program_config):
+    """调用 miniprogram-automator 执行一个操作并返回结果。
+
+    Usage:
+        result = run_mini_action("smoke")
+        assert result["ok"], result["message"]
+        # result["screenshot"] 是截图路径
+    """
+
+    def _run(action: str, **kwargs) -> dict:
+        project_path = mini_program_config.get("project_path", ".")
+        dev_tool = mini_program_config.get("dev_tool_path", "")
+        timeout = mini_program_config.get("timeout_seconds", 120) * 1000
+
+        return _run_runner(
+            project_path,
+            action,
+            dev_tool_path=dev_tool,
+            timeout=timeout,
+            **kwargs,
+        )
+
+    return _run
+
+
+def save_failure_screenshot(name: str) -> str:
+    """保存 Python 端的失败截图（如 runner 调用超时）。"""
+    scenario_dir = SCREENSHOTS_DIR / "mini_program"
+    scenario_dir.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = str(scenario_dir / f"FAIL_{name}_{ts}.png")
+    # 此层无法截图小程序，创建空标记文件用于日志追溯
+    Path(path).touch()
+    return path

tests/web/mini_program/tests/test_smoke.py

@@ -0,0 +1,43 @@
+"""小程序冒烟测试 — 通过 miniprogram-automator 驱动。
+
+用法:
+    pytest tests/web/mini_program/ -m p0
+"""
+
+import pytest
+
+pytestmark = [pytest.mark.mini_program]
+
+
+class TestMiniProgramSmoke:
+    """小程序启动 + 基础页面验证。"""
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_mini_program_launches(self, run_mini_action):
+        """验证小程序能正常启动并展示首页。"""
+        result = run_mini_action("smoke")
+        assert result["ok"], f"小程序启动失败: {result.get('message')}"
+
+    @pytest.mark.p0
+    @pytest.mark.smoke
+    def test_navigate_to_pages(self, run_mini_action):
+        """验证关键页面可正常导航。"""
+        pages = ["/pages/order/index", "/pages/mine/index"]
+        for p in pages:
+            result = run_mini_action("navigate", page=p)
+            assert result["ok"], f"导航到 {p} 失败: {result.get('message')}"
+
+
+class TestMiniProgramElements:
+    """关键元素验证。"""
+
+    @pytest.mark.p1
+    @pytest.mark.parametrize("selector,label", [
+        pytest.param(".order-entry", "运单入口", id="order_entry"),
+        pytest.param(".mine-tab", "我的tab", id="mine_tab"),
+    ])
+    def test_key_element_exists(self, run_mini_action, selector, label):
+        """验证关键 UI 元素存在。"""
+        result = run_mini_action("element", element_action="exists", selector=selector)
+        assert result["ok"], f"{label} ({selector}) 不存在: {result.get('message')}"