Appium+Pytest实现app并发测试

Posted linux超

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Appium+Pytest实现app并发测试相关的知识,希望对你有一定的参考价值。

前言

这个功能已经写完很长时间了,一直没有发出来,今天先把代码发出来吧,有一些代码是参考网上写的,具体的代码说明今天暂时先不发了,代码解释的太详细还得我花点时间^_^, 毕竟想让每个人都能看明白也不容易,所以先放代码,有兴趣的先研究吧,等我有时间再做代码说明(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
base/base_page.py
 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)
common/check_port.py
 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())
common/get_main_js.py
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"
config/desired_caps.yml
 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")
config/root_config.py
 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
app 自动化测试 - 多设备并发 -appium+pytest+ 多线程

pytest学习

Appium 并发多进程基于 Pytest框架

Python+Appium+Pytest+Allure实战APP自动化测试框架,小试牛刀!

python+appium+pytest自动化测试-参数化设置建议收藏

Appium+Python app自动化测试之脚本启动和停止Appium服务