Jenkins教程-10-发送飞书测试报告通知

上一小节我们学习了发送企业微信测试报告通知的方法，本小节我们讲解一下发送飞书测试报告通知的方法。

1、自动化用例执行完后，使用pytest_terminal_summary钩子函数收集测试结果，存入本地status.txt文件中，供Jenkins调用

conftest.py代码如下：

#conftest.py def pytest_terminal_summary(terminalreporter, exitstatus, config):"""收集测试报告summary,并存入status.txt文件中，供Jenkins调用"""print("pytest_terminal_summary")passed_num = len([i for i in terminalreporter.stats.get('passed', []) if i.when != 'teardown'])failed_num = len([i for i in terminalreporter.stats.get('failed', []) if i.when != 'teardown'])error_num = len([i for i in terminalreporter.stats.get('error', []) if i.when != 'teardown'])skipped_num = len([i for i in terminalreporter.stats.get('skipped', []) if i.when != 'teardown'])total_num = passed_num + failed_num + error_num + skipped_numtest_result = '测试通过' if total_num == passed_num + skipped_num else '测试失败'duration = round((time.time() - terminalreporter._sessionstarttime), 2)# 定义目录路径directory_path = './reports/'# 确保文件所在的目录存在os.makedirs(os.path.dirname(directory_path), exist_ok=True)# 定义文件路径file_path = os.path.join(directory_path, 'status.txt')with open(file_path, 'w', encoding='utf-8') as f:f.write(f'TEST_TOTAL={total_num}\n')f.write(f'TEST_PASSED={passed_num}\n')f.write(f'TEST_FAILED={failed_num}\n')f.write(f'TEST_ERROR={error_num}\n')f.write(f'TEST_SKIPPED={skipped_num}\n')f.write(f'TEST_DURATION={duration}\n')f.write(f'TEST_RESULT={test_result}\n')

本地文件status.txt中收集测试结果示例：

2、Jenkins中安装Environment Injector 和description setter 插件

Environment Injector插件用于注入环境变量

自动化测试任务配置中，添加构建步骤

填写测试结果收集文件status.txt的路径

description setter用于构建后设置任务描述

将status.txt中的的测试结果字段映射到任务描述中

执行任务构建后，任务描述中会显示构建的测试结果，如下

3、安装python-jenkins 库，读取自动化测试任务构建后的测试结果描述信息

pip install python-jenkins

代码如下

# qywechat_remind.py
import json
from datetime import datetime
import jenkins
import requests
import jmespathhost = "http://localhost:8080/"
username = 'admin'
password = 'xxxxxxxxxx'
webhook = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=f4444444444-c1d3-47f2-be78-098f80c2d194"
env = "test"
stage = "回归测试"
job = "auto_api_test"
maintainer = "米兔1号"
server = jenkins.Jenkins(host, username=username, password=password)
last_build_number = server.get_job_info(job)['lastCompletedBuild']['number']
build_info = server.get_build_info(job, last_build_number)
console_url = build_info['url'] + "console"
report_url = build_info['url'] + 'allure'
# report_url = ip_host + report_url.split(":")[-1]
test_status = json.loads(build_info['description'])
print("构建测试结果描述信息：", test_status)total = test_status["total"]
passed = test_status["passed"]
passed_ratio = round(passed / total, 4) * 100
failed = test_status["failed"]
failed_ratio = round((100 - passed_ratio), 2)
error = test_status["error"]
skipped = test_status["skipped"]
duration = test_status["duration"]
build_time = datetime.fromtimestamp(build_info['timestamp'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
success = total == (passed + skipped) if passed != 0 else False

执行上述代码，可以看出，已经获取到jenkins任务的测试结果信息了

4、上一步获取到的测试结果信息，包装成消息体，调用飞书机器人发送群消息接口，自动发送消息到群里

飞书机器人的配置和接口，请参考：开发文档 - 飞书开放平台

飞书发送测试结果消息的整体代码如下：

import json
from datetime import datetime
import jenkins
import requests
import jmespathhost = "http://localhost:8080/"
username = 'admin'
password = 'xxxxx'
webhook = "https://open.feishu.cn/open-apis/bot/v2/hook/c82222qqw64de8a-cac3-4523-b234-85269cf4945d"
env = "test"
stage = "回归测试"
job = "auto_api_test"
maintainer = "米兔1号"
server = jenkins.Jenkins(host, username=username, password=password)
last_build_number = server.get_job_info(job)['lastCompletedBuild']['number']
build_info = server.get_build_info(job, last_build_number)
print("构建信息：", build_info)
console_url = build_info['url'] + "console"
print("console:", console_url)
report_url = build_info['url'] + 'allure'
print("report_url:", report_url)
test_status = json.loads(build_info['description'])
print("测试结果：", test_status)
total = test_status["total"]
passed = test_status["passed"]
passed_ratio = round(passed / total, 4) * 100
print("passed_ratio", passed_ratio)
failed = test_status["failed"]
failed_ratio = round((100 - passed_ratio), 2)
print("failed:", failed_ratio)
error = test_status["error"]
skipped = test_status["skipped"]
duration = test_status["duration"]
build_time = datetime.fromtimestamp(build_info['timestamp'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
success = total == (passed + skipped) if passed != 0 else False
# 使用Jenkins API token 模拟登录
USERNAME = "admin"
# Jenkins API token
TOKEN = "113b81ae7fd66046859f1b9833d391621a"
url_suites = f"{report_url}/data/suites.json"
# print("url_suites", url_suites)
res = requests.get(url_suites, auth=(USERNAME, TOKEN))
# print("res", res.content)
s_url = f"{report_url}/#suites/"
# print('s_url', s_url)
url_raw_list = jmespath.search("children[].children[].children[].children[?status=='failed'||status=='broken'].{name:name,parentUid:parentUid,uid:uid,status:status,tags:tags}",res.json())
# print("url_raw_list", url_raw_list)url_list = []
for raw in url_raw_list[0]:url_dict = {"name": raw["name"], "url": s_url + raw["parentUid"] + "/" + raw["uid"] + "/", "uid": raw["uid"],"status": raw["status"], "author": raw["tags"][0]}url_list.append(url_dict)
# print("url_list", url_list)url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"payload = json.dumps({"app_id": "cli_asss6baddbc63b4500c","app_secret": "c39BsssskuRpZqbXzgcyab7fqgRVkTfT7skL"
})
headers = {'Content-Type': 'application/json'
}response = requests.request("POST", url, headers=headers, data=payload).json()
# red = "#FF0000"
# green = "#00ff00"
a = 'green'
b = 'red'
c = 'yellow'
card_demo = {"msg_type": "interactive","card": {"elements": [{"tag": "div","text": {"content": f"-**任务名称**：{job}\n\n-**测试阶段**：{stage}\n\n-**测试结果**：<font color={a if success else b}>{'通过~' if success else '失败!'}</font> {chr(0x1f600) if success else chr(0x1f627)}\n\n-**用例总数**：{total}\n\n-**通过数**：<font color={a}>{passed}</font>\n\n-**通过率**：{passed_ratio}%\n\n-**失败数**：<font color={b}>{failed}</font>\n\n-**失败率**：{failed_ratio}%\n\n-**错误数**：{error}\n\n-**跳过数**：{skipped}\n\n-**执行人**：@{maintainer}\n\n-**执行时间**：{build_time}\n\n-**执行耗时**：{duration}s\n\n","tag": "lark_md"}}, {"actions": [{"tag": "button","text": {"content": "查看测试报告","tag": "lark_md"},"url": report_url,"type": "primary","value": {"key": "value"}}],"tag": "action"}],"header": {"template": "wathet","title": {"content": "钉钉oapi接口测试任务执行报告通知","tag": "plain_text"}}}
}# payload = json.dumps({
#     "msg_type": "post",
#     "content": {
#         "post": {
#             "zh_cn": {
#                 "title": "钉钉oapi接口测试报告",
#                 "content": [
#                     [
#                         {
#                             "tag": "text",
#                             "text": f"【用例总数】:{total} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试通过】:{passed} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试失败】:{failed} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试错误】:{error} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试跳过】:{skipped} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试耗时】:{duration}s \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试时间】:{build_time} \n"
#                         },
#                         {
#                             "tag": "text",
#                             "text": f"【测试结果】: {'通过~' if success else '失败!'}{chr(0x1f600) if success else chr(0x1f627)} \n"
#                         },
#                         {
#                             "tag": "a",
#                             "text": "Allure详细报告，请查看",
#                             "href": f"{report_url}"
#                         }
#                     ]
#                 ]
#             }
#         }
#     }
# })
payload = json.dumps(card_demo)headers = {'Authorization': f"Bearer {response['tenant_access_token']}",'Content-Type': 'application/json'
}requests.request("POST", webhook, headers=headers, data=payload)# 单个报告，详细数据
# http://localhost:8080/job/auto_api_test/76/allure/data/test-cases/9a4eba68509440c8.jsonphone_mapping = {"zhang.san": "13421503860","li.si": "14564591649"
}
single_url = f"{report_url}/data/test-cases/"
for case in url_list:url = single_url + str(case["uid"]) + ".json"res = requests.get(url, auth=(USERNAME, TOKEN)).json()case["message"] = res["statusMessage"]
print("url_list", url_list)
author_list = list(set(jmespath.search("[*].author", url_list)))
# print("author_list",author_list)
failed_list = jmespath.search("[?status=='failed']", url_list)
print("failed_list", failed_list)
broken_list = jmespath.search("[?status=='broken']", url_list)
print("broken_list", broken_list)phone_list = []failed_string = f"<font color={b}>【**失败用例**】:\n</font>"
broken_string = f"<font color={c}>【**错误用例**】:\n</font>"for url_info in url_list:name_text = " " + url_info["name"] + "\n"url_text = "[" + " " + url_info["message"] + "]" + "(" + url_info["url"] + ")" + "\n"single_case_text = name_text + url_text# at_dict = {#     "tag": "at",#     "user_id": 1111,# }if url_info["status"] == "failed":failed_string += single_case_text# failed_string_list.append(url_dict)# failed_string_list.append(at_dict)elif url_info["status"] == "broken":failed_string += single_case_text# broken_string_list.append(name_dict)# broken_string_list.append(url_dict)# broken_string_list.append(at_dict)
null_string = " " + f"无\n"
if not failed_list:failed_string += null_string
if not broken_list:broken_string += null_string
# print("failed_string_list", failed_string_list, type(failed_string_list))
# print("broken_string_list", broken_string_list, type(broken_string_list))
# end_string_list = [{
#     "tag": "a",
#     "text": "Allure详细报告，请查看",
#     "href": f"{report_url}"
# }]
all_string = failed_string + broken_string
# print("all_string", all_string_list, type(all_string_list))data_ca_demo = {"msg_type": "interactive","card": {"elements": [{"tag": "div","text": {"content": all_string,"tag": "lark_md"}}, {"actions": [{"tag": "button","text": {"content": "查看测试报告","tag": "lark_md"},"url": report_url,"type": "primary","value": {"key": "value"}}],"tag": "action"}],"header": {"template": "wathet","title": {"content": "钉钉oapi接口测试任务执行错误日志通知","tag": "plain_text"}}}
}
# data_ca = {
#     "msg_type": "post",
#     "content": {
#         "post": {
#             "zh_cn": {
#                 "title": "钉钉oapi接口测试报错信息汇总",
#                 "content": all_string_list
#             }
#         }
#     }
# }
print("data_ca", data_ca_demo)
for author in author_list:if author in list(phone_mapping.keys()):phone_list.append(phone_mapping[author])
print(phone_list)response_r = requests.request("POST", webhook, headers=headers, data=json.dumps(data_ca_demo))
print("response_r", response_r.json())

4、执行上述脚本，查看飞书通知，如下

测试结果信息：

错误日志信息：

最后感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走，希望可以帮助到大家！