From c4c1692f09e3d8685e0366a16fc11415574dd3be Mon Sep 17 00:00:00 2001 From: CNWei Date: Wed, 14 Jan 2026 19:11:13 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E4=BC=98=E5=8C=96=20Appium=20?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AE=A1=E7=90=86=E9=80=BB=E8=BE=91=E4=B8=8E?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 弃用 npm run 改为直接调用 appium.cmd，消除 Windows 进程残留。 - 统一 _cleanup_process_tree 逻辑，确保异常退出时无孤儿进程。 - 重构 start_appium_service 为单一循环状态机，增强启动探测的健壮性。 - 新增 with_appium 装饰器，确保无论测试结果如何均能干净退出 --- run_appium.py | 237 ++++++++++++++++++++++++++++++++++---------------- 1 file changed, 161 insertions(+), 76 deletions(-) diff --git a/run_appium.py b/run_appium.py index 13730fb..abfacc5 100644 --- a/run_appium.py +++ b/run_appium.py @@ -9,6 +9,8 @@ @date: 2026/1/12 10:21 @desc: """ +import functools +import signal import subprocess import time import os @@ -16,8 +18,9 @@ import sys import http.client import socket import json +from collections import namedtuple from pathlib import Path -from enum import Enum, auto +from enum import Enum # --- 核心配置 --- # 使用 pathlib 获取当前脚本所在的绝对路径 @@ -25,8 +28,10 @@ BASE_DIR = Path(__file__).resolve().parent APPIUM_HOST = "127.0.0.1" APPIUM_PORT = 4723 + + # 使用 npm run 确保 APPIUM_HOME=. 变量和本地版本生效 -NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}" +# NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}" class AppiumStatus(Enum): @@ -39,6 +44,80 @@ class AppiumStatus(Enum): UNKNOWN = "未知状态" +class ServiceRole(Enum): + """服务角色枚举：明确权责""" + MANAGED = "由脚本启动 (受控模式)" # 需要负责清理 + EXTERNAL = "复用外部服务 (共享模式)" # 不负责清理 + NULL = "无效服务 (空模式)" # 占位或失败状态 + + +def get_appium_command(): + """精确定位 Appium 执行文件，避免 npm 包装层""" + 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，引导用户安装 + print(f"\n❌ 错误: 在路径 {appium_bin} 未找到 Appium 执行文件。") + print("💡 请确保已在项目目录下执行过: npm install appium") + sys.exit(1) + # 返回执行列表（用于 shell=False） + return [str(appium_bin), "-p", str(APPIUM_PORT)] + + +# 在全局或 start_appium_service 中获取命令 +APP_CMD_LIST = get_appium_command() + + +def _cleanup_process_tree(process: subprocess.Popen = None): + """核心清理逻辑：针对方案一的跨平台递归关闭""" + if not process or process.poll() is not None: + return + print(f"\n🛑 正在关闭 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) + else: + # Unix/Mac: 获取进程组 ID 并发送终止信号 + pgid = os.getpgid(process.pid) + os.killpg(pgid, signal.SIGTERM) + + process.wait(timeout=5) + print("✅ 所有相关进程已安全退出。") + except Exception as e: + print(f"⚠️ 关闭进程时遇到意外: {e}") + try: + process.kill() + except Exception as e: + print(f"强制: {e}") + # 这里通常保持安静，因为我们已经尝试过清理了 + print("✅ 服务已完全清理。") + + +class AppiumService: + """Appium 服务上下文容器""" + + def __init__(self, role: ServiceRole, process: subprocess.Popen = None): + self.role = role + self.process = process + + def stop(self): + """统一停止接口：根据角色决定是否关闭进程""" + match self.role: + case ServiceRole.EXTERNAL: + print(f"--> [角色: {self.role.value}] 脚本退出，保留原服务运行。") + return + case ServiceRole.MANAGED: + _cleanup_process_tree(self.process) + case ServiceRole.NULL: + print(f"--> [角色: {self.role.value}] 无需清理。") + + def __repr__(self): + return f"" + + def get_appium_status() -> AppiumStatus: """深度探测 Appium 状态""" conn = None @@ -58,7 +137,6 @@ def get_appium_status() -> AppiumStatus: is_ready = data.get("value", {}).get("ready") return AppiumStatus.READY if is_ready else AppiumStatus.INITIALIZING - except (socket.error, ConnectionRefusedError): # 3. 如果通信拒绝，检查端口是否真的空闲 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: @@ -69,7 +147,7 @@ def get_appium_status() -> AppiumStatus: except OSError: return AppiumStatus.CONFLICT # 端口被占用但没响应 HTTP except (http.client.HTTPException, json.JSONDecodeError): - return AppiumStatus.CONFLICT + return AppiumStatus.UNKNOWN except Exception as e: print(e) return AppiumStatus.ERROR @@ -77,96 +155,103 @@ def get_appium_status() -> AppiumStatus: if conn: conn.close() -def start_appium_service(): +def start_appium_service() -> AppiumService: """管理 Appium 服务的生命周期""" - # if check_before_start(): - # return None process = None # 1. 预先初始化变量，防止作用域错误 - status = get_appium_status() - match status: # Python 3.10+ 的模式匹配 - case AppiumStatus.READY: - print(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})") - return None - case AppiumStatus.INITIALIZING: - print("⏳ Appium 正在初始化，等待...") - case AppiumStatus.CONFLICT: - print(f"\n[!] 错误: 端口 {APPIUM_PORT} 被非 Appium 程序占用。") - print("=" * 60) - print("请手动执行以下命令释放端口后重试：") - if sys.platform == "win32": - print( - f" CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a") - else: - print(f" Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9") - print("=" * 60) - sys.exit(1) - case AppiumStatus.OFFLINE: - print("🔌 Appium 未启动") - print(f"🚀 正在准备启动本地 Appium 服务 (Port: {APPIUM_PORT})...") - - # 注入环境变量，确保 Appium 寻找项目本地的驱动 - env_vars = os.environ.copy() - env_vars["APPIUM_HOME"] = str(BASE_DIR) - - try: - process = subprocess.Popen( - NODE_CMD, - shell=True, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=env_vars, - cwd=BASE_DIR - ) - - except Exception as e: - print(f"❌ 启动过程发生异常: {e}") - sys.exit(1) - case _: - print("Appium 启动异常") - sys.exit(1) + managed = False # 轮询等待真正就绪 max_retries = 40 + for i in range(max_retries): status = get_appium_status() + match status: # Python 3.10+ 的模式匹配 + case AppiumStatus.READY: + if managed: + # 安全打印 PID + pid_str = f"PID: {process.pid}" if process else "EXTERNAL" + print(f"✨ Appium 已经完全就绪! ({pid_str})") + return AppiumService(ServiceRole.MANAGED, process) + else: + print(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})") + print(" [注意] 脚本退出时将保留该服务，不会将其关闭。") + return AppiumService(ServiceRole.EXTERNAL, process) + case AppiumStatus.CONFLICT: + print(f"\n[!] 错误: 端口 {APPIUM_PORT} 被非 Appium 程序占用。") + print("=" * 60) + print("请手动执行以下命令释放端口后重试：") + if sys.platform == "win32": + print( + f" CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a") + else: + print(f" Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9") + print("=" * 60) + sys.exit(1) + case AppiumStatus.OFFLINE: - if status == AppiumStatus.READY: - # 安全打印 PID - pid_str = f"PID: {process.pid}" if process else "EXTERNAL" - print(f"✨ Appium 已经完全就绪! ({pid_str})") - return process + if not managed: + print("🔌 Appium 未启动") + print(f"🚀 正在准备启动本地 Appium 服务 (Port: {APPIUM_PORT})...") - if status == AppiumStatus.ERROR: - print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。") - if process: process.terminate() - sys.exit(1) + # 注入环境变量，确保 Appium 寻找项目本地的驱动 + env_vars = os.environ.copy() + env_vars["APPIUM_HOME"] = str(BASE_DIR) - if status == AppiumStatus.INITIALIZING: - if i % 4 == 0: - print("...Appium 正在加载驱动/插件，请稍候...") - - if status == AppiumStatus.OFFLINE: - # 仅当进程是我们启动的（process 不为 None）才检查崩溃(None: 程序正常运行，非None: 程序异常) - if process and process.poll() is not None: - print("❌ Appium 进程启动后异常退出。") + 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: + print(f"❌ 启动过程发生异常: {e}") + sys.exit(1) + else: + if process and process.poll() is not None: + print("❌ Appium 进程启动后异常退出。") + sys.exit(1) + case AppiumStatus.INITIALIZING: + if managed and process and process.poll() is not None: + print("❌ Appium 驱动加载期间进程崩溃。") + _cleanup_process_tree(process) + sys.exit(1) + if i % 4 == 0: # 每 2 秒提醒一次，避免刷屏 + print("⏳ Appium 正在加载驱动/插件，请稍候...") + case AppiumStatus.ERROR: + print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。") + if managed and process: + _cleanup_process_tree(process) + sys.exit(1) + case _: + print("Appium 启动异常") sys.exit(1) time.sleep(0.5) print("❌ 启动超时：Appium 在规定时间内未完成初始化。") - if process: process.terminate() + _cleanup_process_tree(process) sys.exit(1) -def stop_appium_service(process): - """安全关闭服务""" - if process and process.poll() is None: - print(f"\n🛑 正在关闭 Appium 服务 (PID: {process.pid})...") - process.terminate() +def stop_appium_service(server: AppiumService): + # """安全关闭服务""" + server.stop() + +# --- 装饰器实现 --- +def with_appium(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + service = start_appium_service() try: - process.wait(timeout=5) - except subprocess.TimeoutExpired: - process.kill() - print("✅ 服务已安全退出。") + return func(service, *args, **kwargs) + finally: + service.stop() + return wrapper if __name__ == "__main__":