os_util 工具类和方法的实现

一、前置说明

• 总体目录：《从 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() == 'windows'def is_linux(self):return self.os_name.lower() == 'linux'def is_mac(self):return self.os_name.lower() == 'darwin'osname = OSName()class CMDRunner:@staticmethoddef run_command(command, timeout=5, retry_interval=0.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', timeout=60, expected_text='already 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, shell=True, capture_output=True, text=True, timeout=timeout)if result.returncode == 0:# 如果 returncode==0，表示执行成功output = result.stdout.strip()# 如果期望字符串为真，则通过"字符串是否在输出结果中"来判断是否执行成功status = expected_text in output if expected_text else True# 通常情况下，执行成功时，命令行不会返回任何结果，此时result为''，因此添加这个逻辑output = output or 'successful'else:# 否则，从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(f"Execute adb command successfully: {command}"f"\nOutput is: "f"\n=========================="f"\n{output}"f"\n==========================")return output, statusexcept subprocess.TimeoutExpired as e:logger.warning(f"Command timed out: {e}")output, status = '', Falseexcept Exception as e:logger.warning(f"Unexpected error while executing command: {e}")output, status = '', False# 添加等待时间，等待上一条命令执行完成time.sleep(retry_interval)attempts += 1logger.debug(f'Execute adb command failure: {command}'f'\nRetrying... Attempt {attempts}')# 如果超过截止时间，则跳出循环if time.time() > end_time:breaklogger.warning(f"Execute adb command failure: {command}"f"\nOutput is: "f"\n=========================="f"\n{output}"f"\n==========================")return output, statusdef run_command_strict(self, command, timeout=5, retry_interval=0.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(f"Port {port} is in use.")return Truelogger.debug(f"Port {port} is not in use.")return Falseexcept Exception as e:# 处理异常，例如权限不足等情况logger.error(f"Error 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 pids@staticmethoddef 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 pids@staticmethoddef 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 pids@staticmethoddef 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(f"Killed process with PID {process.pid}")# 杀死进程时，可能会遇到权限等问题except Exception as e:logger.debug(f"Failed 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 False@staticmethoddef set_environment_variable(key, value):"""设置系统变量，比如设置: JAVA_HOME、ANDROID_HOME"""os.environ[key] = value@staticmethoddef set_path_env_variable(path):"""设置Path环境变量warning: 这种方式设置环境变量，在windows平台会遇到1024截断的问题, todo"""if 'PATH' in os.environ:os.environ['PATH'] = path + os.pathsep + os.environ['PATH']else:os.environ['PATH'] = path@staticmethoddef start_application(executable_path, is_run_in_command=False):"""启动应用程序，但不必等待其启动完成。:param executable_path: 可执行文件的路径，比如： .exe文件:param is_run_in_command: 是否要在命令行中启动，比如：在命令行中输入 `appium` 启动 appium server"""try:if osname.is_windows():if is_run_in_command:# 如果程序需要在命令行中启动，则需要需要使用 start 启动命令行窗口command = f"start {executable_path}"else:command = executable_path# subprocess.Popen() 方法用于启动命令，但不必等待其完成,# 这对于需要启动长时间运行的程序或不需要等待程序完成的情况非常有用, 例如等待命令完成、发送信号等。subprocess.Popen(command, shell=True)logger.debug(f"Started program '{executable_path}'")else:if is_run_in_command:# 类Unix系统使用nohup命令启动程序command = f"nohup {executable_path} &"else:command = executable_pathsubprocess.Popen(command, shell=True, preexec_fn=os.setsid)logger.debug(f"Started program '{executable_path}'")except Exception as e:logger.warning(f"Error starting program '{executable_path}': {str(e)}")application_manager = ApplicationManager()if __name__ == '__main__':logging.basicConfig(level=logging.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

点击返回主目录