Appium自動化測試同時運行多個設備

做android自動化的時候，啟動一個appium服務，只能匹配一個手機去自動化執行。有時候想同一套代碼，可以在不同的手機上執行，測下app在不同手機上兼容性。 這就需要啟動多個appium服務了，並且android設備和appium要一一對應才行。

一、實現需求

1.為每一台設備准備兩個端口號port和bp-port

• 每一台手機需要在PC端啟動2個線程，每個線程都需要1個端口。

• 端口號前提沒被占用，建議指定1000以上端口號。

2.檢測端口是否被占用

對於上面的端口號可以利用socket連接來檢測是否被其他占用。如果我們的檢測程序順利建立連接，意味着該端口有系統其他程序占用；如果檢測程序無法建立連接，報錯，這意味着該端口無人使用，我們拿來備用

3.利用多線程去分別為每一台設備啟動2個線程

• 一個線程用於運行測試腳本，一個線程用於運行Appium Server。

• 先啟動Appium Server最后再運行腳本。由於Appium Server啟動慢，所以待該線程啟動一段時間后，方可啟動測試腳本。

#Appium Server啟動命令

appium -p {port} -bp {bp_port} --device-name {device_name}' --platform-version {platform_version} --log {log_file} --log-level info' --log-timestamp

二、代碼實現

SingleThreadScript.py

from appium import webdriver

class SingleThreadScript:
    def __init__(self,deviceName, platformVersion, post):
        self.desired_caps = {
            'platformName': 'Android',  # 被測手機是安卓
            'platformVersion': platformVersion,  # 手機安卓版本
            'deviceName': deviceName,  # 設備名，安卓手機可以隨意填寫
            'appPackage': 'com.mobivans.onestrokecharge',  # 啟動APP Package名稱
            'appActivity': 'com.mobivans.onestrokecharge.activitys.MainActivity',  # 啟動Activity名稱
            'unicodeKeyboard': True,  # 使用自帶輸入法，輸入中文時填True
            'resetKeyboard': True,  # 執行完程序恢復原來輸入法
            'noReset': True,  # 不要重置App
            'newCommandTimeout': 60000,
            'automationName': 'UiAutomator2'
        }

        # 連接Appium Server，初始化自動化環境
        self.driver = webdriver.Remote(f'http://127.0.0.1:{post}/wd/hub', self.desired_caps)


    def business(self):
        # 設置缺省等待時間
        self.driver.implicitly_wait(5)

        #點擊 加號
        self.driver.find_element_by_xpath(
            "//android.widget.LinearLayout[@resource-id='com.mobivans.onestrokecharge:id/main_write1']/android.widget.ImageView").click()

        #點擊 日常 按鈕
        self.driver.find_element_by_xpath("//android.support.v7.widget.RecyclerView[@resource-id='com.mobivans.onestrokecharge:id/add_rv_cateGrid']/android.widget.LinearLayout[1]/android.widget.TextView").click()

        #添加備注
        self.driver.find_element_by_id("com.mobivans.onestrokecharge:id/add_et_remark").send_keys("購買了毛衣")

        #輸入 金額 45
        self.driver.find_element_by_id("com.mobivans.onestrokecharge:id/keyb_btn_4").click()
        self.driver.find_element_by_id("com.mobivans.onestrokecharge:id/keyb_btn_5").click()

        #點擊 完成 按鈕
        self.driver.find_element_by_id('com.mobivans.onestrokecharge:id/keyb_btn_finish').click()

CloudTestPlatform.py

import os
import socket
import threading
import time
from SingleThreadScript import SingleThreadScript

class CloudTestPlatform:
    def get_devices_info(self):
        devices = os.popen("adb devices").read().strip().split("\n")
        #從讀取出來的結果列表中去除下標為0的元素
        devices.pop(0)

        port = 5000
        bp_port = 8000

        #用例保存處理后的設置信息，格式如[(設備一，版本，port, bp_port),(設備二，版本，port, bp_port)]
        device_information = []

        for device in devices:
            #取出設置名字
            device_name = device.split("\t")[0].strip()
            #根據上面取出的設備名稱 取出設備版本
            platform_version = os.popen(f"adb -s {device_name} shell getprop ro.build.version.release").read().replace("\n","")

            #從當前系統查找兩個沒有被占用的端口 獲取port bpport兩個端口
            port = self.find_port(port)
            bp_port = self.find_port(bp_port)

            #將每一次設置的循環信息進行保存成一個元組
            device_information.append((device_name, platform_version, port, bp_port))

            #注意：這里要設置兩個端口每次要自增加一，因為本次端口一旦使用，下次就不能在使用，那么從下一個開始查找
            port += 1
            bp_port += 1

        return device_information

    #判斷當前系統中該端口是否可用
    def find_port(self, port):
        #self.check_port(port)返回的是true或者false，true代表占用，如果占用則在當前端口的基礎上加一，繼續判斷
        while self.check_port(port):
            port += 1
        return port

    def check_port(self, port):
        #使用tcp連接， 構造一個socket連接對象，
        #參數 AF_INET 表示該socket網絡層使用IP協議
        # 參數 SOCK_STREAM 表示該socket傳輸層使用tcp協議
        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            #判斷當前的這個端口是否能連接上,如果連接上證明端口有人在用了，連接不上會拋出異常需要使用socket.error捕獲，證明此端口可用
            conn.connect(('127.0.0.1', port))
            #關閉連接 socket.SHUT_RDWR指斷掉讀寫
            conn.shutdown(socket.SHUT_RDWR)
            return True
        except socket.error:
            return False

    #定義一個啟動appium servie的方法
    # appium -p port -bp bp_port --device-name device_name --platform-version platform_version
    def start_appiumservice(self, device_name, platform_version, port, bp_port):
        log_file = os.path.join(os.getcwd(), f'report/{device_name}_appium.log')
        cmd = f'appium -p {port} -bp {bp_port} --device-name {device_name}' \
              f' --platform-version {platform_version} --log {log_file} --log-level info' \
              f' --log-timestamp'
        os.system(cmd)

    #構造測試方法
    def start_thread_test(self):
        devices = self.get_devices_info()
        # print("設備信息", devices)
        #同時取出循環次數和值
        for device_info in devices:

            #[('127.0.0.1:21503', '7.1.2', 10000, 20000)]
            # 注意這里一定不要掉錯了方法，client線程要調用測試腳本的測試方法
            # server線程要調用appium server啟動的方法
            server_thread = threading.Thread(target=self.start_appiumservice, args=(*device_info,))
            server_thread.start()


            ost = SingleThreadScript(*device_info[:3])
            client_thread = threading.Thread(target=ost.business)


            time.sleep(10)
            client_thread.start()


if __name__ == '__main__':
    print(CloudTestPlatform().start_thread_test())