From 4de84039cb716ff741ee36871d61124032b3052c Mon Sep 17 00:00:00 2001 From: CNWei Date: Mon, 2 Feb 2026 17:48:30 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E4=BC=98=E5=8C=96=E4=BB=A3?= =?UTF-8?q?=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 优化 部分核心功能实现。 - 新增 详细的文档字符串（Docstrings）和注释。 - 移除 代码中的冗余注释和无效代码。 --- conftest.py | 25 +- core/base_page.py | 127 ++++++---- core/custom_expected_conditions.py | 61 ++++- core/driver.py | 46 +++- core/modules.py | 7 + core/run_appium.py | 234 +++++++++++------- core/settings.py | 2 + ...{api_demod_home.py => wan_android_home.py} | 0 ..._demos_views.py => wan_android_sidebar.py} | 0 ..._api_demos.py => test_wan_android_home.py} | 6 +- 10 files changed, 344 insertions(+), 164 deletions(-) rename page_objects/{api_demod_home.py => wan_android_home.py} (100%) rename page_objects/{api_demos_views.py => wan_android_sidebar.py} (100%) rename test_cases/{test_api_demos.py => test_wan_android_home.py} (91%) diff --git a/conftest.py b/conftest.py index d2f8e4b..2c7fdc8 100644 --- a/conftest.py +++ b/conftest.py @@ -10,7 +10,9 @@ @desc: """ import logging +import secrets +import allure import pytest from core.run_appium import start_appium_service, stop_appium_service from core.driver import CoreDriver @@ -42,8 +44,10 @@ def driver(app_server): "platformName": "Android", "automationName": "uiautomator2", "deviceName": "Android", - "appPackage": "io.appium.android.apis", - "appActivity": "io.appium.android.apis.ApiDemos", + "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 } @@ -68,6 +72,21 @@ def pytest_exception_interact(node, call, report): if report.failed: # 获取详细的错误堆栈（包含 assert 的对比信息） exc_info = call.excinfo.getrepr(style='no-locals') + name = f"异常截图_{secrets.token_hex(8)}" logger.error(f"TEST FAILED: {node.nodeid}") - logger.error(f"详细错误信息如下:\n{exc_info}") \ No newline at end of file + logger.error(f"截图名称: {name}") + logger.error(f"详细错误信息如下:\n{exc_info}") + + # 3. 自动截图：尝试从 fixture 中获取 driver + # node.funcargs 包含了当前测试用例请求的所有 fixture 实例 + driver = node.funcargs.get("driver") + if driver: + try: + allure.attach( + driver.get_screenshot_as_png(), + name=name, + attachment_type=allure.attachment_type.PNG + ) + except Exception as e: + logger.error(f"执行异常截图失败: {e}") diff --git a/core/base_page.py b/core/base_page.py index 09c3b7f..11b8a31 100644 --- a/core/base_page.py +++ b/core/base_page.py @@ -10,31 +10,70 @@ @desc: """ import logging -import secrets -from typing import Type, TypeVar, List, Tuple, Optional -import allure from pathlib import Path +from typing import Type, TypeVar, Optional + +import allure from appium import webdriver -from selenium.common import TimeoutException from core.driver import CoreDriver -from utils.decorators import exception_capture -# 定义一个泛型，用于类型推断（IDE 依然会有补全提示） +# 定义一个泛型，用于类型推断 T = TypeVar('T', bound='BasePage') + logger = logging.getLogger(__name__) class BasePage(CoreDriver): - + # --- 全局通用的属性 --- def __init__(self, driver: webdriver.Remote): - super().__init__(driver) - # 定义常见弹窗的关闭按钮定位 - - def log_screenshot(self, label: str = "步骤截图"): """ - 业务级截图：执行截图并附加到 Allure 报告。 - 用户可自由手动调用此方法。 + 初始化 BasePage。 + + :param driver: Appium WebDriver 实例 + """ + super().__init__(driver) + + # --- 所有页面通用的元动作 --- + def go_to(self, page_cls: Type[T]) -> T: + """ + 通用的页面跳转/实例化方法 (Page Factory)。 + + :param page_cls: 目标页面类 (BasePage 的子类) + :return: 目标页面的实例 + """ + logger.info(f"跳转到页面: {page_cls.__name__}") + return page_cls(self.driver) + + def handle_permission_popups(self): + """ + 处理通用的系统权限弹窗。 + 遍历预定义的黑名单，尝试关闭出现的系统级弹窗（如权限申请、安装确认等）。 + """ + # 普适性黑名单 + popup_blacklist = [ + ("id", "com.android.packageinstaller:id/permission_allow_button"), + ("xpath", "//*[@text='始终允许']"), + ("xpath", "//*[@text='稍后提醒']"), + ("xpath", "//*[@text='以后再说']"), + ("id", "com.app:id/iv_close_global_ad"), + ("accessibility id", "Close"), # iOS 常用 + ] + self.clear_popups(popup_blacklist) + + def handle_business_ads(self): + """ + 处理全 App 通用的业务广告弹窗。 + 针对应用启动后可能出现的全局广告进行关闭处理。 + """ + ads_blacklist = [("id", "com.app:id/global_ad_close")] + return self.clear_popups(ads_blacklist) + + def save_and_attach_screenshot(self, label: str = "日志截图") -> None: + """ + 保存截图到本地并附加到 Allure 报告。 + + :param label: 截图在报告中显示的名称 """ path_str = self.full_screen_screenshot(name=label) @@ -47,15 +86,16 @@ class BasePage(CoreDriver): attachment_type=allure.attachment_type.PNG ) - def log_screenshot_bytes(self, label: str = "步骤截图"): + def attach_screenshot_bytes(self, label: str = "日志截图") -> None: """ - 业务级截图：执行截图并附加到 Allure 报告。 - 用户可自由手动调用此方法。 + 直接获取内存中的截图数据并附加到 Allure 报告（不存本地文件）。 + + :param label: 截图在报告中显示的名称 """ - _img: bytes = self.driver.get_screenshot_as_png() + screenshot_bytes: bytes = self.driver.get_screenshot_as_png() allure.attach( - _img, + screenshot_bytes, name=label, attachment_type=allure.attachment_type.PNG ) @@ -64,6 +104,7 @@ class BasePage(CoreDriver): def assert_text(self, by: str, value: str, expected_text: str, timeout: Optional[float] = None) -> 'BasePage': """ 断言元素的文本内容是否符合预期。 + :param by: 定位策略。 :param value: 定位值。 :param expected_text: 期望的文本。 @@ -87,45 +128,21 @@ class BasePage(CoreDriver): logger.info(f"断言通过: 文本匹配 '{actual}'") return self - # 这里放全局通用的 Page 属性和逻辑 - def assert_visible(self, by: str, value: str, msg: str = "元素可见性校验"): + def assert_visible(self, by: str, value: str, msg: str = "元素可见性校验") -> 'BasePage': """ - 增强版断言：成功/失败均截图 + 断言元素是否可见。 + + :param by: 定位策略 + :param value: 定位值 + :param msg: 断言描述信息 + :return: self，支持链式调用 """ with allure.step(f"断言检查: {msg}"): - try: - element = self.find_element(by, value) - assert element.is_displayed() - # 成功存证 - except Exception as e: - raise e + element = self.find_element(by, value) + is_displayed = element.is_displayed() - # 封装一些所有页面通用的元动作 - def clear_permission_popups(self): - # 普适性黑名单 - _black_list = [ - ("id", "com.android.packageinstaller:id/permission_allow_button"), - ("xpath", "//*[@text='始终允许']"), - ("xpath", "//*[@text='稍后提醒']"), - ("xpath", "//*[@text='以后再说']"), - ("id", "com.app:id/iv_close_global_ad"), - ("accessibility id", "Close"), # iOS 常用 - ] - self.clear_popups(_black_list) + if is_displayed: + logger.info(f"断言通过: 元素 [{value}] 可见") - def clear_business_ads(self): - """在这里定义一些全 App 通用的业务广告清理""" - _ads = [("id", "com.app:id/global_ad_close")] - return self.clear_popups(_ads) - - def get_toast(self, text): - return self.is_visible("text", text) - - def go_to(self, page_name: Type[T]) -> T: - """ - 通用的页面跳转/获取方法 - :param page_name: 目标页面类 - :return: 目标页面的实例 - """ - logger.info(f"跳转到页面: {page_name.__name__}") - return page_name(self.driver) + assert is_displayed, f"断言失败: 元素 [{value}] 不可见" + return self diff --git a/core/custom_expected_conditions.py b/core/custom_expected_conditions.py index 682bb88..2db3c99 100644 --- a/core/custom_expected_conditions.py +++ b/core/custom_expected_conditions.py @@ -12,13 +12,13 @@ """ import logging -from typing import Tuple, Any, Union +from typing import Any, Union + from appium.webdriver.webdriver import WebDriver from selenium.webdriver.support import expected_conditions as EC -from selenium.common.exceptions import StaleElementReferenceException, NoSuchElementException from selenium.webdriver.common.by import By from selenium.webdriver.remote.webelement import WebElement - +from selenium.common.exceptions import StaleElementReferenceException, NoSuchElementException logger = logging.getLogger(__name__) @@ -34,15 +34,30 @@ logger = logging.getLogger(__name__) # 自定义预期条件(Custom Expected Condition) class BaseCondition: - """基础条件类：负责统一的 WebDriverWait 协议实现和异常拦截""" + """ + 基础条件类：负责统一的 WebDriverWait 协议实现和异常拦截。 + 所有自定义的类形式 EC 都应继承此类 + """ def __call__(self, driver: WebDriver): + """ + WebDriverWait 调用的入口方法。 + + :param driver: WebDriver 实例 + :return: check 方法的返回值，或者在捕获异常时返回 False + """ try: return self.check(driver) except (NoSuchElementException, StaleElementReferenceException): return False def check(self, driver: WebDriver): + """ + 执行具体的检查逻辑，由子类实现。 + + :param driver: WebDriver 实例 + :return: 判定成功返回对象或 True，失败返回 False + """ raise NotImplementedError("子类必须实现 check 方法") @@ -51,10 +66,13 @@ EC_MAPPING: dict[str, Any] = {} def register(name: str = None): """ - 强大的注册装饰器： + 自定义预期条件注册装饰器： 1. @register() -> 使用函数名注册 2. @register("alias") -> 使用别名注册 3. register("name", func) -> 手动注入 + + :param name: 注册别名，默认为函数/类名 + :return: 装饰器函数 """ def decorator(item): @@ -67,7 +85,13 @@ def register(name: str = None): @register("toast_visible") class ToastVisible(BaseCondition): + """检查 Toast 消息是否可见""" + def __init__(self, text: str, partial: Union[str, bool] = True): + """ + :param text: 期望的 Toast 文本 + :param partial: 是否部分匹配 (默认 True) + """ self.text = text # 处理从装饰器传来的字符串 "true"/"false" if isinstance(partial, str): @@ -84,8 +108,16 @@ class ToastVisible(BaseCondition): @register("attr_contains") class ElementHasAttribute(BaseCondition): + """检查元素的属性是否包含特定值""" + # 扁平化参数以支持字符串调用: "attr_contains:id,btn_id,checked,true" def __init__(self, by: str, value: str, attribute: str, expect_value: str): + """ + :param by: 定位策略 + :param value: 定位值 + :param attribute: 属性名 + :param expect_value: 期望包含的属性值 + """ self.locator = (by, value) self.attribute = attribute self.value = expect_value @@ -114,6 +146,13 @@ class ElementCountAtLeast(BaseCondition): @register() # 使用函数名 is_element_present 注册 def is_element_present(by: str, value: str): + """ + 检查元素是否存在于 DOM 中 (不一定可见)。 + + :param by: 定位策略 + :param value: 定位值 + :return: 判定函数 + """ locator = (by, value) def _predicate(driver): @@ -128,6 +167,13 @@ def is_element_present(by: str, value: str): @register() def system_ready(api_client): + """ + 检查外部系统 (API) 是否就绪。 + + :param api_client: API 客户端实例 + :return: 判定函数 + """ + def _predicate(_): # 忽略传入的 driver try: @@ -145,6 +191,11 @@ def get_condition(method: Union[str, Any], *args, **kwargs): 1. 如果 method 是字符串，先查自定义 EC_MAPPING 2. 如果自定义里没有，去官方 selenium.webdriver.support.expected_conditions 找 3. 如果 method 本身就是 Callable (比如 EC.presence_of_element_located)，直接透传 + + :param method: 预期条件名称 (str) 或 可调用对象 + :param args: 传递给条件的参数 + :param kwargs: 传递给条件的关键字参数 + :return: 实例化后的预期条件对象 (Callable) """ # 情况 A: 如果传入的是官方 EC 对象或自定义函数实例，直接返回 diff --git a/core/driver.py b/core/driver.py index 83a542e..8debbf9 100644 --- a/core/driver.py +++ b/core/driver.py @@ -11,7 +11,7 @@ """ import logging import secrets # 原生库，用于生成安全的随机数 -from typing import Optional, Type, TypeVar, Union, Callable +from typing import Optional, Type, TypeVar, Union, Callable, Any from time import sleep from appium import webdriver @@ -431,8 +431,16 @@ class CoreDriver: break return active + def back(self) -> 'CoreDriver': + """ + 模拟设备返回键操作。 + :return: self + """ + self.driver.back() + return self + @property - def session_id(self): + def session_id(self) -> Any | None: """获取当前 Appium 会话的 Session ID。""" return self.driver.session_id @@ -463,7 +471,7 @@ class CoreDriver: def swipe(self, direction: str = "up", duration: int = 1000) -> 'CoreDriver': """ 封装方向滑动 - :param direction: 滑动方向 up/down/left/right + :param direction: 滑动方向 up/down/left/right (指手指滑动的方向) :param duration: 滑动持续时间 (ms) :return: self """ @@ -486,7 +494,16 @@ class CoreDriver: def long_press(self, element: Optional[WebElement] = None, x: Optional[int] = None, y: Optional[int] = None, duration: int = 2000) -> 'CoreDriver': """ - 长按封装：支持传入元素或坐标 + 长按封装：支持传入元素或坐标。 + 如果传入 element，则计算其中心点坐标进行长按。 + 如果传入 x, y，则直接在坐标处长按。 + + :param element: 目标元素 (WebElement)，可选。 + :param x: 绝对坐标 X，可选。 + :param y: 绝对坐标 Y，可选。 + :param duration: 长按持续时间 (ms)，默认 2000ms。 + :return: self + :raises ValueError: 如果既未传入 element 也未传入坐标 (x, y)。 """ if element: rect = element.rect @@ -502,7 +519,13 @@ class CoreDriver: def drag_and_drop(self, source_el: WebElement, target_el: WebElement, duration: int = 1000) -> 'CoreDriver': """ - 将 source_el 拖拽到 target_el + 将 source_el 拖拽到 target_el。 + 计算两个元素的中心点，执行从源元素中心到目标元素中心的拖拽操作。 + + :param source_el: 源元素 (WebElement)。 + :param target_el: 目标元素 (WebElement)。 + :param duration: 拖拽过程持续时间 (ms)，默认 1000ms。 + :return: self """ s_rect = source_el.rect t_rect = target_el.rect @@ -518,6 +541,7 @@ class CoreDriver: 智能滚动：自动识别平台并调用最稳定的原生滚动脚本 :param element: 需要滚动的容器元素 (如 ScrollView, RecyclerView, TableView) :param direction: 滚动方向 'up', 'down', 'left', 'right' + :return: self """ platform = self.driver.capabilities.get('platformName', '').lower() match platform: @@ -537,10 +561,18 @@ class CoreDriver: return self - def swipe_by_percent(self, start_xp: int, start_yp: int, end_xp: int, end_yp: int, + def swipe_by_percent(self, start_xp: float, start_yp: float, end_xp: float, end_yp: float, duration: int = 1000) -> 'CoreDriver': """ - 按屏幕比例滑动 (0.5 = 50%) + 按屏幕比例滑动。 + 坐标值为屏幕宽高的百分比 (0.0 - 1.0)。 + + :param start_xp: 起点 X 比例 (如 0.5)。 + :param start_yp: 起点 Y 比例 (如 0.8)。 + :param end_xp: 终点 X 比例 (如 0.5)。 + :param end_yp: 终点 Y 比例 (如 0.2)。 + :param duration: 滑动持续时间 (ms)，默认 1000ms。 + :return: self """ size = self.driver.get_window_size() w, h = size['width'], size['height'] diff --git a/core/modules.py b/core/modules.py index 7f0a602..390fda3 100644 --- a/core/modules.py +++ b/core/modules.py @@ -13,11 +13,18 @@ from enum import Enum class AppPlatform(Enum): + """ + 定义支持的移动应用平台枚举。 + """ ANDROID = "android" IOS = "ios" class Locator(str, Enum): + """ + 定义元素定位策略枚举。 + 继承 str 以便直接作为参数传递给 Selenium/Appium 方法。 + """ # --- 原有 Selenium 支持 --- ID = "id" NAME = "name" diff --git a/core/run_appium.py b/core/run_appium.py index 4c0467e..9593395 100644 --- a/core/run_appium.py +++ b/core/run_appium.py @@ -20,8 +20,9 @@ import http.client import socket import json from enum import Enum +from typing import List -from core.settings import BASE_DIR, APPIUM_HOST, APPIUM_PORT +from core.settings import BASE_DIR, APPIUM_HOST, APPIUM_PORT, MAX_RETRIES logger = logging.getLogger(__name__) @@ -41,20 +42,26 @@ class AppiumStatus(Enum): class ServiceRole(Enum): - """服务角色枚举：明确权责""" - MANAGED = "由脚本启动 (受控模式)" # 需要负责清理 - EXTERNAL = "复用外部服务 (共享模式)" # 不负责清理 - NULL = "无效服务 (空模式)" # 占位或失败状态 + """服务角色枚举：定义服务的所有权和生命周期""" + MANAGED = "由脚本启动 (托管模式)" # 由本脚本启动，负责清理 + EXTERNAL = "复用外部服务 (共享模式)" # 复用现有服务，不负责清理 + NULL = "无效服务 (空模式)" # 无效或未初始化的服务 -def get_appium_command(): - """精确定位 Appium 执行文件，避免 npm 包装层""" +def resolve_appium_command() -> List[str]: + """ + 解析 Appium 可执行文件的绝对路径。 + 优先查找项目 node_modules 下的本地安装版本，避免 npm 包装层带来的信号传递问题。 + + :return: 用于 subprocess 的命令列表 + :raises SystemExit: 如果找不到 Appium 执行文件 + """ bin_name = "appium.cmd" if sys.platform == "win32" else "appium" appium_bin = BASE_DIR / "node_modules" / ".bin" / bin_name if not appium_bin.exists(): # 报错提示：找不到本地 Appium，引导用户安装 - logger.info(f"\n错误: 在路径 {appium_bin} 未找到 Appium 执行文件。") + logger.error(f"\n错误: 在路径 {appium_bin} 未找到 Appium 执行文件。") logger.info("请确保已在项目目录下执行过: npm install appium") sys.exit(1) # 返回执行列表（用于 shell=False） @@ -62,37 +69,45 @@ def get_appium_command(): # 移除全局调用，防止 import 时因找不到 appium 而直接退出 -# APP_CMD_LIST = get_appium_command() +# APP_CMD_LIST = resolve_appium_command() -def _cleanup_process_tree(process: subprocess.Popen = None): - """核心清理逻辑：针对方案一的跨平台递归关闭""" +def _cleanup_process_tree(process: subprocess.Popen = None) -> None: + """ + 核心清理逻辑：跨平台递归终止进程树。 + + :param process: subprocess.Popen 对象 + """ if not process or process.poll() is not None: return logger.info(f"正在关闭 Appium 进程树 (PID: {process.pid})...") try: if sys.platform == "win32": # Windows 下使用 taskkill 强制关闭进程树 /T - subprocess.run(['taskkill', '/F', '/T', '/PID', str(process.pid)], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + subprocess.run( + ['taskkill', '/F', '/T', '/PID', str(process.pid)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL + ) else: - # Unix/Mac: 获取进程组 ID 并发送终止信号 + # Unix/Linux: 获取进程组 ID 并发送终止信号 pgid = os.getpgid(process.pid) os.killpg(pgid, signal.SIGTERM) process.wait(timeout=5) - logger.info("所有相关进程已安全退出。") + logger.info("Appium 服务已停止,相关进程已安全退出。") except Exception as e: - logger.error(f"关闭进程时遇到意外: {e}") + logger.error(f"停止服务时发生异常: {e}") try: process.kill() + logger.warning(f"已强制杀死进程: {e}") except Exception as e: - logger.warning(f"强制关闭: {e}") + logger.debug(f"强制杀死进程失败 (可能已退出): {e}") # 这里通常保持安静，因为我们已经尝试过清理了 logger.info("服务已完全清理。") class AppiumService: - """Appium 服务上下文容器""" + """Appium 服务实例封装，用于管理服务生命周期""" def __init__(self, role: ServiceRole, process: subprocess.Popen = None): self.role = role @@ -102,25 +117,47 @@ class AppiumService: """统一停止接口：根据角色决定是否关闭进程""" match self.role: case ServiceRole.EXTERNAL: - logger.info(f"--> [角色: {self.role.value}] 脚本退出，保留原服务运行。") + logger.info(f"--> [角色: {self.role.value}] 脚本退出，保留外部服务运行。") return case ServiceRole.MANAGED: _cleanup_process_tree(self.process) case ServiceRole.NULL: - logger.info(f"--> [角色: {self.role.value}] 无需清理。") + logger.info(f"--> [角色: {self.role.value}] 无需执行清理。") def __repr__(self): return f"" +def _check_port_availability() -> AppiumStatus: + """辅助函数：检查端口是否被占用""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + try: + sock.settimeout(1) + # 尝试绑定端口，如果成功说明端口空闲 + sock.bind((APPIUM_HOST, APPIUM_PORT)) + return AppiumStatus.OFFLINE # 真正未启动 + except OSError: + # 绑定失败，说明端口被占用，但此前 HTTP 请求失败，说明不是 Appium + return AppiumStatus.CONFLICT # 端口被占用但没响应 HTTP + + def get_appium_status() -> AppiumStatus: - """深度探测 Appium 状态""" - conn = None + """ + 检测 Appium 服务当前状态。 + + 逻辑： + 1. 尝试 HTTP 连接 /status 接口。 + 2. 如果连接成功，检查 ready 字段。 + 3. 如果连接被拒绝，尝试绑定端口以确认端口是否真正空闲。 + + :return: AppiumStatus 枚举 + """ + connection = None try: # 1. 端口开启，尝试获取 Appium 状态接口 - conn = http.client.HTTPConnection(APPIUM_HOST, APPIUM_PORT, timeout=2) - conn.request("GET", "/status") - response = conn.getresponse() + connection = http.client.HTTPConnection(APPIUM_HOST, APPIUM_PORT, timeout=2) + connection.request("GET", "/status") + response = connection.getresponse() if response.status != 200: return AppiumStatus.CONFLICT @@ -134,95 +171,63 @@ def get_appium_status() -> AppiumStatus: except (socket.error, ConnectionRefusedError): # 3. 如果通信拒绝，检查端口是否真的空闲 - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - try: - s.settimeout(1) - s.bind((APPIUM_HOST, APPIUM_PORT)) - return AppiumStatus.OFFLINE # 真正未启动 - except OSError: - return AppiumStatus.CONFLICT # 端口被占用但没响应 HTTP + return _check_port_availability() except (http.client.HTTPException, json.JSONDecodeError): return AppiumStatus.UNKNOWN except Exception as e: - logger.error(e) + logger.error(f"状态检测异常: {e}") return AppiumStatus.ERROR finally: - if conn: conn.close() + if connection: connection.close() def start_appium_service() -> AppiumService: - """管理 Appium 服务的生命周期""" - process = None # 1. 预先初始化变量，防止作用域错误 - managed = False - # 轮询等待真正就绪 - - # 延迟获取命令，确保只在真正需要启动服务时检查环境 - app_cmd_list = get_appium_command() - max_retries = 40 + """ + 管理 Appium 服务的生命周期 + 如果服务未启动，则启动本地服务；如果已启动，则复用。 - for i in range(max_retries): + :return: AppiumService 对象 + """ + process = None # 1. 预先初始化变量，防止作用域错误 + is_managed = False + # 轮询等待真正就绪 + # 延迟获取命令，确保只在真正需要启动服务时检查环境 + cmd_args = resolve_appium_command() + + for i in range(MAX_RETRIES): status = get_appium_status() match status: # Python 3.10+ 的模式匹配 case AppiumStatus.READY: - if managed: + if is_managed: # 安全打印 PID - pid_str = f"PID: {process.pid}" if process else "EXTERNAL" - logger.info(f"Appium 已经完全就绪! ({pid_str})") + 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: - logger.warning(f"\n[!] 错误: 端口 {APPIUM_PORT} 被非 Appium 程序占用。") - logger.info("=" * 60) - logger.info("请手动执行以下命令释放端口后重试：") - if sys.platform == "win32": - logger.info( - f" CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a") - else: - logger.info(f" Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9") - logger.info("=" * 60) - sys.exit(1) + _handle_port_conflict() case AppiumStatus.OFFLINE: + if not is_managed: + process = _spawn_appium_process(cmd_args) + is_managed = True - if not managed: - logger.info("Appium 服务未启动...") - logger.info(f"正在准备启动本地 Appium 服务 (Port: {APPIUM_PORT})...") - - # 注入环境变量，确保 Appium 寻找项目本地的驱动 - env_vars = os.environ.copy() - env_vars["APPIUM_HOME"] = str(BASE_DIR) - - try: - process = subprocess.Popen( - app_cmd_list, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=env_vars, - cwd=BASE_DIR, - # Windows 和 Linux/Mac 的处理方式不同 - creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0, - preexec_fn=os.setsid if sys.platform != "win32" else None - ) - managed = True - except Exception as e: - logger.error(f"启动过程发生异常: {e}") - sys.exit(1) else: if process and process.poll() is not None: logger.warning("Appium 进程启动后异常退出。") sys.exit(1) case AppiumStatus.INITIALIZING: - if managed and process and process.poll() is not None: - logger.warning("Appium 驱动加载期间进程崩溃。") + 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 managed and process: + if is_managed and process: _cleanup_process_tree(process) sys.exit(1) case _: @@ -236,13 +241,60 @@ def start_appium_service() -> AppiumService: sys.exit(1) -def stop_appium_service(server: AppiumService): - # """安全关闭服务""" - server.stop() +def _handle_port_conflict(): + logger.warning(f"\n[!] 错误: 端口 {APPIUM_PORT} 被占用。") + logger.info("=" * 60) + logger.info("请手动执行以下命令释放端口后重试：") + if sys.platform == "win32": + logger.info( + f" CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a") + else: + logger.info(f" Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9") + logger.info("=" * 60) + sys.exit(1) + + +def _spawn_appium_process(cmd_args: List[str]) -> subprocess.Popen: + """启动 Appium 子进程""" + logger.info(f"正在启动本地 Appium 服务 (Port: {APPIUM_PORT})...") + + # 注入环境变量，确保 Appium 寻找项目本地的驱动 + env_vars = os.environ.copy() + env_vars["APPIUM_HOME"] = str(BASE_DIR) + + try: + return subprocess.Popen( + cmd_args, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=env_vars, + cwd=BASE_DIR, + # Windows 和 Linux/Mac 的处理方式不同 + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0, + preexec_fn=os.setsid if sys.platform != "win32" else None + ) + except Exception as e: + logger.error(f"启动过程发生异常: {e}") + sys.exit(1) + + +def stop_appium_service(service: AppiumService): + """ + 停止 Appium 服务。 + :param service: AppiumService 对象 + """ + service.stop() # --- 装饰器实现 --- def with_appium(func): + """ + 装饰器：在函数执行前后自动启动和停止 Appium 服务。 + 适用于简单的脚本或调试场景。 + :param func: 需要包装的函数 + :return: 包装后的函数 + """ + @functools.wraps(func) def wrapper(*args, **kwargs): service = start_appium_service() @@ -257,18 +309,18 @@ def with_appium(func): if __name__ == "__main__": # 使用示例：作为一个上下文管理器或简单的生命周期示例 - appium_proc = None + appium_service = None try: - appium_proc = start_appium_service() + appium_service = start_appium_service() print(f"\n[项目路径] {BASE_DIR}") - print("\n[提示] 现在可以手动或通过其他脚本运行测试用例。") - print("[提示] 按下 Ctrl+C 可停止由本脚本启动的服务。") + print(f"[服务状态] Appium 运行中 (Port: {APPIUM_PORT})") + print("[操作提示] 按 Ctrl+C 停止服务...") # 保持运行，直到手动停止（在实际测试框架中，这里会被替换为测试执行逻辑） while True: time.sleep(1) except KeyboardInterrupt: - pass + print("\n收到停止信号...") finally: - stop_appium_service(appium_proc) + stop_appium_service(appium_service) diff --git a/core/settings.py b/core/settings.py index c07a1ee..7172038 100644 --- a/core/settings.py +++ b/core/settings.py @@ -31,6 +31,8 @@ for folder in [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR]: # --- 文件路径 --- LOG_SOURCE = LOG_DIR / "pytest.log" +# --- 启动 Appium 最大尝试次数 --- +MAX_RETRIES = 40 # --- 业务常量 (可选) --- IMPLICIT_WAIT_TIMEOUT = 10 EXPLICIT_WAIT_TIMEOUT = 10 diff --git a/page_objects/api_demod_home.py b/page_objects/wan_android_home.py similarity index 100% rename from page_objects/api_demod_home.py rename to page_objects/wan_android_home.py diff --git a/page_objects/api_demos_views.py b/page_objects/wan_android_sidebar.py similarity index 100% rename from page_objects/api_demos_views.py rename to page_objects/wan_android_sidebar.py diff --git a/test_cases/test_api_demos.py b/test_cases/test_wan_android_home.py similarity index 91% rename from test_cases/test_api_demos.py rename to test_cases/test_wan_android_home.py index 0062c91..2b51d6e 100644 --- a/test_cases/test_api_demos.py +++ b/test_cases/test_wan_android_home.py @@ -14,8 +14,8 @@ import logging import allure import pytest -from page_objects.api_demod_home import HomePage -from page_objects.api_demos_views import ViewsPage +from page_objects.wan_android_home import HomePage +from page_objects.wan_android_sidebar import ViewsPage # 配置日志 logger = logging.getLogger(__name__) @@ -23,7 +23,7 @@ logger = logging.getLogger(__name__) @allure.epic("ApiDemos") @allure.feature("登录认证模块") class TestApiDemos: - # @allure.story("常规登录场景") + @allure.story("常规登录场景") @allure.title("使用合法账号登录成功") @allure.severity(allure.severity_level.BLOCKER) @allure.description("""