App 自动化测试:多设备并发 -appium+pytest+ 多线程

发表于:2021-12-21 09:16

字体: | 上一篇 | 下一篇 | 我要投稿

 作者:软件测试狂阿沐    来源:CSDN

  1、appium+python 实现单设备的 app 自动化测试
  启动 appium server,占用端口 4723。
  电脑与一个设备连接,通过 adb devices 获取已连接的设备。
  在 python 代码当中,编写启动参数,通过 pytest 编写测试用例,来进行自动化测试。
  2、若要多设备并发,同时执行自动化测试,那么需要:
  ·确定设备个数
  · 每个设备对应一个 appium server 的端口号,并启动 appium
  · pytest 要获取到每个设备的启动参数,然后执行自动化测试
  3、实现策略
  第一步:从设备池当中,获取当前连接的设备。若设备池为空,则无设备连接。
  第二步:若设备池不为空,启动一个线程,用来启动appium server.与设备个数对应。
  起始server端口为4723,每多一个设备,端口号默认+4。
  第三步:若设备池不为空,则启用多个线程,来执行app自动化测试。
  4、具体实现步骤
  4.1 通过 adb 命令,获取当前已连接的设备数、设备名称、设备的安卓版本号
  ·定义一个 ManageDevices 类。
  · 重启adb服务。
  · 通过adb devices命令获取当前平台中,已连接的设备个数,和设备uuid。
  · 通过adb -P 5037 -s 设备uuid shell getprop ro.build.version.release获取每一个设备的版本号。
  · 将所有已连接设备的设备名称、设备版本号存储在一个列表当中。
  · 通过调用get_devices_info函数,即可获得4中的列表。
  实现的部分代码为:
  class ManageDevices:
      """
         1、重启adb服务。
         2、通过adb devices命令获取当前平台中,已连接的设备个数,和设备uuid.
         3、通过adb -P 5037 -s 设备uuid shell getprop ro.build.version.release获取每一个设备的版本号。
         4、将所有已连接设备的设备名称、设备版本号存储在一个列表当中。
         5、通过调用get_devices_info函数,即可获得4中的列表。
      """
      def __init__(self):
          self.__devices_info = []
          # 重启adb服务
          self.__run_command_and_get_stout("adb kill-server")
          self.__run_command_and_get_stout("adb start-server")
      def get_devices_info(self):
          """
          获取已连接设备的uuid,和版本号。
          :return: 所有已连接设备的uuid,和版本号。
          """
          self.__get_devices_uuid()
          print(self.__devices_info)
          self.__get_device_platform_vesion()
          return self.__devices_info

  4.2 定义一个设备配置池
  设备启动参数管理池。
  每一个设备:对应一个启动参数,以及appium服务的端口号。
  1)desired_caps_config/desired_caps.yaml文件中存储了启动参数模板。
  2)从1中的模板读取出启动参数。
  3)从设备列表当中,获取每个设备的设备uuid、版本号,与2中的启动参数合并。
  4)每一个设备,指定一个appium服务端口号。从4723开始,每多一个设备,默认递增4。
  5)每一个设备,指定一个本地与设备tcp通信的端口号。从8200开始,每多一个设备,默认递增4。
  在启动参数当中,通过systemPort指定。因为appium服务会指定一个本地端口号,将数据转发到安卓设备上。默认都是使用8200端口,当有多个appium服务时就会出现端口冲突。会导致运行过程中出现socket hang up的报错。
  实现的部分代码:
  def devices_pool(port=4723,system_port=8200):
      """
      设备启动参数管理池。含启动参数和对应的端口号
      :param port: appium服务的端口号。每一个设备对应一个。
      :param system_port: appium服务指定的本地端口,用来转发数据给安卓设备。每一个设备对应一个。
      :return: 所有已连接设备的启动参数和appium端口号。
      """
      desired_template = __get_yaml_data()
      devs_pool = []
      # 获取当前连接的所有设备信息
      m = ManageDevices()
      all_devices_info = m.get_devices_info()
      # 补充每一个设备的启动信息,以及配置对应的appium server端口号
      if all_devices_info:
          for dev_info in all_devices_info:
              dev_info.update(desired_template)
              dev_info["systemPort"] = system_port
              new_dict = {
                  "caps": dev_info,
                  "port": port
              }
              devs_pool.append(new_dict)
              port += 4
              system_port += 4
      return devs_pool

  特别注意事项:2 个及 2 个以设备并发时,会遇到设备 socket hang up 的报错。
  原因是什么呢:
  在 appium server 的日志当中,有这样一行 adb 命令:adb -P 5037 -s 08e7c5997d2a forward tcp:8200 tcp:6790
  什么意思呢?
  将本地 8200 端口的数据,转发到安卓设备的 6790 端口。所以,本地启动多个 appium server,都是用的 8200 端口,就会出现冲突。
  解决方案:
  应该设置为,每一个 appium server 用不同的本地端口号,去转发数据给不同的设备。
  启动参数当中:添加 systemPort= 端口号 来设置。
  这样,每个设备都使用不同的本地端口,那么可解决此问题。
  4.3 appium server 启停管理 
  (ps 此处可以使用 appium 命令行版,也可以使用桌面版)
  在自动化用例运行之前,必须让 appium server 启动起来。
  在自动化用例执行完成之后,要 kill 掉 appium 服务。这样才不会影响下一次运行。
  代码实现如下:
  import subprocess
  import os
  from Common.handle_path import appium_logs_dir
  class ManageAppiumServer:
      """
      appium desktop通过命令行启动appium服务。
      不同平台上安装的appium,默认的appium服务路径不一样。
      初始化时,设置appium服务启动路径
      再根据给定的端口号启动appium
      """
      def __init__(self,appium_server_apth):
          self.server_apth = appium_server_apth
      # 启动appium server服务
      def start_appium_server(self,port=4723):
          appium_log_path = os.path.join(appium_logs_dir,"appium_server_{0}.log".format(port))
          command = "node {0} -p {1} -g {2} " \
                    "--session-override " \
                    "--local-timezone " \
                    "--log-timestamp & ".format(self.server_apth, port, appium_log_path)
          subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,shell=True).communicate()
      # 关闭appium服务
      @classmethod
      def stop_appium(cls,pc,post_num=4723):
          '''关闭appium服务'''
          if pc.upper() == 'WIN':
              p = os.popen(f'netstat  -aon|findstr {post_num}')
              p0 = p.read().strip()
              if p0 != '' and 'LISTENING' in p0:
                  p1 = int(p0.split('LISTENING')[1].strip()[0:4])  # 获取进程号
                  os.popen(f'taskkill /F /PID {p1}')  # 结束进程
                  print('appium server已结束')
          elif pc.upper() == 'MAC':
              p = os.popen(f'lsof -i tcp:{post_num}')
              p0 = p.read()
              if p0.strip() != '':
                  p1 = int(p0.split('\n')[1].split()[1])  # 获取进程号
                  os.popen(f'kill {p1}')  # 结束进程
                  print('appium server已结束')

  4.4 pytest 当中根据不同的启动参数来执行自动化测试用例
  在使用 pytest 执行用例时,是通过 pytest.main()会自动收集所有的用例,并自动执行生成结果。
  这种情况下,appium 会话的启动信息是在代码当中给定的。
  以上模式当中,只会读取一个设备的启动信息,并启动与设备的会话。
  虽然 fixture 有参数可以传递多个设备启动信息,但它是串行执行的。
  需要解决的问题的是:
  可以传递多个设备的启动参数,但不是通过 fixture 的参数。
  每传递一个设备启动参数进来,执行一次 pytest.main()
  解决方案:
  通过 pytest 的命令行参数。即在 pytest.main()的参数当中,将设备的启动信息传进来。
  使用 python 的多线程来实现。每接收到一个设备启动参数,就启动一个线程来执行 pytest.main。
  4.4.1 第一个,pytest 的命令行参数。
  首先需要在 conftest.py 添加命令行选项,命令行传入参数”–cmdopt“。
  用例如果需要用到从命令行传入的参数,就调用 cmdopt 函数。

  def pytest_addoption(parser):
      parser.addoption(
          "--cmdopt", action="store", default="{platformName:'Android',platformVersion:'5.1.1'}",
          help="my devices info"
      )
  @pytest.fixture(scope="session")
  def cmdopt(request):
      return request.config.getoption("--cmdopt")
  @pytest.fixture
  def start_app(cmdopt):
      device = eval(cmdopt)
      print("开始与设备 {} 进行会话,并执行测试用例 !!".format(device["caps"]["deviceName"]))
      driver = start_appium_session(device)
      yield driver
      driver.close_app()
      driver.quit()

  4.4.2 使用多线程实现: 每接收到一个设备启动参数,就启动一个线程来执行 pytest.main
  定义一个 main.py。
  run_case 函数。
  此方法主要是:接收设备启动参数,通过 pytest.main 去收集并执行用例。
  # 根据设备启动信息,通过pytest.main来收集并执行用例。
  def run_cases(device):
    """
    参数:device为设备启动参数。在pytest.main当中,传递给--cmdopt选项。
    """
      print(["-s", "-v", "--cmdopt={}".format(device)])
      reports_path = os.path.join(reports_dir,"test_result_{}_{}.html".format(device["caps"]["deviceName"], device["port"]))
      pytest.main(["-s", "-v",
                   "--cmdopt={}".format(device),
                   "--html={}".format(reports_path)]
                  )

  每有一个设备,就启动一个线程,执行 run_cases 方法。
  # 第一步:从设备池当中,获取当前连接的设备。若设备池为空,则无设备连接。
  devices = devices_pool()
  # 第二步:若设备池不为空,启动appium server.与设备个数对应。起始server端口为4723,每多一个设备,端口号默认+4
  if devices and platform_name and appium_server_path:
      # 创建线程池
      T = ThreadPoolExecutor()
      # 实例化appium服务管理类。
      mas = ManageAppiumServer(appium_server_path)
      for device in devices:
          # kill 端口,以免占用
          mas.stop_appium(platform_name,device["port"])
          # 启动appium server
          task = T.submit(mas.start_appium_server,device["port"])
          time.sleep(1)
      # 第三步:若设备池不为空,在appium server启动的情况下,执行app自动化测试。
      time.sleep(15)
      obj_list = []
      for device in devices:
          index = devices.index(device)
          task = T.submit(run_cases,device)
          obj_list.append(task)
          time.sleep(1)
      # 等待自动化任务执行完成
      for future in as_completed(obj_list):
          data = future.result()
          print(f"sub_thread: {data}")
      # kill 掉appium server服务,释放端口。
      for device in devices:
          ManageAppiumServer.stop_appium(platform_name, device["port"])

  本文内容不用于商业目的,如涉及知识产权问题,请权利人联系51Testing小编(021-64471599-8017),我们将立即处理
《2023软件测试行业现状调查报告》独家发布~

关注51Testing

联系我们

快捷面板 站点地图 联系我们 广告服务 关于我们 站长统计 发展历程

法律顾问:上海兰迪律师事务所 项棋律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2024
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪ICP备05003035号

沪公网安备 31010102002173号