Appium+Pytest实现app并发测试

学向勤中得,萤窗万卷书。这篇文章主要讲述Appium+Pytest实现app并发测试相关的知识,希望能为你提供帮助。

Appium+Pytest实现app并发测试

文章图片
这个功能已经写完很长时间了,一直没有发出来,今天先把代码发出来吧,有一些代码是参考网上写的,具体的代码说明今天暂时先不发了,代码解释的太详细还得我花点时间^_^, 毕竟想让每个人都能看明白也不容易,所以先放代码,有兴趣的先研究吧,等我有时间再做代码说明(will doing)前言这个功能已经写完很长时间了,一直没有发出来,今天先把代码发出来吧,有一些代码是参考网上写的,具体的代码说明今天暂时先不发了,代码解释的太详细还得我花点时间^_^, 毕竟想让每个人都能看明白也不容易,所以先放代码,有兴趣的先研究吧,等我有时间再做代码说明(will doing)
目录结构【Appium+Pytest实现app并发测试】
Appium+Pytest实现app并发测试

文章图片

文件源码
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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 22def __init__(self, driver: WebDriver): 23self.driver = driver 24 25@property 26def get_phone_size(self): 27"""获取屏幕的大小""" 28width = self.driver.get_window_size()[\'width\'] 29height = self.driver.get_window_size()[\'height\'] 30return width, height 31 32def swipe_left(self, duration=300): 33"""左滑""" 34width, height = self.get_phone_size 35start = width * 0.9, height * 0.5 36end = width * 0.1, height * 0.5 37return self.driver.swipe(*start, *end, duration) 38 39def swipe_right(self, duration=300): 40"""右滑""" 41width, height = self.get_phone_size 42start = width * 0.1, height * 0.5 43end = width * 0.9, height * 0.5 44return self.driver.swipe(*start, *end, duration) 45 46def swipe_up(self, duration): 47"""上滑""" 48width, height = self.get_phone_size 49start = width * 0.5, height * 0.9 50end = width * 0.5, height * 0.1 51return self.driver.swipe(*start, *end, duration) 52 53def swipe_down(self, duration): 54"""下滑""" 55width, height = self.get_phone_size 56start = width * 0.5, height * 0.1 57end = width * 0.5, height * 0.9 58return self.driver.swipe(*start, *end, duration) 59 60def skip_welcome_page(self, direction, num=3): 61""" 62滑动页面跳过引导动画 63:param direction:str 滑动方向,left, right, up, down 64:param num: 滑动次数 65:return: 66""" 67direction_dic = { 68"left": "swipe_left", 69"right": "swipe_right", 70"up": "swipe_up", 71"down": "swipe_down" 72} 73time.sleep(3) 74if hasattr(self, direction_dic[direction]): 75for _ in range(num): 76getattr(self, direction_dic[direction])()# 使用反射执行不同的滑动方法 77else: 78raise ValueError("参数{}不存在, direction可以为{}任意一个字符串". 79format(direction, direction_dic.keys())) 80 81@staticmethod 82def get_element_size_location(element): 83width = element.rect["width"] 84height = element.rect["height"] 85start_x = element.rect["x"] 86start_y = element.rect["y"] 87return width, height, start_x, start_y 88 89def get_password_location(self, element: WebElement) -> dict: 90width, height, start_x, start_y = self.get_element_size_location(element) 91point_1 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 1)} 92point_2 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 1)} 93point_3 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 1)} 94point_4 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 3)} 95point_5 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 3)} 96point_6 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 3)} 97point_7 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 5)} 98point_8 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 5)} 99point_9 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 5)} 100keys = { 1011: point_1, 1022: point_2, 1033: point_3, 1044: point_4, 1055: point_5, 1066: point_6, 1077: point_7, 1088: point_8, 1099: point_9 110} 111return keys 112 113def gesture_password(self, element: WebElement, *pwd): 114"""手势密码: 直接输入需要链接的点对应的数字,最多9位 115pwd: 1, 2, 3, 6, 9 116""" 117if len(pwd) > 9: 118raise ValueError("需要设置的密码不能超过9位!") 119keys_dict = self.get_password_location(element) 120start_point = "TouchAction(self.driver).press(x={0}, y={1}).wait(200)". \\ 121format(keys_dict[pwd[0]]["x"], keys_dict[pwd[0]]["y"]) 122for index in range(len(pwd) - 1):# 0,1,2,3 123follow_point = ".move_to(x={0}, y={1}).wait(200)". \\ 124format(keys_dict[pwd[index + 1]]["x"], 125keys_dict[pwd[index + 1]]["y"]) 126start_point = start_point + follow_point 127full_point = start_point + ".release().perform()" 128return eval(full_point) 129 130def find_element(self, locator: tuple, timeout=30) -> WebElement: 131wait = WebDriverWait(self.driver, timeout) 132try: 133element = wait.until(lambda driver: driver.find_element(*locator)) 134return element 135except (NoSuchElementException, TimeoutException): 136print(\'no found element {} by {}\', format(locator[1], locator[0])) 137 138 139 if __name__ == \'__main__\': 140pass

base/base_page.py
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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"""检测指定的端口是否被占用""" 18s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 创建socket对象 19try: 20s.connect((host, port)) 21s.shutdown(2) 22except OSError: 23print(\'port %s is available! \' % port) 24return True 25else: 26print(\'port %s already be in use !\' % port) 27return False 28 29 30 def release_port(port): 31"""释放指定的端口""" 32cmd_find = \'netstat -aon | findstr {}\'.format(port)# 查找对应端口的pid 33print(cmd_find) 34 35# 返回命令执行后的结果 36result = os.popen(cmd_find).read() 37print(result) 38 39if str(port) and \'LISTENING\' in result: 40# 获取端口对应的pid进程 41i = result.index(\'LISTENING\') 42start = i + len(\'LISTENING\') + 7 43end = result.index(\'\\n\') 44pid = result[start:end] 45cmd_kill = \'taskkill -f -pid %s\' % pid# 关闭被占用端口的pid 46print(cmd_kill) 47os.popen(cmd_kill) 48else: 49print(\'port %s is available !\' % port) 50 51 52 if __name__ == \'__main__\': 53host = \'127.0.0.1\' 54port = 4723 55if not check_port(host, port): 56print("端口被占用") 57release_port(port)

common/check_port.py
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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 23def __init__(self, cmd: str = "where main.js"): 24self.cmd = cmd 25 26def get_cmd_result(self): 27p = subprocess.Popen(self.cmd, 28stdin=subprocess.PIPE, 29stdout=subprocess.PIPE, 30stderr=subprocess.PIPE, 31shell=True) 32with open(LOG_DIR + "/" + "cmd.txt", "w", encoding="utf-8") as f: 33f.write(p.stdout.read().decode("gbk")) 34with open(LOG_DIR + "/" + "cmd.txt", "r", encoding="utf-8") as f: 35cmd_result = f.read().strip("\\n") 36return cmd_result 37 38 39 if __name__ == \'__main__\': 40main = MainJs("where main.js") 41print(main.get_cmd_result())

common/get_main_js.py
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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
Appium+Pytest实现app并发测试

文章图片
Appium+Pytest实现app并发测试

文章图片
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""" 24def __init__(self, device_info): 25main = MainJs("where main.js") 26with open(CONFIG_PATH, \'r\') as f: 27self.data = https://www.songbingjia.com/android/yaml.load(f, Loader=yaml.FullLoader) 28self.device_info = device_info 29js_path = main.get_cmd_result() 30cmd = r"node {0} -a {1} -p {2} -bp {3} -U {4}:{5}".format( 31js_path, 32self.data["ip"], 33self.device_info["server_port"], 34str(int(self.device_info["server_port"]) + 1), 35self.data["ip"], 36self.device_info["device_port"] 37) 38print(\'%s at %s\' % (cmd, ctime())) 39if not check_port(self.data["ip"], int(self.device_info["server_port"])): 40release_port(self.device_info["server_port"]) 41subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + "/" + device_info["server_port"] + \'.log\', \'a\'), 42stderr=subprocess.STDOUT) 43 44def get_base_driver(self): 45desired_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

    推荐阅读