超詳細(xì)干貨：Appium+Pytest實(shí)現(xiàn)App并發(fā)測試！
開課了：重磅消息 | 2021年最新全棧測試開發(fā)技能實(shí)戰(zhàn)指南（第2期）
1. 前言
Appium結(jié)合Pytest開展App自動化測試時，你知道如何實(shí)現(xiàn)用例并發(fā)執(zhí)行嗎？費(fèi)話不多說，直接上代碼， 畢竟想讓每個人都能看明白也不容易，所以先放代碼，有興趣的先自行研究。
2. 目錄結(jié)構(gòu)
3. 文件源碼
3.1 base/base_page.py
"""
------------------------------------
@File : base_page.py
------------------------------------
"""
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, 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ǎo)動畫
:param direction: str 滑動方向，left, right, up, down
:param num: 滑動次數(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í)行不同的滑動方法
else:
raise ValueError("參數(shù){}不存在, direction可以為{}任意一個字符串".
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):
"""手勢密碼: 直接輸入需要鏈接的點(diǎn)對應(yīng)的數(shù)字，最多9位
pwd: 1, 2, 3, 6, 9
"""
if len(pwd) > 9:
raise ValueError("需要設(shè)置的密碼不能超過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
3.2 common/check_port.py
"""
------------------------------------
@File : check_port.py
------------------------------------
"""
import socket
import os
def check_port(host, port):
"""檢測指定的端口是否被占用"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 創(chuàng)建socket對象
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) # 查找對應(yīng)端口的pid
print(cmd_find)
# 返回命令執(zhí)行后的結(jié)果
result = os.popen(cmd_find).read()
print(result)
if str(port) and 'LISTENING' in result:
# 獲取端口對應(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)
3.3 common/get_main_js.py
"""
------------------------------------
@File : get_main_js.py
@IDE : PyCharm
------------------------------------
"""
import subprocess
from config.root_config import LOG_DIR
"""
獲取main.js的未知，使用main.js啟動appium server
"""
class MainJs(object):
"""獲取啟動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())
3.4 config/desired_caps.yml
automationName: uiautomator2
platformVersion: 5.1.1
platformName: Android
appPackage: com.xxzb.fenwoo
appActivity: .activity.addition.WelcomeActivity
noReset: True
ip: "127.0.0.1"
3.5 config/root_config.py
"""
------------------------------------
@File : root_config.py
------------------------------------
"""
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")
3.6 drivers/app_driver.py
"""
------------------------------------
@File : app_driver.py
------------------------------------
"""
import subprocess
from time import ctime
from appium import webdriver
import yaml
from common.check_port import check_port, release_port
from common.get_main_js import MainJs
from 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
3.7 conftest.py
"""
------------------------------------
@File : conftest.py
------------------------------------
"""
from drivers.app_driver import BaseDriver
import pytest
import 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"])
3.8 run_case.py
"""
------------------------------------
@File : run_case.py
------------------------------------
"""
import pytest
import os
from 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()
3.9 cases/test_concurrent.py
"""
------------------------------------
@File : test_concurrent.py
------------------------------------
"""
import pytest
import time
from appium.webdriver.common.mobileby import MobileBy
from base.base_page import Base
class TestGesture(object):
def test_gesture_password(self, common_driver):
"""這個case我只是簡單的做了一個繪制手勢密碼的過程"""
driver = common_driver
base = 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()
4. 啟動說明
我代碼中使用的是模擬器，如果你需要使用真機(jī)，那么需要修改部分代碼，模擬器是帶著端口號的，而真機(jī)沒有端口號，具體怎么修改先自己研究，后面我再詳細(xì)的介紹
desired_caps.yml文件中的配置需要根據(jù)自己的app配置修改
代碼中沒有包含自動連接手機(jī)的部分代碼，所以執(zhí)行項(xiàng)目前需要先手動使用adb連接上手機(jī)(有條件的，可以自己把這部分代碼寫一下，然后再運(yùn)行項(xiàng)目之前調(diào)用一下adb連接手機(jī)的方法即可)
項(xiàng)目目錄中的allure_report, allure_results目錄是系統(tǒng)自動生成的，一個存放最終的測試報告，一個是存放報告的依賴文件，如果你接觸過allure應(yīng)該知道
log目錄下存放了appium server啟動之后運(yùn)行的日志
5. 效果展示
6. 最后
上述只是初步實(shí)現(xiàn)了這樣一個多手機(jī)并發(fā)的需求，并沒有寫的很詳細(xì)，比如，讓項(xiàng)目更加的規(guī)范還需要引入PO設(shè)計(jì)模式，這里沒寫這部分，其次base_page.py中還可以封裝更多的方法，上述代碼中也只封裝了幾個方法，如果真正的把這個并發(fā)引入到項(xiàng)目中肯定還需要完善的，但是需要添加的東西都是照葫蘆畫瓢了，有問題多思考！
原文鏈接：https://www.cnblogs.com/linuxchao/p/linuxchao-pytest-mult.html
重磅消息: 由狂師老師授課主講的「全棧測試開發(fā)技能訓(xùn)練營」已經(jīng)正式開課了，課程內(nèi)容非常值得推薦！課程大綱：重磅消息 | 2021年最新全棧測試開發(fā)技能實(shí)戰(zhàn)指南（第2期）
END
長按二維碼/微信掃碼 添加作者
閱讀原文
評論
圖片
表情