Yb-QaAutomationHub/tests/app/_shared.py

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