TCP 127.0.0.1:4723 0.0.0.0:0 LISTENING 29532
taskkill -f -pid 29532
Process finished with exit code 0
Appium并发测试综合实践
测试场景
并发启动2个appium服务，再并发启动2台设备测试考研帮App
2个appium服务，端口配置如下：
Appium服务器端口：4723，bp端口为4724
Appium服务器端口：4725，bp端口为4726
2台设备：
127.0.0.1:62025
127.0.0.1:62001
测试app：考研帮Andriod版
场景分析
其实就是将前面所讲的两部分组合起来，先启动appium服务，再分配设备启动app。
代码实现
appium_devices_sync.py
from appium_sync.multi_appium import appium_start
from appium_sync.multi_devices import appium_desired
from appium_sync.check_port import *
from time import sleep
import multiprocessing
devices_list=[\'127.0.0.1:62025\',\'127.0.0.1:62001\']
def start_appium_action(host,port):
\'\'\'检测端口是否被占用，如果没有被占用则启动appium服务\'\'\'
if check_port(host,port):
appium_start(host,port)
return True
else:
print(\'appium %s start failed!\' %port)
return False
def start_devices_action(udid,port):
\'\'\'先检测appium服务是否启动成功，启动成功则再启动App,否则释放端口\'\'\'
host=\'127.0.0.1\'
if start_appium_action(host,port):
appium_desired(udid,port)
else:
release_port(port)
def appium_start_sync():
\'\'\'并发启动appium服务\'\'\'
print(\'====appium_start_sync=====\')
#构建appium进程组
appium_process=[]
#加载appium进程
for i in range(len(devices_list)):
host=\'127.0.0.1\'
port = 4723 + 2 * i
appium=multiprocessing.Process(target=start_appium_action,args=(host,port))
appium_process.append(appium)
# 启动appium服务
for appium in appium_process:
appium.start()
for appium in appium_process:
appium.join()
sleep(5)
def devices_start_sync():
\'\'\'并发启动设备\'\'\'
print(\'===devices_start_sync===\')
#定义desired进程组
desired_process = []
#加载desired进程
for i in range(len(devices_list)):
port = 4723 + 2 * i
desired = multiprocessing.Process(target=start_devices_action, args=(devices_list[i], port))
desired_process.append(desired)
#并发启动App
for desired in desired_process:
desired.start()
for desired in desired_process:
desired.join()
if __name__ == \'__main__\':
appium_start_sync()
devices_start_sync()
补充资料：谈谈TCP中的TIME_WAIT
netstat -ano |findstr 4723
TCP 127.0.0.1:4723 127.0.0.1:63255 TIME_WAIT 0
TCP 127.0.0.1:4723 127.0.0.1:63257 TIME_WAIT 0
TCP 127.0.0.1:4723 127.0.0.1:63260 TIME_WAIT 0
TCP 127.0.0.1:62998 127.0.0.1:4723 TIME_WAIT 0
port 4723 is available
并发用例执行
测试场景
再上面的场景基础之上，并发启动设备后然后执行跳过引导页面操作。
代码实现
kyb_test.py
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
from appium import webdriver
import yaml
from time import ctime
from appium_sync.kyb_test import KybTest
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)
k=KybTest(driver)
k.skip_update_guide()
return driver
if __name__ == \'__main__\':
appium_desired(devices_list[0],4723)
appium_desired(devices_list[1],4725)