diff --git a/.gitignore b/.gitignore index 12f3cc3..35d7c9b 100644 --- a/.gitignore +++ b/.gitignore @@ -19,7 +19,7 @@ outputs/screenshots/ # --- Allure 报告 --- temp/ -report/ +reports/ .allure/ # --- pytest缓存 --- diff --git a/README.md b/README.md index df27fbc..e154a6b 100644 --- a/README.md +++ b/README.md @@ -62,4 +62,18 @@ newCommandTimeout：设置Appium服务器等待客户端发送新命令的超时 | @allure.severity | "严重程度（BLOCKER, CRITICAL, NORMAL...）" | 用于筛选高优先级用例 | | @allure.description | 详细描述（支持 Markdown） | 解释测试背景或前提条件 | | @allure.link | 外部链接（Bug系统、需求文档） | 快速点击跳转 | -| @allure.issue | 缺陷链接（通常会自动带上 ISSUE 前缀） | 追踪已知 Bug | \ No newline at end of file +| @allure.issue | 缺陷链接（通常会自动带上 ISSUE 前缀） | 追踪已知 Bug | + +Pytest 原生高频参数 和 你的自定义参数。 + +| **分类** | **长参数** | **短参数/别名** | **作用说明** | +|-----------|------------------|------------|------------------------------------| +| **基础运行** | `--verbose` | `-v` | 打印详细运行过程（显示用例名称） | +| | (无) | `-s` | 允许在控制台打印代码里的 `print` 内容 | +| **调试控制** | `--exitfirst` | `-x` | 遇到第一个失败的用例就立即停止测试 | +| | `--maxfail=n` | (无) | 累计失败 `n` 个用例后停止运行 | +| | `--last-failed` | `--lf` | 只运行上次运行失败的用例 | +| **自定义参数** | **`--platform`** | **`-P`** | 指定运行平台 \[Android\/IOS\/别名\]\(可自定义) | +| | **`--udid`** | **`-U`** | 指定手机唯一标识 (可自定义) | +| | **`--host`** | **`-H`** | 指定 Appium 服务器地址 (可自定义) | +| **报告相关** | `--alluredir` | (无) | 指定 Allure 原始数据的存放路径 | diff --git a/config/caps.yaml b/config/caps.yaml new file mode 100644 index 0000000..07418d4 --- /dev/null +++ b/config/caps.yaml @@ -0,0 +1,19 @@ +# Appium Capabilities Configuration +android: + platformName: "Android" + automationName: "uiautomator2" + deviceName: "Android" + appPackage: "com.manu.wanandroid" + appActivity: "com.manu.wanandroid.ui.main.activity.MainActivity" + noReset: false + newCommandTimeout: 60 + # udid: "emulator-5554" # Can be injected via CLI + +ios: + platformName: "iOS" + automationName: "XCUITest" + deviceName: "iPhone 14" + bundleId: "com.example.app" + autoAcceptAlerts: true + waitForQuiescence: false + newCommandTimeout: 60 diff --git a/conftest.py b/conftest.py index 59ff4bd..49677bf 100644 --- a/conftest.py +++ b/conftest.py @@ -7,74 +7,110 @@ @contact: t6g888@163.com @file: conftest @date: 2026/1/16 10:52 -@desc: +@desc: Pytest 核心配置与 Fixture 管理 """ import logging import secrets +from pathlib import Path +from typing import Generator, Any -import allure import pytest +import allure + from core.run_appium import start_appium_service, stop_appium_service from core.driver import CoreDriver -from utils.dirs_manager import ensure_dirs_ok +from core.settings import APPIUM_HOST, APPIUM_PORT +from core.config_loader import get_caps -def pytest_configure(config): +# 注册命令行参数 +def pytest_addoption(parser: Any) -> None: """ - Pytest 钩子函数：在测试收集开始前执行 + 注册自定义命令行参数 + :param parser: Pytest 参数解析器对象 """ - # ensure_dirs_ok() - ... + parser.addoption("--platform", action="store", default="Android", help="目标平台: Android or IOS") + parser.addoption("--udid", action="store", default=None, help="设备唯一标识") + parser.addoption("--host", action="store", default=APPIUM_HOST, help="Appium Server Host") + parser.addoption("--port", action="store", default=str(APPIUM_PORT), help="Appium Server Port") @pytest.fixture(scope="session") -def app_server(): +def app_server(request: pytest.FixtureRequest) -> Generator[Any, None, None]: """ 第一层：管理 Appium Server 进程。 - 利用start_appium_service 里的 40 次轮询和 sys.exit(1) 逻辑。 + :param request: Pytest 请求对象 + :return: Appium 服务进程实例 """ - # 启动服务 - service = start_appium_service() + # 获取命令行参数 + host = request.config.getoption("--host") + port = int(request.config.getoption("--port")) + + service = start_appium_service(host, port) yield service - # 所有测试结束，清理进程 stop_appium_service(service) -@pytest.fixture(scope="module") -def driver(app_server): +@pytest.fixture(scope="session") +def driver_session(request: pytest.FixtureRequest, app_server: Any) -> Generator[CoreDriver, None, None]: """ - 第二层：管理 WebDriver 会话。 - 依赖 app_server，确保服务 Ready 后才创建连接。 + 第二层：全局单例 Driver 管理 (Session Scope)。 + 负责创建和销毁 Driver，整个测试过程只启动一次 App。 + :param request: Pytest 请求对象 + :param app_server: Appium 服务 fixture 依赖 + :return: CoreDriver 实例 """ - # 实例化你提供的类结构 + platform = request.config.getoption("--platform") + ud_id = request.config.getoption("--udid") + host = request.config.getoption("--host") + port = int(request.config.getoption("--port")) + + # 1. 获取基础 Caps + caps = get_caps(platform) + + # 2. 动态注入参数 + if ud_id: caps["udid"] = ud_id + + # 将最终生效的 caps 存入 pytest 配置，方便报告读取 + request.config._final_caps = caps + + # 3. 初始化 Driver app_helper = CoreDriver() - caps = { - "platformName": "Android", - "automationName": "uiautomator2", - "deviceName": "Android", - "appPackage": "com.manu.wanandroid", - # "appPackage": "com.bocionline.ibmp", - "appActivity": "com.manu.wanandroid.ui.main.activity.MainActivity", - # "appActivity": "com.bocionline.ibmp.app.main.launcher.LauncherActivity", - "noReset": False, # 不清除应用数据 - "newCommandTimeout": 60 - } - # 连接并获取原生 driver 实例 - # 这里可以根据需要扩展，比如通过命令行参数选择平台 - app_helper.connect(platform="android", caps=caps) + app_helper.server_config(host=host, port=port) - yield app_helper.driver + try: + app_helper.connect(platform=platform, caps=caps) + except Exception as e: + pytest.exit(f"无法初始化 Driver: {e}") - # 用例结束，只关 session，不关 server + yield app_helper + + # 4. 清理 app_helper.quit() -def pytest_exception_interact(node, call, report): +@pytest.fixture(scope="function") +def driver(driver_session: CoreDriver) -> Generator[Any, None, None]: + """ + 第三层：用例级 Driver 注入。 + 每个用例直接获取已存在的 Driver 实例。 + 可以在这里添加 reset_app() 逻辑，确保用例间独立性。 + :param driver_session: CoreDriver 会话实例 + :return: 原始 Appium Driver 对象 (webdriver.Remote) + """ + # 可选：如果需要在每个用例前重置 App 状态 + # driver_session.driver.reset() + yield driver_session.driver + + +def pytest_exception_interact(node: Any, call: Any, report: Any) -> None: """ 当测试用例抛出异常（断言失败或代码报错）时，Pytest 会调用这个钩子。 我们在这里手动把错误信息喂给 logging。 + :param node: 测试节点 + :param call: 调用信息 + :param report: 测试报告 """ - # 获取名为 'Error' 的记录器，它会遵循 pytest.ini 中的 log_file 配置 logger = logging.getLogger("pytest") if report.failed: @@ -83,23 +119,71 @@ def pytest_exception_interact(node, call, report): exc_info = call.excinfo.getrepr(style='short') screenshot_name = f"异常截图_{secrets.token_hex(4)}" - logger.error("\n" + "=" * 40 + " TEST FAILED " + "=" * 40) - logger.error(f"Node ID: {node.nodeid}") - logger.error(f"截图名称: {screenshot_name}") - logger.error(f"详细错误信息如下:\n{exc_info}") + logger.error(f"\n{'=' * 40} TEST FAILED {'=' * 40}\n" + f"Node ID: {node.nodeid}\n" + f"截图名称: {screenshot_name}\n" + f"详细错误信息:\n{exc_info}" + ) - # 3. 自动截图：尝试从 fixture 中获取 driver - # node.funcargs 包含了当前测试用例请求的所有 fixture 实例 - driver = node.funcargs.get("driver") - if driver: + # 尝试获取 driver_session (CoreDriver 实例) + if "driver_session" in node.funcargs: + helper = node.funcargs["driver_session"] try: - # 同时保存到 Allure 和本地（如果你需要本地有一份备份） - screenshot_data = driver.get_screenshot_as_png() + # 截图并附加到 Allure + screenshot_bytes = helper.driver.get_screenshot_as_png() allure.attach( - screenshot_data, - name=name, + screenshot_bytes, + name=screenshot_name, attachment_type=allure.attachment_type.PNG ) except Exception as e: logger.error(f"执行异常截图失败: {e}") logger.error("=" * 93 + "\n") + + +def pytest_sessionfinish(session: Any, exitstatus: int) -> None: + """ + 测试会话结束时，收集环境信息到 Allure 报告 + :param session: Pytest 会话对象 + :param exitstatus: 退出状态码 + """ + match exitstatus: + case pytest.ExitCode.OK: + logging.info("测试全部通过！") + case pytest.ExitCode.TESTS_FAILED: + logging.warning("部分测试用例执行失败，请检查报告。") + case pytest.ExitCode.INTERRUPTED: + logging.error("测试被用户手动中断（Ctrl+C）。") + case pytest.ExitCode.INTERNAL_ERROR: + logging.critical("Pytest 发生内部错误！") + case pytest.ExitCode.USAGE_ERROR: + logging.error("Pytest 命令行参数错误或用法不当。") + case pytest.ExitCode.NO_TESTS_COLLECTED: + logging.warning("未发现任何测试用例。") + case _: + logging.error(f"未知错误状态码: {exitstatus}") + + report_dir = session.config.getoption("--alluredir") + final_caps = getattr(session.config, "_final_caps", {}) + if not report_dir: + return + report_path = Path(report_dir) + # 收集环境信息 + env_info = { + "Platform": session.config.getoption("--platform"), + "UDID": final_caps.get("udid") or session.config.getoption("--udid") or "未指定", + "Host": session.config.getoption("--host"), + "Python": "3.11+" + } + + try: + if not report_path.exists(): + report_path.mkdir(parents=True, exist_ok=True) + # 生成 environment.properties 文件 + env_file = report_path / "environment.properties" + with env_file.open("w", encoding="utf-8") as f: + for k, v in env_info.items(): + f.write(f"{k}={v}\n") + logging.info("Allure 环境信息已生成。") + except Exception as e: + logging.error(f"无法写入环境属性: {e}") diff --git a/core/config_loader.py b/core/config_loader.py new file mode 100644 index 0000000..aca1dce --- /dev/null +++ b/core/config_loader.py @@ -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}") diff --git a/core/driver.py b/core/driver.py index 8debbf9..2d3914f 100644 --- a/core/driver.py +++ b/core/driver.py @@ -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: """ 截取当前完整屏幕内容 (自愈逻辑、异常报错首选) diff --git a/core/run_appium.py b/core/run_appium.py index 9593395..8a263cc 100644 --- a/core/run_appium.py +++ b/core/run_appium.py @@ -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"" + + 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"" - -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 停止服务...") # 保持运行，直到手动停止（在实际测试框架中，这里会被替换为测试执行逻辑） diff --git a/core/settings.py b/core/settings.py index 1f28678..1fe540b 100644 --- a/core/settings.py +++ b/core/settings.py @@ -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" + } } diff --git a/page_objects/wan_android_home.py b/page_objects/wan_android_home.py index d29c02e..6f45ecf 100644 --- a/page_objects/wan_android_home.py +++ b/page_objects/wan_android_home.py @@ -10,6 +10,7 @@ @desc: """ import logging +import os import allure from appium import webdriver @@ -22,13 +23,18 @@ logger = logging.getLogger(__name__) class HomePage(BasePage): # 定位参数 menu = ("accessibility id", "开启") - home = ("-android uiautomator", 'new UiSelector().resourceId("com.manu.wanandroid:id/icon").instance(0)') + home = ("id", 'com.manu.wanandroid:id/largeLabel') project = ("-android uiautomator", 'new UiSelector().text("项目")') system = ("-android uiautomator", 'new UiSelector().text("体系")') - tv_name = ("id", "com.manu.wanandroid:id/tvName") + account=("-android uiautomator",'new UiSelector().text("账号")') + pass_word=("-android uiautomator",'new UiSelector().text("密码")') + + + login_button = ("accessibility id", '登录') + def __init__(self, driver: webdriver.Remote): super().__init__(driver) @@ -41,17 +47,23 @@ class HomePage(BasePage): self.click(*self.tv_name) @allure.step("登录账号：{1}") - def click_unicode(self, username,password): + def login(self, username, password): """执行登录业务逻辑""" + account_input = {"elementId": '00000000-0000-0ecb-0000-0017000002d8', "text": username} + pass_word_input = {"elementId": '00000000-0000-0ecb-0000-0018000002d8', "text": password} + if self.wait_until_visible(*self.login_button): + self.click(*self.account).driver.execute_script('mobile: type', account_input) + self.click(*self.pass_word).driver.execute_script('mobile: type', pass_word_input) - if self.wait_until_visible(*self.unicode): - self.swipe("left") + self.click(*self.login_button).full_screen_screenshot() - self.click(*self.unicode).log_screenshot() - - @allure.step("获取 “Text ”文本") - def get_home_text(self): - """执行登录业务逻辑""" - # 调用继承自 CoreDriver 的方法（假设你的 CoreDriver 已经被注入或组合） - - return self.get_text(*self.text) + # @allure.step("获取 “Text ”文本") + # def get_home_text(self): + # """执行登录业务逻辑"""os.getenv("USER_NAME") os.getenv("PASS_WORD") + # # 调用继承自 CoreDriver 的方法（假设你的 CoreDriver 已经被注入或组合） + # """ + # { "elementId": "00000000-0000-0ecb-0000-0017000002d8", "text": "admintest123456" } + # { "elementId": "00000000-0000-0ecb-0000-0018000002d8", "text": "admin123456" } + # """ + # + # return self.get_text(*self.text) diff --git a/pyproject.toml b/pyproject.toml index 3d87057..2e2a15a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,8 @@ dependencies = [ "pytest>=8.3.5", "PyYAML>=6.0.1", "openpyxl>=3.1.2", + "pytest-rerunfailures>=16.1", + "python-dotenv>=1.2.1", ] [[tool.uv.index]] diff --git a/pytest.ini b/pytest.ini index 43cfca2..96b642f 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,5 +1,5 @@ [pytest] -addopts = -q --show-capture=no +addopts = -q --show-capture=no --reruns 2 --reruns-delay 1 # 1. 开启实时控制台日志 log_cli = True diff --git a/test_cases/test_wan_android_home.py b/test_cases/test_wan_android_home.py index 2b51d6e..afa50e9 100644 --- a/test_cases/test_wan_android_home.py +++ b/test_cases/test_wan_android_home.py @@ -10,16 +10,20 @@ @desc: """ import logging +import os import allure import pytest - +from dotenv import load_dotenv from page_objects.wan_android_home import HomePage from page_objects.wan_android_sidebar import ViewsPage +load_dotenv() + # 配置日志 logger = logging.getLogger(__name__) + @allure.epic("ApiDemos") @allure.feature("登录认证模块") class TestApiDemos: @@ -34,20 +38,20 @@ class TestApiDemos: """) @allure.link("https://docs.example.com/login_spec", name="登录业务说明文档") @allure.issue("BUG-1001", "已知偶发：部分机型广告Banner无法滑动") - def test_api_demos_success(self, driver,user): + def test_api_demos_success(self, driver): """ 测试场景：使用正确的用户名和密码登录成功 """ - api_demos = HomePage(driver) + wan = HomePage(driver) # 执行业务逻辑 - api_demos.click_text() - api_demos.click_unicode() + wan.click_open() + wan.login(os.getenv("USER_NAME"),os.getenv("PASS_WORD")) # 断言部分使用 allure.step 包装，使其在报告中也是一个可读的步骤 with allure.step("最终校验：检查是否进入首页并显示‘交易’标题"): - actual_text = api_demos.get_home_text() - assert actual_text == "Text" - + # actual_text = api_demos.get_home_text() + # assert actual_text == "Text" + print("开发中。。。") # 页面跳转 - api_demos.go_to(ViewsPage).screenshot_views() - api_demos.delay(5) \ No newline at end of file + # wan.go_to(ViewsPage).screenshot_views() + wan.delay(5) diff --git a/utils/data_loader.py b/utils/data_loader.py new file mode 100644 index 0000000..ad101fa --- /dev/null +++ b/utils/data_loader.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python +# coding=utf-8 + +""" +@author: CNWei,ChenWei +@Software: PyCharm +@contact: t6g888@163.com +@file: data_loader +@date: 2026/1/27 10:00 +@desc: 数据加载工具 +""" +import yaml +from pathlib import Path +from typing import Any, List + + +def load_yaml(file_path: Path | str) -> dict[str, Any] | List[Any]: + """ + 加载 YAML 文件 + :param file_path: 文件路径 + :return: 数据字典或列表 + """ + path = Path(file_path) + if not path.exists(): + raise FileNotFoundError(f"YAML file not found: {path}") + + with open(path, "r", encoding="utf-8") as f: + return yaml.safe_load(f)