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

Pytest+requests进行接口自动化测试5.0（5种assert断言的封装 + pymysql）

news 2025/9/24 7:38:17

5 种常用断言模式

  • 一、assert封装目标：
    • 目录结构
  • 二、5 种常见 assert 断言封装
    • 1. 字符串包含断言（ if str1 in str2）
    • 2. 结果相等断言（if dict1 == dict2）
    • 3. 结果不相等断言（if dict1 != dict2）
    • 4. 断言接口返回值里面的任意一个值（ if str1 in str2 ）
    • 5. 数据库断言
      • 1）PyMySQL
      • 2） 前置准备（整体架构）
      • 3）数据库断言代码实现
    • 终：assert_result：统一断言入口

一、assert封装目标：

  • 构建一个高可维护、易扩展、企业级的接口自动化测试框架，具备以下能力：
  • 实现 5 种常用断言模式
  • 支持从 YAML 配置驱动
  • 日志清晰、异常可追溯
  • Allure 报告正确显示失败/成功
  • 工业级健壮性（防 jsonpath 返回 False 等陷阱）

核心理念：让断言失败直接抛出 AssertionError，不手动管理 flag

目录结构

interface_automation/
├── common/                     # 公共模块
│   ├── assertion.py            # 断言封装（本文核心）
│   ├── recordlog.py            # 日志工具
│   └── connection.py           # 数据库操作
│   └── operationConfig.py      # config.yaml操作文件
├── conf/                       # 配置文件
│   ├── setting.py              # 日志配置
│   └── config.yaml             # 环境、数据库配置
├── testcase/                   # 测试用例文件
│   ├── login.yaml              # 登录测试用例

二、5 种常见 assert 断言封装

在 common/assertion.py 中实现以下 5 种断言模式

YAML 数据驱动文件示例：

- baseInfo:api_name: 根据id查找叶片数据url: /api/aqc/leaf/getByIdmethod: GETheader:Authorization: "{{get_data(token)}}"test_case:- case_name: 正确查询该id风机叶片的数据params:id: "{{get_data(id)}}"validation:- contains: {code: "200"}- contains: {message: "操作成功"}- contains: {success: true}extract_list:batchNo: $.data.batchNo

外部调用三件套：

from common.assertion import Assertions  # 导入工具validation = to.pop('validation')  # 获取 validationAssertions().assert_result(validation,response,status_code) # 调用该函数

1. 字符串包含断言（ if str1 in str2）

  • 功能说明：断言预期结果的字符串是否包含在接口的实际返回结果当中（模糊匹配），常用于：
  • 消息提示（举例：message 包含“成功”）
  • Token 是否以 Bearer 开头
  • 错误信息校验
  • 代码实现：

common/assertion.py 中 字符串包含断言封装代码：

import jsonpath
from common.recordlog import logs
import allureclass Assertions:# 1.字符串包含断言def contains_assert(self,value_dict,response,status_code):""":param value_dict:  预期结果: yaml文件当中validation关键字下的结果:param response: 接口返回的 JSON 数据:param status_code: HTTP 状态码"""# 断言状态标识：0代表成功，其他代表失败flag = 0for assert_key,assert_value in value_dict.items():if assert_key == 'code':if str(assert_value) != str(status_code):flag = flag + 1allure.attach(f"预期结果：{assert_value}\n,实际结果:{status_code},响应代码断言结果：失败",allure.attachment_type.TEXT)logs.error(f'validation断言失败：接口返回码{status_code}不等于{assert_value}')else:resp_list = jsonpath.jsonpath(response,f'$..{assert_key}')# 安全判断：jsonpath 找不到时返回 Falseif not resp_list:logs.error(f"字段 '{assert_key}' 在响应中未找到")raise AssertionError(f"断言失败：字段 '{assert_key}' 不存在于响应中")# 安全使用 resp_list[0]if isinstance(resp_list[0],str):resp_list = ''.join(resp_list)else:resp_list = str(resp_list[0])# 执行包含断言if str(assert_value) in str(resp_list):logs.info(f'字符串包含断言成功：预期结果：{assert_value}\n,实际结果:{resp_list}')else:flag = flag + 1allure.attach(f"预期结果：{assert_value}\n,实际结果:{resp_list},响应文本断言结果：失败",allure.attachment_type.TEXT)logs.error(f"响应文本断言结果：失败。预期结果：{assert_value}\n,实际结果:{resp_list}")return flag
  • 代码解析
关键点说明
jsonpath.jsonpath(…)用来从复杂的 JSON 数据中找某个字段（任意层级，比如找 message 的值）
“计数器”：flagflag = 0：全部通过；flag = 1, 2, 3…：有失败的
for循环获取到的yaml的数据validation

检查 非code 的其他字段， 如果找不到 resp_list （resp_list 是空的）报错，直接停止测试

            else:resp_list = jsonpath.jsonpath(response,f'$..{assert_key}')# 安全判断：jsonpath 找不到时返回 Falseif not resp_list:logs.error(f"字段 '{assert_key}' 在响应中未找到")raise AssertionError(f"断言失败：字段 '{assert_key}' 不存在于响应中")

如果找到了则将之全部处理为字符串，方便后续使用 in

                # 安全使用 resp_list[0]if isinstance(resp_list[0],str):resp_list = ''.join(resp_list)else:resp_list = str(resp_list[0])

关键代码！！包含 断言

                if str(assert_value) in str(resp_list):logs.info(f'字符串包含断言成功：预期结果：{assert_value}\n,实际结果:{resp_list}')else:flag = flag + 1allure.attach(f"预期结果：{assert_value}\n,实际结果:{resp_list},响应文本断言结果：失败",allure.attachment_type.TEXT)logs.error(f"响应文本断言结果：失败。预期结果：{assert_value}\n,实际结果:{resp_list}")

2. 结果相等断言（if dict1 == dict2）

  • 断言接口的实际返回结果是否与预期结果完全一致（精确匹配），常用于：
  • 状态码校验（举例：{“code”: 200} 必须严格等于实际返回的 code）
  • 登录成功后返回的标准信息比对（如：{“msg”: “登录成功”}）
  • 接口数据结构固定字段的精确验证（如分页信息 {“page”: 1, “size”: 10}）
  • 代码实现：

common/assertion.py 中 结果相等断言封装代码：

    def equal_assert(self,value,response):"""相等断言模式:param value:  预期结果，也就是yaml文件里面的validation关键字下的参数，必须为dict类型:param response: 接口的实际返回结果:return:  flag标识，0表示测试通过，非0表示测试未通过"""if not isinstance(value, dict) or not isinstance(response, dict):raise TypeError("预期结果和响应必须为字典")# 提取 response 中 value 对应的字段filtered = {k: response[k] for k in value if k in response}if filtered == value:logs.info(f"相等断言成功: {filtered}")return 0else:error_msg = f"相等断言失败: 期望={value}, 实际={filtered}"logs.error(error_msg)allure.attach(error_msg, "相等断言失败", allure.attachment_type.TEXT)raise AssertionError(error_msg)
  • 代码解析

检查两个参数是不是都是字典格式

if not isinstance(value, dict) or not isinstance(response, dict):raise TypeError("预期结果和响应必须为字典")

从实际返回的数据 response 中，只取出你在 value 里提到的那些字段

filtered = {k: response[k] for k in value if k in response}

如果过滤后的结果和期望的一样，就打印日志：“成功”，并返回 0 表示通过

if filtered == value:logs.info(f"相等断言成功: {filtered}")return 0

3. 结果不相等断言（if dict1 != dict2）

  • 断言接口的实际返回结果是否与预期结果不一致（反向匹配），常用于：
  • Token 刷新前后对比（新旧 token 不能相等）
  • 验证码更新后内容变化校验（刷新后的验证码应不同于之前）
  • 防重复提交场景（两次请求的唯一标识或时间戳不应相同）
  • 代码实现：

common/assertion.py 中 结果相等断言封装代码：

    def equal_assert(self,value,response):"""不相等断言模式:param value:  预期结果，也就是yaml文件里面的validation关键字下的参数，必须为dict类型:param response: 接口的实际返回结果:return:  flag标识，0表示测试通过，非0表示测试未通过"""if not isinstance(value, dict) or not isinstance(response, dict):raise TypeError("必须是字典类型")filtered = {k: response[k] for k in value if k in response}if filtered != value:logs.info(f"不相等断言成功: {filtered} ≠ {value}")return 0else:error_msg = f"不相等断言失败: {filtered} == {value} (不应相等)"logs.error(error_msg)allure.attach(error_msg, "不相等断言失败", allure.attachment_type.TEXT)raise AssertionError(error_msg)
  • 代码解析

检查两个参数是不是都是字典格式

if not isinstance(value, dict) or not isinstance(response, dict):raise TypeError("预期结果和响应必须为字典")

从实际返回的数据 response 中，只取出你在 value 里提到的那些字段

filtered = {k: response[k] for k in value if k in response}

如果“实际” ≠ “期望”，就打印日志：“成功”，并返回 0 表示通过

if filtered != value:logs.info(f"不相等断言成功: {filtered} ≠ {value}")return 0

4. 断言接口返回值里面的任意一个值（ if str1 in str2 ）

  • 在整个响应体中搜索是否包含某些关键词，适用于：
  • 全文搜索用户信息（如：响应中包含 “张三”）
  • 校验 Token 或用户名出现在任意位置
  • 错误信息中是否包含关键字（如 “参数错误”）
  • 代码实现：

common/assertion.py 中 结果相等断言封装代码：

def any_value_assert(self, expected_list, response):"""断言响应中任意位置包含预期值（全文模糊匹配）:param expected_list: ['admin', '超级管理员', 200]:param response: 响应 JSON"""# 提取所有值all_values = jsonpath.jsonpath(response, "$..*")if not all_values or not isinstance(all_values, list):raise AssertionError("响应为空或格式错误")for expected in expected_list:found = Falsefor val in all_values:if str(expected) in str(val):found = Truelogs.info(f"找到匹配值: '{expected}' in '{val}'")breakif not found:logs.error(f"未找到值: '{expected}'")raise AssertionError(f"值 '{expected}' 未在响应中找到")
  • 代码解析（和 字符串包含断言写法类似 ）
关键点说明
jsonpath.jsonpath(…)用来从复杂的 JSON 数据中找某个字段（任意层级，比如找 message 的值）
for循环获取到的yaml的数据validation

5. 数据库断言

1）PyMySQL

pymysql 是 Python 连接 MySQL 的“桥梁”

  • 自动化测试中的用途
  • 验证数据库字段是否更新（如订单状态）
  • 清理测试数据（DELETE FROM temp_data）
  • 准备测试数据（INSERT 测试用户）
  • 校验接口是否正确写入数据库

安装：

pip install pymysql
功能方法
连接数据库pymysql.connect()
创建游标（必须创建）.cursor()
执行 SQL.cursor.execute(sql) / .cursor.executemany(sql, seq_of_params)
获取查询结果.cursor.fetchone() /.cursor.fetchall()
提交事务（常见）.cursor.commit()
回滚事务（常见）.cursor.rollback()
防止注入使用 %s 占位符
返回字典cursorclass=DictCursor

验证过程中若遇到报错：需下载cryptography库

报错：

RuntimeError: 'cryptography' package is required for sha256_password or caching_sha2_password auth methods

安装：

pip install cryptography

问题原因：

MySQL 服务器使用了 caching_sha2_password 作为用户认证方式（MySQL 8.0+ 的默认认证插件），而 PyMySQL 在这种模式下需要一个额外的加密库：cryptography

2） 前置准备（整体架构）

config.yaml 数据库配置↓
operationConfig.py → 读取配置↓
connection.py → 连接数据库 + 执行SQL↓
assertion.py → 封装所有断言（包含db断言）↓
测试用例中调用 Assertions().assert_result(...) → 自动完成所有验证
    1. config.yaml：数据库等配置驱动
# 数据库配置
mysql:host: 192.192.2.6port: 3306user: fdafdapassword: "fdakj@2023.."database: ai_quality_controlcharset: utf8mb4
    1. login.yaml：测试用例中的预期结果
      validation:- contains: {code: "200"}- contains: {message: "成功"}- eq: {'msg': '登录成功'}- ne: {'msg': '登录失败'}- db: 'select * from w_leaf where id = 1'
    1. operationConfig.py：通用配置读取工具（可直接获取到数据库配置）
import yaml
from conf.setting import FILE_PATHclass OperationYaml:"""封装读取 YAML 配置文件"""def __init__(self, file_path=None):if file_path is None:self.__file_path = FILE_PATH['conf']else:self.__file_path = file_pathself.__data = None  # 缓存数据，避免重复读文件self.__load_data()def __load_data(self):"""私有方法：加载 YAML 文件"""try:with open(self.__file_path, 'r', encoding='utf-8') as f:self.__data = yaml.safe_load(f)  # 安全解析 YAMLexcept Exception as e:print(f"读取 YAML 文件失败：{e}")self.__data = {}def get(self, *keys):"""通用获取方法，支持多层嵌套:param keys: 键的路径，如 get('api_envi', 'host'):return: 对应值"""if not self.__data:return Nonedata = self.__datafor key in keys:if isinstance(data, dict) and key in data:data = data[key]else:print(f"找不到路径: {keys}")return Nonereturn datadef get_envi(self, option):"""快捷方法：获取接口环境地址"""return self.get('api_envi', option)def get_mysql_conf(self, key):"""快捷方法：获取数据库配置"""return self.get('mysql', key)
    1. connection.py：数据库操作封装文件
from common.recordlog import logs
from conf.operationConfig import OperationYaml
import pymysqlclass ConnectMysql:def __init__(self):self.connection = OperationYaml()mysql_conf = {'host': self.connection.get_mysql_conf('host'),'port': self.connection.get_mysql_conf('port'),'user': self.connection.get_mysql_conf('user'),'password': self.connection.get_mysql_conf('password'),'database': self.connection.get_mysql_conf('database'),'charset': self.connection.get_mysql_conf('charset') or 'utf8mb4'  # 防空}try:self.conn = pymysql.connect(**mysql_conf)# cursor=pymysql.cursors.DictCursor:将数据库表字段显示：以key-value形式显示self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)logs.info("""成功链接到MySql数据库host:{host}port:{port}database:{database}""".format(**mysql_conf))except Exception as e:logs.error(e)def close(self):if self.conn and self.cursor:self.cursor.close()self.conn.close()def query(self, sql):"""查询数据"""try:self.cursor.execute(sql)self.conn.commit()res = self.cursor.fetchall()return resexcept Exception as e:logs.error(e)finally:self.close()def insert(self, sql):"""新增"""passdef update(self, sql):"""修改"""passdef delete(self, sql):"""删除"""pass
  • 代码解释

使用 operationConfig.py 中封装好的解析方法读取 config.yaml 数据库配置

from conf.operationConfig import OperationYamlself.connection = OperationYaml()mysql_conf = {'host': self.connection.get_mysql_conf('host'),'port': self.connection.get_mysql_conf('port'),'user': self.connection.get_mysql_conf('user'),'password': self.connection.get_mysql_conf('password'),'database': self.connection.get_mysql_conf('database'),'charset': self.connection.get_mysql_conf('charset') or 'utf8mb4'  # 防空}

pymysql.connect 连接数据库， pymysql.cursors.DictCursor （以字典形式清晰展示返回内容）

            self.conn = pymysql.connect(**mysql_conf)# cursor=pymysql.cursors.DictCursor:将数据库表字段显示：以key-value形式显示self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)logs.info("""成功链接到MySql数据库host:{host}port:{port}database:{database}""".format(**mysql_conf))

自动关闭数据库连接

    def close(self):if self.conn and self.cursor:self.cursor.close()self.conn.close()

封装数据库查询方法 （.cursor( ) 是操作数据库的“操作手柄”）

    def query(self, sql):"""查询数据"""try:self.cursor.execute(sql)self.conn.commit()res = self.cursor.fetchall()return resexcept Exception as e:logs.error(e)finally:self.close()
创建游标.cursor()
执行 SQL.cursor.execute(sql) / .cursor.executemany(sql, seq_of_params)
提交事务（常见）.cursor.commit()
获取查询结果.cursor.fetchone() /.cursor.fetchall()

3）数据库断言代码实现

from common.connection import ConnectMysqldef assert_mysql(self,expected_sql):"""数据库断言:param expected_sql:预期结果，也就是yaml文件的SQL语句:return: 返回flag标识，0标识测试通过，非0表示测试失败"""flag = 0conn = ConnectMysql()db_value = conn.query(expected_sql)if db_value is not None:logs.info(f"数据库断言成功")else:flag = flag + 1logs.error("数据库断言失败，请检查数据库是否存在该数据！")return flag

终：assert_result：统一断言入口

assertion.py 文件

    def assert_result(self,expected,response,status_code):"""断言模式，通过all_flag标记:param expected: 预期结果:param response:  接口实际返回结果,需要json格式:param status_code: 接口实际返回状态码:return:"""all_flag = 0# 断言状态标识：0 代表成功，其他代表失败try:for yq in expected:for key,value in yq.items():if key == 'contains':flag = self.contains_assert(value,response,status_code)all_flag = all_flag + flagelif key == 'eq':flag = self.equal_assert(value,response)all_flag = all_flag + flagelif key == 'ne':flag = self.not_equal_assert(value,response)all_flag = all_flag + flagelif key == 'db':flag = self.assert_mysql(value)all_flag = all_flag + flagassert all_flag == 0logs.info('测试成功')except Exception as e:logs.error(f'测试失败，异常信息：{e}')raise

→ 遍历每一项
→ 判断 key 是哪种断言
→ 调用对应方法
→ 返回 flag（0 或 1）
→ 累加到 all_flag
→ 最后 assert all_flag == 0 → 测试通过
  • 外部使用方法
from common.assertion import Assertions  # 导入工具validation = to.pop('validation')  # 获取 validationAssertions().assert_result(validation,response,status_code) # 调用该函数
http://www.dtcms.com/a/398419.html

相关文章：

  • C# MVC 模型绑定全解析：从基础机制到自定义绑定器实战指南
  • 企业网站网页设计专业的团队网站建设
  • 网站建设可上传视频的wordpress 数据库类型
  • 广州南沙区建设和交通局网站个人建立网站要多少钱
  • Vue3 》》vite》》vite-plugin-mock mock 模拟数据 ，loadEnv
  • 宝塔面板搭建RustDesk教程：告别命令行，一键拥有私有远程桌面
  • Docker + IDEA 一键部署！
  • Rust开源HyperSwitch实战指南
  • Chrome性能优化指南
  • Chrome高危类型混淆0-Day漏洞（CVE-2025-10585）技术分析
  • 教做面点的网站广州百度竞价托管
  • 网站推广方案合肥房产网安居客
  • 【算法专题训练】24、单调栈
  • 【开题答辩全过程】以 IRWT考试预约系统为例，包含答辩的问题和答案
  • 在字典和列表相互嵌套的结构体中搜索指定元素
  • 文献阅读 | iMetaMed | FigureYa：一个标准化可视化框架，用于增强生物医学数据解释和研究效率
  • wordpress自由拖拽同ip网站做排名seo
  • 面向运动障碍患者的语音识别新突破：零样本实时专家混合自适应方法详解
  • 校园网站建设的维护制作触屏版网站开发
  • 零衍门户组件联邦模式：重新定义组件开发新体验！
  • 【Web前端|第一篇】HTML、CSS与JavaScript
  • 有手机网站了还要微网站吗所有的网站都要用htmlu做吗
  • 面向对象设计：构建可维护、可扩展的软件系统
  • 52.haproxy负载均衡
  • 什么是“智能体”？
  • 负载均衡式在线OJ项目复盘
  • 【Golang】数据设计模式
  • 新建免费网站软件关键词排名
  • 小迪安全v2023学习笔记（八十六讲）—— FridaHOOK证书提取SSL双向校验绕过
  • 律师事务所网站方案网站 建设 初期规划