UI自动化-多手机并发及交互
多手机并发
通常UI自动化都是单机执行，当需要测试兼容性时，则需要多机执行（多机跑同一套脚本）。这时候如何处理呢?
方案一:for 循环-多机型依次执行，即A手机执行完后B手机再执行
def run_parallel(user_info):
#pytest 执行入口
report = f"report-{user_info['device']['deviceName']}"
try:
os.system(f"del /s /q " + projectPath + f"\\{report}")
time.sleep(1)
os.system(f"rd /s /q " + projectPath + f"\\{report}")
time.sleep(1)
print(f"{report} report has deleted")
except:
print("error occur")
else:
pytest.main(["../TestCases/", f"--cmdopt={user_info}",
"--alluredir", f"../{report}/xml"])
os.system(f"allure generate ../{report}/xml -o ../{report}/html")
if __name__ == '__main__':
for i in user_info:
run_parallel(i)以上方案虽然也实现了，但明显的执行时长随着设备的增多而增多。理想状态应是多机同时执行，那如何实现呢?这就要用到多进程。
方案二:多进程-多机型并发执行，即A,B同时执行
if __name__ == '__main__':
pool = Pool()
for i in user_info:
#注意要真正实现多进程，需使用非阻塞式，即pool.apply_async;若使用阻塞式，即pool.apply，则效果同方案1
pool.apply_async(run_parallel, (i, ))
pool.close()
pool.join()【踩坑】当使用了以上多进程方案后，若发现并没有起到想象中的作用，是同一套脚本在同一台手机多次执行，则是配置不当。
2.1 配置
1.driver配置一定要加udid和systemPort，否则脚本会默认在第一台手机多次执行
desired_caps["udid"] = self.device_info["udid"]
desired_caps["systemPort"] = self.device_info["server_port"]
2.启动appium_server配置:加udid
cmd = "start appium -p {0} -bp {1} -U {2}".format(self.device_info["server_port"], self.device_info["server_port"] + 1, self.device_info["udid"])
os.system(cmd)多手机交互
上面我们处理了多机并发的问题，那如果需要多机交互的场景，该如何处理呢?比如，直播间交互，需要一台手机做主播，另一台做观众。
方案一: 定义多diver
# conftest.py
@pytest.fixture(scope='function')
def driver1_reset():
global action1s
if action1s is None:
global different
different = 1.5
with allure.step('Start and Reset APP'):
action1s = BaseDriver.base_driver(0,noReset=False)
yield action1s
with allure.step('Close_APP'):
time.sleep(3)
action1s.close_app()
different=None
action1s=None
@pytest.fixture(scope='function')
def driver2_noreset():
global action2
if action2 is None:
global different
different =2
with allure.step('Start second phone and noReset APP'):
action2 = BaseDriver.base_driver(1,noReset=True)
yield action2
with allure.step('Close_APP2'):
time.sleep(3)
action2.close_app()
different=None
action2=None#test_live.py
def test_comment(self, page, page2):
page.loginpage.if_not_login('94724068', '123456')
page2.loginpage.if_not_login('61042217', '123456')
# 主播开播
page.livePage1.click_liveBtn()
page.livePage1.click_liveBtn()
page.livePage1.clear_liveTitle()
live_title = Tools.generate_five_randomStr()
page.livePage1.input_liveTitle(live_title)
page.livePage1.click_permissions()
page.livePage1.click_configureBtn()
page.livePage1.click_liveTagBtn()
page.livePage2.select_a_liveTag(1)
page.livePage2.click_confirm_TagBtn()
page.livePage1.start_live()
# 观众进入直播间
page2.livePage2.click_live_class('颜值')
page2.livePage1.swipe_down()
page2.livePage1.enter_liveroom(live_title)以上也可实现多机交互，但可以明显的看到conftest.py中代码冗余，手机的执行顺序也是每一步都按顺序执行。即A登录后，B登录。A开播后，B当观众端。由此考虑，仍使用多进程，需要按序执行的步骤，加锁即可。
方案二：多进程 And 加锁
# run.py
def run_parallel(user_info, lock):
report = f"report-{user_info['device']['deviceName']}"
try:
os.system(f"del /s /q " + projectPath + f"\\{report}")
time.sleep(1)
os.system(f"rd /s /q " + projectPath + f"\\{report}")
time.sleep(1)
print(f"{report} report has deleted")
except:
print("error occur")
else:
pytest.main(["../TestCases/", f"--cmdopt={user_info}", f"--input={id(lock)}",
"--alluredir", f"../{report}/xml"])
os.system(f"allure generate ../{report}/xml -o ../{report}/html")
if __name__ == '__main__':
manager = Manager()
lock = manager.Lock()
pool = Pool()
for i in user_info:
pool.apply_async(run_parallel, (i, lock))
pool.close()
pool.join()# conftest.py
def pytest_addoption(parser): # 定义命令行传参参数
parser.addoption("--cmdopt", action="store", default="device", help="None")
parser.addoption("--input", default="default")
@pytest.fixture(scope="session") # 命令行参数传递给pytest
def cmdopt(request):
return request.config.getoption("--cmdopt")
@pytest.fixture(scope="session")
def input(request):
lock = int(request.config.getoption("--input"))
l = _ctypes.PyObj_FromPtr(lock)
return l
# test_live.py
def live_contact(self):
self.page.livePage2.click_live_class('颜值')
self.page.livePage1.swipe_down()
flag = self.page.livePage1.the_element_exist((By.XPATH, "//*[@text='{}' and @index='2']".format(live_title)))
if not flag:
print('start live')
print('pid is {}'.format(os.getpid()))
self.page.livePage1.click_liveBtn()
self.page.livePage1.click_liveBtn()
self.page.livePage1.clear_liveTitle()
self.page.livePage1.input_liveTitle(live_title)
self.page.livePage1.click_permissions()
self.page.livePage1.click_configureBtn()
self.page.livePage1.click_liveTagBtn()
self.page.livePage2.select_a_liveTag(1)
self.page.livePage2.click_confirm_TagBtn()
self.page.livePage1.start_live()
else:
print('join live')
print('pid is {}'.format(os.getpid()))
self.page.livePage1.enter_liveroom(live_title)
# 等待10秒观众进入直播间
time.sleep(10)
def test_comment(self, get_info,input):
info = eval(get_info)
user =info['user']
pwd = info['pwd']
device = info['device']
lock = input
self.login(user, pwd, device)
lock.acquire()
self.live_contact()
lock.release()
self.message_contact()
self.verify_message()执行以上代码，可看到执行顺序为:两手机同时登录，一台手机开播后，另一台手机进入当观众端。即不加锁时，多机同步并发，加锁时，仅一部手机执行操作后，后续手机才可依次执行该操作。利用以上逻辑，可最大化提升手机执行效率，且实现了交互的目的。
【踩坑】pytest 框架不同于Unitest框架，其传参虽有fixture可灵活使用，但其测试类中不可有init.故而要传特殊类型的参数时，需做特殊处理。因此以上难点即在如何传lock对象。
2.1 传lock对象
manage对象只可在主进程定义。以上代码使用manager.Lock()来控制进程操作，如需进程共用同一份数据，可使用manager.List(),manager.Dict()等。
pytest通过命令行传参数，默认参数类型都为字符串。所以在test_live.py拿到的参数也为字符串，并不能发挥作用。故在取参时需做处理，将内存地址转为对象。
lock = int(request.config.getoption("--input"))
l = _ctypes.PyObj_FromPtr(lock)