簡介
?接著上一篇繼續(xù)看一下如何并發(fā)測試以及并發(fā)測試的過程中，可能遇到的問題，在這里宏哥把宏哥遇到的和小伙伴或者童鞋們，一起分享一下。
Appium端口檢測
問題思考
經(jīng)過前面學習，我們已經(jīng)能夠使用 python啟動appium服務，但是啟動Appium服務之前必須保證對應的端口沒有被占用，否則會出現(xiàn)如下報錯：
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.如果端口被占用則自動關(guān)閉對應端口的進程
端口檢測
端口檢測需要使用到 socket 模塊來校驗端口是否被占用。
python socket模塊官方文檔
什么是 socket？
網(wǎng)絡(luò)上的兩個程序通過一個雙向的通信連接實現(xiàn)數(shù)據(jù)的交換，這個連接的一端稱為一個 socket。建立網(wǎng)絡(luò)通信連接至少要一對端口號(socket)。
socket本質(zhì)是編程接口(API)，對TCP/IP的封裝，TCP/IP也要提供可供程序員做網(wǎng)絡(luò)開發(fā)所用的接口，這就是Socket編程接口；HTTP是轎車，提供了封裝或者顯示數(shù)據(jù)的具體形式；Socket是發(fā)動機，提供了網(wǎng)絡(luò)通信的能力。
例如當你用瀏覽器打開我要博客園主頁時，你的瀏覽器會創(chuàng)建一個 socket并命令它去連接博客園的服務器主機，服務器也對客戶端的請求創(chuàng)建一個socket進行監(jiān)聽。兩端使用各自的socket來發(fā)送和接收信息。在socket通信的時候，每個socket都被綁定到一個特定的IP地址和端口。
補充資料：? 網(wǎng)絡(luò)工程師視頻教程（自己網(wǎng)上搜一下哈）
代碼實現(xiàn)
?參考代碼
check_port.py
1
# coding=utf-
8
2
#
1
.先設(shè)置編碼，utf-
8可支持中英文，如上，一般放在第一行
3
4
#
2
.注釋：包括記錄創(chuàng)建時間，創(chuàng)建人，項目名稱。
5
'''
6
Created on
2019
-
9
-
15
7
@author: 北京-宏哥 QQ交流群：
707699217
8
Project:學習和使用appium自動化測試-
并發(fā)測試
9
'''
10
#
3
.導入模塊
11
import socket
12
13
14
def check_port(host, port):
15
"""
檢測指定的端口是否被占用
"""
16
17
# 創(chuàng)建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上進行數(shù)據(jù)的接收與發(fā)送。利用shutdown()函數(shù)使socket雙向數(shù)據(jù)傳輸變?yōu)閱蜗驍?shù)據(jù)傳輸。shutdown()需要一個單獨的參數(shù)， 該參數(shù)表示了如何關(guān)閉socket
參數(shù)
- 0表示禁止將來讀；
- 1表示禁止將來寫
- 2表示禁止將來讀和寫。
當端口不可以使用時，運行上邊代碼，控制臺輸出如下：此使說明服務端已經(jīng)開啟這個端口服務，所以不可用。
這個端口不可用是由于我用命令行啟動這個端口的appium服務
將appium服務關(guān)閉后，
端口可以使用時，運行上邊代碼，控制臺輸出如下：此使說明服務端沒有開啟這個端口服務，所以可用。
端口釋放
如果端口被占用，則需要釋放該端口。那么怎么樣去釋放被占用的端口呢？
代碼實現(xiàn)
?參考代碼
check_port.py
1
# coding=utf-
8
2
#
1
.先設(shè)置編碼，utf-
8可支持中英文，如上，一般放在第一行
3
4
#
2
.注釋：包括記錄創(chuàng)建時間，創(chuàng)建人，項目名稱。
5
'''
6
Created on
2019
-
9
-
15
7
@author: 北京-宏哥 QQ交流群：
707699217
8
Project:學習和使用appium自動化測試-
并發(fā)測試
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
# 返回命令執(zhí)行后的結(jié)果
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
# 關(guān)閉被占用端口的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并發(fā)測試綜合實踐
測試場景
并發(fā)啟動 2個appium服務，再并發(fā)啟動2臺設(shè)備測試考研幫App
2個appium服務，端口配置如下：
Appium服務器端口：4723，bp端口為4724
Appium服務器端口：4725，bp端口為4726
2臺設(shè)備：
設(shè)備1：127.0.0.1:62001（夜神模擬器）
設(shè)備2：emulator-5554（AVD模擬器）
測試 app：考研幫Andriod版
場景分析
其實就是將前面所講的兩部分組合起來，先啟動 appium服務，再分配設(shè)備啟動app。
代碼實現(xiàn)
?
?參考代碼
appium_devices_sync.py
1
# coding=utf-
8
2
#
1
.先設(shè)置編碼，utf-
8可支持中英文，如上，一般放在第一行
3
4
#
2
.注釋：包括記錄創(chuàng)建時間，創(chuàng)建人，項目名稱。
5
'''
6
Created on
2019
-
9
-
15
7
@author: 北京-宏哥 QQ交流群：
707699217
8
Project:學習和使用appium自動化測試-
并發(fā)測試
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
'''
并發(fā)啟動appium服務
'''
58
59
print(
'
====appium_start_sync=====
'
)
60
61
# 構(gòu)建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
'''
并發(fā)啟動設(shè)備
'''
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
# 并發(fā)啟動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，所以會出現(xiàn)下邊的樣子。
?再次運行代碼控制臺輸出如下日志，這又是怎么回事了？？？
這個是因為第一步啟動appium服務已經(jīng)將端口4723和4725兩個端口占用了，第二步appium服務連接設(shè)備再次使用的還是同樣的端口，所以才會出現(xiàn)如下錯誤，這個是代碼里的bug。宏哥考考你們能不能自己找到修改。
修改bug后，再次運行代碼再看一下，如下就正常了，說明你找到bug并已經(jīng)修改好了。
并發(fā)用例執(zhí)行
測試場景
再上面的場景基礎(chǔ)之上，并發(fā)啟動設(shè)備后然后執(zhí)行跳過引導頁面操作。
代碼實現(xiàn)
參考代碼
kyb_test.py
# coding=utf-
8
#
1
.先設(shè)置編碼，utf-
8可支持中英文，如上，一般放在第一行
#
2
.注釋：包括記錄創(chuàng)建時間，創(chuàng)建人，項目名稱。
'''
Created on
2019
-
9
-
15
@author: 北京
-宏哥 QQ交流群：
707699217
Project:學習和使用appium自動化測試
-
并發(fā)測試
'''
#
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()
?
將執(zhí)行的用例集成到 multi_devices.py
代碼實現(xiàn)
參考代碼
multi_devices.py
?
# coding=utf-
8
#
1
.先設(shè)置編碼，utf-
8可支持中英文，如上，一般放在第一行
#
2
.注釋：包括記錄創(chuàng)建時間，創(chuàng)建人，項目名稱。
'''
Created on
2019
-
9
-
14
@author: 北京
-宏哥 QQ交流群：
707699217
Project:學習和使用appium自動化測試
-
并發(fā)測試
'''
#
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并發(fā)測試（有興趣的可以了解一下）
Docker
STF
實踐案例： https://github.com/haifengrundadi/DisCartierEJ
小結(jié)
?這一篇和上一篇合起來是一個微型的demo，有興趣的童鞋和小伙伴們可以自己完善一下這個demo，最好是應用在實際工作中。
好了并發(fā)測試就分享到這里吧！
?
?個人公眾號? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 微信群 ?（微信群已滿100，可以加宏哥的微信拉你進群，請備注：進群）? ? ? ? ??
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
您的肯定就是我進步的動力。 如果你感覺還不錯，就請鼓勵一下吧！記得點波? 推薦 ?哦！?。。c擊右邊的小球即可！(^__^)?嘻嘻……）
更多文章、技術(shù)交流、商務合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯(lián)系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
