Yb-QaAutomationHub/tests/core/appium_support.py

"""Appium 支持 — driver 工厂 + 登录态复用 + 失败截图。

各 App 端 conftest 调用本模块即可，无需重复实现。
"""

from datetime import datetime

import pytest

from tests.core.allure_support import allure, attach_png, attach_text
from tests.core.paths import SCREENSHOTS_DIR


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):
    """创建一个 ensure_logged_in 闭包。

    策略：
        1. 回到首页（优先 start_activity 跳转，fallback back 键循环）
        2. 点击"我的"tab，检查"登录/注册"文本是否存在
        3. 不存在 → 已登录；存在 → 未登录

    Args:
        section_config: App 端配置
        config: 全局配置
        account_role: 默认账号角色名（如 "driver"、"shipper"）
    """
    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)

        app_package = section_config.get("app_package", "")
        app_activity = section_config.get("app_activity", "")

        # ── 底部导航栏三个 tab 定位（使用 text 定位，更可靠）──
        home_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("首页")')
        freight_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("货源")')
        mine_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("我的")')

        # ── 步骤1: 回到首页 ──
        # 策略：优先用 start_activity 直接跳转到主 Activity（最快最可靠），
        # 失败时 fallback 到 back 键循环
        _navigate_to_home(driver, app_package, app_activity, home_tab, freight_tab, mine_tab)

        # ── 步骤2: 点击"我的"tab ──
        wait.until(EC.visibility_of_element_located(mine_tab))
        driver.find_element(*mine_tab).click()
        driver.implicitly_wait(1)

        # ── 步骤3: 判断登录态 ──
        login_check = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
        try:
            el = driver.find_element(*login_check)
            if el.is_displayed():
                if skip_login:
                    raise RuntimeError(
                        "APP 未登录！请在设备上手动完成一次登录（含验证码），"
                        "之后重新运行用例即可。\n如需自动化登录，请设置 skip_login: false。"
                    )
                _perform_ui_login(driver, config["accounts"].get(account_role, {}), wait)
                return
        except RuntimeError:
            raise
        except Exception:
            pass  # 没找到"登录/注册" → 已登录

        # ── 步骤4: 回到首页 ──
        try:
            wait.until(EC.visibility_of_element_located(home_tab))
            driver.find_element(*home_tab).click()
        except Exception:
            pass

    return _ensure_logged_in


def _navigate_to_home(driver, app_package: str, app_activity: str,
                      home_tab, freight_tab, mine_tab) -> None:
    """回到 APP 主页。

    策略:
        1) 优先用 start_activity 直接跳到主 Activity（深层页面中 back 不可靠）
        2) fallback: back 键循环（适用于浅层页面）
    """
    # ── 策略1: 直接跳到主 Activity ──
    if app_package and app_activity:
        try:
            driver.start_activity(app_package, app_activity)
            driver.implicitly_wait(3)
            if _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
                return
        except Exception:
            pass

    # ── 策略2: back 键循环 ──
    for i in range(30):
        if _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
            return
        try:
            driver.back()
            driver.implicitly_wait(1.5)
        except Exception:
            break

    # 最终检查：如果仍然不在一级页面，抛异常而不是让后续步骤无意义超时
    if not _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
        try:
            current = driver.current_activity
        except Exception:
            current = "unknown"
        raise RuntimeError(
            f"无法回到 APP 首页！当前 Activity: {current}。"
            "请检查模拟器中 APP 是否正常运行，或手动将 APP 退到首页后重试。"
        )


def _all_tabs_visible(driver, home_tab, freight_tab, mine_tab) -> bool:
    """检查底部三个导航 tab 是否全部可见。"""
    return (
        _is_displayed(driver, home_tab)
        and _is_displayed(driver, freight_tab)
        and _is_displayed(driver, mine_tab)
    )


def _is_displayed(driver, locator) -> bool:
    """检查元素是否可见（不抛异常）。"""
    try:
        el = driver.find_element(*locator)
        return el.is_displayed()
    except Exception:
        return False


def _perform_ui_login(driver, account: dict, wait) -> None:
    """执行 UI 自动登录（skip_login: false 时使用）。当前仅适配司机 APP。"""
    from appium.webdriver.common.appiumby import AppiumBy

    phone = account.get("phone", "")
    password = account.get("password", "")

    # 点击登录入口
    login_entry = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
    wait.until(lambda d: d.find_element(*login_entry))
    driver.find_element(*login_entry).click()
    driver.implicitly_wait(1)

    # 填写手机号
    phone_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_phone")
    wait.until(lambda d: d.find_element(*phone_input))
    driver.find_element(*phone_input).send_keys(phone)

    # 填写密码
    password_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_pwd")
    driver.find_element(*password_input).send_keys(password)

    # 点击登录
    login_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_login")
    driver.find_element(*login_btn).click()
    driver.implicitly_wait(2)

    # 回到首页
    home_tab = (
        AppiumBy.ANDROID_UIAUTOMATOR,
        'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)',
    )
    try:
        wait.until(lambda d: d.find_element(*home_tab))
        driver.find_element(*home_tab).click()
    except Exception:
        pass


def save_failure_screenshot(driver, section_config: dict, platform: str, request) -> str | None:
    """用例失败时截图 + 页面源码快照（存盘/attach Allure），返回路径；未失败返回 None。"""
    if not (
        section_config.get("screenshot_on_failure", True)
        and hasattr(request.node, "rep_call")
        and request.node.rep_call.failed
    ):
        return None
    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)
    attach_png(path, name=f"失败截图-{platform}")
    # page_source 快照: 定位器失效时直接看到当时的控件层级
    try:
        attach_text(driver.page_source, name=f"页面源码-{platform}",
                    attachment_type=allure.attachment_type.XML)
    except Exception:
        pass
    return path