无锡专业做网站的公司,宁波公司建网站哪家,贵州铁路投资建设网站,wordpress路径错误一、前置说明
总体目录#xff1a;《从 0-1 搭建企业级 APP 自动化测试框架》上节回顾#xff1a;在 init_appium_and_devices 的实现思路分析 小节中#xff0c;分析了实现 init_appium_and_devices 的思路#xff0c;梳理出了必要的工具类和方法。本节目标#xff1a;完…一、前置说明
总体目录《从 0-1 搭建企业级 APP 自动化测试框架》上节回顾在 init_appium_and_devices 的实现思路分析 小节中分析了实现 init_appium_and_devices 的思路梳理出了必要的工具类和方法。本节目标完成 os_util 模块工具类和方法的编写为具体实现做准备。其它说明 工具类和方法是实现目的的手段不是框架的重点可以使用 ChatGPT 工具辅助实现工具类和方法验证即可对工具类和方法进行分类整理使用途更明确。
二、代码实现
utils/os_util.py
import os
import platform
import logging
import shutil
import subprocess
import timeimport psutil
import win32gui
import win32processlogger logging.getLogger(__name__)class RunCMDError(Exception):...class OSName:propertydef os_name(self):return platform.system()def is_windows(self):return self.os_name.lower() windowsdef is_linux(self):return self.os_name.lower() linuxdef is_mac(self):return self.os_name.lower() darwinosname OSName()class CMDRunner:staticmethoddef run_command(command, timeout5, retry_interval0.5, expected_text):执行命令并等待返回结果。无论是执行成功还是执行失败都会返回结果。:param command: 待执行的命令:param timeout: 超时时间该参数的作用说明- 在连续执行多条命令时会遇到因为执行速度过快上一条命令未执行完成就执行了下一条命令导致下一条命令执行失败的问题- 因此需要增加失败重跑的机制只要未执行成功就进入失败重跑直到到达指定的超时时间为止。:param retry_interval: 失败重试的间隔等待时间:param expected_text: 期望字符串如果期望字符串为真则以字符串是否在输出结果中来判断是否执行成功该参数的作用举例说明- 如果你在命令行中输入adb connect 127.0.0.1:7555- 返回的结果是* daemon not running; starting now at tcp:5037* daemon started successfullycannot connect to 127.0.0.1:7555: 由于目标计算机积极拒绝无法连接。 (10061)- 虽然命令已执行成功但是并没有达到 连接成功 的预期- 此时你可以使用 run_command(adb connect 127.0.0.1:62001, timeout60, expected_textalready connected to 127.0.0.1:7555):return: (output, status), 返回输出结果和状态无论是成功还是失败都会返回输出结果和状态# 定义失败重跑的截止时间end_time time.time() timeoutattempts 0while True:try:# subprocess.run() 方法用于执行命令并等待其完成然后返回一个 CompletedProcess 对象该对象包含执行结果的属性# 它适用于需要等待命令完成并获取结果的情况。result subprocess.run(command, shellTrue, capture_outputTrue, textTrue, timeouttimeout)if result.returncode 0:# 如果 returncode0表示执行成功output result.stdout.strip()# 如果期望字符串为真则通过字符串是否在输出结果中来判断是否执行成功status expected_text in output if expected_text else True# 通常情况下执行成功时命令行不会返回任何结果此时result为因此添加这个逻辑output output or successfulelse:# 否则从result.stderr.strip() 获取 outputoutput result.stderr.strip()status expected_text in output if expected_text else False# 如果 status 为真则返回 output, status, 否则进入 while 循环进行失败重试if status:logger.debug(fExecute adb command successfully: {command}f\nOutput is: f\nf\n{output}f\n)return output, statusexcept subprocess.TimeoutExpired as e:logger.warning(fCommand timed out: {e})output, status , Falseexcept Exception as e:logger.warning(fUnexpected error while executing command: {e})output, status , False# 添加等待时间等待上一条命令执行完成time.sleep(retry_interval)attempts 1logger.debug(fExecute adb command failure: {command}f\nRetrying... Attempt {attempts})# 如果超过截止时间则跳出循环if time.time() end_time:breaklogger.warning(fExecute adb command failure: {command}f\nOutput is: f\nf\n{output}f\n)return output, statusdef run_command_strict(self, command, timeout5, retry_interval0.5, expected_text):严格执行命令如果执行失败则抛出RunCMDError异常如果执行成功则返回输出结果.:param command: 待执行的命令:param timeout: 超时时间:param retry_interval: 失败重试的间隔等待时间:param expected_text: 期望字符串:return: output, 返回命令执行成功之后的输出结果output, status self.run_command(command, timeout, retry_interval, expected_text)if not status:raise RunCMDError(output)return outputcmd_runner CMDRunner()class PortManager:staticmethoddef is_port_in_use(port) - bool:判断指定端口是否被占用try:# 获取当前系统中所有正在运行的网络连接connections psutil.net_connections()# 检查是否有连接占用指定端口for conn in connections:if conn.laddr.port int(port):logger.debug(fPort {port} is in use.)return Truelogger.debug(fPort {port} is not in use.)return Falseexcept Exception as e:# 处理异常例如权限不足等情况logger.error(fError checking port {port}: {e})return Falsedef generate_available_ports(self, start_port, count):生成可用的端口号:param start_port: 起始端口号:param count: 生成端口的个数:return: 列表available_ports []port start_portwhile len(available_ports) count:if not self.is_port_in_use(port):available_ports.append(port)port 1return available_portsport_manager PortManager()class ProcessManager:staticmethoddef find_pid_by_window_name(window_name):根据窗口名称查找对应的进程ID列表pids [] # 存储找到的进程ID列表system platform.system() # 获取操作系统类型if system Windows:# 使用 win32gui 库查找窗口def callback(hwnd, hwnd_list):try:if win32gui.IsWindowVisible(hwnd): # 检查窗口是否可见window_text win32gui.GetWindowText(hwnd) # 获取窗口标题文本if window_name.lower() in window_text.lower(): # 如果窗口标题包含指定的窗口名称不区分大小写_, pid win32process.GetWindowThreadProcessId(hwnd) # 获取窗口对应的进程IDhwnd_list.append(pid) # 将进程ID添加到列表中except:passreturn Truewin32gui.EnumWindows(callback, pids) # 枚举所有窗口并调用回调函数进行查找else:# 在 macOS 和 Linux 上使用 psutil 库查找进程 PIDfor proc in psutil.process_iter([pid, name]): # 遍历所有进程并获取进程的PID和名称try:process_name proc.info[name].lower() # 获取进程名称转换为小写if window_name.lower() in process_name: # 如果进程名称包含指定的窗口名称不区分大小写pids.append(proc.info[pid]) # 将进程ID添加到列表中except (psutil.NoSuchProcess, psutil.AccessDenied):passreturn pidsstaticmethoddef find_pid_by_port(port):根据端口号查找对应的进程ID列表pids []for conn in psutil.net_connections():try:if conn.laddr.port port:pids.append(conn.pid)except (psutil.NoSuchProcess, psutil.AccessDenied):passreturn pidsstaticmethoddef find_pid_by_process_name(process_name):根据进程名称查找对应的进程ID列表pids []for proc in psutil.process_iter([pid, name]):try:if process_name.lower() in proc.info[name].lower():pids.append(proc.info[pid])except (psutil.NoSuchProcess, psutil.AccessDenied):passreturn pidsstaticmethoddef kill_process_by_pid(pid):根据进程PID查杀进程。此方法是 kill_process_by_port、kill_process_by_name、kill_process_by_window_name 的底层方法。try:process psutil.Process(pid)process.kill()logger.debug(fKilled process with PID {process.pid})# 杀死进程时可能会遇到权限等问题except Exception as e:logger.debug(fFailed to kill process: {e})def kill_process_by_port(self, port):根据端口号查杀进程pids self.find_pid_by_port(int(port))for pid in pids:self.kill_process_by_pid(pid)def kill_process_by_name(self, process_name):根据进程名查杀进程pids self.find_pid_by_process_name(process_name)for pid in pids:self.kill_process_by_pid(pid)def kill_process_by_window_name(self, window_name):根据窗口名称查杀进程pids self.find_pid_by_window_name(window_name)for pid in pids:self.kill_process_by_pid(pid)process_manager ProcessManager()class ApplicationManager:staticmethoddef check_app_installed(app_name):使用 shutil.which() 方法来检查应用是否已安装,可适用于所有平台。只会从Path环境变量中判断如果已安装但未设置为环境变量会返回False.if shutil.which(app_name) is not None:return Trueelse:return Falsestaticmethoddef set_environment_variable(key, value):设置系统变量比如设置: JAVA_HOME、ANDROID_HOMEos.environ[key] valuestaticmethoddef set_path_env_variable(path):设置Path环境变量warning: 这种方式设置环境变量在windows平台会遇到1024截断的问题, todoif PATH in os.environ:os.environ[PATH] path os.pathsep os.environ[PATH]else:os.environ[PATH] pathstaticmethoddef start_application(executable_path, is_run_in_commandFalse):启动应用程序但不必等待其启动完成。:param executable_path: 可执行文件的路径比如 .exe文件:param is_run_in_command: 是否要在命令行中启动比如在命令行中输入 appium 启动 appium servertry:if osname.is_windows():if is_run_in_command:# 如果程序需要在命令行中启动则需要需要使用 start 启动命令行窗口command fstart {executable_path}else:command executable_path# subprocess.Popen() 方法用于启动命令但不必等待其完成,# 这对于需要启动长时间运行的程序或不需要等待程序完成的情况非常有用, 例如等待命令完成、发送信号等。subprocess.Popen(command, shellTrue)logger.debug(fStarted program {executable_path})else:if is_run_in_command:# 类Unix系统使用nohup命令启动程序command fnohup {executable_path} else:command executable_pathsubprocess.Popen(command, shellTrue, preexec_fnos.setsid)logger.debug(fStarted program {executable_path})except Exception as e:logger.warning(fError starting program {executable_path}: {str(e)})application_manager ApplicationManager()if __name__ __main__:logging.basicConfig(levellogging.DEBUG)# 按窗口名称查杀 uiautomatorviewerprocess_manager.kill_process_by_window_name(automator)# 按端口号杀进 appiumprocess_manager.kill_process_by_port(4723)# 按进程名称查杀夜神模拟器process_manager.kill_process_by_name(nox.exe)
三、要点小结
请注意
工具类和方法是实现目的的手段不是框架的重点内容可以使用 ChatGPT 工具辅助实现工具类和方法验证即可对工具类和方法进行分类整理使用途更明确。 以本节为例以上所有方法都与系统操作相关因此将模块文件命名为 os_util.py再将方法进行分类CMDRunner、PortManager、ProcessManager、ApplicationManager
点击返回主目录
