利用pytest hook函数实现自动化测试结果推送企业微信

import redis


class RedisHandler:
def __init__(self, host, port=6379, db=0):
self.client = redis.StrictRedis(host=host, port=port, db=db) # 生成客户端连接,StrictRedis()默认使用连接池，不必再单独使用ConnectPool

def set_string(self, name: str, value, ex=None, px=None, nx=False, xx=False) -> None:
"""
缓存中写入str（单个）
:param name: 缓存名称
:param value: 缓存值
:param ex: 过期时间（秒）
:param px: 过期时间（毫秒）
:param nx: 如果设置为True，则只有name不存在时，当前set操作才执行（新增）
:param xx: 如果设置为True，则只有name不存在时，当前set操作才执行（修改）
:return:
"""
self.client.set(name, value=value, ex=ex, px=px, nx=nx, xx=xx)

def incr(self, key):
"""
使用incr方法，处理并发问题
当key不存在时，会先初始为0，每次调用，则会+1
:param key:
:return:
"""
self.client.incr(key)

def get_key(self, name):
"""读取缓存"""
result = self.client.get(name)
return result

from api_test.common.redis_handler import RedisHandler
from api_test.config.db_config import DBConfig


class CaseCountName:
ERROR: str = "error_count"
FAILED: str = "failed_count"
PASSED: str = "passed_count"
SKIP: str = "skip_count"
TOTAL: str = "total_count"


class CaseCount:
"""
redis 缓存统计用例执行情况
"""

def __init__(self):
self.redis = RedisHandler(host=DBConfig.redis_config.get('host')) # redis主机地址可以写死在这里，也可以从配置类中获取

def init_process(self):
"""
初始化进度、总数、成功数、失败数
"""
self.redis.set_string(CaseCountName.TOTAL, 0)
self.redis.set_string(CaseCountName.SKIP, 0)
self.redis.set_string(CaseCountName.PASSED, 0)
self.redis.set_string(CaseCountName.FAILED, 0)
self.redis.set_string(CaseCountName.ERROR, 0)

def failed_count(self):
"""失败用例数"""
return int(self.redis.get_key(CaseCountName.FAILED))

def passed_count(self):
"""通过用例总数"""
return int(self.redis.get_key(CaseCountName.PASSED))

def skip_count(self):
"""跳过用例数"""
return int(self.redis.get_key(CaseCountName.SKIP))

def error_count(self):
"""报错用例数"""
return int(self.redis.get_key(CaseCountName.ERROR))

def total_count(self):
"""用例总数"""
return int(self.redis.get_key(CaseCountName.TOTAL))

def pass_rate(self):
"""用例成功率"""
try:
rate = round((self.passed_count() + self.skip_count()) / self.total_count() * 100, 2)
return rate
except ZeroDivisionError:
raise Exception("执行失败，未检测到用例执行数量")

import os
import json
import requests
import platform


def get_env_from_jenkins(name, base=''):
"""从Jenkins中获取全局环境变量"""
return os.getenv(name) and os.getenv(name).strip() or base


ProjectName = get_env_from_jenkins("JOB_NAME") # Jenkins构建项目名称
BUILD_URL = get_env_from_jenkins("BUILD_URL") # Jenkins构建项目URL
BUILD_NUMBER = get_env_from_jenkins("BUILD_NUMBER") # Jenkins构建编号


class EnterpriseWechatNotification:
def __init__(self, hook: list):
# 企业微信群机器人的hook地址，一个机器人就一个，多个就定义多个，可以写死，也可以写在配置类中
self.hook_url_list = [f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={i}" for i in hook]
# allure生成报告的地址，Jenkins执行时会用到，Windows暂未配置allure地址
self.allure_url = f"http://192.168.1.122:8088/jenkins/job/{ProjectName}/{BUILD_NUMBER}/allure/"
self.header = {'Content-Type': 'application/json'}

def send_msg(self, result=''):
"""发送企业微信消息通知"""
global payload
linux_content = f"""** 【{ProjectName}】**
> 项目名称:{ProjectName}
> 构件编号:#{BUILD_NUMBER}
> 测试环境:{platform.system()}
> [报告链接]({self.allure_url})
> [控制台链接]({BUILD_URL})
{result}"""

windows_content = f"""** 【auto_test_project】**
> 测试环境:{platform.system()}
{result}"""
if platform.system() == "Linux":
payload = {
"msgtype": "markdown",
"markdown": {
"content": linux_content
}
}
elif platform.system() == "Windows":
payload = {
"msgtype": "markdown",
"markdown": {
"content": windows_content
}
}
for hook_url in self.hook_url_list:
requests.post(url=hook_url, headers=self.header, data=json.dumps(payload))

import time
import pytest
from api_test.config.config import HookUrlConf
from api_test.common.send_enterprise_wechat import EnterpriseWechatNotification
from api_test.common.redis_handler import RedisHandler
from api_test.config.db_config import DBConfig
from api_test.common.case_count_control import CaseCountName, CaseCount

redis = RedisHandler(host=DBConfig.redis_config.get('host'))
case_count = CaseCount()
case_count.init_process() # 初始化Redis中的用例统计缓存数据


@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
"""获取测试结果、生成测试报告"""
print('------------------------------------')
out = yield
report = out.get_result()
if report.when == 'call':
# print(f"测试报告：{report}")
# print(f"步骤：{report.when}")
print(f"用例id：{report.nodeid}")
print(f"用例描述：{str(item.function.__doc__)}")
print(f"运行结果：{report.outcome}")
"""将用例执行结果写入缓存"""
if report.outcome == 'passed':
redis.incr(CaseCountName.PASSED)
if report.outcome == 'failed':
redis.incr(CaseCountName.FAILED)

if report.when == 'setup':
if report.outcome == 'skipped':
redis.incr(CaseCountName.SKIP)

CaseCount().total_run_count()


def pytest_terminal_summary(terminalreporter, exitstatus, config):
"""收集测试结果，从Redis缓存数据中获取"""
total_case = case_count.total_count()
pass_case = case_count.passed_count()
fail_case = case_count.failed_count()
skip_case = case_count.skip_count()
error_case = case_count.error_count()
pass_rate = case_count.pass_rate()
run_time = round((time.time() - terminalreporter._sessionstarttime), 2)
print("******用例执行结果统计******")
print(f"总用例数：{total_case}条")
print(f"通过：{pass_case}条")
print(f"失败：{fail_case}条")
print(f"跳过：{skip_case}条")
print(f"报错：{error_case}条")
print(f"用例通过率：{pass_rate}%")
print(f"用时：{run_time}s")
desc = """
本次执行情况如下：
总用例数为：{}
通过用例数：<font color=\"info\">{}条</font>
失败用例数：<font color=\"warning\">{}条</font>
错误用例数：{}
跳过用例数：{}
通过率为：{} %
用时：{}s
""".format(total_case, pass_case, fail_case, error_case, skip_case, pass_rate, run_time)
EnterpriseWechatNotification(hook=HookUrlConf.HOOK_URL.value).send_msg(desc) # 执行结果发送企业微信