Yb-QaAutomationHub/tests/cross_end/conftest.py

"""跨端 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)