feat(core,utils): 新增CoreDriver基础操作,更新文档

- 新增 switch_to_webview/switch_to_native 切换视图。
- 新增 config_loader.py 配置文件系统
- 优化 conftest.py,支持获取设备信息和默认参数。
- 优化 run_appium.py
- 更新 README.md
- 其他优化
This commit is contained in:
2026-02-06 17:28:04 +08:00
parent 483a31793d
commit 52758940ed
13 changed files with 495 additions and 199 deletions

67
core/config_loader.py Normal file
View File

@@ -0,0 +1,67 @@
#!/usr/bin/env python
# coding=utf-8
"""
@author: CNWei,ChenWei
@Software: PyCharm
@contact: t6g888@163.com
@file: config_loader
@date: 2026/1/16 10:52
@desc: Pytest 核心配置与 Fixture 管理
"""
from typing import Any
from utils.data_loader import load_yaml
from core.settings import CAPS_CONFIG_PATH, ENV_CONFIG, CURRENT_ENV
from core.modules import AppPlatform
def get_env_config() -> dict[str, str]:
"""
根据当前环境 (CURRENT_ENV) 获取对应的业务配置
"""
return ENV_CONFIG.get(CURRENT_ENV, ENV_CONFIG["test"])
def get_caps(platform: str) -> dict[str, Any]:
"""
从 YAML 文件加载 Capabilities 配置
"""
try:
all_caps = load_yaml(CAPS_CONFIG_PATH)
all_caps = {k.lower(): v for k, v in all_caps.items()}
platform_key = platform.lower()
if platform_key not in all_caps:
base_caps: dict[str, Any] = {
"noReset": False,
"newCommandTimeout": 60,
}
match platform_key:
case AppPlatform.ANDROID:
android_caps = {
"platformName": "Android",
"automationName": "uiautomator2",
"deviceName": "Android",
"appPackage": "com.manu.wanandroid",
"appActivity": "com.manu.wanandroid.ui.main.activity.MainActivity",
}
return base_caps | android_caps
case AppPlatform.IOS:
ios_caps = {
"platformName": "iOS",
"automationName": "XCUITest",
"deviceName": "iPhone 14",
"bundleId": "com.example.app",
"autoAcceptAlerts": True,
"waitForQuiescence": False,
}
return base_caps | ios_caps
return all_caps[platform_key]
except Exception as e:
raise RuntimeError(f"无法加载“caps”内容 {CAPS_CONFIG_PATH}: {e}")

View File

@@ -585,6 +585,44 @@ class CoreDriver:
duration
)
# --- 上下文切换 (Context Switch) ---
def switch_to_context(self, context_name: str) -> 'CoreDriver':
"""
切换到指定的上下文 (Context)。
:param context_name: 上下文名称 (如 'NATIVE_APP', 'WEBVIEW_com.example.app')
:return: self
"""
logger.info(f"尝试切换到上下文: {context_name}")
try:
self.driver.switch_to.context(context_name)
logger.info(f"成功切换到上下文: {context_name}")
except Exception as e:
logger.error(f"切换上下文失败: {e}")
available_contexts = self.driver.contexts
logger.info(f"当前可用上下文: {available_contexts}")
raise e
return self
def switch_to_webview(self) -> 'CoreDriver':
"""
自动切换到第一个可用的 WebView 上下文。
:return: self
"""
logger.info("尝试自动切换到 WebView 上下文...")
contexts = self.driver.contexts
for context in contexts:
if 'WEBVIEW' in context:
return self.switch_to_context(context)
logger.warning("未找到 WebView 上下文,保持在当前上下文。")
return self
def switch_to_native(self) -> 'CoreDriver':
"""
切换回原生应用上下文 (NATIVE_APP)。
:return: self
"""
return self.switch_to_context('NATIVE_APP')
def full_screen_screenshot(self, name: str | None = None) -> str:
"""
截取当前完整屏幕内容 (自愈逻辑、异常报错首选)

View File

@@ -29,6 +29,30 @@ logger = logging.getLogger(__name__)
# 使用 npm run 确保 APPIUM_HOME=. 变量和本地版本生效
# NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}"
# --- 1. 异常体系设计 (精确分类) ---
class AppiumStartupError(Exception):
"""所有 Appium 启动相关异常的基类"""
pass
class AppiumPortConflictError(AppiumStartupError):
"""端口被其他非 Appium 程序占用"""
pass
class AppiumProcessCrashError(AppiumStartupError):
"""进程启动后立即崩溃或在运行中意外退出"""
pass
class AppiumTimeoutError(AppiumStartupError):
"""服务在规定时间内未进入 READY 状态"""
pass
class AppiumInternalError(AppiumStartupError):
"""探测接口返回了无法解析的数据或网络异常"""
pass
class AppiumStatus(Enum):
@@ -43,12 +67,12 @@ class AppiumStatus(Enum):
class ServiceRole(Enum):
"""服务角色枚举:定义服务的所有权和生命周期"""
MANAGED = "由脚本启动 (托管模式)" # 由本脚本启动,负责清理
EXTERNAL = "复用外部服务 (共享模式)" # 复用现有服务,不负责清理
NULL = "无效服务 (空模式)" # 无效或未初始化的服务
MANAGED = "托管模式" # 由本脚本启动,负责清理
EXTERNAL = "共享模式" # 复用现有服务,不负责清理
NULL = "空模式" # 无效或未初始化的服务
def resolve_appium_command() -> List[str]:
def resolve_appium_command(host: str, port: int | str) -> List[str]:
"""
解析 Appium 可执行文件的绝对路径。
优先查找项目 node_modules 下的本地安装版本,避免 npm 包装层带来的信号传递问题。
@@ -64,8 +88,8 @@ def resolve_appium_command() -> List[str]:
logger.error(f"\n错误: 在路径 {appium_bin} 未找到 Appium 执行文件。")
logger.info("请确保已在项目目录下执行过: npm install appium")
sys.exit(1)
# 返回执行列表(用于 shell=False
return [str(appium_bin), "-p", str(APPIUM_PORT)]
# 返回执行列表(用于 shell=False--address 127.0.0.1 --port 4723 appium -a 127.0.0.1 -p 4723
return [str(appium_bin), "--address", host, "--port", str(port)]
# 移除全局调用,防止 import 时因找不到 appium 而直接退出
@@ -79,21 +103,27 @@ def _cleanup_process_tree(process: subprocess.Popen = None) -> None:
"""
if not process or process.poll() is not None:
return
logger.info(f"正在关闭 Appium 进程树 (PID: {process.pid})...")
pid = process.pid
logger.info(f"正在关闭 Appium 进程树 (PID: {pid})...")
try:
if sys.platform == "win32":
# Windows 下使用 taskkill 强制关闭进程树 /T
subprocess.run(
['taskkill', '/F', '/T', '/PID', str(process.pid)],
['taskkill', '/F', '/T', '/PID', str(pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
else:
# Unix/Linux: 获取进程组 ID 并发送终止信号
pgid = os.getpgid(process.pid)
pgid = os.getpgid(pid)
os.killpg(pgid, signal.SIGTERM)
process.wait(timeout=5)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
# 5秒没关掉强制杀掉整个进程组
# 如果 SIGTERM 不起作用,直接 SIGKILL 整个组
logger.warning(f"进程 {pid} 未响应 SIGTERM发送 SIGKILL...")
os.killpg(pgid, signal.SIGKILL)
logger.info("Appium 服务已停止,相关进程已安全退出。")
except Exception as e:
logger.error(f"停止服务时发生异常: {e}")
@@ -109,10 +139,21 @@ def _cleanup_process_tree(process: subprocess.Popen = None) -> None:
class AppiumService:
"""Appium 服务实例封装,用于管理服务生命周期"""
def __init__(self, role: ServiceRole, process: subprocess.Popen = None):
def __init__(self, role: ServiceRole, host: str, port: int | str, process: subprocess.Popen = None):
self.role = role
self.host = host
self.port = port
self.process = process
def __repr__(self):
return f"<AppiumService 角色='{self.role.value}' 地址=http://{self.host}:{self.port}>"
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
def stop(self):
"""统一停止接口:根据角色决定是否关闭进程"""
match self.role:
@@ -120,28 +161,27 @@ class AppiumService:
logger.info(f"--> [角色: {self.role.value}] 脚本退出,保留外部服务运行。")
return
case ServiceRole.MANAGED:
logger.info(f"正在关闭托管的 Appium 服务 (PID: {self.process.pid})...")
_cleanup_process_tree(self.process)
self.process = None
case ServiceRole.NULL:
logger.info(f"--> [角色: {self.role.value}] 无需执行清理。")
def __repr__(self):
return f"<AppiumService 角色='{self.role.value}' 端口={APPIUM_PORT}>"
def _check_port_availability() -> AppiumStatus:
def _check_port_availability(host: str, port: int) -> AppiumStatus:
"""辅助函数:检查端口是否被占用"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.settimeout(1)
# 尝试绑定端口,如果成功说明端口空闲
sock.bind((APPIUM_HOST, APPIUM_PORT))
sock.bind((host, port))
return AppiumStatus.OFFLINE # 真正未启动
except OSError:
# 绑定失败,说明端口被占用,但此前 HTTP 请求失败,说明不是 Appium
return AppiumStatus.CONFLICT # 端口被占用但没响应 HTTP
def get_appium_status() -> AppiumStatus:
def get_appium_status(host: str, port: int | str) -> AppiumStatus:
"""
检测 Appium 服务当前状态。
@@ -153,16 +193,17 @@ def get_appium_status() -> AppiumStatus:
:return: AppiumStatus 枚举
"""
connection = None
port = int(port) # 确保端口是整数
try:
# 1. 端口开启,尝试获取 Appium 状态接口
connection = http.client.HTTPConnection(APPIUM_HOST, APPIUM_PORT, timeout=2)
connection = http.client.HTTPConnection(host, port, timeout=2)
connection.request("GET", "/status")
response = connection.getresponse()
if response.status != 200:
return AppiumStatus.CONFLICT
data = json.loads(response.read().decode())
data = json.loads(response.read().decode('utf-8'))
# 2. 解析 Appium 3.x 标准响应结构
# 即使服务响应了,也要看驱动是否加载完成 (ready 字段)
@@ -170,8 +211,8 @@ def get_appium_status() -> AppiumStatus:
return AppiumStatus.READY if is_ready else AppiumStatus.INITIALIZING
except (socket.error, ConnectionRefusedError):
# 3. 如果通信拒绝,检查端口是否真的空闲
return _check_port_availability()
# 3. 如果通信拒绝,检查端口是否被占用
return _check_port_availability(host, port)
except (http.client.HTTPException, json.JSONDecodeError):
return AppiumStatus.UNKNOWN
except Exception as e:
@@ -181,7 +222,7 @@ def get_appium_status() -> AppiumStatus:
if connection: connection.close()
def start_appium_service() -> AppiumService:
def start_appium_service(host: str = APPIUM_HOST, port: int | str = APPIUM_PORT) -> AppiumService:
"""
管理 Appium 服务的生命周期
如果服务未启动,则启动本地服务;如果已启动,则复用。
@@ -192,71 +233,65 @@ def start_appium_service() -> AppiumService:
is_managed = False
# 轮询等待真正就绪
# 延迟获取命令,确保只在真正需要启动服务时检查环境
cmd_args = resolve_appium_command()
cmd_args = resolve_appium_command(host, port)
try:
for i in range(MAX_RETRIES):
status = get_appium_status(host, port)
match status: # Python 3.10+ 的模式匹配
case AppiumStatus.READY:
if is_managed:
# 安全打印 PID
pid_info = f"PID: {process.pid}" if process else "EXTERNAL"
logger.info(f"Appium 服务启动成功! ({pid_info})")
return AppiumService(ServiceRole.MANAGED, host, port, process)
else:
logger.info(f"--> [复用] 有效的 Appium 服务已在运行 (http://{host}:{port})")
logger.info("--> [注意] 脚本退出时将保留该服务,不会将其关闭。")
return AppiumService(ServiceRole.EXTERNAL, host, port, process)
case AppiumStatus.CONFLICT:
_handle_port_conflict(port)
case AppiumStatus.OFFLINE:
if not is_managed:
process = _spawn_appium_process(cmd_args)
is_managed = True
for i in range(MAX_RETRIES):
status = get_appium_status()
match status: # Python 3.10+ 的模式匹配
case AppiumStatus.READY:
if is_managed:
# 安全打印 PID
pid_info = f"PID: {process.pid}" if process else "EXTERNAL"
logger.info(f"Appium 服务启动成功! ({pid_info})")
return AppiumService(ServiceRole.MANAGED, process)
else:
logger.info(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})")
logger.info("--> [注意] 脚本退出时将保留该服务,不会将其关闭。")
return AppiumService(ServiceRole.EXTERNAL, process)
case AppiumStatus.CONFLICT:
_handle_port_conflict()
case AppiumStatus.OFFLINE:
if not is_managed:
process = _spawn_appium_process(cmd_args)
is_managed = True
elif process and process.poll() is not None:
raise AppiumProcessCrashError("Appium 进程启动后异常退出。")
else:
if process and process.poll() is not None:
logger.warning("Appium 进程启动后异常退出")
sys.exit(1)
case AppiumStatus.INITIALIZING:
if is_managed and process and process.poll() is not None:
logger.warning("Appium 在初始化期间崩溃。")
_cleanup_process_tree(process)
sys.exit(1)
if i % 4 == 0: # 每 2 秒提醒一次,避免刷屏
logger.info("Appium 正在加载驱动/插件,请稍候...")
case AppiumStatus.ERROR:
logger.error("探测接口发生内部错误(可能是解析失败或严重网络异常),脚本终止。")
if is_managed and process:
_cleanup_process_tree(process)
sys.exit(1)
case _:
logger.error("Appium 启动异常")
sys.exit(1)
case AppiumStatus.INITIALIZING:
if is_managed and process and process.poll() is not None:
raise AppiumProcessCrashError("Appium 在初始化期间崩溃")
time.sleep(0.5)
if i % 4 == 0: # 每 2 秒提醒一次,避免刷屏
logger.info("Appium 正在加载驱动/插件,请稍候...")
case AppiumStatus.ERROR:
raise AppiumInternalError("探测接口发生内部错误(可能是解析失败或严重网络异常),脚本终止。")
case _:
raise AppiumInternalError("Appium 启动异常")
logger.warning("启动超时Appium 在规定时间内未完成初始化。")
_cleanup_process_tree(process)
sys.exit(1)
time.sleep(0.5)
raise AppiumTimeoutError("启动超时Appium 在规定时间内未完成初始化。")
except AppiumStartupError as e:
if process:
logger.warning(f"检测到启动阶段异常,正在回收进程资源 (PID: {process.pid})...")
_cleanup_process_tree(process)
# 将清理后的异常继续向上传递给装饰器
raise e
def _handle_port_conflict():
logger.warning(f"\n[!] 错误: 端口 {APPIUM_PORT} 被占用。")
logger.info("=" * 60)
logger.info("请手动执行以下命令释放端口后重试:")
def _handle_port_conflict(port: int | str) -> AppiumStartupError:
hand = f"端口 {port} 被其他程序占用,建议清理命令:\n"
if sys.platform == "win32":
logger.info(
f" CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a")
raise AppiumPortConflictError(
f"{hand}CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{port}') do taskkill /F /PID %a")
else:
logger.info(f" Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9")
logger.info("=" * 60)
sys.exit(1)
raise AppiumPortConflictError(f"{hand}Unix: lsof -ti:{port} | xargs kill -9")
def _spawn_appium_process(cmd_args: List[str]) -> subprocess.Popen:
"""启动 Appium 子进程"""
logger.info(f"正在启动本地 Appium 服务 (Port: {APPIUM_PORT})...")
logger.info(f"正在启动本地 Appium 服务...")
# 注入环境变量,确保 Appium 寻找项目本地的驱动
env_vars = os.environ.copy()
@@ -281,30 +316,28 @@ def _spawn_appium_process(cmd_args: List[str]) -> subprocess.Popen:
def stop_appium_service(service: AppiumService):
"""
停止 Appium 服务。
如果已经使用了 with 语句,通常不需要手动调用此函数。
:param service: AppiumService 对象
"""
service.stop()
if service:
service.stop()
# --- 装饰器实现 ---
def with_appium(func):
"""
装饰器:在函数执行前后自动启动和停止 Appium 服务。
适用于简单的脚本或调试场景。
:param func: 需要包装的函数
:return: 包装后的函数
"""
def managed_appium(host: str = APPIUM_HOST, port: int | str = APPIUM_PORT):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 1. start_appium_service 返回 AppiumService 实例
# 2. with 触发 __enter__
# 3. 结束或异常触发 __exit__ -> stop()
with start_appium_service(host, port) as service:
logger.debug(f"装饰器环境就绪: {service}")
return func(*args, **kwargs)
@functools.wraps(func)
def wrapper(*args, **kwargs):
service = start_appium_service()
try:
# return func(service, *args, **kwargs)
return func(*args, **kwargs)
finally:
service.stop()
return wrapper
return wrapper
return decorator
if __name__ == "__main__":
@@ -313,7 +346,7 @@ if __name__ == "__main__":
try:
appium_service = start_appium_service()
print(f"\n[项目路径] {BASE_DIR}")
print(f"[服务状态] Appium 运行中 (Port: {APPIUM_PORT})")
print(f"[服务状态] Appium 运行中...)")
print("[操作提示] 按 Ctrl+C 停止服务...")
# 保持运行,直到手动停止(在实际测试框架中,这里会被替换为测试执行逻辑)

View File

@@ -7,14 +7,14 @@
@contact: t6g888@163.com
@file: settings
@date: 2026/1/19 16:54
@desc:
@desc: 全局配置管理
"""
import os
from pathlib import Path
# 项目根目录 (core 的上一级)
# BASE_DIR = Path(__file__).parent.parent
BASE_DIR = Path(__file__).resolve().parents[1] # 获取根路径(绝对路径)
BASE_DIR = Path(__file__).resolve().parents[1]
# --- 目录配置 ---
OUTPUT_DIR = BASE_DIR / "outputs"
@@ -23,48 +23,43 @@ LOG_DIR = OUTPUT_DIR / "logs"
LOG_BACKUP_DIR = LOG_DIR / "backups"
ALLURE_TEMP = BASE_DIR / "temp"
REPORT_DIR = BASE_DIR / "report"
CONFIG_DIR = BASE_DIR / "config"
DATA_DIR = BASE_DIR / "data"
# 需要初始化的目录列表 (将在入口文件(mani.py)或 conftest.py 中被调用创建)
# 需要初始化的目录列表
REQUIRED_DIRS = [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR]
# --- 文件路径 ---
LOG_SOURCE = LOG_DIR / "pytest.log"
CAPS_CONFIG_PATH = CONFIG_DIR / "caps.yaml"
# --- 启动 Appium 最大尝试次数 ---
MAX_RETRIES = 40
# --- 业务常量 (可选) ---
APPIUM_SERVER = "http://127.0.0.1:4723"
# --- 核心配置 ---
IMPLICIT_WAIT_TIMEOUT = 10
EXPLICIT_WAIT_TIMEOUT = 10
# 默认 Appium Server 地址 (可通过命令行参数覆盖)
APPIUM_HOST = "127.0.0.1"
APPIUM_PORT = 4723
# --- 设备能力配置 (Capabilities) ---
ANDROID_CAPS = {
"platformName": "Android",
"automationName": "uiautomator2",
"deviceName": "Android",
"appPackage": "com.android.settings",
"appActivity": ".Settings",
"noReset": False
}
# ANDROID_CAPS = {
# "platformName": "Android",
# "automationName": "uiautomator2",
# "deviceName": "Android",
# "appPackage": "com.bocionline.ibmp",
# "appActivity": "com.bocionline.ibmp.app.main.launcher.LauncherActivity",
# "noReset":False
# }
IOS_CAPS = {
"platformName": "iOS",
"automationName": "XCUITest",
"autoAcceptAlerts": True, # 自动接受系统权限请求
"waitForQuiescence": False, # 设为 False 可加速扫描
# 如果是某些特定的业务弹窗 autoAcceptAlerts 无效,
# 此时就会触发我们代码里的 PopupManager.solve()
# --- 环境配置 (Environment Switch) ---
CURRENT_ENV = os.getenv("APP_ENV", "test")
# base_url业务接口的入口地址。主要用于通过 API 快速构造测试数据(前置条件)或查询数据库/接口状态来验证 App 操作是否生效(数据断言)
# source_address后端服务器的物理/源站地址。通常用于绕过负载均衡或 DNS 直接指定访问特定的服务器节点。
ENV_CONFIG = {
"test": {
"base_url": "https://test.example.com",
"source_address": "192.168.1.100"
},
"uat": {
"base_url": "https://uat.example.com",
"source_address": "192.168.1.200"
},
"prod": {
"base_url": "https://www.example.com",
"source_address": "10.0.0.1"
}
}