当前位置：首页 > news >正文

深入解析：自动化接口框架搭建分享-pytest

news 2025/9/26 18:30:31

深入解析：自动化接口框架搭建分享-pytest

我们要介绍的是一个基于 Python 和 Pytest 的 API 自动化测试框架。

框架设计(不分先后)

环境与依赖 (requirements.txt)
配置管理 (config/setting.py)
HTTP 请求封装 (utils/requests_helper.py)
数据加载器 (utils/data_loader.py 和 data/)
测试框架基础 (pytest.ini, conftest.py)
API 响应解析与断言 (utils/api_parser.py, utils/api_assertion.py)
编写第一个测试用例 (modules/ 或 testset/)
日志记录 (utils/logger_helper.py) : 用于记录测试执行过程中的信息。
数据库操作 (utils/mysql_helper.py) : 如果 API 测试需要与数据库进行交互（例如，准备测试数据或验证数据变更）。

①环境配置

Python 3.9.13 版本

一些插件

或者我们可以将一些所需要的库或者插件封装放在一个requirement.txt当中，确定好需要的python与版本之后，运行 pip install -r requirement.txt，确保所有的第三方库已安装，

②配置管理setting.py

定义不同环境的配置，还有一些认证token、数据库连接，方便在不同环境下测试

import os
# 项目根目录
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 日志目录
LOG_DIR = os.path.join(BASE_DIR, 'logs')
# 报告目录
REPORT_DIR = os.path.join(BASE_DIR, 'reports')
# 测试数据目录
DATA_DIR = os.path.join(BASE_DIR, 'data')
# 默认请求头
DEFAULT_HEADERS = {
'Content-Type': 'application/json'
}
# 你的swagger文档地址
Swagger_URL = "https://company.com"
# 基础URL (根据环境动态加载) -> 获取认证
DOMAIN = "https://pre.company.com"
Test_URL = DOMAIN  # 测试环境的地址
Auth_URL = "/getToken" # 拿到token的接口
User_Id = 99 # getToken的接口的入参
# Uat环境
Uat_Domain = "https://uat.company.com/"
Uat_URL = Uat_Domain
# Prod
Prod_Domain = "https://prod.company.com/"
Prod_URL = Prod_Domain + "/api"
# 飞书机器人Webhook+Jenkins配置
Jenkins_Url = "http://ip:port"
Job_Name = "demo"
Jenkins_User = "userName"
Jenkins_Pwd = "passWord"
webhook = "信息发送地址" # 群
# SIT_MySQL数据库连接
DB_Host= ""
DB_User= ""
DB_Pwd= ""
DB_Port = 80
DB_Schema = ""

③HTTP 请求封装 (utils/requests_helper.py)

#/usr/bin/python3
# coding=utf-8
import json
import sys
sys.path.append("..")
sys.path.append(".")
sys.dont_write_bytecode = True
import requests
from utils.logger_helpper import logger
class RequestUtil:
def __init__(self, base_url, default_headers=None, timeout=None):
self.base_url = base_url
self.default_headers = default_headers or {'Content-Type': 'application/json'}
self.timeout = timeout or 60
self.session = requests.Session()
def send_request(self, method, endpoint, params=None, data=None, json=None, headers=None, **kwargs):
"""
发送HTTP请求
:param method: 请求方法 (get/post/put/delete)
:param url: 请求URL
:param params: 查询参数
:param data: 表单数据
:param json: JSON数据
:param headers: 请求头
:param kwargs: 其他requests参数
:return: 响应对象
"""
method = method.lower()
url = f"{self.base_url}{endpoint}"
req_headers = headers if headers else self.default_headers
logger.info(f"请求URL:{url}")
logger.info(f"请求Method: {method.upper()}")
logger.info(f"请求Header: {req_headers}")
logger.info(f"请求Params: {params}")
logger.info(f"请求Data: {data}")
try:
if method == 'get':
response = self.session.get(url, params=params, data=data, json=json, headers=req_headers, **kwargs)
elif method == 'post':
response = self.session.post(url, params=params, data=data, json=json, headers=req_headers, **kwargs)
elif method == 'put':
response = self.session.put(url,  params=params, data=data, json=json, headers=req_headers, **kwargs)
elif method == 'delete':
response = self.session.delete(url, params=params, data=data, json=json, headers=req_headers, **kwargs)
else:
raise ValueError(f"不支持的请求方法: {method}")
logger.debug(f"响应内容: {response.json()}")
return response
except Exception as e:
logger.error(f"请求发生异常: {str(e)}")
raise

简单来说，“封装 HTTP 请求”就是把发送 HTTP 请求的复杂细节隐藏起来，提供一个更简单、更高级的接口供其他代码调用。

想象一下，每次你要发送一个 HTTP 请求时，你都需要：

导入 requests 库。
构建完整的 URL。
设置请求方法（GET, POST, PUT, DELETE 等）。
添加请求头（ Content-Type , Authorization 等）。
处理请求参数（ params , data , json ）。
处理可能的异常（网络错误、超时等）。
解析响应。

而这个 requests_helper.py 文件就是为了避免你在每个测试用例中重复这些步骤。它创建了一个 RequestUtil 类，把这些通用的逻辑都集中在这个类里面，你只需要调用这个类的方法，就能轻松地发送 HTTP 请求。

文件中定义了一个名为 RequestUtil 的类，它负责处理所有的 HTTP 请求。

json : 用于处理 JSON 数据。
sys : 用于修改 Python 路径，确保可以导入项目中的其他模块。
requests : 这是 Python 中用于发送 HTTP 请求的第三方库
logger : 从 utils.logger_helper 导入，用于记录请求和响应的日志信息。

RequestUtil 类：

- __init__(self, base_url, default_headers=None, timeout=None) 方法：
- 这是类的构造函数，在创建 RequestUtil 对象时会被调用。
- base_url : API 的基础 URL，例如 https://pre.xiulie-sh.com 。这样在发送请求时，你只需要提供接口的路径（ endpoint ），它会自动拼接成完整的 URL。
- default_headers : 默认的请求头，例如 {'Content-Type': 'application/json'} 。这样每次请求都会自动带上这些头，除非你特别指定。
- timeout : 请求的超时时间，防止请求长时间无响应。
- self.session = requests.Session() : 这是 requests 库的一个重要特性。使用 Session 对象可以让你在多次请求之间保持某些参数（如 cookies、headers）的持久性，并且可以提高性能（例如，重用底层 TCP 连接）。这对于接口测试非常有用，特别是当需要处理登录状态或连续请求时。
- send_request(self, method, endpoint, params=None, data=None, json=None, headers=None, **kwargs) 方法：

- 发送实际的 HTTP 请求。
- method : HTTP 请求方法，如 'get' , 'post' , 'put' , 'delete' 。
- endpoint : 接口的路径，例如 /getToken 。它会与 base_url 拼接成完整的请求 URL。
- params : URL 查询参数，通常用于 GET 请求。
- data : 请求体数据，通常用于 POST/PUT 请求，发送表单数据。
- json : 请求体数据，通常用于 POST/PUT 请求，发送 JSON 格式数据。
- headers : 当前请求的请求头，如果提供了，会覆盖 default_headers 。
- **kwargs : 允许你传递任何其他 requests 库支持的参数，增加了灵活性。
- 内部逻辑：
- 将 method 转换为小写。
- 拼接完整的 url 。
- 合并请求头。
- 使用 logger.info 记录请求的详细信息，这对于调试和问题排查非常重要。
- 根据 method 调用 self.session 对象的相应方法（ get , post , put , delete ）来发送请求。
- 如果遇到不支持的请求方法，会抛出 ValueError 。
- 使用 logger.debug 记录响应内容（这里是 JSON 格式）。
- 返回 requests 库的 response 对象。
- 使用 try...except 块捕获请求过程中可能发生的异常，并使用 logger.error 记录错误信息。

总的来说

这个 requests_helper.py 文件通过 RequestUtil 类，将 requests 库的使用进行了高级封装，做到了以下这些：

- 统一的请求入口：所有 HTTP 请求都通过 send_request 方法发送。
- 配置的集中管理： base_url 和 default_headers 在初始化时设置，避免重复。
- 会话管理：使用 requests.Session 保持会话状态，方便处理认证和连续请求。
- 日志记录：自动记录请求和响应的详细信息，便于调试。
- 错误处理：统一捕获和记录请求异常。
通过这种封装，其他模块在编写测试用例时，只需要关注业务逻辑和接口参数，而不需要关心底层的 HTTP 请求细节

④logger_helper封装Python的日志

这个文件主要封装了 Python 的日志记录功能，目的是提供一个统一、方便的日志管理机制，使得在整个测试框架中可以轻松地记录各种信息，包括调试信息、普通信息、警告和错误等。

import sys
sys.path.append("..")
sys.path.append(".")
sys.dont_write_bytecode = True
import logging
import os
from datetime import datetime
from config.setting import LOG_DIR
class Logger:
def __init__(self):
# 确保日志目录存在
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR)
# 配置日志
self.logger = logging.getLogger('api_test')
self.logger.setLevel(logging.DEBUG)
# 日志文件名
log_file = os.path.join(LOG_DIR, f"test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")
# 文件处理器
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 日志格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# 添加处理器
self.logger.addHandler(file_handler)
self.logger.addHandler(console_handler)
def get_logger(self):
return self.logger
# 全局日志实例
logger = Logger().get_logger()