appium+pytest實(shí)現(xiàn)APP并發(fā)測(cè)試

"""------------------------------------@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   : [email protected]@GROUP: 878565760------------------------------------"""import timefrom appium.webdriver import WebElementfrom appium.webdriver.webdriver import WebDriverfrom appium.webdriver.common.touch_action import TouchActionfrom selenium.webdriver.support.wait import WebDriverWaitfrom selenium.common.exceptions import NoSuchElementException, TimeoutException  class Base(object):     def __init__(self, driver: WebDriver):         self.driver = driver     @property    def get_phone_size(self):         """獲取屏幕的大小"""         width = self.driver.get_window_size()['width']         height = self.driver.get_window_size()['height']          return width, height     def swipe_left(self, duration=300):        """左滑"""         width, height = self.get_phone_size         start = width * 0.9, height * 0.5         end = width * 0.1, height * 0.5         return self.driver.swipe(*start, *end, duration)     def swipe_right(self, duration=300):          """右滑"""        width, height = self.get_phone_size        start = width * 0.1, height * 0.5        end = width * 0.9, height * 0.5        return self.driver.swipe(*start, *end, duration)     def swipe_up(self, duration):         """上滑"""         width, height = self.get_phone_size         start = width * 0.5, height * 0.9         end = width * 0.5, height * 0.1        return self.driver.swipe(*start, *end, duration)     def swipe_down(self, duration):        """下滑"""        width, height = self.get_phone_size        start = width * 0.5, height * 0.1        end = width * 0.5, height * 0.9        return self.driver.swipe(*start, *end, duration)       def skip_welcome_page(self, direction, num=3):        """        滑動(dòng)頁(yè)面跳過(guò)引導(dǎo)動(dòng)畫(huà)        :param direction:  str 滑動(dòng)方向，left, right, up, down        :param num: 滑動(dòng)次數(shù)        :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])()  # 使用反射執(zhí)行不同的滑動(dòng)方法        else:             raise ValueError("參數(shù){}不存在, direction可以為{}任意一個(gè)字符串".                              format(direction, direction_dic.keys()))      @staticmethod    def 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_y      def 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 keys
    def gesture_password(self, element: WebElement, *pwd):        """手勢(shì)密碼: 直接輸入需要鏈接的點(diǎn)對(duì)應(yīng)的數(shù)字，最多9位        pwd: 1, 2, 3, 6, 9        """        if len(pwd) > 9:            raise ValueError("需要設(shè)置的密碼不能超過(guò)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,3            follow_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_point        full_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 element        except (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   : [email protected]@GROUP: 878565760------------------------------------"""import socketimport os

def check_port(host, port):    """檢測(cè)指定的端口是否被占用"""    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 創(chuàng)建socket對(duì)象    try:        s.connect((host, port))        s.shutdown(2)    except OSError:        print('port %s is available! ' % port)        return True    else:        print('port %s already be in use !' % port)        return False

def release_port(port):    """釋放指定的端口"""    cmd_find = 'netstat -aon | findstr {}'.format(port)  # 查找對(duì)應(yīng)端口的pid    print(cmd_find)
    # 返回命令執(zhí)行后的結(jié)果    result = os.popen(cmd_find).read()    print(result)
    if str(port) and 'LISTENING' in result:        # 獲取端口對(duì)應(yīng)的pid進(jìn)程        i = result.index('LISTENING')        start = i + len('LISTENING') + 7        end = result.index('\n')        pid = result[start:end]        cmd_kill = 'taskkill -f -pid %s' % pid  # 關(guān)閉被占用端口的pid        print(cmd_kill)        os.popen(cmd_kill)    else:        print('port %s is available !' % port)

if __name__ == '__main__':    host = '127.0.0.1'    port = 4723    if 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   : [email protected]@GROUP: 878565760------------------------------------"""import subprocessfrom config.root_config import LOG_DIR
"""獲取main.js的未知，使用main.js啟動(dòng)appium server"""

class MainJs(object):    """獲取啟動(dòng)appium服務(wù)的main.js命令"""
    def __init__(self, cmd: str = "where main.js"):        self.cmd = cmd
    def 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_result

if __name__ == '__main__':    main = MainJs("where main.js")????print(main.get_cmd_result())

automationName: uiautomator2platformVersion: 5.1.1platformName: AndroidappPackage: com.xxzb.fenwooappActivity: .activity.addition.WelcomeActivitynoReset: Trueip: "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   : [email protected]@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   : [email protected]@GROUP: 878565760------------------------------------"""import subprocessfrom time import ctimefrom appium import webdriverimport yaml
from common.check_port import check_port, release_portfrom common.get_main_js import MainJsfrom config.root_config import CONFIG_PATH, LOG_DIR

class 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_info        js_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 driver

if __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   : [email protected]@GROUP: 878565760------------------------------------"""from drivers.app_driver import BaseDriverimport pytestimport time
from common.check_port import release_port
base_driver = None

def 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_driver    base_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   : [email protected]@GROUP: 878565760------------------------------------"""import pytestimport osfrom multiprocessing import Pool

device_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   : [email protected]@GROUP: 878565760------------------------------------"""import pytestimport timefrom appium.webdriver.common.mobileby import MobileBy
from base.base_page import Base

class TestGesture(object):
    def test_gesture_password(self, common_driver):        """這個(gè)case我只是簡(jiǎn)單的做了一個(gè)繪制手勢(shì)密碼的過(guò)程"""        driver = common_driver        base = Base(driver)        base.skip_welcome_page('left', 3)  # 滑動(dòng)屏幕        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()