Appium+python自動化（三十七）- 士兵突擊許三多 - 多個appium服務啟動，多個設備啟動，多進程並發啟動設備-並發測試 - 下（超詳解）
- 2019 年 10 月 3 日
- 筆記
簡介
接著上一篇繼續看一下如何並發測試以及並發測試的過程中，可能遇到的問題，在這裡宏哥把宏哥遇到的和小夥伴或者童鞋們，一起分享一下。
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模組來校驗埠是否被佔用。
什麼是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並發測試（有興趣的可以了解一下）
實踐案例：https://github.com/haifengrundadi/DisCartierEJ
小結
這一篇和上一篇合起來是一個微型的demo，有興趣的童鞋和小夥伴們可以自己完善一下這個demo，最好是應用在實際工作中。
好了並發測試就分享到這裡吧！
個人公眾號 微信群 （微信群已滿100，可以加宏哥的微信拉你進群，請備註：進群）
您的肯定就是我進步的動力。如果你感覺還不錯，就請鼓勵一下吧！記得點波 推薦 哦！！！（點擊右邊的小球即可！(^__^) 嘻嘻……）
