Files
Yb-QaAutomationHub/tests/app/_shared.py
T
xst f75a6f5ec9 feat(cross_end): add driver-to-settlement E2E test with SQL retry
- Add DriverFreightFlowPage (Android): 28-step flow covering accept/load/unload
- Add FreightAccountingPage (Web): accounting/audit/finance-payment/shipper-payment
- Add db_helper: update_risk_control_status via pymysql
- Fix ensure_logged_in: coordinate tap on 'mine' tab + login check
- Add test_driver_to_settlement.py: full cross-end E2E with SQL retry

Co-authored-by: xst <xst@ntocc.com>
2026-07-22 17:18:52 +08:00

174 lines
6.3 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 闭包。
策略:坐标点击底部"我的"tab(第4个图标)→ 检查"登录/注册"文本。
无法通过 resourceId 或 text 定位底部导航时用坐标点击兜底。
Args:
section_config: App 端配置
config: 全局配置
account_role: 默认账号角色名(如 "driver"、"shipper"
home_indicator: 未使用,保留兼容
"""
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)
driver.implicitly_wait(3)
# 坐标点击底部"我的"tab4 tab 均分宽度,第4个中心 x≈0.875*w, y≈h-50
try:
w = driver.get_window_size()["width"]
h = driver.get_window_size()["height"]
driver.tap([(int(w * 0.875), h - 50)])
driver.implicitly_wait(1)
except Exception as e:
raise RuntimeError(f"无法点击底部导航栏: {e}")
# 判断登录态:检查"登录/注册"文本
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 # 没找到"登录/注册" → 已登录
# 回到首页
home_tab = (
AppiumBy.ANDROID_UIAUTOMATOR,
'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)',
)
try:
wait.until(EC.visibility_of_element_located(home_tab))
driver.find_element(*home_tab).click()
except Exception:
pass
return _ensure_logged_in
def _perform_ui_login(driver, account: dict, wait) -> None:
"""执行 UI 自动登录(skip_login: false 时使用)。"""
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, 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)