前言

这个功能已经写完很长时间了，一直没有发出来，今天先把代码发出来吧，有一些代码是参考网上写的，具体的代码说明今天暂时先不发了，代码解释的太详细还得我花点时间, 毕竟想让每个人都能看明白也不容易，所以先放代码，有兴趣的先研究吧。

目录结构

文件源码

"""
------------------------------------
@Time : 2019/9/22 12:19
@Auth : linux超
@File : base_page.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import time
from appium.webdriver import WebElement
from appium.webdriver.webdriver import WebDriver
from appium.webdriver.common.touch_action import TouchAction
from selenium.webdriver.support.wait import WebDriverWait
from selenium.common.exceptions import NoSuchElementException, TimeoutExceptionclass Base(object):def __init__(self, driver: WebDriver):self.driver = driver@propertydef get_phone_size(self):"""获取屏幕的大小"""width = self.driver.get_window_size()['width']height = self.driver.get_window_size()['height']return width, heightdef swipe_left(self, duration=300):"""左滑"""width, height = self.get_phone_sizestart = width * 0.9, height * 0.5end = width * 0.1, height * 0.5return self.driver.swipe(*start, *end, duration)def swipe_right(self, duration=300):"""右滑"""width, height = self.get_phone_sizestart = width * 0.1, height * 0.5end = width * 0.9, height * 0.5return self.driver.swipe(*start, *end, duration)def swipe_up(self, duration):"""上滑"""width, height = self.get_phone_sizestart = width * 0.5, height * 0.9end = width * 0.5, height * 0.1return self.driver.swipe(*start, *end, duration)def swipe_down(self, duration):"""下滑"""width, height = self.get_phone_sizestart = width * 0.5, height * 0.1end = width * 0.5, height * 0.9return self.driver.swipe(*start, *end, duration)def skip_welcome_page(self, direction, num=3):"""滑动页面跳过引导动画:param direction:  str 滑动方向，left, right, up, down:param num: 滑动次数:return:"""direction_dic = {"left": "swipe_left","right": "swipe_right","up": "swipe_up","down": "swipe_down"}time.sleep(3)if hasattr(self, direction_dic[direction]):for _ in range(num):getattr(self, direction_dic[direction])()  # 使用反射执行不同的滑动方法else:raise ValueError("参数{}不存在, direction可以为{}任意一个字符串".format(direction, direction_dic.keys()))@staticmethoddef get_element_size_location(element):width = element.rect["width"]height = element.rect["height"]start_x = element.rect["x"]start_y = element.rect["y"]return width, height, start_x, start_ydef get_password_location(self, element: WebElement) -> dict:width, height, start_x, start_y = self.get_element_size_location(element)point_1 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 1)}point_2 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 1)}point_3 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 1)}point_4 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 3)}point_5 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 3)}point_6 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 3)}point_7 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 5)}point_8 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 5)}point_9 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 5)}keys = {1: point_1,2: point_2,3: point_3,4: point_4,5: point_5,6: point_6,7: point_7,8: point_8,9: point_9}return keysdef gesture_password(self, element: WebElement, *pwd):"""手势密码: 直接输入需要链接的点对应的数字，最多9位pwd: 1, 2, 3, 6, 9"""if len(pwd) > 9:raise ValueError("需要设置的密码不能超过9位!")keys_dict = self.get_password_location(element)start_point = "TouchAction(self.driver).press(x={0}, y={1}).wait(200)". \format(keys_dict[pwd[0]]["x"], keys_dict[pwd[0]]["y"])for index in range(len(pwd) - 1):  # 0,1,2,3follow_point = ".move_to(x={0}, y={1}).wait(200)". \format(keys_dict[pwd[index + 1]]["x"],keys_dict[pwd[index + 1]]["y"])start_point = start_point + follow_pointfull_point = start_point + ".release().perform()"return eval(full_point)def find_element(self, locator: tuple, timeout=30) -> WebElement:wait = WebDriverWait(self.driver, timeout)try:element = wait.until(lambda driver: driver.find_element(*locator))return elementexcept (NoSuchElementException, TimeoutException):print('no found element {} by {}', format(locator[1], locator[0]))if __name__ == '__main__':pass

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:17
@Auth : linux超
@File : check_port.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import socket
import osdef check_port(host, port):"""检测指定的端口是否被占用"""s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 创建socket对象try:s.connect((host, port))s.shutdown(2)except OSError:print('port %s is available! ' % port)return Trueelse:print('port %s already be in use !' % port)return Falsedef release_port(port):"""释放指定的端口"""cmd_find = 'netstat -aon | findstr {}'.format(port)  # 查找对应端口的pidprint(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') + 7end = result.index('\n')pid = result[start:end]cmd_kill = 'taskkill -f -pid %s' % pid  # 关闭被占用端口的pidprint(cmd_kill)os.popen(cmd_kill)else:print('port %s is available !' % port)if __name__ == '__main__':host = '127.0.0.1'port = 4723if not check_port(host, port):print("端口被占用")release_port(port)

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 13:47
@Auth : linux超
@File : get_main_js.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import subprocess
from config.root_config import LOG_DIR"""
获取main.js的未知，使用main.js启动appium server
"""class MainJs(object):"""获取启动appium服务的main.js命令"""def __init__(self, cmd: str = "where main.js"):self.cmd = cmddef get_cmd_result(self):p = subprocess.Popen(self.cmd,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,shell=True)with open(LOG_DIR + "/" + "cmd.txt", "w", encoding="utf-8") as f:f.write(p.stdout.read().decode("gbk"))with open(LOG_DIR + "/" + "cmd.txt", "r", encoding="utf-8") as f:cmd_result = f.read().strip("\n")return cmd_resultif __name__ == '__main__':main = MainJs("where main.js")print(main.get_cmd_result())

（左右滑动查看完整代码）

automationName: uiautomator2
platformVersion: 5.1.1
platformName: Android
appPackage: com.xxzb.fenwoo
appActivity: .activity.addition.WelcomeActivity
noReset: True
ip: "127.0.0.1"

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:29
@Auth : linux超
@File : root_config.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import os"""
project dir and path
"""
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LOG_DIR = os.path.join(ROOT_DIR, "log")
CONFIG_DIR = os.path.join(ROOT_DIR, "config")
CONFIG_PATH = os.path.join(CONFIG_DIR, "desired_caps.yml")

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:23
@Auth : linux超
@File : app_driver.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import subprocess
from time import ctime
from appium import webdriver
import yamlfrom common.check_port import check_port, release_port
from common.get_main_js import MainJs
from config.root_config import CONFIG_PATH, LOG_DIRclass BaseDriver(object):"""获取driver"""def __init__(self, device_info):main = MainJs("where main.js")with open(CONFIG_PATH, 'r') as f:self.data = yaml.load(f, Loader=yaml.FullLoader)self.device_info = device_infojs_path = main.get_cmd_result()cmd = r"node {0} -a {1} -p {2} -bp {3} -U {4}:{5}".format(js_path,self.data["ip"],self.device_info["server_port"],str(int(self.device_info["server_port"]) + 1),self.data["ip"],self.device_info["device_port"])print('%s at %s' % (cmd, ctime()))if not check_port(self.data["ip"], int(self.device_info["server_port"])):release_port(self.device_info["server_port"])subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + "/" + device_info["server_port"] + '.log', 'a'),stderr=subprocess.STDOUT)def get_base_driver(self):desired_caps = {'platformName': self.data['platformName'],'platformVerion': self.data['platformVersion'],'udid': self.data["ip"] + ":" + self.device_info["device_port"],"deviceName": self.data["ip"] + ":" + self.device_info["device_port"],'noReset': self.data['noReset'],'appPackage': self.data['appPackage'],'appActivity': self.data['appActivity'],"unicodeKeyboard": True}print('appium port:%s start run %s at %s' % (self.device_info["server_port"],self.data["ip"] + ":" + self.device_info["device_port"],ctime()))driver = webdriver.Remote('http://' + self.data['ip'] + ':' + self.device_info["server_port"] + '/wd/hub',desired_caps)return driverif __name__ == '__main__':pass

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:16
@Auth : linux超
@File : conftest.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
from drivers.app_driver import BaseDriver
import pytest
import timefrom common.check_port import release_portbase_driver = Nonedef pytest_addoption(parser):parser.addoption("--cmdopt", action="store", default="device_info", help=None)@pytest.fixture(scope="session")
def cmd_opt(request):return request.config.getoption("--cmdopt")@pytest.fixture(scope="session")
def common_driver(cmd_opt):cmd_opt = eval(cmd_opt)print("cmd_opt", cmd_opt)global base_driverbase_driver = BaseDriver(cmd_opt)time.sleep(1)driver = base_driver.get_base_driver()yield driver# driver.close_app()driver.quit()release_port(cmd_opt["server_port"])

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:17
@Auth : linux超
@File : run_case.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import pytest
import os
from multiprocessing import Pooldevice_infos = [{"platform_version": "5.1.1","server_port": "4723","device_port": "62001",},{"platform_version": "5.1.1","server_port": "4725","device_port": "62025",}
]def main(device_info):pytest.main(["--cmdopt={}".format(device_info),"--alluredir", "./allure-results", "-vs"])os.system("allure generate allure-results -o allure-report --clean")if __name__ == "__main__":with Pool(2) as pool:pool.map(main, device_infos)pool.close()pool.join()

（左右滑动查看完整代码）

"""
------------------------------------
@Time : 2019/9/22 12:17
@Auth : linux超
@File : test_concurrent.py
@IDE  : PyCharm
@Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error!
@QQ   : 28174043@qq.com
@GROUP: 878565760
------------------------------------
"""
import pytest
import time
from appium.webdriver.common.mobileby import MobileByfrom base.base_page import Baseclass TestGesture(object):def test_gesture_password(self, common_driver):"""这个case我只是简单的做了一个绘制手势密码的过程"""driver = common_driverbase = Base(driver)base.skip_welcome_page('left', 3)  # 滑动屏幕time.sleep(3)  # 为了看滑屏的效果driver.start_activity(app_package="com.xxzb.fenwoo",app_activity=".activity.user.CreateGesturePwdActivity")commit_btn = (MobileBy.ID, 'com.xxzb.fenwoo:id/right_btn')password_gesture = (MobileBy.ID, 'com.xxzb.fenwoo:id/gesturepwd_create_lockview')element_commit = base.find_element(commit_btn)element_commit.click()password_element = base.find_element(password_gesture)base.gesture_password(password_element, 1, 2, 3, 6, 5, 4, 7, 8, 9)time.sleep(5)  # 看效果if __name__ == '__main__':pytest.main()

（左右滑动查看完整代码）

启动说明

我代码中使用的是模拟器，如果你需要使用真机，那么需要修改部分代码，模拟器是带着端口号的，而真机没有端口号，具体怎么修改先自己研究，后面我再详细的介绍。

desired_caps.yml文件中的配置需要根据自己的app配置修改。

代码中没有包含自动连接手机的部分代码，所以执行项目前需要先手动使用adb连接上手机（有条件的，可以自己把这部分代码写一下，然后再运行项目之前调用一下adb连接手机的方法即可）。

项目目录中的allure_report, allure_results目录是系统自动生成的，一个存放最终的测试报告，一个是存放报告的依赖文件，如果你接触过allure应该知道。

log目录下存放了appium server启动之后运行的日志。

效果展示

最后

我只是初步实现了这样一个多手机并发的需求，并没有写的很详细。

比如，让项目更加的规范还需要引入PO设计模式，我这里没写这部分，其次base_page.py中还可以封装更多的方法，我也只封装了几个方法。

如果真正的把这个并发引入到项目中肯定还需要完善的，但是需要添加的东西都是照葫芦画瓢了，有问题多思考！yes I can！

来源：

https://www.cnblogs.com/linuxchao/p/linuxchao-pytest-mult.html

end