Pytest+Allure+飞书机器人

时间:2024-10-27 09:04:25

分享一个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.py

Pytest框架我们都很熟悉了之前也分享过,所以本文不再详细讲解,只是讲一下如何获取测试结果并且发送到飞书,生成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)

最终结果:

7c6ff2d6786bcb47c8dad76d9b431623.png

05f81509107cc4e72d747c79fe066b3e.png

最后,本文也要感谢大佬分享的文章,我也是参考,然后稍微完善简化了一下!
参考文章:https://blog.****.net/qq_22357323/article/details/140024783