一. 前言

Appium结合Pytest开展App自动化测试时，你知道如何实现用例并发执行吗？费话不多说，直接上代码， 毕竟想让每个人都能看明白也不容易，所以先放代码，有兴趣的先自行研究。

二、.目录结构

三、文件源码

3.1 base/base_page.py

"""n------------------------------------n@File : base_page.pyn------------------------------------n"""nimport timenfrom appium.webdriver import WebElementnfrom appium.webdriver.webdriver import WebDrivernfrom appium.webdriver.common.touch_action import TouchActionnfrom selenium.webdriver.support.wait import WebDriverWaitnfrom selenium.common.exceptions import NoSuchElementException, TimeoutExceptionn n nclass Base(object):n n def __init__(self, driver: WebDriver):n self.driver = drivern n @propertyn def get_phone_size(self):n """获取屏幕的大小"""n width = self.driver.get_window_size()['width']n height = self.driver.get_window_size()['height']n return width, heightn n def swipe_left(self, duration=300):n """左滑"""n width, height = self.get_phone_sizen start = width * 0.9, height * 0.5n end = width * 0.1, height * 0.5n return self.driver.swipe(*start, *end, duration)n n def swipe_right(self, duration=300):n """右滑"""n width, height = self.get_phone_sizen start = width * 0.1, height * 0.5n end = width * 0.9, height * 0.5n return self.driver.swipe(*start, *end, duration)n n def swipe_up(self, duration):n """上滑"""n width, height = self.get_phone_sizen start = width * 0.5, height * 0.9n end = width * 0.5, height * 0.1n return self.driver.swipe(*start, *end, duration)n n def swipe_down(self, duration):n """下滑"""n width, height = self.get_phone_sizen start = width * 0.5, height * 0.1n end = width * 0.5, height * 0.9n return self.driver.swipe(*start, *end, duration)n n def skip_welcome_page(self, direction, num=3):n """n 滑动页面跳过引导动画n :param direction: str 滑动方向，left, right, up, downn :param num: 滑动次数n :return:n """n direction_dic = {n "left": "swipe_left",n "right": "swipe_right",n "up": "swipe_up",n "down": "swipe_down"n }n time.sleep(3)n if hasattr(self, direction_dic[direction]):n for _ in range(num):n getattr(self, direction_dic[direction])() # 使用反射执行不同的滑动方法n else:n raise ValueError("参数{}不存在, direction可以为{}任意一个字符串".n format(direction, direction_dic.keys()))n n @staticmethodn def get_element_size_location(element):n width = element.rect["width"]n height = element.rect["height"]n start_x = element.rect["x"]n start_y = element.rect["y"]n return width, height, start_x, start_yn n def get_password_location(self, element: WebElement) -> dict:n width, height, start_x, start_y = self.get_element_size_location(element)n point_1 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 1)}n point_2 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 1)}n point_3 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 1)}n point_4 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 3)}n point_5 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 3)}n point_6 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 3)}n point_7 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 5)}n point_8 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 5)}n point_9 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 5)}n keys = {n 1: point_1,n 2: point_2,n 3: point_3,n 4: point_4,n 5: point_5,n 6: point_6,n 7: point_7,n 8: point_8,n 9: point_9n }n return keysnn def gesture_password(self, element: WebElement, *pwd):n """手势密码: 直接输入需要链接的点对应的数字，最多9位n pwd: 1, 2, 3, 6, 9n """n if len(pwd) > 9:n raise ValueError("需要设置的密码不能超过9位!")n keys_dict = self.get_password_location(element)n start_point = "TouchAction(self.driver).press(x={0}, y={1}).wait(200)". n format(keys_dict[pwd[0]]["x"], keys_dict[pwd[0]]["y"])n for index in range(len(pwd) - 1): # 0,1,2,3n follow_point = ".move_to(x={0}, y={1}).wait(200)". n format(keys_dict[pwd[index + 1]]["x"],n keys_dict[pwd[index + 1]]["y"])n start_point = start_point + follow_pointn full_point = start_point + ".release().perform()"n return eval(full_point)nn def find_element(self, locator: tuple, timeout=30) -> WebElement:n wait = WebDriverWait(self.driver, timeout)n try:n element = wait.until(lambda driver: driver.find_element(*locator))n return elementn except (NoSuchElementException, TimeoutException):n print('no found element {} by {}', format(locator[1], locator[0]))nnnif __name__ == '__main__':n pass

3.2 common/check_port.py

"""n------------------------------------n@File : check_port.pyn------------------------------------n"""nimport socketnimport osnnndef check_port(host, port):n """检测指定的端口是否被占用"""n s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建socket对象n try:n s.connect((host, port))n s.shutdown(2)n except OSError:n print('port %s is available! ' % port)n return Truen else:n print('port %s already be in use !' % port)n return Falsennndef release_port(port):n """释放指定的端口"""n cmd_find = 'netstat -aon | findstr {}'.format(port) # 查找对应端口的pidn print(cmd_find)nn # 返回命令执行后的结果n result = os.popen(cmd_find).read()n print(result)nn if str(port) and 'LISTENING' in result:n # 获取端口对应的pid进程n i = result.index('LISTENING')n start = i + len('LISTENING') + 7n end = result.index('n')n pid = result[start:end]n cmd_kill = 'taskkill -f -pid %s' % pid # 关闭被占用端口的pidn print(cmd_kill)n os.popen(cmd_kill)n else:n print('port %s is available !' % port)nnnif __name__ == '__main__':n host = '127.0.0.1'n port = 4723n if not check_port(host, port):n print("端口被占用")n release_port(port)

3.3 common/get_main_js.py

"""n------------------------------------n@File : get_main_js.pyn@IDE : PyCharmn------------------------------------n"""nimport subprocessnfrom config.root_config import LOG_DIRnn"""n获取main.js的未知，使用main.js启动appium servern"""nnnclass MainJs(object):n """获取启动appium服务的main.js命令"""nn def __init__(self, cmd: str = "where main.js"):n self.cmd = cmdnn def get_cmd_result(self):n p = subprocess.Popen(self.cmd,n stdin=subprocess.PIPE,n stdout=subprocess.PIPE,n stderr=subprocess.PIPE,n shell=True)n with open(LOG_DIR + "/" + "cmd.txt", "w", encoding="utf-8") as f:n f.write(p.stdout.read().decode("gbk"))n with open(LOG_DIR + "/" + "cmd.txt", "r", encoding="utf-8") as f:n cmd_result = f.read().strip("n")n return cmd_resultnnnif __name__ == '__main__':n main = MainJs("where main.js")n print(main.get_cmd_result())

3.4 config/desired_caps.yml

automationName: uiautomator2nplatformVersion: 5.1.1nplatformName: AndroidnappPackage: com.xxzb.fenwoonappActivity: .activity.addition.WelcomeActivitynnoReset: Truenip: "127.0.0.1"

3.5 config/root_config.py

"""n------------------------------------n@File : root_config.pyn------------------------------------n"""nimport osnn"""nproject dir and pathn"""nROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))nLOG_DIR = os.path.join(ROOT_DIR, "log")nCONFIG_DIR = os.path.join(ROOT_DIR, "config")nCONFIG_PATH = os.path.join(CONFIG_DIR, "desired_caps.yml")

3.6 drivers/app_driver.py

"""n------------------------------------n@File : app_driver.pyn------------------------------------n"""nimport subprocessnfrom time import ctimenfrom appium import webdrivernimport yamlnnfrom common.check_port import check_port, release_portnfrom common.get_main_js import MainJsnfrom config.root_config import CONFIG_PATH, LOG_DIRnnnclass BaseDriver(object):n """获取driver"""n def __init__(self, device_info):n main = MainJs("where main.js")n with open(CONFIG_PATH, 'r') as f:n self.data = yaml.load(f, Loader=yaml.FullLoader)n self.device_info = device_infon js_path = main.get_cmd_result()n cmd = r"node {0} -a {1} -p {2} -bp {3} -U {4}:{5}".format(n js_path,n self.data["ip"],n self.device_info["server_port"],n str(int(self.device_info["server_port"]) + 1),n self.data["ip"],n self.device_info["device_port"]n )n print('%s at %s' % (cmd, ctime()))n if not check_port(self.data["ip"], int(self.device_info["server_port"])):n release_port(self.device_info["server_port"])n subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + "/" + device_info["server_port"] + '.log', 'a'),n stderr=subprocess.STDOUT)nn def get_base_driver(self):n desired_caps = {n 'platformName': self.data['platformName'],n 'platformVerion': self.data['platformVersion'],n 'udid': self.data["ip"] + ":" + self.device_info["device_port"],n "deviceName": self.data["ip"] + ":" + self.device_info["device_port"],n 'noReset': self.data['noReset'],n 'appPackage': self.data['appPackage'],n 'appActivity': self.data['appActivity'],n "unicodeKeyboard": Truen }n print('appium port:%s start run %s at %s' % (n self.device_info["server_port"],n self.data["ip"] + ":" + self.device_info["device_port"],n ctime()n ))n driver = webdriver.Remote(n 'http://' + self.data['ip'] + ':' + self.device_info["server_port"] + '/wd/hub',n desired_capsn )n return drivernnnif __name__ == '__main__':n pass

3.7 conftest.py

"""n------------------------------------n@File : conftest.pyn------------------------------------n"""nfrom drivers.app_driver import BaseDrivernimport pytestnimport timennfrom common.check_port import release_portnnbase_driver = Nonennndef pytest_addoption(parser):n parser.addoption("--cmdopt", action="store", default="device_info", help=None)nnn@pytest.fixture(scope="session")ndef cmd_opt(request):n return request.config.getoption("--cmdopt")nnn@pytest.fixture(scope="session")ndef common_driver(cmd_opt):n cmd_opt = eval(cmd_opt)n print("cmd_opt", cmd_opt)n global base_drivern base_driver = BaseDriver(cmd_opt)n time.sleep(1)n driver = base_driver.get_base_driver()n yield drivern # driver.close_app()n driver.quit()n release_port(cmd_opt["server_port"])

3.8 run_case.py

"""n------------------------------------n@File : run_case.pyn------------------------------------n"""nimport pytestnimport osnfrom multiprocessing import Poolnnndevice_infos = [n {n "platform_version": "5.1.1",n "server_port": "4723",n "device_port": "62001",n },n {n "platform_version": "5.1.1",n "server_port": "4725",n "device_port": "62025",n }n]nnndef main(device_info):n pytest.main(["--cmdopt={}".format(device_info),n "--alluredir", "./allure-results", "-vs"])n os.system("allure generate allure-results -o allure-report --clean")nnnif __name__ == "__main__":n with Pool(2) as pool:n pool.map(main, device_infos)n pool.close()n pool.join()

3.9 cases/test_concurrent.py

"""n------------------------------------n@File : test_concurrent.pyn------------------------------------n"""nimport pytestnimport timenfrom appium.webdriver.common.mobileby import MobileBynnfrom base.base_page import Basennnclass TestGesture(object):nn def test_gesture_password(self, common_driver):n """这个case我只是简单的做了一个绘制手势密码的过程"""n driver = common_drivern base = Base(driver)n base.skip_welcome_page('left', 3) # 滑动屏幕n time.sleep(3) # 为了看滑屏的效果n driver.start_activity(app_package="com.xxzb.fenwoo",n app_activity=".activity.user.CreateGesturePwdActivity")n commit_btn = (MobileBy.ID, 'com.xxzb.fenwoo:id/right_btn')n password_gesture = (MobileBy.ID, 'com.xxzb.fenwoo:id/gesturepwd_create_lockview')n element_commit = base.find_element(commit_btn)n element_commit.click()n password_element = base.find_element(password_gesture)n base.gesture_password(password_element, 1, 2, 3, 6, 5, 4, 7, 8, 9)n time.sleep(5) # 看效果nnnif __name__ == '__main__':n pytest.main()

四、启动说明

1. 我代码中使用的是模拟器，如果你需要使用真机，那么需要修改部分代码，模拟器是带着端口号的，而真机没有端口号，具体怎么修改先自己研究，后面我再详细的介绍
2. desired_caps.yml文件中的配置需要根据自己的app配置修改
3. 代码中没有包含自动连接手机的部分代码，所以执行项目前需要先手动使用adb连接上手机(有条件的，可以自己把这部分代码写一下，然后再运行项目之前调用一下adb连接手机的方法即可)
4. 项目目录中的allure_report, allure_results目录是系统自动生成的，一个存放最终的测试报告，一个是存放报告的依赖文件，如果你接触过allure应该知道
5. log目录下存放了appium server启动之后运行的日志

五、 效果展示

上述只是初步实现了这样一个多手机并发的需求，并没有写的很详细，比如，让项目更加的规范还需要引入PO设计模式，这里没写这部分，其次base_page.py中还可以封装更多的方法，上述代码中也只封装了几个方法，如果真正的把这个并发引入到项目中肯定还需要完善的，但是需要添加的东西都是照葫芦画瓢了，有问题多思考！

最后我也整理了一些软件测试学习资料，对于学软件测试的小伙伴来说应该会很有帮助，为了更好地整理每个模块

需要的私信我关键字【555】免费获取哦 注意关键字是：555

全套软件测试自动化测试教学视频

300G教程资料下载【视频教程+PPT+项目源码】

全套软件测试自动化测试大厂面经