Appium並發測試

本文轉載自查看原文 2020-04-19 15:46 789 Appium

多設備啟動

前面我們已經啟動了多個appium服務，那么接下來我們可以基於這些服務來啟動不同的設備。

測試場景

連接以下2台設備，然后分別啟動App

設備1：127.0.0.1:62001
設備2：127.0.0.1:62025

代碼實現

multi_device.py

from appium import webdriver
import yaml
from time import ctime
import os
with open('kyb_caps.yaml','r',encoding="utf-8")as file:
    data=yaml.load(file,Loader=yaml.FullLoader)

devices_list=['127.0.0.1:62001','127.0.0.1:62025']

def appium_desire(udid,port):
    desired_caps={}
    desired_caps['platformName']=data['platformName']
    desired_caps['platformVersion']=data['platformVersion']
    desired_caps['deviceName']=data['deviceName'] #devicesName 雖然是強制要求寫的，但是沒啥用，只要寫一個就行如下yaml的配置
    desired_caps['udid']=udid

    base_dir = os.path.dirname(os.path.dirname(__file__))
    app_path = os.path.join(base_dir, 'app', data['appname'])
    desired_caps['app'] = app_path

    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)
    return driver

if __name__ == '__main__':
    appium_desire(devices_list[0],4723)
    appium_desire(devices_list[1],4725)

yaml 的配置

platformName: Android
#模擬器
platformVersion: 5.1.1 deviceName: 127.0.0.1:62025 #deviceName: 127.0.0.1:62001 #mx4真機 #platformVersion: 5.1 #udid: 750BBKL22GDN #deviceName: MX4  appname: EGStar1.03.200316.beta.apk noReset: False unicodeKeyboard: True resetKeyboard: True appPackage: com.southgnss.egstar3 appActivity: com.southgnss.egstar.EGStarSplash ip: 127.0.0.1 #port: 4723

python 多進程與多線程的區別所在 ✔點擊

代碼實現 python多進程與多線程

multi_devices_sync.py

from appium import webdriver
import yaml
from time import ctime
import os
import multiprocessing
from threading import Thread


devices_list=['127.0.0.1:62001','127.0.0.1:62025']

def appium_desire(udid,port):
    with open('kyb_caps.yaml', 'r', encoding="utf-8")as file:
        data = yaml.load(file, Loader=yaml.FullLoader)
    desired_caps={}
    desired_caps['platformName']=data['platformName']
    desired_caps['platformVersion']=data['platformVersion']
    desired_caps['deviceName']=data['deviceName']
    desired_caps['udid']=udid

    base_dir = os.path.dirname(os.path.dirname(__file__))
    app_path = os.path.join(base_dir, 'app', data['appname'])
    desired_caps['app'] = app_path

    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)
    return driver

#構建一個desired進程組
desired_process=[]

#加載desied進程
for i in range(len(devices_list)):
    port = 4723 + 2 * i
    desired = multiprocessing.Process(target=appium_desire,args=(devices_list[i],port))
    # desired = Thread(target=appium_desire,args=(devices_list[i],port))
    desired_process.append(desired)

if __name__ == '__main__':
    for desired in desired_process:
        desired.start()
    for desired in desired_process:
        desired.join()

Python啟動Appium 服務

目前我們已經實現了並發啟動設備，但是我們的Appium服務啟動還是手動檔，比如使用Dos命令或者bat批處理來手動啟動appium服務，啟動效率低下。如何將啟動Appium服務也實現自動化呢？

方案分析

我們可以使用python啟動appium服務，這里需要使用subprocess模塊，該模塊可以創建新的進程，並且連接到進程的輸入、輸出、錯誤等管道信息，並且可以獲取進程的返回值。

subprocess模塊官方文檔

測試場景

使用Python啟動2台appium服務，端口配置如下：

Appium服務器端口：4723，bp端口為4724
Appium服務器端口：4725，bp端口為4726

說明：bp端口（ --bootstrap-port）是appium和設備之間通信的端口，如果不指定到時無法操作多台設備運行腳本。

代碼實現

首先我們使用Python腳本啟動單個appium服務：

host：127.0.0.1
port：4723

multi_appium.py

import subprocess
from time import ctime

def appium_start(host,port):
    '''啟動appium server'''
    bootstrap_port = str(port + 1)
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

    print('%s at %s' %(cmd,ctime()))
    subprocess.Popen(cmd, shell=True,stdout=open('./appium_log/'+str(port)+'.log','a'),stderr=subprocess.STDOUT)


if __name__ == '__main__':
    host = '127.0.0.1'
    port=4723
    appium_start(host,port)

啟動校驗

啟動后我們需要校驗服務是否啟動成功，校驗方法如下：

首先查看有沒有生成對應的log文件，查看log里面的內容。
使用如下命令來查看

netstat -ano |findstr 端口號

netstat 命令解釋

netstat命令是一個監控TCP/IP網絡的非常有用的工具，它可以顯示路由表、實際的網絡連接以及每一個網絡接口設備的狀態信息。輸入 netstat -ano 回車.可以查看本機開放的全部端口；輸入命令 netstat -h可以查看全部參數含義。

C:\Users\Shuqing>netstat -ano |findstr "4723"
  TCP    127.0.0.1:4723         0.0.0.0:0              LISTENING       8224

關閉Appium服務

關閉進程有2種方式，具體如下：

通過netstat命令找到對應的Appium進程pid然后可以在系統任務管理器去關閉進程；

win7系統任務管理器PID顯示

使用如下命令來關閉：

taskkill -f -pid appium進程id

多個appium服務啟動

多個appium服務啟動非常簡單，只需在執行環境使用循環調用即可。

if __name__ == '__main__':
    host = '127.0.0.1'
    for i in range(2):
        port=4723+2*i
        appium_start(host,port)

多進程並發啟動appium服務

上面的案例還不是並發執行啟動appium,因此我們需要使用多進程來實現並發啟動。同樣需要引入multiprocessing多進程模塊。

muti_appium_sync.py

import multiprocessing
import subprocess
from  time import  ctime

def appium_start(host,port):
    '''啟動appium server'''

    bootstrap_port = str(port + 1)
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' --bootstrap-port ' + str(bootstrap_port)

    print('%s at %s' %(cmd,ctime()))
    subprocess.Popen(cmd, shell=True,stdout=open('./appium_log/'+str(port)+'.log','a'),stderr=subprocess.STDOUT)
    
    
#構建appium進程組
appium_process=[]

#加載appium進程
for i in range(2):
    host='127.0.0.1'
    port = 4723 + 2 * i
    appium=multiprocessing.Process(target=appium_start,args=(host,port))
    appium_process.append(appium)
    

if __name__ == '__main__':
    #並發啟動appium服務
    for appium in appium_process:
        appium.start()
    for appium in appium_process:
        appium.join()

Appium端口檢測

問題思考

經過前面學習，我們已經能夠使用python啟動appium服務，但是啟動Appium服務之前必須保證對應的端口沒有被占用，否則會出現如下報錯：

C:\Users\Shuqing>appium -a 127.0.0.1 -p 4723
[Appium] Welcome to Appium v1.7.2
[Appium] Non-default server args:
[Appium]   address: 127.0.0.1
[HTTP] Could not start REST http interface listener. The requested port may already be in use. Please make sure there is no other instance of this server running already.
uncaughtException: listen EADDRINUSE 127.0.0.1:4723

針對以上這種情況，我們在啟動appium服務前該如何檢測端口是否可用呢？對於被占用的端口我們又該如何釋放？

需求分析

自動檢測端口是否被占用
如果端口被占用則自動關閉對應端口的進程

端口檢測

端口檢測需要使用到socket模塊來校驗端口是否被占用。

python socket模塊官方文檔

什么是socket？

網絡上的兩個程序通過一個雙向的通信連接實現數據的交換，這個連接的一端稱為一個socket。建立網絡通信連接至少要一對端口號(socket)。

socket本質是編程接口(API)，對TCP/IP的封裝，TCP/IP也要提供可供程序員做網絡開發所用的接口，這就是Socket編程接口；HTTP是轎車，提供了封裝或者顯示數據的具體形式；Socket是發動機，提供了網絡通信的能力。

例如當你用瀏覽器打開我要自學網主頁時，你的瀏覽器會創建一個socket並命令它去連接自學網的服務器主機，服務器也對客戶端的請求創建一個socket進行監聽。兩端使用各自的socket來發送和接收信息。在socket通信的時候，每個socket都被綁定到一個特定的IP地址和端口。

補充資料：網絡工程師視頻教程

代碼實現

check_port.py

import socket

def check_port(host, port):
    """檢測指定的端口是否被占用"""

    #創建socket對象
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.connect((host, port))
        s.shutdown(2)
    except OSError as msg:
        print('port %s is available! ' %port)
        print(msg)
        return True
    else:
        print('port %s already be in use !' % port)
        return False


if __name__ == '__main__':
    host='127.0.0.1'
    port=4723
    check_port(host,port)

方法

shutdown(self, flag)：禁止在一個Socket上進行數據的接收與發送。利用shutdown()函數使socket雙向數據傳輸變為單向數據傳輸。shutdown()需要一個單獨的參數，該參數表示了如何關閉socket

參數

0表示禁止將來讀；
1表示禁止將來寫
2表示禁止將來讀和寫。

當端口可以使用時，控制台輸出如下：此使說明服務端沒有開啟這個端口服務，所以可用。

C:\Python35\python.exe E:/AppiumScript/advance/appium_cmd/appium_multiProcess.py
port 4723 is available! 
[WinError 10061] 由於目標計算機積極拒絕，無法連接。 

Process finished with exit code 0

端口釋放

如果端口被占用，則需要釋放該端口。那么怎么樣去釋放被占用的端口呢？

代碼實現

check_port.py

import os

def release_port(port):
    """釋放指定的端口"""

    #查找對應端口的pid
    cmd_find='netstat -aon | findstr %s' %port
    print(cmd_find)

    #返回命令執行后的結果
    result = os.popen(cmd_find).read()
    print(result)

    if  str(port) and 'LISTENING' in result:
        #獲取端口對應的pid進程
        i=result.index('LISTENING')
        start=i+len('LISTENING')+7
        end=result.index('\n')
        pid=result[start:end]

        # 關閉被占用端口的pid
        cmd_kill='taskkill -f -pid %s' %pid
        print(cmd_kill)
        os.popen(cmd_kill)

    else:
        print('port %s is available !' %port)


if __name__ == '__main__':
    host='127.0.0.1'
    port=4723
    # check_port(host,port)
    release_port(port)

控制台顯示：

C:\Python35\python.exe E:/AppiumScript/advance/appium_cmd/appium_multiProcess.py
netstat -aon | findstr "4723"
  TCP    127.0.0.1:4723         0.0.0.0:0              LISTENING       29532

taskkill -f -pid 29532

Process finished with exit code 0