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

importredisclassRedisHandler:
def__init__(self, host, port=6379, db=0):
# 生成客户端连接,StrictRedis()默认使用连接池，不必再单独使用ConnectPoolself.client=redis.StrictRedis(host=host, port=port, db=db)
defset_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)
defincr(self, key):
"""        使用incr方法，处理并发问题        当key不存在时，会先初始为0，每次调用，则会+1        :param key:        :return:        """self.client.incr(key)
defget_key(self, name):
"""读取缓存"""result=self.client.get(name)
returnresult

fromapi_test.common.redis_handlerimportRedisHandlerfromapi_test.config.db_configimportDBConfigclassCaseCountName:
ERROR: str="error_count"FAILED: str="failed_count"PASSED: str="passed_count"SKIP: str="skip_count"TOTAL: str="total_count"classCaseCount:
"""    redis 缓存统计用例执行情况    """def__init__(self):
self.redis=RedisHandler(host=DBConfig.redis_config.get('host'))  # redis主机地址可以写死在这里，也可以从配置类中获取definit_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)
deffailed_count(self):
"""获取失败用例数"""returnint(self.redis.get_key(CaseCountName.FAILED))
defpassed_count(self):
"""获取通过用例总数"""returnint(self.redis.get_key(CaseCountName.PASSED))
defskip_count(self):
"""获取跳过用例数"""returnint(self.redis.get_key(CaseCountName.SKIP))
deferror_count(self):
"""报错用例数"""returnint(self.redis.get_key(CaseCountName.ERROR))
deftotal_count(self):
"""用例总数"""returnint(self.redis.get_key(CaseCountName.TOTAL))
defpass_rate(self):
"""计算用例成功率"""try:
rate=round((self.passed_count() +self.skip_count()) /self.total_count() *100, 2)
returnrateexceptZeroDivisionError:
raiseException("执行失败，未检测到用例执行数量")

importosimportjsonimportrequestsimportplatformdefget_env_from_jenkins(name):
"""从Jenkins中获取全局环境变量"""returnos.getenv(name) andos.getenv(name).strip()
ProjectName=get_env_from_jenkins("JOB_NAME")  # Jenkins构建项目名称BUILD_URL=get_env_from_jenkins("BUILD_URL")  # Jenkins构建项目URLBUILD_NUMBER=get_env_from_jenkins("BUILD_NUMBER")  # Jenkins构建编号classEnterpriseWechatNotification:
def__init__(self, hook: list):
# 企业微信群机器人的hook地址，一个机器人就一个，多个就定义多个，可以写死，也可以写在配置类中self.hook_url_list= [f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={i}"foriinhook]
# 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'}
defsend_msg(self, result=''):
"""发送企业微信消息通知"""globalpayloadlinux_content=f"""** 【{ProjectName}】**> 项目名称:{ProjectName}
> 构件编号:#{BUILD_NUMBER}
> 测试环境:{platform.system()}
> [报告链接]({self.allure_url})> [控制台链接]({BUILD_URL}){result}"""windows_content=f"""** 【auto_test_project】**> 测试环境:{platform.system()}
{result}"""ifplatform.system() =="Linux":
payload= {
"msgtype": "markdown",
"markdown": {
"content": linux_content                }
            }
elifplatform.system() =="Windows":
payload= {
"msgtype": "markdown",
"markdown": {
"content": windows_content                }
            }
forhook_urlinself.hook_url_list:
requests.post(url=hook_url, headers=self.header, data=json.dumps(payload))

importtimeimportpytestfromapi_test.config.configimportHookUrlConffromapi_test.common.send_enterprise_wechatimportEnterpriseWechatNotificationfromapi_test.common.redis_handlerimportRedisHandlerfromapi_test.config.db_configimportDBConfigfromapi_test.common.case_count_controlimportCaseCountName, CaseCountredis=RedisHandler(host=DBConfig.redis_config.get('host'))
case_count=CaseCount()
case_count.init_process()  # 初始化Redis中的用例统计缓存数据@pytest.hookimpl(hookwrapper=True, tryfirst=True)
defpytest_runtest_makereport(item, call):
"""获取测试结果、生成测试报告"""print('------------------------------------')
out=yieldreport=out.get_result()
ifreport.when=='call':
# print(f"测试报告：{report}")# print(f"步骤：{report.when}")print(f"用例id：{report.nodeid}")
print(f"用例描述：{str(item.function.__doc__)}")
print(f"运行结果：{report.outcome}")
"""将用例执行结果写入缓存"""ifreport.outcome=='passed':
redis.incr(CaseCountName.PASSED)
ifreport.outcome=='failed':
redis.incr(CaseCountName.FAILED)
ifreport.when=='setup':
ifreport.outcome=='skipped':
redis.incr(CaseCountName.SKIP)
CaseCount().total_run_count()
defpytest_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)  # 执行结果发送企业微信