279 lines
8.8 KiB
Python
279 lines
8.8 KiB
Python
import json
|
||
import os
|
||
import time
|
||
|
||
import allure
|
||
import pytest
|
||
from filelock import FileLock
|
||
|
||
from selenium import webdriver
|
||
from selenium.webdriver import Remote
|
||
from selenium.webdriver.chrome.options import Options as CH_Options
|
||
from selenium.webdriver.firefox.options import Options as FF_Options
|
||
from common.config import RunConfig, PathConfig
|
||
from page.login_page import LoginPage
|
||
from tools.assert_tool.assert_control import assert_compare
|
||
from tools.read_file_rool.cache_control import CacheControl
|
||
from poium.common import logging
|
||
from tools.read_file_rool.yaml_control import YamlControl
|
||
|
||
root_path = PathConfig.root_path
|
||
cache_path = PathConfig.cache_path
|
||
cookies_path = os.path.join(cache_path, "cookies")
|
||
base_url = RunConfig.url
|
||
case_result = None
|
||
username = YamlControl().get_yaml_data("$.account.username")
|
||
password = YamlControl().get_yaml_data("$.account.password")
|
||
|
||
|
||
# 项目首页URL
|
||
@pytest.fixture(scope='function')
|
||
def url():
|
||
return base_url
|
||
|
||
|
||
def clear_cache():
|
||
"""
|
||
清空缓存文件夹
|
||
"""
|
||
cache_list = os.listdir(cache_path)
|
||
for cache in cache_list:
|
||
path = os.path.join(cache_path, cache)
|
||
os.remove(path)
|
||
|
||
|
||
def login(browser, url):
|
||
"""
|
||
登录函数
|
||
"""
|
||
try:
|
||
page = LoginPage(browser)
|
||
page.open(url)
|
||
page.login_a.click()
|
||
page.username_input = username
|
||
page.password_input = password
|
||
page.login_span.click()
|
||
assert_compare(page.username_span.text, '==', username)
|
||
cookies = page.get_cookies()
|
||
browser.quit()
|
||
except Exception as e:
|
||
pytest.exit("登录失败")
|
||
raise e
|
||
return cookies
|
||
|
||
|
||
@pytest.fixture(scope="session", autouse=True)
|
||
def session_fixture(tmp_path_factory, worker_id):
|
||
# 如果是单机运行 则运行这里的代码块【不可删除、修改】
|
||
if worker_id == "master":
|
||
"""
|
||
【自定义代码块】
|
||
这里就写你要本身应该要做的操作，比如：登录请求、新增数据、清空数据库历史数据等等
|
||
"""
|
||
cookies = login(get_driver(), base_url)
|
||
CacheControl("cookies").cache = str(cookies)
|
||
|
||
# 如果是分布式运行
|
||
else:
|
||
# 获取所有子节点共享的临时目录，无需修改【不可删除、修改】
|
||
root_tmp_dir = tmp_path_factory.getbasetemp().parent
|
||
# 【不可删除、修改】
|
||
fn = root_tmp_dir / "data.json"
|
||
# 【不可删除、修改】
|
||
with FileLock(str(fn) + ".lock"):
|
||
# 【不可删除、修改】
|
||
if fn.is_file():
|
||
# 缓存文件中读取数据，像登录操作的话就是 token 【不可删除、修改】
|
||
cookies = json.loads(fn.read_text())
|
||
else:
|
||
"""
|
||
【自定义代码块】
|
||
跟上面 if 的代码块一样就行
|
||
"""
|
||
clear_cache()
|
||
cookies = login(get_driver(), base_url)
|
||
# 【不可删除、修改】
|
||
fn.write_text(json.dumps(cookies))
|
||
|
||
# 最好将后续需要保留的数据存在缓存文件
|
||
CacheControl("cookies").cache = str(cookies)
|
||
|
||
|
||
@pytest.fixture(scope='function', autouse=True)
|
||
def set_up_function(browser):
|
||
# 用例执行前获取cookies缓存
|
||
cookies = CacheControl("cookies").cache
|
||
browser.get(base_url)
|
||
for i in eval(cookies):
|
||
browser.add_cookie(i)
|
||
browser.refresh()
|
||
time.sleep(2)
|
||
|
||
|
||
# 控制浏览器启动和关闭
|
||
@pytest.fixture(scope='function')
|
||
def browser():
|
||
global d
|
||
d = get_driver()
|
||
RunConfig.driver = d
|
||
yield d
|
||
d.quit()
|
||
|
||
|
||
def get_driver():
|
||
"""
|
||
全局定义浏览器驱动
|
||
:return:
|
||
"""
|
||
if RunConfig.driver_type == "chrome":
|
||
# 本地chrome浏览器
|
||
driver = webdriver.Chrome()
|
||
driver.maximize_window()
|
||
|
||
elif RunConfig.driver_type == "firefox":
|
||
# 本地firefox浏览器
|
||
driver = webdriver.Firefox()
|
||
driver.maximize_window()
|
||
|
||
elif RunConfig.driver_type == "chrome-headless":
|
||
# chrome headless模式
|
||
chrome_options = CH_Options()
|
||
chrome_options.add_argument("--headless")
|
||
chrome_options.add_argument('--disable-gpu')
|
||
chrome_options.add_argument("--window-size=1920x1080")
|
||
driver = webdriver.Chrome(options=chrome_options)
|
||
|
||
elif RunConfig.driver_type == "firefox-headless":
|
||
# firefox headless模式
|
||
firefox_options = FF_Options()
|
||
firefox_options.headless = True
|
||
driver = webdriver.Firefox(firefox_options=firefox_options)
|
||
|
||
elif RunConfig.driver_type == "grid":
|
||
# 通过远程节点运行
|
||
driver = Remote(command_executor='http://localhost:4444/wd/hub',
|
||
desired_capabilities={
|
||
"browserName": "chrome",
|
||
})
|
||
driver.set_window_size(1920, 1080)
|
||
|
||
else:
|
||
raise NameError("driver驱动类型定义错误！")
|
||
|
||
return driver
|
||
|
||
|
||
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
|
||
def pytest_runtest_makereport(item, call):
|
||
"""
|
||
hook pytest失败
|
||
:param item:
|
||
:param call:
|
||
:return:
|
||
"""
|
||
global case_result
|
||
global item_name
|
||
# 执行所有其他钩子以获取报告对象
|
||
outcome = yield
|
||
rep = outcome.get_result()
|
||
if rep.when == "call":
|
||
case_result = rep.outcome
|
||
item_name = item.name
|
||
# 只关注实际失败的测试调用, 不关心setup/teardown
|
||
if rep.when == "call" and rep.failed:
|
||
mode = "a" if os.path.exists("../../failures") else "w"
|
||
with open(root_path+"failures", mode) as f:
|
||
if "tmpdir" in item.fixturenames:
|
||
extra = " (%s)" % item.funcargs["tmpdir"]
|
||
else:
|
||
extra = ""
|
||
f.write(rep.nodeid + extra + "\n")
|
||
with allure.step('添加失败截图...'):
|
||
allure.attach(d.get_screenshot_as_png(), "失败截图", allure.attachment_type.PNG)
|
||
|
||
|
||
@pytest.fixture(scope="function", autouse=True)
|
||
def print_request(request):
|
||
split_line = "="*40
|
||
logging.info(split_line+request.function.__name__+split_line)
|
||
logging.info("用例路径："+str(request.node.nodeid))
|
||
logging.info("使用夹具："+str(request.fixturenames))
|
||
logging.info(split_line+'steps'+split_line)
|
||
yield
|
||
logging.info(split_line+'case_result'+split_line+"\n")
|
||
|
||
|
||
def pytest_terminal_summary(terminalreporter):
|
||
"""
|
||
收集测试结果
|
||
"""
|
||
_PASSED = len([i for i in terminalreporter.stats.get('passed', []) if i.when != 'teardown'])
|
||
_ERROR = len([i for i in terminalreporter.stats.get('error', []) if i.when != 'teardown'])
|
||
_FAILED = len([i for i in terminalreporter.stats.get('failed', []) if i.when != 'teardown'])
|
||
_SKIPPED = len([i for i in terminalreporter.stats.get('skipped', []) if i.when != 'teardown'])
|
||
_TOTAL = terminalreporter._numcollected
|
||
_TIMES = time.time() - terminalreporter._sessionstarttime
|
||
logging.info(""+"="*50+"自动化测试结束！"+"="*50)
|
||
logging.info(f"成功用例数: {_PASSED}")
|
||
logging.error(f"异常用例数: {_ERROR}")
|
||
logging.error(f"失败用例数: {_FAILED}")
|
||
logging.warning(f"跳过用例数: {_SKIPPED}")
|
||
logging.info("用例执行时长: %.2f" % _TIMES + " s")
|
||
|
||
# try:
|
||
# _RATE = round(_PASSED / _TOTAL * 100, 2)
|
||
# logging.info("用例成功率: %.2f" % _RATE + " %")
|
||
# except ZeroDivisionError:
|
||
# logging.info("用例成功率: 0.00 %")
|
||
|
||
|
||
# @pytest.fixture(autouse=True)
|
||
# def reset():
|
||
# """
|
||
# 每条用例执行后，根据运行结果处理
|
||
# """
|
||
# print() # 控制台打印与用例路径换行
|
||
# yield
|
||
# if case_result != 'passed':
|
||
# driver.get(base_url)
|
||
# driver.refresh()
|
||
|
||
|
||
# # 通过fixture调用上一个用例的运行结果
|
||
# @pytest.fixture(scope='function')
|
||
# def last_result():
|
||
# return case_result
|
||
|
||
|
||
# 判断运行结果并处理
|
||
# @pytest.fixture(scope='function')
|
||
# def login_teardown():
|
||
# CacheControl("cookies").cache = ""
|
||
# yield
|
||
# # 判断登录用例是否通过
|
||
# if case_result == 'passed':
|
||
# cookies_ = driver.get_cookies()
|
||
# CacheControl("cookies").cache = str(cookies_)
|
||
# else:
|
||
# pytest.exit("登录用例失败！强制停止测试！")
|
||
|
||
|
||
# @pytest.fixture(scope='session')
|
||
# def debug_browser():
|
||
# """
|
||
# 定义指定chrome账号浏览器驱动
|
||
# """
|
||
# global debug_driver
|
||
# if RunConfig.driver_type == 'chrome':
|
||
# options = webdriver.ChromeOptions()
|
||
# options.add_argument(r'--user-common-dir=C:\Users\Admin\Desktop\AutoTest\maap_v1\chromeData')
|
||
# options.add_argument("--profile-directory=Profile 22")
|
||
# debug_driver = webdriver.Chrome(options=options)
|
||
# else:
|
||
# raise NameError("调试模式需使用Chrome浏览器")
|
||
# # 所有用例执行接受后关闭浏览器
|
||
# yield debug_driver
|
||
# debug_driver.quit()
|
||
# print("\ntest end!")
|
||
# return debug_driver
|