一、接口断言封装（用例驱动 + 断言工具）

⭐ 核心目标

让测试用例（YAML 或代码）里的断言规则，能自动校验接口响应（比如判断状态码是否 200、响应是否包含某字段）。

⭐ 流程拆解（用例 → 执行 → 断言）

（1）用例定义（YAML 文件：get_token.yaml）

url: https://api.weixin.qq.com/cgi-bin/token
params:
  grant_type: client_credential
  appid: wx8a9de033e93f77ab
  secret: 8326fc915928dee3165720c910effb86
extract:  # 提取响应数据（后续用例可能用到）
  access_token: [json, "$.access_token", 0]
  expires_in: [json, "$.expires_in", 0]
validate:  # 断言规则
  equals:  # 相等断言
    断言状态码为200: [200, status_code]
  contains:  # 包含断言
    断言包含access_token: [access_token, text]

• extract：从接口响应提取数据（如 access_token），存为变量。
• validate：定义断言规则，支持 equals（相等）、contains（包含）等。

（2）用例执行（main_util.py 的 stand_case_flow 函数）

def stand_case_flow(caseinfo):
    # 1. 校验 YAML 数据（转成 Python 对象）
    case_obj = verify_yaml(caseinfo)  
    # 2. 发送接口请求（ru 是 RequestUtil 实例，之前讲过的请求封装）
    res = ru.send_all_request(**eu.change(case_obj.request))  
    # 3. 提取变量（如果有 extract）
    if case_obj.extract:
        for key, value in case_obj.extract.items():
            eu.extract(res, key, *value)  
    # 4. 执行断言（如果有 validate）
    if case_obj.validate:
        for assert_type, value in case_obj.validate.items():
            au.assert_all_case(res, assert_type, value)  # au 是 AssertUtil 实例
    else:
        print("此用例没有断言！")

第一步：YAML 用例中定义断言规则（源头）

在 YAML 测试用例里，validate 节点下会定义所有断言规则，例如：

validate:  # 断言总配置
  equals:  # 断言类型1（对应 assert_type）
    断言状态码为200: [200, status_code]  # 该类型下的具体规则（对应 value 中的内容）
    断言过期时间为7200: [7200, json.expires_in]
  contains:  # 断言类型2（对应 assert_type）
    断言响应含token: [access_token, text]  # 该类型下的具体规则（对应 value 中的内容）

• 这里的 equals、contains 是断言类型（后续会成为 assert_type）。
• 每个断言类型下面的键值对（如 {"断言状态码为200": [200, status_code], ...}）是具体的断言规则（后续会成为 value）。

第二步：框架解析 YAML 用例，提取断言规则

框架在执行用例时（如 main_util.py 的 stand_case_flow 函数），会先将 YAML 用例解析成 Python 可识别的对象（如 case_obj），其中：

• case_obj.validate 对应 YAML 中的 validate 节点，是一个嵌套字典：

  case_obj.validate = {
      "equals": {
          "断言状态码为200": [200, "status_code"],
          "断言过期时间为7200": [7200, "json.expires_in"]
      },
      "contains": {
          "断言响应含token": ["access_token", "text"]
      }
  }
  

第三步：遍历解析结果，传递参数到 assert_all_case 函数

在解析完成后，框架会遍历 case_obj.validate 中的所有断言类型和规则，并调用 assert_all_case 函数执行断言：

# 遍历断言总配置（case_obj.validate）
for assert_type, value in case_obj.validate.items():
    # 调用断言函数，传入参数
    au.assert_all_case(res, assert_type, value)

• assert_type：遍历字典时获取的键，即断言类型（如 "equals"、"contains"）。
• value：遍历字典时获取的值，即该断言类型下的具体规则（如 {"断言状态码为200": [200, "status_code"], ...}）。
• res：接口响应对象（包含状态码、响应体等，用于提取实际值）。

（3）断言工具（assert_util.py 的 AssertUtil 类）

import copy
import logging
from functools import singledispatch

class AssertUtil:
    def assert_all_case(self, res, assert_type, value):
        # 1. 深拷贝响应对象（避免修改原数据）
        new_res = copy.deepcopy(res)  
        # 2. 处理响应为 JSON（方便断言 JSON 数据）
        try:
            new_res.json = new_res.json()  # 把 res.json() 结果存为属性
        except Exception:
            new_res.json = {"msg": "response not json data"}  # 非 JSON 响应的兼容处理
        
        # 3. 遍历断言规则
        for msg, yq_and_sj_data in value.items():
            yq, sj = yq_and_sj_data[0], yq_and_sj_data[1]  # yq=预期值，sj=实际值的来源（如 status_code）
            # 从响应对象中获取实际值（如 res.status_code → sj_value）
            sj_value = getattr(new_res, sj)  
            print(assert_type, msg, yq, sj_value)  # 打印断言信息（调试用）
            
            # 4. 根据断言类型执行判断
            match assert_type:
                case "equals":
                    assert yq == sj_value, msg  # 相等断言
                case "contains":
                    assert yq in sj_value, msg   # 包含断言

• 关键逻辑：
  • new_res.json = new_res.json()：把响应的 JSON 数据存为属性，方便后续断言。
  • getattr(new_res, sj)：动态获取响应的属性（如 status_code、text、json）。
  • match assert_type：根据断言类型（equals/contains）执行不同的断言逻辑。

⭐ 断言流程总结

1. 用例定义：在 YAML 里写清楚 “怎么请求、提取什么、断言什么”。
2. 用例执行：框架解析 YAML，发送请求，拿到响应。
3. 断言执行：断言工具根据规则，自动对比 “预期值” 和 “实际响应值”，不通过则抛出异常。

---

二、数据库操作封装（连接 + 执行 SQL）

⭐ 核心目标

让框架能连接数据库，执行 SQL 查询，用于：

• 断言 “接口操作是否同步到数据库”（如接口创建订单后，数据库是否新增记录）。
• 初始化测试数据（如测试前插入一条数据）。

⭐ 代码拆解（assert_util.py 的 AssertUtil 类扩展）

（1）数据库连接（conn_database 方法）

import pymysql

class AssertUtil:
    # 连接数据库
    def conn_database(self):
        self.conn = pymysql.connect(
            user="sdm723416659",      # 数据库用户名
            password="Msiy123456",    # 数据库密码
            host="sdm723416659.my3w.com",  # 数据库地址
            database="sdm723416659_db",    # 数据库名
            port=3306                 # 端口（MySQL 默认 3306）
        )
        return self.conn

关键说明：

• import pymysql：pymysql 是 Python 操作 MySQL 的专用库，没有它就无法通过代码连接 MySQL，所以必须先安装（命令：pip install pymysql）。
• pymysql.connect(...)：这是核心函数，传入数据库的登录信息后，返回一个 “连接对象（conn）”，这个对象就代表 “Python 与数据库之间的一条通路”。
• self.conn：把连接对象存到类的属性里，方便类内部的其他方法（比如 execute_sql）使用。

（2）执行 SQL（execute_sql 方法）

class AssertUtil:
    # 执行 SQL 查询
    def execute_sql(self, sql):
        # 1. 获取数据库连接
        conn = self.conn_database()  
        # 2. 创建游标（用于执行 SQL）
        cs = conn.cursor()  
        # 3. 执行 SQL
        cs.execute(sql)  
        # 4. 获取查询结果（只取第一条，适合单行查询）
        value = cs.fetchone()  
        # 5. 关闭资源（避免连接泄漏）
        cs.close()  
        conn.close()  
        # 6. 返回查询结果
        return value

关键说明：

• 游标（cursor）的作用：游标就像 “数据库的遥控器”，所有 SQL 语句都必须通过游标来执行。你可以理解为：连接（conn）是 “打通了通路”，但执行 SQL 还需要一个 “操作工具”，就是游标（cs）。
• cs.execute(sql)：执行 SQL 语句的核心代码。比如传入 sql="SELECT name FROM user WHERE id=1"，这行代码就会在数据库中执行这个查询。
• cs.fetchone()：获取查询结果。如果 SQL 是查询语句（SELECT），这个方法会返回结果中的第一条数据（格式是元组，比如 ('张三',)）；如果没有结果，返回 None。
• 关闭资源：cs.close() 和 conn.close() 必须写，否则数据库的连接和游标会一直被占用，导致后续无法连接（类似 “用完数据库客户端后要关闭窗口”）。

（3）如何用于断言？（扩展思路）

假设要断言 “接口创建的用户，数据库是否存在”：

def test_create_user():
    # 1. 发送创建用户的接口请求
    res = requests.post("http://example.com/create_user", json={"name": "test"})
    
    # 2. 执行 SQL 查询（查数据库是否有该用户）
    au = AssertUtil()
    sql = "SELECT * FROM user WHERE name='test'"
    db_result = au.execute_sql(sql)
    
    # 3. 断言数据库结果
    assert db_result is not None, "用户创建后数据库未新增记录"

---

三、数据库封装的价值（为什么需要它？）

1. 接口与数据库的双向校验：
   接口返回 “操作成功”≠ 数据库真的变更。通过数据库断言，能确保数据最终落库（如创建订单接口，需查数据库确认订单存在）。

2. 初始化测试数据：
   测试前用 execute_sql 插入数据（如 INSERT INTO user ...），保证测试环境干净。

3. 复杂业务校验：
   某些业务逻辑（如积分扣除、库存减少）需结合数据库数据才能验证，单纯接口响应无法覆盖。

四、数据库断言 

⭐ YAML 用例定义（以 db_equals 为例）

validate:
  db_equals:  # 断言类型：数据库结果相等
    # 描述信息: [SQL语句, 接口响应的实际值来源]
    判断返回结果等于SQL查询内容: ["select email from pw_user where uid=1", json.email]

⭐ 代码逻辑（assert_util.py 的 AssertUtil 类）

class AssertUtil:
    def assert_all_case(self, res, assert_type, value):
        # ... 其他断言逻辑（equals/contains）...
        
        match assert_type:
            # 新增：数据库相等断言
            case "db_equals":
                # 1. 执行 SQL 查询（yq 是 SQL 语句）
                yq_value = self.execute_sql(yq)  
                # 2. 断言：数据库查询结果的第一个值 == 接口响应的实际值
                assert yq_value[0] == sj_value, msg  
            
            # 新增：数据库包含断言
            case "db_contains":
                # 1. 执行 SQL 查询（yq 是 SQL 语句）
                yq_value = self.execute_sql(yq)  
                # 2. 断言：数据库查询结果的第一个值 在 接口响应的实际值中
                assert yq_value[0] in sj_value, msg  

• 核心流程：
  1. 从 YAML 中拿到 SQL 语句 和 接口响应来源。
  2. 调用 self.execute_sql(yq) 执行 SQL，获取数据库实际值。
  3. 从接口响应中提取 实际值（sj_value）。
  4. 对比 数据库值（yq_value[0]） 和 接口值（sj_value），断言是否符合预期。

⭐ db_equals 断言

def assert_all_case(self, res, assert_type, value):
    for msg, yq_and_sj_data in value.items():
        yq, sj = yq_and_sj_data[0], yq_and_sj_data[1]  # yq=SQL，sj=接口值来源
        sj_value = getattr(new_res, sj)  # 从接口响应提取实际值（如 new_res.json["email"]）
        
        match assert_type:
            case "db_equals":
                # 1. 执行 SQL，获取数据库值
                yq_value = self.execute_sql(yq)  # yq 是 SQL："select email from pw_user where uid=1"
                # 2. 断言：数据库结果（yq_value[0]） == 接口值（sj_value）
                assert yq_value[0] == sj_value, msg  

⭐ db_contains 断言

validate:
  db_contains:  # 断言类型：数据库结果包含
    判断返回结果包含SQL内容: ["select email from pw_user where uid=1", text]

• 逻辑：
  执行 SQL 查询数据库，拿查询结果的第一个值，判断是否 包含在接口响应中。

• case "db_contains":
      yq_value = self.execute_sql(yq)  
      assert yq_value[0] in sj_value, msg  
  

• 应用场景：
  接口响应是长文本（如 操作成功，影响用户 test@example.com），需验证 “数据库查询的 email” 是否包含在响应中。

不是在执行sql函数的时候用了fetchone()了吗，为什么这里还要用assert yq_value[0] == sj_value, msg

⭐ fetchone() 的返回值格式

当执行查询 SQL（如 SELECT email from pw_user where uid=1）时：

• 情况 1：查询到数据
  数据库返回结果是 元组（即使只有一个字段），例如：

  yq_value = ('test@example.com',)  # fetchone() 返回的结果
  

   
  • 这里的 ('test@example.com',) 是元组类型，不是字符串。如果直接用 yq_value == sj_value（假设 sj_value 是字符串 'test@example.com'），会因为类型不匹配导致断言失败。

• 情况 2：查询无数据
  fetchone() 返回 None，此时 yq_value is None，也需要通过断言逻辑判断是否符合预期。

⭐ 为什么不直接改 execute_sql 返回值？

你可能会想：为什么不在 execute_sql 里直接返回 yq_value[0]，简化断言逻辑？

这是因为：

1. 通用性需求：
   execute_sql 是通用数据库操作函数，可能被用于各种场景（如查询多条结果、判断是否存在数据）。如果强制返回 yq_value[0]，会破坏函数的通用性（比如其他逻辑需要完整的元组结果）。

2. 语义清晰：
   在断言逻辑中显式写 yq_value[0]，能清晰表达 “取数据库查询结果的第一个字段值”，让阅读代码的人明白对比的是 “单字段值”，而非复杂结构。

⭐ 完整流程示例（结合实际场景）

假设测试 “创建用户后，数据库是否新增用户”：

1. YAML 用例定义

validate:
  db_equals:
    断言数据库新增用户: ["SELECT email FROM user WHERE name='test'", json.email]

2. 代码执行流程

# 1. 执行 SQL 查询数据库
yq = "SELECT email FROM user WHERE name='test'"
yq_value = self.execute_sql(yq)  # 返回元组，如 ('test@example.com',)

# 2. 提取接口响应的实际值
sj_value = res.json().get('email')  # 假设是 'test@example.com'

# 3. 断言：数据库查询结果的第一个值 == 接口响应值
assert yq_value[0] == sj_value, "数据库用户与接口响应不一致"

3. 断言通过的条件

只有当：

yq_value[0] == sj_value  # 即 'test@example.com' == 'test@example.com'

时，断言才会通过，确保 “接口响应的 email” 和 “数据库查询的 email” 完全一致。