簡介

 接著上一篇繼續看一下如何併發測試以及併發測試的過程中，可能遇到的問題，在這裡巨集哥把巨集哥遇到的和小夥伴或者童鞋們，一起分享一下。

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，最好是應用在實際工作中。

好了併發測試就分享到這裡吧！

您的肯定就是我進步的動力。如果你感覺還不錯，就請鼓勵一下吧！記得點波 推薦 哦！！！（點選右邊的小球即可！(^__^) 嘻嘻……）