Appium 并发测试基于unitest

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

前言：

在回归测试阶段，UI测试，兼容测试是测试的必要步骤。UI自动化的本身是比较冗余的测试，但是换个角度思考，UI自动化同时连接多台设备，那么在回归测试时，在同一个脚本下产生的测试结果是非常有价值的。

不同设备在并发下的测试结果可以为我们提供：

1. 兼容性测试（不同的手机品牌，Android版本，分辨率等）

2. 性能测试（通过安装Emmagee，监控不同手机在同脚本下，性能的变化）

3. 界面对比（通过图像识别opencv，截图对比等查看在相同页面的变化）

思路：

1. 启动多路appium服务

2. 启动并连接多路手机端

3. 运行并生成测试报告

问题：

1. python的unittest框架和java不同，不支持参数传入，可以通过重写unittest.TestCase的init添加参数

2. appium 通过命令行启动，需要安装非desktop的版本，且最好安装1.9版本appium，1.0我启动不了

框架代码截取：

1. 重写unittest的初始化函数

class ParametrizedCase(unittest.TestCase):
def __init__(self, methodName='runTest', param=None):
super(ParametrizedCase, self).__init__(methodName)
global devices
devices = param @classmethod
def setUpClass(cls):
cls.driver = connect_device(devices) @classmethod
def tearDownClass(cls):
cls.driver.close_app()
cls.driver.quit()

2. 封装启动appium的服务方法：

基于 appium 的启动命令

appium -p -bp -U

封装多线程启动

class AppiumServer:
def __init__(self, kwargs=None):
self.kwargs = kwargs def start_server(self):
"""start the appium server
"""
for i in range(0, len(self.kwargs)):
cmd = "appium --session-override -p %s -bp %s -U %s" % (
self.kwargs[i]["port"], self.kwargs[i]["bport"], self.kwargs[i]["devices"])
print(cmd)
if platform.system() == "Windows": # windows下启动server
t1 = RunServer(cmd)
p = Process(target=t1.start())
p.start()
while True:
print("--------start_win_server-------------")
if self.win_is_runnnig("http://127.0.0.1:" + self.kwargs[i]["port"] + "/wd/hub" + "/status"):
print("-------win_server_ 成功--------------")
break
else:
appium = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1,
close_fds=True)
while True:
appium_line = appium.stdout.readline().strip().decode()
time.sleep(1)
print("---------start_server----------")
if 'listener started' in appium_line or 'Error: listen' in appium_line:
print("----server_ 成功---")
break def win_is_runnnig(self, url):
"""Determine whether server is running
:return:True or False
"""
response = None
time.sleep(1)
try:
response = urllib.request.urlopen(url, timeout=5) if str(response.getcode()).startswith(""):
return True
else:
return False
except URLError:
return False
except socket.timeout:
return False
finally:
if response:
response.close() def stop_server(self, devices):
sysstr = platform.system() if sysstr == 'Windows':
os.popen("taskkill /f /im node.exe")
else:
for device in devices:
# mac
cmd = "lsof -i :{0}".format(device["port"])
plist = os.popen(cmd).readlines()
plisttmp = plist[1].split(" ")
plists = plisttmp[1].split(" ")
# print plists[0]
os.popen("kill -9 {0}".format(plists[0]))class RunServer(threading.Thread):
def __init__(self, cmd):
threading.Thread.__init__(self)
self.cmd = cmd def run(self):
os.system(self.cmd)

3. 封装连接Android设备的方法：

def connect_device(devices):
desired_caps = {}
desired_caps['platformVersion'] = devices["platformVersion"]
desired_caps['platformName'] = devices["platformName"]
desired_caps["automationName"] = devices['automationName']
desired_caps['deviceName'] = devices["deviceName"]
desired_caps["appPackage"] = devices["appPackage"]
desired_caps["appActivity"] = devices["appActivity"]
desired_caps["noReset"] = True
desired_caps['noSign'] = True
desired_caps["unicodeKeyboard"] = True
desired_caps["resetKeyboard"] = True
desired_caps["systemPort"] = devices["systemPort"] # desired_caps['app'] = devices["app"]
remote = "http://127.0.0.1:" + str(devices["port"]) + "/wd/hub"
# remote = "http://127.0.0.1:" + "4723" + "/wd/hub"
driver = webdriver.Remote(remote, desired_caps)
return driver

4. 多线程启动服务和多线程连接多终端，生成日志报告

def runnerPool(getDevices):
devices_Pool = [] for i in range(0, len(getDevices)):
_initApp = {}
_initApp["deviceName"] = getDevices[i]["devices"]
_initApp["platformVersion"] = getPhoneInfo(devices=_initApp["deviceName"])["release"]
_initApp["platformName"] = "Android"
_initApp["port"] = getDevices[i]["port"]
_initApp["automationName"] = "UiAutomator2"
_initApp["systemPort"] = getDevices[i]["systemPort"]
_initApp["appPackage"] = 'cn.vsx.vc'
_initApp["appActivity"] = '.activity.RegistActivity'
devices_Pool.append(_initApp)
print(f'devices pool are {devices_Pool}')
with ProcessPoolExecutor(len(devices_Pool)) as pool:
pool.map(runnerCaseApp, devices_Pool)def runnerCaseApp(devices):
suite = unittest.TestSuite()
suite.addTest(ParametrizedCase.parametrize(group_call, param=devices)) # 加入测试类
now = time.strftime('%Y-%m-%d %H_%M_%S')
result = BeautifulReport(suite)
result.report(filename=now + f"_{devices['deviceName'].split(':', 1)[0]}.html", log_path='E:\\TestReports\\',
description=f"{devices['deviceName'].split(':', 1)[0]}")if __name__ == '__main__':
devices = attached_devices()
if len(devices) > 0:
l_devices = []
for dev in devices:
app = {}
app["devices"] = dev
app["port"] = str(random.randint(4700, 4900))
app["bport"] = str(random.randint(4700, 4900))
app["systemPort"] = random.randint(4700, 4900)
l_devices.append(app)
print(f'list of server:{l_devices}')
appium_server = AppiumServer(l_devices)
appium_server.start_server()
runnerPool(l_devices)
appium_server.stop_server(l_devices)
else:
print("没有可用的安卓设备")