Appium+Pytest实现app并发测试

发表于:2020-1-13 11:28

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

 作者:linux超    来源:博客园

#
Appium
分享:
  前言
  这个功能已经写完很长时间了,一直没有发出来,今天先把代码发出来吧,有一些代码是参考网上写的,具体的代码说明今天暂时先不发了,代码解释的太详细还得我花点时间^_^, 毕竟想让每个人都能看明白也不容易,所以先放代码,有兴趣的先研究吧,等我有时间再做代码说明(will doing)
  目录结构
  文件源码
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:19
  4 @Auth : linux超
  5 @File : base_page.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import time
  13 from appium.webdriver import WebElement
  14 from appium.webdriver.webdriver import WebDriver
  15 from appium.webdriver.common.touch_action import TouchAction
  16 from selenium.webdriver.support.wait import WebDriverWait
  17 from selenium.common.exceptions import NoSuchElementException, TimeoutException
  18
  19
  20 class Base(object):
  21
  22     def __init__(self, driver: WebDriver):
  23         self.driver = driver
  24
  25     @property
  26     def get_phone_size(self):
  27         """获取屏幕的大小"""
  28         width = self.driver.get_window_size()['width']
  29         height = self.driver.get_window_size()['height']
  30         return width, height
  31
  32     def swipe_left(self, duration=300):
  33         """左滑"""
  34         width, height = self.get_phone_size
  35         start = width * 0.9, height * 0.5
  36         end = width * 0.1, height * 0.5
  37         return self.driver.swipe(*start, *end, duration)
  38
  39     def swipe_right(self, duration=300):
  40         """右滑"""
  41         width, height = self.get_phone_size
  42         start = width * 0.1, height * 0.5
  43         end = width * 0.9, height * 0.5
  44         return self.driver.swipe(*start, *end, duration)
  45
  46     def swipe_up(self, duration):
  47         """上滑"""
  48         width, height = self.get_phone_size
  49         start = width * 0.5, height * 0.9
  50         end = width * 0.5, height * 0.1
  51         return self.driver.swipe(*start, *end, duration)
  52
  53     def swipe_down(self, duration):
  54         """下滑"""
  55         width, height = self.get_phone_size
  56         start = width * 0.5, height * 0.1
  57         end = width * 0.5, height * 0.9
  58         return self.driver.swipe(*start, *end, duration)
  59
  60     def skip_welcome_page(self, direction, num=3):
  61         """
  62         滑动页面跳过引导动画
  63         :param direction:  str 滑动方向,left, right, up, down
  64         :param num: 滑动次数
  65         :return:
  66         """
  67         direction_dic = {
  68             "left": "swipe_left",
  69             "right": "swipe_right",
  70             "up": "swipe_up",
  71             "down": "swipe_down"
  72         }
  73         time.sleep(3)
  74         if hasattr(self, direction_dic[direction]):
  75             for _ in range(num):
  76                 getattr(self, direction_dic[direction])()  # 使用反射执行不同的滑动方法
  77         else:
  78             raise ValueError("参数{}不存在, direction可以为{}任意一个字符串".
  79                              format(direction, direction_dic.keys()))
  80
  81     @staticmethod
  82     def get_element_size_location(element):
  83         width = element.rect["width"]
  84         height = element.rect["height"]
  85         start_x = element.rect["x"]
  86         start_y = element.rect["y"]
  87         return width, height, start_x, start_y
  88
  89     def get_password_location(self, element: WebElement) -> dict:
  90         width, height, start_x, start_y = self.get_element_size_location(element)
  91         point_1 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 1)}
  92         point_2 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 1)}
  93         point_3 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 1)}
  94         point_4 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 3)}
  95         point_5 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 3)}
  96         point_6 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 3)}
  97         point_7 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 5)}
  98         point_8 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 5)}
  99         point_9 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 5)}
  100         keys = {
  101             1: point_1,
  102             2: point_2,
  103             3: point_3,
  104             4: point_4,
  105             5: point_5,
  106             6: point_6,
  107             7: point_7,
  108             8: point_8,
  109             9: point_9
  110         }
  111         return keys
  112
  113     def gesture_password(self, element: WebElement, *pwd):
  114         """手势密码: 直接输入需要链接的点对应的数字,最多9位
  115         pwd: 1, 2, 3, 6, 9
  116         """
  117         if len(pwd) > 9:
  118             raise ValueError("需要设置的密码不能超过9位!")
  119         keys_dict = self.get_password_location(element)
  120         start_point = "TouchAction(self.driver).press(x={0}, y={1}).wait(200)". \
  121             format(keys_dict[pwd[0]]["x"], keys_dict[pwd[0]]["y"])
  122         for index in range(len(pwd) - 1):  # 0,1,2,3
  123             follow_point = ".move_to(x={0}, y={1}).wait(200)". \
  124                 format(keys_dict[pwd[index + 1]]["x"],
  125                        keys_dict[pwd[index + 1]]["y"])
  126             start_point = start_point + follow_point
  127         full_point = start_point + ".release().perform()"
  128         return eval(full_point)
  129
  130     def find_element(self, locator: tuple, timeout=30) -> WebElement:
  131         wait = WebDriverWait(self.driver, timeout)
  132         try:
  133             element = wait.until(lambda driver: driver.find_element(*locator))
  134             return element
  135         except (NoSuchElementException, TimeoutException):
  136             print('no found element {} by {}', format(locator[1], locator[0]))
  137
  138
  139 if __name__ == '__main__':
  140     pass
  
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:17
  4 @Auth : linux超
  5 @File : check_port.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import socket
  13 import os
  14
  15
  16 def check_port(host, port):
  17     """检测指定的端口是否被占用"""
  18     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 创建socket对象
  19     try:
  20         s.connect((host, port))
  21         s.shutdown(2)
  22     except OSError:
  23         print('port %s is available! ' % port)
  24         return True
  25     else:
  26         print('port %s already be in use !' % port)
  27         return False
  28
  29
  30 def release_port(port):
  31     """释放指定的端口"""
  32     cmd_find = 'netstat -aon | findstr {}'.format(port)  # 查找对应端口的pid
  33     print(cmd_find)
  34
  35     # 返回命令执行后的结果
  36     result = os.popen(cmd_find).read()
  37     print(result)
  38
  39     if str(port) and 'LISTENING' in result:
  40         # 获取端口对应的pid进程
  41         i = result.index('LISTENING')
  42         start = i + len('LISTENING') + 7
  43         end = result.index('\n')
  44         pid = result[start:end]
  45         cmd_kill = 'taskkill -f -pid %s' % pid  # 关闭被占用端口的pid
  46         print(cmd_kill)
  47         os.popen(cmd_kill)
  48     else:
  49         print('port %s is available !' % port)
  50
  51
  52 if __name__ == '__main__':
  53     host = '127.0.0.1'
  54     port = 4723
  55     if not check_port(host, port):
  56         print("端口被占用")
  57         release_port(port)
 
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 13:47
  4 @Auth : linux超
  5 @File : get_main_js.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import subprocess
  13 from config.root_config import LOG_DIR
  14
  15 """
  16 获取main.js的未知,使用main.js启动appium server
  17 """
  18
  19
  20 class MainJs(object):
  21     """获取启动appium服务的main.js命令"""
  22
  23     def __init__(self, cmd: str = "where main.js"):
  24         self.cmd = cmd
  25
  26     def get_cmd_result(self):
  27         p = subprocess.Popen(self.cmd,
  28                              stdin=subprocess.PIPE,
  29                              stdout=subprocess.PIPE,
  30                              stderr=subprocess.PIPE,
  31                              shell=True)
  32         with open(LOG_DIR + "/" + "cmd.txt", "w", encoding="utf-8") as f:
  33             f.write(p.stdout.read().decode("gbk"))
  34         with open(LOG_DIR + "/" + "cmd.txt", "r", encoding="utf-8") as f:
  35             cmd_result = f.read().strip("\n")
  36         return cmd_result
  37
  38
  39 if __name__ == '__main__':
  40     main = MainJs("where main.js")
  41     print(main.get_cmd_result())
 
   1 automationName: uiautomator2
  2 platformVersion: 5.1.1
  3 platformName: Android
  4 appPackage: com.xxzb.fenwoo
  5 appActivity: .activity.addition.WelcomeActivity
  6 noReset: True
  7 ip: "127.0.0.1"
  1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:29
  4 @Auth : linux超
  5 @File : root_config.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import os
  13
  14 """
  15 project dir and path
  16 """
  17 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  18 LOG_DIR = os.path.join(ROOT_DIR, "log")
  19 CONFIG_DIR = os.path.join(ROOT_DIR, "config")
  20 CONFIG_PATH = os.path.join(CONFIG_DIR, "desired_caps.yml")
  
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:23
  4 @Auth : linux超
  5 @File : app_driver.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import subprocess
  13 from time import ctime
  14 from appium import webdriver
  15 import yaml
  16
  17 from common.check_port import check_port, release_port
  18 from common.get_main_js import MainJs
  19 from config.root_config import CONFIG_PATH, LOG_DIR
  20
  21
  22 class BaseDriver(object):
  23     """获取driver"""
  24     def __init__(self, device_info):
  25         main = MainJs("where main.js")
  26         with open(CONFIG_PATH, 'r') as f:
  27             self.data = yaml.load(f, Loader=yaml.FullLoader)
  28         self.device_info = device_info
  29         js_path = main.get_cmd_result()
  30         cmd = r"node {0} -a {1} -p {2} -bp {3} -U {4}:{5}".format(
  31             js_path,
  32             self.data["ip"],
  33             self.device_info["server_port"],
  34             str(int(self.device_info["server_port"]) + 1),
  35             self.data["ip"],
  36             self.device_info["device_port"]
  37         )
  38         print('%s at %s' % (cmd, ctime()))
  39         if not check_port(self.data["ip"], int(self.device_info["server_port"])):
  40             release_port(self.device_info["server_port"])
  41         subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + "/" + device_info["server_port"] + '.log', 'a'),
  42                          stderr=subprocess.STDOUT)
  43
  44     def get_base_driver(self):
  45         desired_caps = {
  46             'platformName': self.data['platformName'],
  47             'platformVerion': self.data['platformVersion'],
  48             'udid': self.data["ip"] + ":" + self.device_info["device_port"],
  49             "deviceName": self.data["ip"] + ":" + self.device_info["device_port"],
  50             'noReset': self.data['noReset'],
  51             'appPackage': self.data['appPackage'],
  52             'appActivity': self.data['appActivity'],
  53             "unicodeKeyboard": True
  54         }
  55         print('appium port:%s start run %s at %s' % (
  56             self.device_info["server_port"],
  57             self.data["ip"] + ":" + self.device_info["device_port"],
  58             ctime()
  59         ))
  60         driver = webdriver.Remote(
  61             'http://' + self.data['ip'] + ':' + self.device_info["server_port"] + '/wd/hub',
  62             desired_caps
  63         )
  64         return driver
  65
  66
  67 if __name__ == '__main__':
  68     pass
  
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:16
  4 @Auth : linux超
  5 @File : conftest.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 from drivers.app_driver import BaseDriver
  13 import pytest
  14 import time
  15
  16 from common.check_port import release_port
  17
  18 base_driver = None
  19
  20
  21 def pytest_addoption(parser):
  22     parser.addoption("--cmdopt", action="store", default="device_info", help=None)
  23
  24
  25 @pytest.fixture(scope="session")
  26 def cmd_opt(request):
  27     return request.config.getoption("--cmdopt")
  28
  29
  30 @pytest.fixture(scope="session")
  31 def common_driver(cmd_opt):
  32     cmd_opt = eval(cmd_opt)
  33     print("cmd_opt", cmd_opt)
  34     global base_driver
  35     base_driver = BaseDriver(cmd_opt)
  36     time.sleep(1)
  37     driver = base_driver.get_base_driver()
  38     yield driver
  39     # driver.close_app()
  40     driver.quit()
  41     release_port(cmd_opt["server_port"])
 
   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:17
  4 @Auth : linux超
  5 @File : run_case.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import pytest
  13 import os
  14 from multiprocessing import Pool
  15
  16
  17 device_infos = [
  18     {
  19         "platform_version": "5.1.1",
  20         "server_port": "4723",
  21         "device_port": "62001",
  22     },
  23     {
  24         "platform_version": "5.1.1",
  25         "server_port": "4725",
  26         "device_port": "62025",
  27     }
  28 ]
  29
  30
  31 def main(device_info):
  32     pytest.main(["--cmdopt={}".format(device_info),
  33                  "--alluredir", "./allure-results", "-vs"])
  34     os.system("allure generate allure-results -o allure-report --clean")
  35
  36
  37 if __name__ == "__main__":
  38     with Pool(2) as pool:
  39         pool.map(main, device_infos)
  40         pool.close()
  41         pool.join()

   1 """
  2 ------------------------------------
  3 @Time : 2019/9/22 12:17
  4 @Auth : linux超
  5 @File : test_concurrent.py
  6 @IDE  : PyCharm
  7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
  8 @QQ   : 28174043@qq.com
  9 @GROUP: 878565760
  10 ------------------------------------
  11 """
  12 import pytest
  13 import time
  14 from appium.webdriver.common.mobileby import MobileBy
  15
  16 from base.base_page import Base
  17
  18
  19 class TestGesture(object):
  20
  21     def test_gesture_password(self, common_driver):
  22         """这个case我只是简单的做了一个绘制手势密码的过程"""
  23         driver = common_driver
  24         base = Base(driver)
  25         base.skip_welcome_page('left', 3)  # 滑动屏幕
  26         time.sleep(3)  # 为了看滑屏的效果
  27         driver.start_activity(app_package="com.xxzb.fenwoo",
  28                               app_activity=".activity.user.CreateGesturePwdActivity")
  29         commit_btn = (MobileBy.ID, 'com.xxzb.fenwoo:id/right_btn')
  30         password_gesture = (MobileBy.ID, 'com.xxzb.fenwoo:id/gesturepwd_create_lockview')
  31         element_commit = base.find_element(commit_btn)
  32         element_commit.click()
  33         password_element = base.find_element(password_gesture)
  34         base.gesture_password(password_element, 1, 2, 3, 6, 5, 4, 7, 8, 9)
  35         time.sleep(5)  # 看效果
  36
  37
  38 if __name__ == '__main__':
  39     pytest.main()
  启动说明
  1. 我代码中使用的是模拟器,如果你需要使用真机,那么需要修改部分代码,模拟器是带着端口号的,而真机没有端口号,具体怎么修改先自己研究,后面我再详细的介绍
  2. desired_caps.yml文件中的配置需要根据自己的app配置修改
  3. 代码中没有包含自动连接手机的部分代码,所以执行项目前需要先手动使用adb连接上手机(有条件的,可以自己把这部分代码写一下,然后再运行项目之前调用一下adb连接手机的方法即可)
  4. 项目目录中的allure_report, allure_results目录是系统自动生成的,一个存放最终的测试报告,一个是存放报告的依赖文件,如果你接触过allure应该知道
  5. log目录下存放了appium server启动之后运行的日志
  效果展示
  最后
  我只是初步实现了这样一个多手机并发的需求,并没有写的很详细,比如,让项目更加的规范还需要引入PO设计模式,我这里没写这部分,其次base_page.py中还可以封装更多的方法,我也只封装了几个方法,如果真正的把这个并发引入到项目中肯定还需要完善的,但是需要添加的东西都是照葫芦画瓢了,有问题多思考!yes i can!

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

关注51Testing

联系我们

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

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

沪ICP备05003035号

沪公网安备 31010102002173号