Appium 并发测试

启动多个Appium服务

前面课程只是启动了单个appium服务，只能控制单台设备。如果需要针对多台设备测试那么该如何处理？

首先看下面两个启动appium服务案例。

启动appium服务1

C:\Users\Shuqing>appium -p 4723

[Appium] Welcome ***\*to\**** Appium v1.7.2

[Appium] Appium REST http interface listener started on 0.0.0.0:4723

启动appium 服务2

C:\Users\Shuqing>appium -p 4725

[Appium] Welcome ***\*to\**** Appium v1.7.2

[Appium] Non-default server args:

[Appium]  port: 4725

[Appium] Appium REST http interface listener started on 0.0.0.0:4725

上面案例我们启动了2个不同的appium服务器，他们通过不同的端口来区分不同的服务；如同百米赛跑要给不同的运动员安排不同的赛道，每个运动员也只能在自己指定的赛道进行比赛。

Appium常用参数****

参数****	默认值****	含义****
-U, --udid	null	连接物理设备的唯一设备标识符
-a, --address	0.0.0.0	监听的 ip 地址
-p, --port	4723	监听的端口
-bp, --bootstrap-port	4724	连接Android设备的端口号(Android-only)
-g, --log	null	将日志输出到指定文件
--no-reset	false	session 之间不重置应用状态
--session-override	false	允许 session 被覆盖 (冲突的话)
--app-activity	null	打开Android应用时，启动的 Activity(Android-only) 的名字
--app	null	本地绝对路径或远程 http URL 所指向的一个安装包

更多参数请输入命令： appium -h

Appium安卓手机每次运行时都要安装 Unlock、Setting解决方案****

首先通过如下命令找到appium的安装路径

C:\Users\Shuqing>where appium

C:\Users\Shuqing\AppData\Roaming\npm\appium

C:\Users\Shuqing\AppData\Roaming\npm\appium.cmd

1.打开 C:\Users\Shuqing\AppData\Roaming\npm\node_modules\appium\node_modules\appium-android-driver\lib 中的android-helpers.js

***\*#\*******\*注释\*******\*475\*******\*行如下代码\****

**//await helpers.pushSettingsApp(adb);**



***\*#\*******\*注释\*******\*486\*******\*行如下代码\****

**//await helpers.pushUnlock(adb);**

2.打开C:\Users\Shuqing\AppData\Roaming\npm\node_modules\appium\node_modules\appium-android-driver\build\lib中的android-helpers.js

***\*#\*******\*注释\*******\*1128\*******\*行下面这行代码\****

//***\*return\**** _regeneratorRuntime.awrap(helpers.pushSettingsApp(adb))**;**



***\*#\*******\*修改为如下：\****

***\*return\**** context$1$0.abrupt('return', defaultIME)**;**



***\*#\*******\*注释\*******\*1163\*******\*行下面这行代码\****

//***\*return\**** _regeneratorRuntime.awrap(helpers.pushUnlock(adb))**;**



***\*#\*******\*修改如下：\****

***\*return\**** context$1$0.abrupt('return', defaultIME)**;**

修改完成后重启Appium服务即可，如果新设备没有这个两个守护App可以手动安装这两个App到设备。

Appium Setting路径**** :

{appium安装路径}\node_modules\_io.appium.settings@2.4.0@io.appium.settings\app\build\outputs\apk\settings_apk-debug.apk



eg:

C:\Users\Shuqing\AppData\Roaming\npm\node_modules\appium\node_modules\_io.appium.settings@2.4.0@io.appium.settings\app\build\outputs\apk\settings_apk-debug.apk

Unlock app路径:

appium安装路径\node_modules\_appium-unlock@2.0.0@appium-unlock\bin \unlock_apk-debug.apk



eg：

C:\Users\Shuqing\AppData\Roaming\npm\node_modules\appium\node_modules\_io.appium.settings@2.4.0@io.appium.settings\app\build\outputs\apk\unlock_apk-debug.apk

多设备启动****

前面我们已经启动了多个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



with open('desired_caps.yaml','r')as file:

 data=yaml.load(file)



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']

 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)

 return driver



if __name__ == '__main__':

 appium_desire(devices_list[0],4723)

 appium_desire(devices_list[1],4725)

多进程并发启动设备****

上面的案例设备启动并不是并发进行的，而是先后执行。如何实现2台设备同时启动，并启动App呢？

测试场景****

同时启动2台设备：'127.0.0.1:62025'和'127.0.0.1:62001'并打开考研帮app

实现思路****

可以使用Python多线程或者多进程实现。这里我们推荐使用多进程( multiprocessing) 原因如下：

· 多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响。

· 而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，容易把内容给改乱了。

知识点补充：

· 线程与进程

· Python多进程

· python多线程

代码实现****

multi_devices_sync.py

from appium ***\*import\**** webdriver

***\*import\**** yaml

from  time ***\*import\**** ctime

***\*import\**** multiprocessing

***\*with\**** open('desired_caps.yaml','r') ***\*as\**** file:

  ***\*data\****=yaml.load(file)

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

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)

  return driver

\#构建desired进程租

desired_process=[]

\#加载desied进程

***\*for\**** i in range(len(devices_list)):

  port = 4723 + 2 * i

  desired=multiprocessing.Process(target=appium_desired,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)

启动校验****

启动后我们需要校验服务是否启动成功，校验方法如下：

\1. 首先查看有没有生成对应的log文件，查看log里面的内容。

\2. 使用如下命令来查看

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种方式，具体如下：

\1. 通过netstat命令找到对应的Appium进程pid然后可以在系统任务管理器去关闭进程；

win7系统任务管理器PID显示

\2. 使用如下命令来关闭：

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服务前该如何检测端口是否可用呢？对于被占用的端口我们又该如何释放？

需求分析****

\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

***\*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