Pytest+Allure+飞书机器人
分享一个Python的接口测试自动化框架
首先来看目录结构
my_project/
├──config.py
├── test_cases/
│ ├── conftest.py # test_cases 目录下的 conftest.py
│ └── test_example.py
└── test_data/
│ └──data_read.py
└── reports/ #存放测试结果
│ └──status.txt
└──allure-results/
└── allure-report/
└── public_fun/
│ └──feishu_robot.py
│ └──public_fun.pyPytest框架我们都很熟悉了之前也分享过，所以本文不再详细讲解，只是讲一下如何获取测试结果并且发送到飞书，生成allure测试报告，能够在飞书访问。
环境必备：
pytest、allure-pytest、allure命令行工具、requests
现在来定义conftest.py
import shutil
import subprocess
import pytest
import os
import time
from config import Config
from public_fun.public_log import CustomLogger
from public_fun.feishu_robot import send_report, results
from yyjhqypt_test.test_data.public_url import *
report_dir = Config.REPORT_DIR
output_dir = Config.OUTPUT_DIR
feishu_webhook = Config.FEISHU_WEBHOOK
logger = CustomLogger()
report_port = Config.REPORT_PORT
def clear_directory(directory):
"""清空指定目录函数"""
if os.path.exists(directory):
shutil.rmtree(directory)
os.makedirs(directory)
# @pytest.hookimpl(tryfirst=True)
# def pytest_sessionstart(session):
# """清空之前测试报告"""
# try:
# clear_directory(report_dir)
# logger.info(f"已清空目录: {report_dir}")
# except Exception as e:
# print(f"清空之前测试报告: {e}")
@pytest.fixture(scope='session')
def set_up():
"""清空之前测试报告"""
try:
clear_directory(report_dir)
logger.info(f"已清空目录: {report_dir}")
except Exception as e:
print(f"清空之前测试报告: {e}")
public_url = PublicUrl('beta')
url = public_url.yyjhqtpt_url
return url
def generate_allure_report():
"""生成 Allure 报告"""
try:
command = f'allure generate {report_dir} -o {output_dir} --clean'
subprocess.run(f'powershell -Command "{command}"', shell=True)
except FileNotFoundError as e:
print("Error: Allure 命令未找到。请确保 Allure 已安装并添加到系统 PATH。")
print(e)
except subprocess.CalledProcessError as e:
print("Error: Allure 生成报告时出错。")
print(e)
except Exception as e:
print("An unexpected error occurred:")
print(e)
def stop_existing_allure_servers():
try:
# 使用 PowerShell 获取 Allure Serve 进程
get_process_command = [
"powershell",
"-Command",
"Get-Process -Name allure -ErrorAction SilentlyContinue"
]
result = subprocess.run(get_process_command, capture_output=True, text=True)
if result.stdout:
# 解析进程 ID 并终止进程
processes = result.stdout.strip().split('\n')
for proc in processes:
proc = proc.strip()
if proc:
parts = proc.split()
if len(parts) >= 2 and parts[1].isdigit():
pid = parts[1] # 默认输出格式，进程名后为 PID
subprocess.run(["powershell", "-Command", f"Stop-Process -Id {pid} -Force"])
print(f"已停止 Allure 进程，PID: {pid}")
else:
print("未检测到正在运行的 Allure 服务器。")
except Exception as e:
print(f"停止 Allure 服务器时发生错误: {e}")
def start_allure_server():
"""启动 Allure 服务器"""
command = f'allure open {output_dir} -p {report_port}'
subprocess.Popen(f'powershell -Command "{command}"', shell=True)
def send_feishu_report():
send_report(webhook=feishu_webhook, results=results)
@pytest.hookimpl(tryfirst=True)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
"""收集测试报告summary,并存入status.txt文件中"""
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_num
test_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')
time.sleep(5)
"""在测试会话结束时生成报告并发送飞书通知"""
print("Report directory exists:", os.path.exists(report_dir))
print("Output directory exists:", os.path.exists(output_dir))
# 生成 Allure 报告
try:
generate_allure_report()
except subprocess.CalledProcessError as e:
print(f"生成 Allure 报告失败: {e}")
# 启动 Allure 服务器
try:
stop_existing_allure_servers()
start_allure_server()
except Exception as e:
print(f"启动 Allure 服务器失败: {e}")
# 发送飞书通知
try:
send_feishu_report()
except Exception as e:
print(f"发送飞书报告失败: {e}")
# 可选：在终端输出一些总结信息
terminalreporter.write_sep("=", "测试会话总结")
terminalreporter.write(f"退出状态码: {exitstatus}\n")这样，我们在执行 pytest --alluredir=.\allure-results的时候就能够自动生成测试报告，提取测试结果发送到飞书，并且打开allure的服务器，飞书可以通过连接访问allure测试报告
下面来看飞书如何定义消息体和发送结果通知。飞书消息体定义可以参考飞书官方文档开发文档 - 飞书开放平台
import json
import time
import datetime
import requests
import socket
import hashlib
import base64
import hmac
from config import Config
config = Config()
# 拼接签证字符串
def gen_sign(timestamp, secret):
# 拼接timestamp和secret
string_to_sign = '{}\n{}'.format(timestamp, secret)
hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
# 对结果进行base64处理
sign = base64.b64encode(hmac_code).decode('utf-8')
return sign
# 获取宿主机的ip地址
def get_host_ip():
try:
# 获取主机名
host_name = socket.gethostname()
# 使用gethostbyname获取IP地址
# 注意：这会返回第一个解析的IP地址，可能是环回地址
host_ip = socket.gethostbyname(host_name)
# 更准确的方法是使用getaddrinfo，它可以返回多个地址
# 下面的代码会过滤掉环回地址，并尝试找到第一个非环回IPv4地址
for addr in socket.getaddrinfo(host_name, None):
if addr[4][0] != '127.0.0.1': # 过滤掉环回地址
if ':' not in addr[4][0]: # 过滤掉IPv6地址
return addr[4][0]
# 如果没有找到非环回IPv4地址，则返回之前可能获得的环回地址
return host_ip
except socket.gaierror:
return "IP address could not be determined"
# 读取status.txt中的变量
def read_variables_from_txt(file_path):
variables = {}
try:
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
# 去除行尾的换行符，并分割键和值
key, value = line.strip().split('=')
# 对于数字和非数字（如字符串）类型，尝试进行类型转换
try:
# 尝试将值转换为整数
value = int(value)
except ValueError:
pass
# 存储到字典中
variables[key] = value
return variables
except FileNotFoundError:
print(f"文件 {file_path} 未找到。")
return {}
except Exception as e:
print(f"读取文件时发生错误: {e}")
return {}
# 使用函数
file_path = config.STATUS_FILE # 请替换为你的txt文件路径
results = read_variables_from_txt(file_path)
def send_report(webhook, results):
# 定义一些变量
pass_color = 'green'
failed_color = 'red'
wrong_color = 'yellow'
report_url = "http://" + get_host_ip() + f":{config.REPORT_PORT}/"
webhook = webhook
env = "beta"
stage = "回归测试"
job = "接口自动化测试"
maintainer = "**" #执行测试人员
failed_string = f"<font color={failed_color}>【**失败用例**】:\n</font>"
broken_string = f"<font color={wrong_color}>【**错误用例**】:\n</font>"
all_string = failed_string + broken_string
total = results['TEST_TOTAL']
passed = results['TEST_PASSED']
passed_ratio = round(passed / total, 4) * 100
print("passed_ratio", passed_ratio)
failed = results['TEST_FAILED']
failed_ratio = round((100 - passed_ratio), 2)
print("failed:", failed_ratio)
error = results['TEST_ERROR']
skipped = results['TEST_SKIPPED']
duration = results['TEST_DURATION']
current_time_stamp = int(time.time())
# 将时间戳转换为datetime对象
dt_object = datetime.datetime.fromtimestamp(current_time_stamp)
build_time = dt_object.strftime("%Y-%m-%d %H:%M:%S")
success = total == (passed + skipped) if passed != 0 else False
seret = 'fVxwtxCaYjoeLzRbwOGhjb'
signature = gen_sign(current_time_stamp, seret)
print(current_time_stamp)
print(signature)
# 定义消息体
card_demo = {
"msg_type": "interactive",
"timestamp": current_time_stamp,
"sign": signature,
"card": {
"elements": [{
"tag": "div",
"text": {
"content": f"-**任务名称**：{job}\n\n-**测试阶段**：{stage}\n\n-**测试结果**：<font color={pass_color if success else failed_color}>{'通过~' if success else '失败!'}</font> {chr(0x1f600) if success else chr(0x1f627)}\n\n-**用例总数**：{total}\n\n-**通过数**：<font color={pass_color}>{passed}</font>\n\n-**通过率**：{passed_ratio}%\n\n-**失败数**：<font color={failed_color}>{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": "飞书接口测试任务执行报告通知",
"tag": "plain_text"
}
}
}
}
headers = {
"Content-Type": "application/json"
}
payload = json.dumps(card_demo)
# 发送请求
r = requests.post(webhook, data=payload, headers=headers)
print(r.text)最终结果：
最后，本文也要感谢大佬分享的文章，我也是参考，然后稍微完善简化了一下！
参考文章：https://blog.csdn.net/qq_22357323/article/details/140024783