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.
2026-07-21 09:10:08 +08:00
parent b822428a2a
commit 6f5668b7ec
15 changed files with 1065 additions and 0 deletions
@@ -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)