第二十三:Appium+Pytest实现app并发测试
1.目录结构
2.文件源码
base/base_page.py
12 import time13 from appium.webdriver import WebElement14 from appium.webdriver.webdriver import WebDriver15 from appium.webdriver.common.touch_action import TouchAction16 from selenium.webdriver.support.wait import WebDriverWait17 from selenium.common.exceptions import NoSuchElementException, TimeoutException18 19 20 class Base(object):21 22 def __init__(self, driver: WebDriver):23 self.driver = driver24 25 @property26 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, height31 32 def swipe_left(self, duration=300):33 """左滑"""34 width, height = self.get_phone_size35 start = width * 0.9, height * 0.536 end = width * 0.1, height * 0.537 return self.driver.swipe(*start, *end, duration)38 39 def swipe_right(self, duration=300):40 """右滑"""41 width, height = self.get_phone_size42 start = width * 0.1, height * 0.543 end = width * 0.9, height * 0.544 return self.driver.swipe(*start, *end, duration)45 46 def swipe_up(self, duration):47 """上滑"""48 width, height = self.get_phone_size49 start = width * 0.5, height * 0.950 end = width * 0.5, height * 0.151 return self.driver.swipe(*start, *end, duration)52 53 def swipe_down(self, duration):54 """下滑"""55 width, height = self.get_phone_size56 start = width * 0.5, height * 0.157 end = width * 0.5, height * 0.958 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, down64 :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 @staticmethod82 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_y88 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
common/check_port.py
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/get_main_js.py
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())
config/desired_caps.yml
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/root_config.py
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")
drivers/app_driver.py
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
conftest.py
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"])
run_case.py
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()
cases/test_concurrent.py
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()
3.启动说明
3.1.我代码中使用的是模拟器,如果你需要使用真机,那么需要修改部分代码,模拟器是带着端口号的,而真机没有端口号。
3.2.desired_caps.yml文件中的配置需要根据自己的app配置修改。
3.3.代码中没有包含自动连接手机的部分代码,所以执行项目前需要先手动使用adb连接上手机(有条件的,可以自己把这部分代码写一下,然后再运行项目之前调用一下adb连接手机的方法即可)。
3.4.项目目录中的allure_report, allure_results目录是系统自动生成的,一个存放最终的测试报告,一个是存放报告的依赖文件,如果你接触过allure应该知道。
3.5.log目录下存放了appium server启动之后运行的日志。
4.效果展示
***5.最后:***只是初步实现了这样一个多手机并发的需求,并没有写的很详细,比如,让项目更加的规范还需要引入PO设计模式,我这里没写这部分,其次base_page.py中还可以封装更多的方法,我也只封装了几个方法,如果真正的把这个并发引入到项目中肯定还需要完善的,但是需要添加的东西都是照葫芦画瓢了,有问题多思考!yes i can!
第二十三:Appium+Pytest实现app并发测试相关推荐
- appium+pytest实现APP并发测试
前言 这个功能已经写完很长时间了,一直没有发出来,今天先把代码发出来吧,有一些代码是参考网上写的,具体的代码说明今天暂时先不发了,代码解释的太详细还得我花点时间, 毕竟想让每个人都能看明白也不容易,所 ...
- 利用Appium对Android App进行测试
文章目录 前言 一.软件 二.环境配置 1.安装node.js (Appium 1.11以上版本不需要安装此环境) 2.Android虚拟手机和Java环境 3.安装Appium 4.测试项目的创建 ...
- Appium并发测试Capability配置
Appium并发测试 参考官方文档解释: Appium 给用户提供了在一个机器上启动多个 Android sessions 的方案.该方案只需要通过不同参数来启动的多个 Appium 服务. 以下是启 ...
- appium 多个APP进行切换测试
appium 启动多个APP 1.desired_caps配置 autoLaunch为False ,表示初始化driver后不自动启动APP 'autoLaunch':False #是否让Appium ...
- 17.第二十三章.测试管理
文章目录 23.1 测试基础 23.2 软件测试技术 23.3 信息系统测试管理 23.1 测试基础 1.软件测试模型 V模型特点: 存在一定的局限性,它将测试过程作为在需求分析.概要设计.详细设计及 ...
- 基于Appium+Pytest的UI自动化实例(Android)
基于Python3 Appium+Pytest的UI自动化实例(Android) 春有百花秋有月,夏有凉风冬有雪 若无闲事挂心头,便是人间好时节 第一部分:所需环境的配置 所需软件网盘链接(提取码19 ...
- appium python 抓包_Python学习教程:另辟蹊径,appium抓取app应用数据了解一下
原标题:Python学习教程:另辟蹊径,appium抓取app应用数据了解一下 作为爬虫工程师,没有价格不知道selenium的. 什么是selenium? Selenium原本是一个用于Web应用程 ...
- 说透APP稳定性测试
最近面试了一些测试候选人,聊到app的稳定性测试相关话题时,比如什么是app稳定性测试,app稳定性测试的目的,在什么时候执行稳定性测试最佳等,发现还是有很多的测试同学对app稳定性测试了解比较浅,甚 ...
- Appium+Pytest+Allure集成PO项目管理模式实现自动化测试
Appium+Pytest+Allure集成PO项目管理模式实现自动化测试 环境配置 Appium环境配置 Pytest环境配置 Allure环境配置 使用与集成 Appium使用 Pytest使用 ...
最新文章
- 笔记2011.7.12
- mysql dba失业_DBA要失业了?AI优化水平超DBA老炮儿
- NYOJ 20 吝啬的国度 (搜索)
- boost::process::pipe相关的测试程序
- 如何改造UE4用于赛璐璐3D卡渲?这里有一份日本大厂的实操分享
- 【bfs】Replication G(P7151)
- oracle agile 性能,Oracle Agile PLM安全漏洞(CVE-2016-3554)
- Python 高级--面向对象
- python手机号定位_利用python进行人肉搜索(一) [获取手机号]
- BIOS中的内存测试memtest
- NB-IOT基础模型搭建思路
- Github TOP100 Android开源,android开发环境搭建实验报心得
- fn键台式计算机在哪,fn键在哪?小鱼教您fn键使用方法
- [C语言]if语句的常见用法
- 【机翻】xenomai RTnet – 灵活的硬实时网络框架
- 杭州电子科技大学ACM竞赛试题----百步穿杨
- 申请 icp经营性许可证/增值电信业务许可证
- 苹果微信换行怎么打_微信空白朋友圈怎么发?安卓苹果双平台教程
- Linux学习过程感悟
- RoboMaster视觉教程(6)目标位置解算(PnP求解目标与摄像头间的相对位置)
热门文章
- 14、高可用keepalived搭建及切换
- 抖音申请企业蓝v认证的流程是怎样的?
- vm15安装MACOS
- linux服务sendmail邮件服务
- eclipse启动报错 Problems occurred when invoking code from plug-in: org.eclipse.jface
- 对AngularJS的编译和链接过程讲解一步到位的文章
- DesignPattern_Java:Proxy Pattern
- Sql server锁,独占锁,共享锁,更新锁,乐观锁,悲观锁
- 《linux c编程指南》学习手记1
- [置顶] EasyUI提交表单