标签：war 目录 占用 页面 包含 src tom lob star
这个功能已经写完很长时间了，一直没有发出来，今天先把代码发出来吧，有一些代码是参考网上写的，具体的代码说明今天暂时先不发了，代码解释的太详细还得我花点时间^_^, 毕竟想让每个人都能看明白也不容易，所以先放代码，有兴趣的先研究吧，等我有时间再做代码说明(will doing)
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:19 4 @Auth : linux超 5 @File : base_page.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import time 13 from appium.webdriver import WebElement 14 from appium.webdriver.webdriver import WebDriver 15 from appium.webdriver.common.touch_action import TouchAction 16 from selenium.webdriver.support.wait import WebDriverWait 17 from selenium.common.exceptions import NoSuchElementException, TimeoutException 18 19 20 class Base(object): 21 22 def __init__(self, driver: WebDriver): 23 self.driver = driver 24 25 @property 26 def get_phone_size(self): 27 """获取屏幕的大小""" 28 width = self.driver.get_window_size()[‘width‘] 29 height = self.driver.get_window_size()[‘height‘] 30 return width, height 31 32 def swipe_left(self, duration=300): 33 """左滑""" 34 width, height = self.get_phone_size 35 start = width * 0.9, height * 0.5 36 end = width * 0.1, height * 0.5 37 return self.driver.swipe(*start, *end, duration) 38 39 def swipe_right(self, duration=300): 40 """右滑""" 41 width, height = self.get_phone_size 42 start = width * 0.1, height * 0.5 43 end = width * 0.9, height * 0.5 44 return self.driver.swipe(*start, *end, duration) 45 46 def swipe_up(self, duration): 47 """上滑""" 48 width, height = self.get_phone_size 49 start = width * 0.5, height * 0.9 50 end = width * 0.5, height * 0.1 51 return self.driver.swipe(*start, *end, duration) 52 53 def swipe_down(self, duration): 54 """下滑""" 55 width, height = self.get_phone_size 56 start = width * 0.5, height * 0.1 57 end = width * 0.5, height * 0.9 58 return self.driver.swipe(*start, *end, duration) 59 60 def skip_welcome_page(self, direction, num=3): 61 """ 62 滑动页面跳过引导动画 63 :param direction: str 滑动方向，left, right, up, down 64 :param num: 滑动次数 65 :return: 66 """ 67 direction_dic = { 68 "left": "swipe_left", 69 "right": "swipe_right", 70 "up": "swipe_up", 71 "down": "swipe_down" 72 } 73 time.sleep(3) 74 if hasattr(self, direction_dic[direction]): 75 for _ in range(num): 76 getattr(self, direction_dic[direction])() # 使用反射执行不同的滑动方法 77 else: 78 raise ValueError("参数{}不存在, direction可以为{}任意一个字符串". 79 format(direction, direction_dic.keys())) 80 81 @staticmethod 82 def get_element_size_location(element): 83 width = element.rect["width"] 84 height = element.rect["height"] 85 start_x = element.rect["x"] 86 start_y = element.rect["y"] 87 return width, height, start_x, start_y 88 89 def get_password_location(self, element: WebElement) -> dict: 90 width, height, start_x, start_y = self.get_element_size_location(element) 91 point_1 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 1)} 92 point_2 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 1)} 93 point_3 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 1)} 94 point_4 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 3)} 95 point_5 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 3)} 96 point_6 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 3)} 97 point_7 = {"x": int(start_x + width * (1 / 6) * 1), "y": int(start_y + height * (1 / 6) * 5)} 98 point_8 = {"x": int(start_x + width * (1 / 6) * 3), "y": int(start_y + height * (1 / 6) * 5)} 99 point_9 = {"x": int(start_x + width * (1 / 6) * 5), "y": int(start_y + height * (1 / 6) * 5)} 100 keys = { 101 1: point_1, 102 2: point_2, 103 3: point_3, 104 4: point_4, 105 5: point_5, 106 6: point_6, 107 7: point_7, 108 8: point_8, 109 9: point_9 110 } 111 return keys 112 113 def gesture_password(self, element: WebElement, *pwd): 114 """手势密码: 直接输入需要链接的点对应的数字，最多9位 115 pwd: 1, 2, 3, 6, 9 116 """ 117 if len(pwd) > 9: 118 raise ValueError("需要设置的密码不能超过9位!") 119 keys_dict = self.get_password_location(element) 120 start_point = "TouchAction(self.driver).press(x={0}, y={1}).wait(200)". 121 format(keys_dict[pwd[0]]["x"], keys_dict[pwd[0]]["y"]) 122 for index in range(len(pwd) - 1): # 0,1,2,3 123 follow_point = ".move_to(x={0}, y={1}).wait(200)". 124 format(keys_dict[pwd[index + 1]]["x"], 125 keys_dict[pwd[index + 1]]["y"]) 126 start_point = start_point + follow_point 127 full_point = start_point + ".release().perform()" 128 return eval(full_point) 129 130 def find_element(self, locator: tuple, timeout=30) -> WebElement: 131 wait = WebDriverWait(self.driver, timeout) 132 try: 133 element = wait.until(lambda driver: driver.find_element(*locator)) 134 return element 135 except (NoSuchElementException, TimeoutException): 136 print(‘no found element {} by {}‘, format(locator[1], locator[0])) 137 138 139 if __name__ == ‘__main__‘: 140 pass
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:17 4 @Auth : linux超 5 @File : check_port.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import socket 13 import os 14 15 16 def check_port(host, port): 17 """检测指定的端口是否被占用""" 18 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建socket对象 19 try: 20 s.connect((host, port)) 21 s.shutdown(2) 22 except OSError: 23 print(‘port %s is available! ‘ % port) 24 return True 25 else: 26 print(‘port %s already be in use !‘ % port) 27 return False 28 29 30 def release_port(port): 31 """释放指定的端口""" 32 cmd_find = ‘netstat -aon | findstr {}‘.format(port) # 查找对应端口的pid 33 print(cmd_find) 34 35 # 返回命令执行后的结果 36 result = os.popen(cmd_find).read() 37 print(result) 38 39 if str(port) and ‘LISTENING‘ in result: 40 # 获取端口对应的pid进程 41 i = result.index(‘LISTENING‘) 42 start = i + len(‘LISTENING‘) + 7 43 end = result.index(‘\n‘) 44 pid = result[start:end] 45 cmd_kill = ‘taskkill -f -pid %s‘ % pid # 关闭被占用端口的pid 46 print(cmd_kill) 47 os.popen(cmd_kill) 48 else: 49 print(‘port %s is available !‘ % port) 50 51 52 if __name__ == ‘__main__‘: 53 host = ‘127.0.0.1‘ 54 port = 4723 55 if not check_port(host, port): 56 print("端口被占用") 57 release_port(port)
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 13:47 4 @Auth : linux超 5 @File : get_main_js.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import subprocess 13 from config.root_config import LOG_DIR 14 15 """ 16 获取main.js的未知，使用main.js启动appium server 17 """ 18 19 20 class MainJs(object): 21 """获取启动appium服务的main.js命令""" 22 23 def __init__(self, cmd: str = "where main.js"): 24 self.cmd = cmd 25 26 def get_cmd_result(self): 27 p = subprocess.Popen(self.cmd, 28 stdin=subprocess.PIPE, 29 stdout=subprocess.PIPE, 30 stderr=subprocess.PIPE, 31 shell=True) 32 with open(LOG_DIR + "/" + "cmd.txt", "w", encoding="utf-8") as f: 33 f.write(p.stdout.read().decode("gbk")) 34 with open(LOG_DIR + "/" + "cmd.txt", "r", encoding="utf-8") as f: 35 cmd_result = f.read().strip("\n") 36 return cmd_result 37 38 39 if __name__ == ‘__main__‘: 40 main = MainJs("where main.js") 41 print(main.get_cmd_result())
1 automationName: uiautomator2 2 platformVersion: 5.1.1 3 platformName: Android 4 appPackage: com.xxzb.fenwoo 5 appActivity: .activity.addition.WelcomeActivity 6 noReset: True 7 ip: "127.0.0.1"
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:29 4 @Auth : linux超 5 @File : root_config.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import os 13 14 """ 15 project dir and path 16 """ 17 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 18 LOG_DIR = os.path.join(ROOT_DIR, "log") 19 CONFIG_DIR = os.path.join(ROOT_DIR, "config") 20 CONFIG_PATH = os.path.join(CONFIG_DIR, "desired_caps.yml")
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:23 4 @Auth : linux超 5 @File : app_driver.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import subprocess 13 from time import ctime 14 from appium import webdriver 15 import yaml 16 17 from common.check_port import check_port, release_port 18 from common.get_main_js import MainJs 19 from config.root_config import CONFIG_PATH, LOG_DIR 20 21 22 class BaseDriver(object): 23 """获取driver""" 24 def __init__(self, device_info): 25 main = MainJs("where main.js") 26 with open(CONFIG_PATH, ‘r‘) as f: 27 self.data = yaml.load(f, Loader=yaml.FullLoader) 28 self.device_info = device_info 29 js_path = main.get_cmd_result() 30 cmd = r"node {0} -a {1} -p {2} -bp {3} -U {4}:{5}".format( 31 js_path, 32 self.data["ip"], 33 self.device_info["server_port"], 34 str(int(self.device_info["server_port"]) + 1), 35 self.data["ip"], 36 self.device_info["device_port"] 37 ) 38 print(‘%s at %s‘ % (cmd, ctime())) 39 if not check_port(self.data["ip"], int(self.device_info["server_port"])): 40 release_port(self.device_info["server_port"]) 41 subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + "/" + device_info["server_port"] + ‘.log‘, ‘a‘), 42 stderr=subprocess.STDOUT) 43 44 def get_base_driver(self): 45 desired_caps = { 46 ‘platformName‘: self.data[‘platformName‘], 47 ‘platformVerion‘: self.data[‘platformVersion‘], 48 ‘udid‘: self.data["ip"] + ":" + self.device_info["device_port"], 49 "deviceName": self.data["ip"] + ":" + self.device_info["device_port"], 50 ‘noReset‘: self.data[‘noReset‘], 51 ‘appPackage‘: self.data[‘appPackage‘], 52 ‘appActivity‘: self.data[‘appActivity‘], 53 "unicodeKeyboard": True 54 } 55 print(‘appium port:%s start run %s at %s‘ % ( 56 self.device_info["server_port"], 57 self.data["ip"] + ":" + self.device_info["device_port"], 58 ctime() 59 )) 60 driver = webdriver.Remote( 61 ‘http://‘ + self.data[‘ip‘] + ‘:‘ + self.device_info["server_port"] + ‘/wd/hub‘, 62 desired_caps 63 ) 64 return driver 65 66 67 if __name__ == ‘__main__‘: 68 pass
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:16 4 @Auth : linux超 5 @File : conftest.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 from drivers.app_driver import BaseDriver 13 import pytest 14 import time 15 16 from common.check_port import release_port 17 18 base_driver = None 19 20 21 def pytest_addoption(parser): 22 parser.addoption("--cmdopt", action="store", default="device_info", help=None) 23 24 25 @pytest.fixture(scope="session") 26 def cmd_opt(request): 27 return request.config.getoption("--cmdopt") 28 29 30 @pytest.fixture(scope="session") 31 def common_driver(cmd_opt): 32 cmd_opt = eval(cmd_opt) 33 print("cmd_opt", cmd_opt) 34 global base_driver 35 base_driver = BaseDriver(cmd_opt) 36 time.sleep(1) 37 driver = base_driver.get_base_driver() 38 yield driver 39 # driver.close_app() 40 driver.quit() 41 release_port(cmd_opt["server_port"])
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:17 4 @Auth : linux超 5 @File : run_case.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import pytest 13 import os 14 from multiprocessing import Pool 15 16 17 device_infos = [ 18 { 19 "platform_version": "5.1.1", 20 "server_port": "4723", 21 "device_port": "62001", 22 }, 23 { 24 "platform_version": "5.1.1", 25 "server_port": "4725", 26 "device_port": "62025", 27 } 28 ] 29 30 31 def main(device_info): 32 pytest.main(["--cmdopt={}".format(device_info), 33 "--alluredir", "./allure-results", "-vs"]) 34 os.system("allure generate allure-results -o allure-report --clean") 35 36 37 if __name__ == "__main__": 38 with Pool(2) as pool: 39 pool.map(main, device_infos) 40 pool.close() 41 pool.join()
1 """ 2 ------------------------------------ 3 @Time : 2019/9/22 12:17 4 @Auth : linux超 5 @File : test_concurrent.py 6 @IDE : PyCharm 7 @Motto: Real warriors,dare to face the bleak warning,dare to face the incisive error! 8 @QQ : 28174043@qq.com 9 @GROUP: 878565760 10 ------------------------------------ 11 """ 12 import pytest 13 import time 14 from appium.webdriver.common.mobileby import MobileBy 15 16 from base.base_page import Base 17 18 19 class TestGesture(object): 20 21 def test_gesture_password(self, common_driver): 22 """这个case我只是简单的做了一个绘制手势密码的过程""" 23 driver = common_driver 24 base = Base(driver) 25 base.skip_welcome_page(‘left‘, 3) # 滑动屏幕 26 time.sleep(3) # 为了看滑屏的效果 27 driver.start_activity(app_package="com.xxzb.fenwoo", 28 app_activity=".activity.user.CreateGesturePwdActivity") 29 commit_btn = (MobileBy.ID, ‘com.xxzb.fenwoo:id/right_btn‘) 30 password_gesture = (MobileBy.ID, ‘com.xxzb.fenwoo:id/gesturepwd_create_lockview‘) 31 element_commit = base.find_element(commit_btn) 32 element_commit.click() 33 password_element = base.find_element(password_gesture) 34 base.gesture_password(password_element, 1, 2, 3, 6, 5, 4, 7, 8, 9) 35 time.sleep(5) # 看效果 36 37 38 if __name__ == ‘__main__‘: 39 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中还可以封装更多的方法，我也只封装了几个方法，如果真正的把这个并发引入到项目中肯定还需要完善的，但是需要添加的东西都是照葫芦画瓢了，有问题多思考！yes i can！
明天就是2020年了，大家加油！
标签：war 目录 占用 页面 包含 src tom lob star
原文地址：https://www.cnblogs.com/linuxchao/p/linuxchao-pytest-mult.html
