分享一个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_urlreturn urldef 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] # 默认输出格式，进程名后为 PIDsubprocess.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_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')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 Configconfig = Config()
# 拼接签证字符串
def gen_sign(timestamp, secret):# 拼接timestamp和secretstring_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_ipexcept 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] = valuereturn variablesexcept 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 = webhookenv = "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_stringtotal = results['TEST_TOTAL']passed = results['TEST_PASSED']passed_ratio = round(passed / total, 4) * 100print("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 Falseseret = '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