简介
接着上一篇继续看一下如何并发测试以及并发测试的过程中，可能遇到的问题，在这里宏哥把宏哥遇到的和小伙伴或者童鞋们，一起分享一下。
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，最好是应用在实际工作中。
好了并发测试就分享到这里吧！
个人公众号 微信群 （微信群已满100，可以加宏哥的微信拉你进群，请备注：进群）
您的肯定就是我进步的动力。 如果你感觉还不错，就请鼓励一下吧！记得点波 推荐 哦！！！（点击右边的小球即可！(^__^) 嘻嘻……）
