专栏首页北京宏哥Appium+python自动化(三十七)- 士兵突击许三多 - 多个appium服务启动,多个设备启动,多进程并发启动设备-并发测试 - 下(超详解)

Appium+python自动化(三十七)- 士兵突击许三多 - 多个appium服务启动,多个设备启动,多进程并发启动设备-并发测试 - 下(超详解)

简介

接着上一篇继续看一下如何并发测试以及并发测试的过程中,可能遇到的问题,在这里宏哥把宏哥遇到的和小伙伴或者童鞋们,一起分享一下。

Appium端口检测

问题思考

经过前面学习,我们已经能够使用python启动appium服务,但是启动Appium服务之前必须保证对应的端口没有被占用,否则会出现如下报错:

error: Couldn't start Appium REST http interface listener. Requested port is already in use. Please make sure there's no other instance of Appium running already.

针对以上这种情况,我们在启动appium服务前该如何检测端口是否可用呢?对于被占用的端口我们又该如何释放?

需求分析

1.自动检测端口是否被占用

2.如果端口被占用则自动关闭对应端口的进程

端口检测

端口检测需要使用到socket模块来校验端口是否被占用。

python socket模块官方文档

什么是socket?

网络上的两个程序通过一个双向的通信连接实现数据的交换,这个连接的一端称为一个socket。建立网络通信连接至少要一对端口号(socket)。

socket本质是编程接口(API),对TCP/IP的封装,TCP/IP也要提供可供程序员做网络开发所用的接口,这就是Socket编程接口;HTTP是轿车,提供了封装或者显示数据的具体形式;Socket是发动机,提供了网络通信的能力。

例如当你用浏览器打开我要博客园主页时,你的浏览器会创建一个socket并命令它去连接博客园的服务器主机,服务器也对客户端的请求创建一个socket进行监听。两端使用各自的socket来发送和接收信息。在socket通信的时候,每个socket都被绑定到一个特定的IP地址和端口。

补充资料: 网络工程师视频教程(自己网上搜一下哈)

代码实现

参考代码

check_port.py

 1 # coding=utf-8
 2 # 1.先设置编码,utf-8可支持中英文,如上,一般放在第一行
 3 
 4 # 2.注释:包括记录创建时间,创建人,项目名称。
 5 '''
 6 Created on 2019-9-15
 7 @author: 北京-宏哥   QQ交流群:707699217
 8 Project:学习和使用appium自动化测试-并发测试
 9 '''
10 # 3.导入模块
11 import socket
12 
13 
14 def check_port(host, port):
15     """检测指定的端口是否被占用"""
16 
17     # 创建socket对象
18 
19     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
20 
21     try:
22 
23         s.connect((host, port))
24 
25         s.shutdown(2)
26 
27     except OSError as msg:
28 
29         print('port %s is available! ' % port)
30 
31         print(msg)
32 
33         return True
34 
35     else:
36 
37         print('port %s already be in use !' % port)
38 
39         return False
40 
41 
42 if __name__ == '__main__':
43     host = '127.0.0.1'
44 
45     port = 4723
46 
47     check_port(host, port)

方法

shutdown(self, flag):禁止在一个Socket上进行数据的接收与发送。利用shutdown()函数使socket双向数据传输变为单向数据传输。shutdown()需要一个单独的参数, 该参数表示了如何关闭socket

参数

  • 0表示禁止将来读;
  • 1表示禁止将来写
  • 2表示禁止将来读和写。

当端口不可以使用时,运行上边代码,控制台输出如下:此使说明服务端已经开启这个端口服务,所以不可用。

这个端口不可用是由于我用命令行启动这个端口的appium服务

将appium服务关闭后,

端口可以使用时,运行上边代码,控制台输出如下:此使说明服务端没有开启这个端口服务,所以可用。

端口释放

如果端口被占用,则需要释放该端口。那么怎么样去释放被占用的端口呢?

代码实现

参考代码

check_port.py

 1 # coding=utf-8
 2 # 1.先设置编码,utf-8可支持中英文,如上,一般放在第一行
 3 
 4 # 2.注释:包括记录创建时间,创建人,项目名称。
 5 '''
 6 Created on 2019-9-15
 7 @author: 北京-宏哥   QQ交流群:707699217
 8 Project:学习和使用appium自动化测试-并发测试
 9 '''
10 # 3.导入模块
11 import os
12 
13 def release_port(port):
14     """释放指定的端口"""
15 
16     # 查找对应端口的pid
17     cmd_find = 'netstat -aon | findstr %s' % port
18     print(cmd_find)
19     # 返回命令执行后的结果
20     result = os.popen(cmd_find).read()
21     print(result)
22     if str(port) and 'LISTENING' in result:
23 
24         # 获取端口对应的pid进程
25         i = result.index('LISTENING')
26         start = i + len('LISTENING') + 7
27         end = result.index('\n')
28         pid = result[start:end]
29         # 关闭被占用端口的pid
30         cmd_kill = 'taskkill -f -pid %s' % pid
31         print(cmd_kill)
32         os.popen(cmd_kill)
33     else:
34 
35         print('port %s is available !' % port)
36 
37 if __name__ == '__main__':
38     host = '127.0.0.1'
39 
40     port = 4723
41 
42     # check_port(host,port)
43     release_port(port)

appium服务端口4723未启动时,控制台显示:

appium服务端口4723启动时,控制台显示:

Appium并发测试综合实践

测试场景

并发启动2个appium服务,再并发启动2台设备测试考研帮App

2个appium服务,端口配置如下:

Appium服务器端口:4723,bp端口为4724

Appium服务器端口:4725,bp端口为4726

2台设备:

设备1:127.0.0.1:62001(夜神模拟器)

设备2:emulator-5554(AVD模拟器)

测试app:考研帮Andriod版

场景分析

其实就是将前面所讲的两部分组合起来,先启动appium服务,再分配设备启动app。

代码实现

参考代码

appium_devices_sync.py

  1 # coding=utf-8
  2 # 1.先设置编码,utf-8可支持中英文,如上,一般放在第一行
  3 
  4 # 2.注释:包括记录创建时间,创建人,项目名称。
  5 '''
  6 Created on 2019-9-15
  7 @author: 北京-宏哥   QQ交流群:707699217
  8 Project:学习和使用appium自动化测试-并发测试
  9 '''
 10 # 3.导入模块
 11 appium_devices_sync.py
 12 
 13 from appium_sync.multi_appium import appium_start
 14 
 15 from appium_sync.multi_devices import appium_desired
 16 
 17 from appium_sync.check_port import *
 18 
 19 from time import sleep
 20 
 21 import multiprocessing
 22 
 23 devices_list = ['emulator-5554', '127.0.0.1:62001']
 24 
 25 
 26 def start_appium_action(host, port):
 27     '''检测端口是否被占用,如果没有被占用则启动appium服务'''
 28 
 29     if check_port(host, port):
 30 
 31         appium_start(host, port)
 32 
 33         return True
 34 
 35     else:
 36 
 37         print('appium %s start failed!' % port)
 38 
 39         return False
 40 
 41 
 42 def start_devices_action(udid, port):
 43     '''先检测appium服务是否启动成功,启动成功则再启动App,否则释放端口'''
 44 
 45     host = '127.0.0.1'
 46 
 47     if start_appium_action(host, port):
 48 
 49         appium_desired(udid, port)
 50 
 51     else:
 52 
 53         release_port(port)
 54 
 55 
 56 def appium_start_sync():
 57     '''并发启动appium服务'''
 58 
 59     print('====appium_start_sync=====')
 60 
 61     # 构建appium进程组
 62 
 63     appium_process = []
 64 
 65     # 加载appium进程
 66 
 67     for i in range(len(devices_list)):
 68         host = '127.0.0.1'
 69 
 70         port = 4723 + 2 * i
 71 
 72         appium = multiprocessing.Process(target=start_appium_action, args=(host, port))
 73 
 74         appium_process.append(appium)
 75 
 76     # 启动appium服务
 77 
 78     for appium in appium_process:
 79         appium.start()
 80 
 81     for appium in appium_process:
 82         appium.join()
 83 
 84     sleep(5)
 85 
 86 
 87 def devices_start_sync():
 88     '''并发启动设备'''
 89 
 90     print('===devices_start_sync===')
 91 
 92     # 定义desired进程组
 93 
 94     desired_process = []
 95 
 96     # 加载desired进程
 97 
 98     for i in range(len(devices_list)):
 99         port = 4723 + 2 * i
100 
101         desired = multiprocessing.Process(target=start_devices_action, args=(devices_list[i], port))
102 
103         desired_process.append(desired)
104 
105     # 并发启动App
106 
107     for desired in desired_process:
108         desired.start()
109 
110     for desired in desired_process:
111         desired.join()
112 
113 
114 if __name__ == '__main__':
115     appium_start_sync()
116 
117     devices_start_sync()

补充资料:谈谈TCP中的TIME_WAIT

运行代码控制台输出如下日志,这是怎么回事了???

这个是因为宏哥一开始用cmd命令窗口启动了appium,所以会出现下边的样子。

再次运行代码控制台输出如下日志,这又是怎么回事了???

这个是因为第一步启动appium服务已经将端口4723和4725两个端口占用了,第二步appium服务连接设备再次使用的还是同样的端口,所以才会出现如下错误,这个是代码里的bug。宏哥考考你们能不能自己找到修改。

修改bug后,再次运行代码再看一下,如下就正常了,说明你找到bug并已经修改好了。

并发用例执行

测试场景

再上面的场景基础之上,并发启动设备后然后执行跳过引导页面操作。

代码实现

参考代码

kyb_test.py
# coding=utf-8
# 1.先设置编码,utf-8可支持中英文,如上,一般放在第一行

# 2.注释:包括记录创建时间,创建人,项目名称。
'''
Created on 2019-9-15
@author: 北京-宏哥   QQ交流群:707699217
Project:学习和使用appium自动化测试-并发测试
'''
# 3.导入模块
from selenium.common.exceptions import NoSuchElementException

class KybTest(object):
    def __init__(self,driver):
        self.driver=driver

    def check_cancelBtn(self):
        print('check cancelBtn')

        try:
            cancelBtn = self.driver.find_element_by_id('android:id/button2')
        except NoSuchElementException:
            print('no cancelBtn')
        else:
            cancelBtn.click()

    def check_skipBtn(self):
        print('check skipBtn')

        try:
            skipBtn = self.driver.find_element_by_id('com.tal.kaoyan:id/tv_skip')
        except NoSuchElementException:
            print('no skipBtn')
        else:
            skipBtn.click()

    def skip_update_guide(self):
        self.check_cancelBtn()
        self.check_skipBtn()

将执行的用例集成到 multi_devices.py

代码实现

参考代码

multi_devices.py

# coding=utf-8
# 1.先设置编码,utf-8可支持中英文,如上,一般放在第一行

# 2.注释:包括记录创建时间,创建人,项目名称。
'''
Created on 2019-9-14
@author: 北京-宏哥   QQ交流群:707699217
Project:学习和使用appium自动化测试-并发测试
'''
# 3.导入模块
from appium import webdriver

import yaml

from time import ctime

from kyb_test import KybTest

with open('desired_caps.yaml', 'r')as file:
    data = yaml.load(file, Loader=yaml.FullLoader)

devices_list = ['emulator-5554','127.0.0.1:62001' ]


def appium_desired(udid, port):
    desired_caps = {}

    desired_caps['platformName'] = data['platformName']

    desired_caps['platformVersion'] = data['platformVersion']

    desired_caps['deviceName'] = data['deviceName']

    desired_caps['udid'] = udid

    desired_caps['app'] = data['app']

    desired_caps['appPackage'] = data['appPackage']

    desired_caps['appActivity'] = data['appActivity']

    desired_caps['noReset'] = data['noReset']

    print('appium port: %s start run %s at %s' % (port, udid, ctime()))

    driver = webdriver.Remote('http://' + str(data['ip']) + ':' + str(port) + '/wd/hub', desired_caps)

    driver.implicitly_wait(5)

    k = KybTest(driver)

    k.skip_update_guide()

    return driver


if __name__ == '__main__':
    appium_desired(devices_list[0], 4723)

    appium_desired(devices_list[1], 4725)

基于Docker+STF Appium并发测试(有兴趣的可以了解一下)

Docker

STF

实践案例:https://github.com/haifengrundadi/DisCartierEJ

小结

这一篇和上一篇合起来是一个微型的demo,有兴趣的童鞋和小伙伴们可以自己完善一下这个demo,最好是应用在实际工作中。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Appium+python自动化(三十六)- 士兵突击许三多 - 多个appium服务启动,多个设备启动,多进程并发启动设备-并发测试 - 上(超详解)

    前面课程只是启动了单个appium服务,只能控制单台设备。如果需要针对多台设备测试那么该如何处理?而且发现群里的小伙伴们也在时不时地在讨论这个问题,想知道怎么实...

    北京-宏哥
  • Appium+python自动化(三十五)- 命令启动appium之 appium服务命令行参数(超详解)

    前边介绍的都是通过按钮点击启动按钮来启动appium服务,有的小伙伴或者童鞋们乍一听可能不信,或者会问如何通过命令行启动appium服务呢?且听宏哥一一道来。

    北京-宏哥
  • Appium+python自动化(二十六)- 烟花一瞬,昙花一现 -Toast提示(超详解)

      今天宏哥在这里首先给小伙伴们和童鞋们分享一个有关昙花的小典故:话说昙花原是一位花神,她每天都开花,四季都灿烂。她还爱上了每天给她浇水除草的年轻人。后来,此事...

    北京-宏哥
  • 苹果申请加州测试牌照获批,无人驾驶项目Titan再加速

    大数据文摘
  • 苹果自动驾驶发生首起车祸,泰坦项目路测曝光

    苹果公司在向加州车管局(DMV)提交的一份文件中披露,它的一辆自动测试车辆发生了撞车事故:被另一辆车追尾。

    新智元
  • 进军自动驾驶,苹果来的未免太迟了

    镁客网
  • 苹果+雷克萨斯,自动驾驶系统路测照曝光

    据外媒macrumors报道,近期,苹果获得了在公共场所展开自动驾驶软件测试的批准,现在,彭博社分享了关于这一测试的新消息。本周早些时候,一辆白色的雷克萨斯RX...

    机器人网
  • Perceptln获三星数百万美元投资,将切入自动驾驶领域

    李根 发自 凹非寺 量子位 报道 | 公众号 QbitAI Perceptln,一家不算完全陌生的公司。 量子位曾先后两次提过这家位于硅谷的华人背景公司。一次...

    量子位
  • 各种字符编码详解

    相信很多人编程中经常遇到编码问题,也在这上面踩了不少坑。花了点时间把所有常用的编码搞清楚。

    serena
  • 业界 | 苹果自动驾驶汽车遭遇首次车祸

    苹果的高管们从未公开谈论过该公司的自动驾驶汽车计划,但今年 7 月份前员工张晓浪涉嫌窃取商业机密被 FBI 逮捕事件中的法律文件曾经向我们透露了一些信息:苹果公...

    机器之心

扫码关注云+社区

领取腾讯云代金券

玩转腾讯云 有奖征文活动