构建一个 Python +Pytest API 自动化测试框架

前言

• 🤔 每次都要手写复杂的 HTTP 请求代码
• 🤔 如何高效地管理数百个 API 测试用例
• 🤔 参数如何在多个接口间传递和依赖
• 🤔 如何快速定位测试失败原因
• 🤔 测试报告如何让非技术人员也能理解

可以通过基于 Python + Pytest + YAML 的 API 自动化测试框架来解决这些问题。

---

一、框架概览

1.1 框架特色一览

┌─────────────────────────────────────────────────────────┐
│         API 自动化测试框架核心特性                       │
├─────────────────────────────────────────────────────────┤
│ ✅ 配置化驱动：YAML 文件定义测试用例，无需编写代码      │
│ ✅ 参数依赖管理：支持接口间参数传递，链式调用           │
│ ✅ 多种断言模式：contains、eq、ne、rv、db 数据库断言    │
│ ✅ 智能参数替换：${函数名(参数)} 动态参数替换            │
│ ✅ 数据库集成：支持 MySQL、MongoDB、ClickHouse         │
│ ✅ 完整的报告系统：Allure + 钉钉通知                    │
│ ✅ 自定义函数库：debugtalk.py 扩展能力                  │
│ ✅ 企业级特性：文件上传、Cookie 管理、代理支持          │
└─────────────────────────────────────────────────────────┘

1.2 核心技术栈

技术组件	用途	特点
Pytest	测试框架	强大的 fixture、plugin、mark 机制
Requests	HTTP 库	简洁API、完整的功能
PyYAML	YAML 解析	配置化测试用例定义
JSONPath	JSON 提取	灵活的数据提取能力
SQLAlchemy	ORM	支持多种数据库
Allure	测试报告	美观的 HTML 报告
Pandas	数据处理	Excel 数据读写
Faker	数据生成	随机测试数据

---

二、架构设计

2.1 项目目录结构

python-pytest-/
├── base/                           # 基础类封装层
│   ├── apiutil.py                 # ⭐ API 请求核心类
│   ├── apiutil_business.py        # 业务相关的 API 操作
│   ├── generateId.py              # ID 生成工具
│   ├── new_testcase_tools.py      # UI 工具（生成测试用例）
│   └── removefile.py              # 文件清理工具
│
├── common/                         # 公共模块层
│   ├── sendrequest.py             # HTTP 请求发送
│   ├── assertions.py              # ⭐ 多种断言模式实现
│   ├── debugtalk.py               # ⭐ 自定义函数库
│   ├── readyaml.py                # YAML 读写
│   ├── recordlog.py               # 日志记录
│   ├── connection.py              # 数据库连接（MySQL、MongoDB）
│   ├── handleExcel.py             # Excel 处理
│   ├── semail.py                  # 邮件发送
│   ├── dingRobot.py               # 钉钉通知
│   ├── operxml.py                 # XML 处理
│   └── two_dimension_data.py      # 二维数据处理
│
├── conf/                           # 配置管理层
│   ├── setting.py                 # 全局配置
│   ├── config.ini                 # 环境配置
│   └── operationConfig.py         # 配置操作类
│
├── data/                           # 测试数据层
│   ├── testdata.xls               # Excel 测试数据
│   ├── loginName.yaml             # 登录用例
│   └── ...其他 YAML 用例
│
├── testcase/                       # 测试用例集
│   ├── conftest.py                # Pytest 配置
│   └── test_*.py                  # 测试文件
│
├── logs/                           # 日志输出
├── report/                         # 测试报告
│   ├── temp/                      # Allure 原始数据
│   └── tmreport/                  # 生成的报告
│
├── extract.yaml                    # ⭐ 参数提取文件（用于参数依赖）
├── environment.xml                 # 环境信息配置
├── pytest.ini                      # Pytest 配置
├── conftest.py                     # 全局 Pytest fixture
├── run.py                          # ⭐ 框架运行入口
├── requirements.txt                # 项目依赖
└── README.md                       # 文档

2.2 分层架构

┌──────────────────────────────────────────────┐
│  Test Case Layer (Pytest + YAML)             │
│  - 测试用例定义                              │
│  - 配置化管理                                │
│  - 参数数据驱动                              │
└─────────────────┬──────────────────────────┘
                  │
┌─────────────────▼──────────────────────────┐
│  Business Layer (base/ + common/)           │
│  - API 请求处理 (apiutil.py)                │
│  - 参数替换和提取 (replace_load)            │
│  - 多种断言模式 (assertions.py)             │
│  - 自定义函数库 (debugtalk.py)              │
│  - 数据库操作 (connection.py)               │
│  - 日志和报告 (recordlog.py)                │
└─────────────────┬──────────────────────────┘
                  │
┌─────────────────▼──────────────────────────┐
│  Infrastructure Layer                       │
│  - HTTP 请求 (requests library)            │
│  - 数据库连接 (SQLAlchemy)                  │
│  - YAML 解析 (PyYAML)                       │
│  - 日志系统 (logging)                       │
│  - 报告生成 (Allure)                        │
└──────────────────────────────────────────────┘

---

三、核心特性深度解析

3.1 配置化测试用例定义

框架最大的特色是使用 YAML 文件定义测试用例，无需编写代码。

测试用例结构：

# data/轨迹查询.yaml
baseInfo:
  api_name: 轨迹查询                    # 接口名称（用于报告）
  url: /monitor/vehicle/getMileageFrom  # 接口路径
  method: post                          # HTTP 方法
  header:
    Content-Type: application/x-www-form-urlencoded;charset=UTF-8
    token: ${get_extract_data(token)}   # 参数替换（来自上一个接口）
    userid: ${get_extract_data(userid)}
  cookies:
    SESSION: ${get_extract_data(Cookie,SESSION)}

testCase:
  - case_name: 有效车牌号轨迹查询    # 用例名称
    data:                             # 表单参数
      vno: 鲁E00098
      color: 2
      startDate: ${start_time()}       # 动态生成当前时间
      endDate: ${end_time()}
      ruleIds: ["${get_extract_data_lst(forbiddenRule,-2)}"]
    
    files:                             # 文件上传
      file: ./data/heimingdan.xlsx
    
    validation:                        # 断言
      - contains: {status_code: 200}   # 状态码为 200
      - contains: {'message':'success'} # 响应包含 success
      - eq: {'state': '已入网'}         # 状态字段等于
      - ne: {'state': '已入网'}         # 状态字段不等于
      - rv: {"data": 2}                # 任意值断言
      - db: select * from sys_user where login_name='test999'  # 数据库断言
    
    extract:                           # 提取参数（用于后续接口）
      id: $.data                       # JSONPath 提取
      status: '"status":"(.*?)"'       # 正则表达式提取
    
    extract_list:                      # 提取列表参数
      ids: $.result[*].id              # 提取数组

优势对比：

# ❌ 传统方式：需要编写大量 Python 代码
def test_vehicle_track():
    response = requests.post(
        url='http://api.example.com/monitor/vehicle/getMileageFrom',
        headers={
            'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
            'token': get_token(),
            'userid': get_userid()
        },
        cookies={'SESSION': get_session()},
        data={
            'vno': '鲁E00098',
            'color': 2,
            'startDate': datetime.now().strftime('%Y-%m-%d'),
            'endDate': (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d'),
            'ruleIds': [get_forbidden_rule()]
        }
    )
    
    assert response.status_code == 200
    assert 'success' in response.json()['message']
    assert response.json()['state'] == '已入网'
    
    # 提取参数
    data = response.json()
    save_extract_data('id', data['data'])
    save_extract_data('status', re.search(r'"status":"(.*?)"', response.text).group(1))

# ✅ 使用框架：YAML 文件定义，一目了然
# 只需在 轨迹查询.yaml 中配置，无需代码

3.2 参数替换机制（核心创新）

这是框架最强大的特性——反射机制 + 动态参数替换。

工作原理：

YAML 中的参数占位符 → 正则识别 ${} → 提取函数名和参数 → 
反射调用对应函数 → 获取返回值 → 替换原始值

实现代码（base/apiutil.py）：

def replace_load(self, data):
    """yaml 数据替换解析"""
    str_data = data
    if not isinstance(data, str):
        str_data = json.dumps(data, ensure_ascii=False)
    
    # 循环处理所有 ${ } 表达式
    for i in range(str_data.count('${')):
        if '${' in str_data and '}' in str_data:
            start_index = str_data.index('$')
            end_index = str_data.index('}', start_index)
            ref_all_params = str_data[start_index:end_index + 1]
            
            # 提取函数名：从 ${ 到 (
            func_name = ref_all_params[2:ref_all_params.index("(")]
            
            # 提取参数：从 ( 到 )
            func_params = ref_all_params[ref_all_params.index("(") + 1:ref_all_params.index(")")]
            
            # 反射调用函数：相当于调用 DebugTalk().get_extract_data('token')
            extract_data = getattr(DebugTalk(), func_name)(
                *func_params.split(',') if func_params else ""
            )
            
            # 替换原始值
            if extract_data and isinstance(extract_data, list):
                extract_data = ','.join(e for e in extract_data)
            str_data = str_data.replace(ref_all_params, str(extract_data))
    
    # 还原数据类型
    if data and isinstance(data, dict):
        data = json.loads(str_data)
    else:
        data = str_data
    return data

使用场景示例：

# 场景1：动态生成当前时间
startDate: ${start_time()}
endDate: ${end_time()}

# 场景2：获取上一个接口提取的参数
token: ${get_extract_data(token)}
userid: ${get_extract_data(userid)}

# 场景3：生成随机数据
phone: ${faker_phone()}
name: ${faker_name()}

# 场景4：从列表中获取指定索引
ruleId: ${get_extract_data_lst(forbiddenRule,-2)}  # 获取列表的倒数第2个

# 场景5：自定义函数调用
custom_param: ${my_custom_function(arg1,arg2)}

自定义函数库（common/debugtalk.py）：

class DebugTalk:
    """自定义函数库"""
    
    def get_extract_data(self, key, *args):
        """获取提取的参数"""
        # 从 extract.yaml 中读取
        extract_data = self.read_yaml_data()
        if args and args[0] == 'randoms':
            # 从列表中随机获取一个
            return random.choice(extract_data[key])
        return extract_data.get(key)
    
    def get_extract_data_lst(self, key, index=None):
        """获取列表参数"""
        extract_data = self.read_yaml_data()
        data = extract_data.get(key)
        if index is not None:
            return data[int(index)]
        return data
    
    def start_time(self):
        """获取当前时间"""
        return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    def end_time(self):
        """获取明天时间"""
        tomorrow = datetime.now() + timedelta(days=1)
        return tomorrow.strftime('%Y-%m-%d %H:%M:%S')
    
    def faker_phone(self):
        """生成随机手机号"""
        faker = Faker('zh_CN')
        return faker.phone_number()
    
    def faker_name(self):
        """生成随机姓名"""
        faker = Faker('zh_CN')
        return faker.name()
    
    @staticmethod
    def md5(text):
        """MD5 加密"""
        import hashlib
        return hashlib.md5(text.encode()).hexdigest()
    
    @staticmethod
    def base64_encode(text):
        """Base64 编码"""
        import base64
        return base64.b64encode(text.encode()).decode()

参数依赖示例：

# 第一个接口：登录，提取 token
# data/login.yaml
baseInfo:
  api_name: 用户登录
  url: /api/login
  method: post

testCase:
  - case_name: 用户登录
    data:
      username: admin
      password: 123456
    validation:
      - contains: {status_code: 200}
    extract:
      token: $.data.token        # 从响应提取 token
      userid: $.data.userid

---

# 第二个接口：查询用户信息，使用 token
# data/user_info.yaml
baseInfo:
  api_name: 查询用户信息
  url: /api/user/info
  method: get
  header:
    Authorization: Bearer ${get_extract_data(token)}  # 使用上一个接口的 token

testCase:
  - case_name: 查询用户信息
    validation:
      - contains: {status_code: 200}

执行流程：

1. 执行登录接口
   - 发送请求
   - 获取响应：{"data": {"token": "xxx", "userid": 123}}
   - 提取参数：写入 extract.yaml
     token: xxx
     userid: 123

2. 执行查询接口
   - 替换参数：get_extract_data(token) → xxx
   - 请求头变为：Authorization: Bearer xxx
   - 发送请求

---

3.3 五种断言模式

框架支持 5 种灵活的断言方式，满足不同场景需求。

1. Contains（包含断言）- 最常用

validation:
  - contains: {status_code: 200}        # 状态码为 200
  - contains: {'message':'success'}     # 响应包含 success
  - contains: {'csrfToken': 'xxx'}      # 验证 token 存在
  - contains: {'data': None}            # 数据为空

实现代码：

def contains_assert(self, value, response, status_code):
    """字符串包含断言"""
    flag = 0
    for assert_key, assert_value in value.items():
        if assert_key == "status_code":
            if assert_value != status_code:
                flag += 1
                logs.error(f"状态码{status_code}不等于{assert_value}")
        else:
            # 使用 JSONPath 从响应中提取值
            resp_list = jsonpath.jsonpath(response, "$..%s" % assert_key)
            if resp_list:
                assert_value = None if assert_value.upper() == 'NONE' else assert_value
                if assert_value in resp_list:
                    logs.info(f"包含断言成功")
                else:
                    flag += 1
                    logs.error(f"包含断言失败")
    return flag

2. Eq（相等断言）

validation:
  - eq: {'state': '已入网'}      # 状态字段等于指定值
  - eq: {'count': 10}            # 数量等于 10

3. Ne（不相等断言）

validation:
  - ne: {'status': 'error'}      # 状态不是 error
  - ne: {'code': 400}            # 状态码不是 400

4. Rv（响应值断言）- 验证任意字段

validation:
  - rv: {"data": 2}              # 验证响应中 data 字段值为 2
  - rv: {"count": 0}             # 验证返回记录数为 0

5. Db（数据库断言）- 企业级特性

validation:
  - db: select * from sys_user where login_name='test999'
  # SQL 查询结果不为空则断言成功
  # 可用于验证数据是否正确写入数据库

实现代码：

def assert_mysql_data(self, expected_results):
    """数据库断言"""
    flag = 0
    conn = ConnectMysql()
    db_value = conn.query_all(expected_results)  # 执行 SQL
    if db_value is not None:
        logs.info("数据库断言成功")
    else:
        flag += 1
        logs.error("数据库断言失败，请检查数据库是否存在该数据")
    return flag

完整断言示例：

testCase:
  - case_name: 完整的断言示例
    data:
      username: admin
      password: 123456
    
    validation:
      # 必须先验证状态码
      - contains: {status_code: 200}
      
      # 然后验证消息
      - contains: {'message':'success'}
      
      # 验证数据字段
      - eq: {'state': '已入网'}
      
      # 验证状态不是错误
      - ne: {'error': 'exists'}
      
      # 验证响应中的任意值
      - rv: {"data": 1}
      
      # 验证数据是否写入数据库
      - db: select * from sys_user where id=1

断言执行逻辑：

def assert_result(self, expected, response, status_code):
    """
    断言主方法
    - 多个断言顺序执行
    - 使用 all_flag 累计失败次数
    - 只要有一个断言失败，整个测试用例失败
    """
    all_flag = 0
    
    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 + flag
            elif key == "eq":
                flag = self.equal_assert(value, response)
                all_flag = all_flag + flag
            elif key == 'ne':
                flag = self.not_equal_assert(value, response)
                all_flag = all_flag + flag
            elif key == 'rv':
                flag = self.assert_response_any(response, value)
                all_flag = all_flag + flag
            elif key == 'db':
                flag = self.assert_mysql_data(value)
                all_flag = all_flag + flag
    
    # 所有断言都通过（all_flag == 0）才算成功
    if all_flag == 0:
        logs.info("测试成功")
        assert True
    else:
        logs.error("测试失败")
        assert False

---

3.4 数据库集成（企业级特性）

框架原生支持多种数据库，可以进行数据预置、数据清理、数据库断言。

支持的数据库：

# common/connection.py
class ConnectMysql:
    """MySQL 连接"""
    def query_one(self, sql)      # 查询单条
    def query_all(self, sql)      # 查询多条
    def insert(self, sql)         # 插入数据
    def delete(self, sql)         # 删除数据
    def update(self, sql)         # 更新数据

class ConnectMongodb:
    """MongoDB 连接"""
    # 支持 MongoDB 的所有基本操作

class ClickhouseSQL:
    """ClickHouse 连接"""
    # 支持 ClickHouse 的数据查询

使用场景：

# 场景1：测试前预置数据
def setup(self):
    conn = ConnectMysql()
    # 插入测试数据
    sql = "insert into test_users values (1, 'test_user', 'test@example.com')"
    conn.insert(sql)

# 场景2：断言数据库数据
validation:
  - db: select * from test_users where username='test_user'

# 场景3：测试后清理数据
def teardown(self):
    conn = ConnectMysql()
    sql = "delete from test_users where username='test_user'"
    conn.delete(sql)

Conftest 中的数据清理示例：

# testcase/conftest.py
@pytest.fixture(scope='session', autouse=True)
def datadb_init():
    """
    后置处理：测试后清理数据
    这样不会产生脏数据，也不会对系统造成影响
    """
    yield  # 测试执行
    
    # 测试完成后清理
    conn = ConnectMysql()
    sql = "delete from test_users where username like 'test_%'"
    conn.delete(sql)
    
    allure.attach('将测试数据清空', 'fixture后置', allure.attachment_type.TEXT)

---

3.5 参数提取与数据处理

框架支持两种参数提取方式：JSONPath 和正则表达式。

JSON 提取（推荐）：

extract:
  id: $.data                      # 提取 data 字段
  name: $.data.user.name          # 提取嵌套字段
  status: $.status                # 提取顶级字段

extract_list:
  ids: $.result[*].id             # 提取数组中所有 id
  names: $.users[0:3].name        # 提取前3个用户的名称

正则提取：

extract:
  status: '"status":"(.*?)"'      # 提取 status 的值
  token: 'token":"([^"]*)"'       # 提取 token 的值
  data: 'data":(\d*)'             # 提取数字

extract_list:
  ids: 'id":"(\d+)"'              # 提取所有 id（列表形式）

实现代码：

def extract_data(self, testcase_extarct, response):
    """参数提取"""
    try:
        for key, value in testcase_extarct.items():
            # 处理 JSON 提取
            if '$' in value:
                ext_json = jsonpath.jsonpath(json.loads(response), value)[0]
                if ext_json:
                    extract_data = {key: ext_json}
                    logs.info(f'提取接口的返回值：{extract_data}')
                    self.read.write_yaml_data(extract_data)  # 写入 extract.yaml
            
            # 处理正则表达式提取
            pattern_lst = ['(.*?)', '(.+?)', r'(\d)', r'(\d*)']
            for pat in pattern_lst:
                if pat in value:
                    ext_lst = re.search(value, response)
                    if ext_lst:
                        if pat in [r'(\d+)', r'(\d*)']:
                            extract_data = {key: int(ext_lst.group(1))}
                        else:
                            extract_data = {key: ext_lst.group(1)}
                        self.read.write_yaml_data(extract_data)
    except Exception as e:
        logs.error(e)

---

3.6 文件上传处理

框架原生支持文件上传功能。

testCase:
  - case_name: 导入数据文件
    data:
      import_type: user_list
    
    files:                         # 文件上传
      file: ./data/user_list.xlsx  # 相对路径或绝对路径
    
    validation:
      - contains: {status_code: 200}
      - contains: {'message': 'success'}

实现代码：

def specification_yaml(self, base_info, test_case):
    """处理接口请求"""
    
    # 处理文件上传接口
    file, files = test_case.pop('files', None), None
    if file is not None:
        for fk, fv in file.items():
            allure.attach(json.dumps(file), '导入文件')
            # 打开文件并传递给 requests
            files = {fk: open(fv, mode='rb')}
    
    # 发送请求（包含文件）
    res = self.run.run_main(
        name=api_name,
        url=url,
        case_name=case_name,
        header=header,
        method=method,
        file=files,  # 文件参数
        cookies=cookie,
        **test_case
    )

---

3.7 完整的测试报告系统

框架集成了企业级的报告系统。

Allure 报告配置：

# run.py
if REPORT_TYPE == 'allure':
    pytest.main([
        '-s', '-v',
        '--alluredir=./report/temp',
        './testcase',
        '--clean-alluredir',
        '--junitxml=./report/results.xml'
    ])
    
    # 复制环境信息配置文件
    shutil.copy('./environment.xml', './report/temp')
    
    # 生成并打开报告
    os.system(f'allure serve ./report/temp')

环境信息配置（environment.xml）：

<?xml version="1.0" encoding="UTF-8"?>
<environment>
    <parameter>
        <name>测试环境</name>
        <value>测试服务器</value>
    </parameter>
    <parameter>
        <name>API 基础 URL</name>
        <value>http://api.example.com</value>
    </parameter>
    <parameter>
        <name>数据库</name>
        <value>MySQL 8.0</value>
    </parameter>
    <parameter>
        <name>测试框架</name>
        <value>Pytest + Python</value>
    </parameter>
</environment>

钉钉通知集成：

# conftest.py
from common.dingRobot import send_dd_msg

def pytest_terminal_summary(terminalreporter, exitstatus, config):
    """测试完成后发送钉钉通知"""
    
    total = terminalreporter._numcollected
    passed = len(terminalreporter.stats.get('passed', []))
    failed = len(terminalreporter.stats.get('failed', []))
    error = len(terminalreporter.stats.get('error', []))
    
    summary = f"""
    自动化测试结果，请查看
    测试用例总数：{total}
    测试通过数：{passed}
    测试失败数：{failed}
    错误数量：{error}
    """
    
    # 发送钉钉通知
    if dd_msg:  # 配置中开启钉钉通知
        send_dd_msg(summary)

---

四、最佳实践与应用场景

4.1 场景一：链式 API 调用（参数依赖）

# 场景：用户注册 → 用户登录 → 查询用户信息 → 修改用户信息

# 1. 用户注册 - data/register.yaml
baseInfo:
  api_name: 用户注册
  url: /api/register
  method: post

testCase:
  - case_name: 注册新用户
    data:
      username: ${faker_name()}
      password: 123456
      email: ${faker_email()}
    
    validation:
      - contains: {status_code: 200}
    
    extract:
      user_id: $.data.id          # 提取用户 ID
      username: $.data.username   # 提取用户名
      token: $.data.token         # 提取 token

---

# 2. 用户登录 - data/login.yaml
baseInfo:
  api_name: 用户登录
  url: /api/login
  method: post
  header:
    Content-Type: application/json

testCase:
  - case_name: 使用新注册的账号登录
    data:
      username: ${get_extract_data(username)}  # 使用注册时的用户名
      password: 123456
    
    validation:
      - contains: {status_code: 200}
    
    extract:
      token: $.data.token         # 提取登录 token

---

# 3. 查询用户信息 - data/user_info.yaml
baseInfo:
  api_name: 查询用户信息
  url: /api/users/${get_extract_data(user_id)}
  method: get
  header:
    Authorization: Bearer ${get_extract_data(token)}

testCase:
  - case_name: 查询新注册用户的信息
    validation:
      - contains: {status_code: 200}
      - contains: {'username': '${get_extract_data(username)}'}

---

# 4. 修改用户信息 - data/update_user.yaml
baseInfo:
  api_name: 修改用户信息
  url: /api/users/${get_extract_data(user_id)}
  method: put
  header:
    Authorization: Bearer ${get_extract_data(token)}
    Content-Type: application/json

testCase:
  - case_name: 修改用户信息
    data:
      email: newemail@example.com
      phone: ${faker_phone()}
    
    validation:
      - contains: {status_code: 200}
      - contains: {'message': 'success'}

4.2 场景二：异常场景测试（多断言）

# 数据验证 - data/user_validation.yaml
baseInfo:
  api_name: 创建用户
  url: /api/users
  method: post
  header:
    Content-Type: application/json

testCase:
  # 场景1：用户名为空
  - case_name: 用户名为空
    data:
      username: ""
      password: 123456
    
    validation:
      - contains: {status_code: 400}
      - contains: {'error': 'username required'}

  # 场景2：用户名过长
  - case_name: 用户名过长（超过 50 字符）
    data:
      username: "abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz"
      password: 123456
    
    validation:
      - contains: {status_code: 400}
      - contains: {'error': 'username too long'}

  # 场景3：密码过弱
  - case_name: 密码过弱
    data:
      username: newuser
      password: 123
    
    validation:
      - contains: {status_code: 400}
      - contains: {'error': 'password too weak'}

  # 场景4：用户已存在
  - case_name: 用户名已存在
    data:
      username: admin
      password: 123456
    
    validation:
      - contains: {status_code: 409}
      - contains: {'error': 'user already exists'}
      - db: select * from users where username='admin'

4.3 场景三：数据库相关测试

# data/order_business.yaml
baseInfo:
  api_name: 创建订单业务
  url: /api/orders
  method: post

testCase:
  - case_name: 创建订单后验证数据库
    data:
      user_id: 1
      product_id: 101
      quantity: 5
      price: 99.99
    
    validation:
      # 验证接口响应
      - contains: {status_code: 201}
      - contains: {'message': 'order created'}
      
      # 验证数据库中订单是否创建
      - db: select * from orders where product_id=101
      
      # 验证库存是否扣减
      - db: select quantity from products where id=101

4.4 场景四：文件上传与数据导入

# data/import_users.yaml
baseInfo:
  api_name: 批量导入用户
  url: /api/users/import
  method: post

testCase:
  - case_name: 导入用户列表 Excel 文件
    files:
      file: ./data/users_import.xlsx
    
    data:
      import_type: user_list
      skip_errors: true
    
    validation:
      - contains: {status_code: 200}
      - contains: {'message': 'import success'}
      
      # 验证导入记录数
      - rv: {'imported_count': 100}
      
      # 验证数据库中用户数量
      - db: select count(*) from users where created_date=today()

---

五、框架代码走查

5.1 RequestBase 类 - 核心请求处理

# base/apiutil.py
class RequestBase:
    """API 请求基础类 - 框架核心"""
    
    def __init__(self):
        self.run = SendRequest()
        self.conf = OperationConfig()
        self.read = ReadYamlData()
        self.asserts = Assertions()
    
    def replace_load(self, data):
        """参数替换 - 处理所有 ${ } 表达式"""
        # 识别、提取、替换参数
        pass
    
    def specification_yaml(self, base_info, test_case):
        """
        接口请求处理 - 主方法
        流程：
        1. 替换参数
        2. 发送请求
        3. 提取数据
        4. 执行断言
        """
        # 1. 处理请求头和 Cookie
        header = self.replace_load(base_info['header'])
        cookie = eval(self.replace_load(base_info['cookies']))
        
        # 2. 处理请求参数
        for key, value in test_case.items():
            if key in ['data', 'json', 'params']:
                test_case[key] = self.replace_load(value)
        
        # 3. 处理文件上传
        if 'files' in test_case:
            files = {fk: open(fv, mode='rb') for fk, fv in test_case['files'].items()}
        
        # 4. 发送请求
        res = self.run.run_main(
            url=url,
            method=method,
            header=header,
            cookies=cookie,
            file=files,
            **test_case
        )
        
        # 5. 提取参数
        extract = test_case.get('extract')
        if extract:
            self.extract_data(extract, res.text)
        
        # 6. 执行断言
        validation = test_case.get('validation')
        if validation:
            self.asserts.assert_result(validation, res.json(), res.status_code)
    
    def extract_data(self, testcase_extract, response):
        """参数提取 - 支持 JSONPath 和正则"""
        for key, value in testcase_extract.items():
            if '$' in value:
                # JSONPath 提取
                ext_json = jsonpath.jsonpath(json.loads(response), value)[0]
                self.read.write_yaml_data({key: ext_json})
            else:
                # 正则提取
                ext_lst = re.search(value, response)
                self.read.write_yaml_data({key: ext_lst.group(1)})

5.2 Assertions 类 - 多种断言实现

# common/assertions.py
class Assertions:
    """断言类 - 支持 5 种断言模式"""
    
    def assert_result(self, expected, response, status_code):
        """主断言方法"""
        all_flag = 0
        
        for yq in expected:
            for key, value in yq.items():
                if key == "contains":
                    flag = self.contains_assert(value, response, status_code)
                elif key == "eq":
                    flag = self.equal_assert(value, response)
                elif key == 'ne':
                    flag = self.not_equal_assert(value, response)
                elif key == 'rv':
                    flag = self.assert_response_any(response, value)
                elif key == 'db':
                    flag = self.assert_mysql_data(value)
                
                all_flag += flag
        
        # 所有断言都必须通过
        assert all_flag == 0, "测试失败"
    
    def contains_assert(self, value, response, status_code):
        """包含断言"""
        # 实现逻辑...
        pass
    
    def equal_assert(self, value, response):
        """相等断言"""
        # 实现逻辑...
        pass
    
    def assert_mysql_data(self, sql):
        """数据库断言"""
        conn = ConnectMysql()
        result = conn.query_all(sql)
        return 0 if result else 1

5.3 DebugTalk 类 - 自定义函数库

# common/debugtalk.py
class DebugTalk:
    """自定义函数库 - 用于参数替换"""
    
    def get_extract_data(self, key, *args):
        """获取提取的参数"""
        extract_data = ReadYamlData().read_yaml_data()
        
        if args and args[0] == 'randoms':
            # 从列表中随机取
            return random.choice(extract_data[key])
        
        if args:
            # 支持多层级访问
            return extract_data[key][args[0]]
        
        return extract_data.get(key)
    
    def start_time(self):
        """当前时间"""
        return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    def end_time(self):
        """明天时间"""
        tomorrow = datetime.now() + timedelta(days=1)
        return tomorrow.strftime('%Y-%m-%d %H:%M:%S')
    
    def faker_phone(self):
        """生成随机手机号"""
        return Faker('zh_CN').phone_number()
    
    def faker_email(self):
        """生成随机邮箱"""
        return Faker().email()
    
    @staticmethod
    def md5(text):
        """MD5 加密"""
        return hashlib.md5(text.encode()).hexdigest()

---

六、框架的安装与使用

6.1 快速安装

# 克隆项目
git clone https://github.com/ismezed/python-pytest-.git
cd python-pytest-

# 安装依赖（使用国内镜像加速）
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt

# 验证安装
pytest --version

6.2 项目配置

# conf/config.ini
[api_envi]
host = http://127.0.0.1:8080
timeout = 60

[mysql]
host = 127.0.0.1
port = 3306
user = root
password = 123456
database = test_db

[mongodb]
host = 127.0.0.1
port = 27017
database = test_db

6.3 运行测试

# 运行所有测试
python run.py

# 或使用 Pytest 命令
pytest ./testcase -v --alluredir=./reports

# 生成并打开 Allure 报告
allure serve ./reports

# 运行特定测试文件
pytest ./testcase/test_login.py -v

# 运行特定测试用例
pytest ./testcase/test_login.py::TestLogin::test_login_success -v

# 使用 mark 运行指定标记的用例
pytest -m smoke ./testcase -v

6.4 查看 Allure 报告

# 自动生成并打开报告
allure serve ./report/temp

# 或手动生成报告
allure generate ./report/temp -o ./allure_report --clean
# 打开 allure_report/index.html

---

七、常见问题与调试技巧

7.1 问题一：参数替换失败

现象： 参数替换后仍然是原始值 ${get_extract_data(token)}

原因： 函数未定义或参数错误

解决方案：

# 1. 检查 debugtalk.py 中是否定义了函数
class DebugTalk:
    def get_extract_data(self, key):
        pass  # ✅ 确保函数存在

# 2. 检查 extract.yaml 中是否有对应的数据
# extract.yaml
token: "xxx_token_value"

# 3. 添加日志调试
def replace_load(self, data):
    print(f"替换前：{data}")
    # 替换逻辑
    print(f"替换后：{data}")

7.2 问题二：断言失败但看不到详细原因

现象： 测试失败，但不知道期望值和实际值是什么

解决方案：

# 在 assertions.py 中添加详细日志
def contains_assert(self, value, response, status_code):
    flag = 0
    for assert_key, assert_value in value.items():
        # 输出详细信息
        print(f"期望值：{assert_value}")
        print(f"实际值：{response}")
        
        # 使用日志记录
        logs.info(f"期望：{assert_value}，实际：{response}")

7.3 问题三：数据库连接失败

现象： 数据库断言报错 Connection refused

解决方案：

# 1. 检查配置
# conf/config.ini
[mysql]
host = 127.0.0.1
port = 3306
user = root
password = 123456  # ✅ 检查密码是否正确

# 2. 检查 MySQL 是否运行
mysql -u root -p -h 127.0.0.1

# 3. 检查防火墙
# 确保 3306 端口开放

7.4 问题四：文件上传路径问题

现象： 文件上传失败 FileNotFoundError

解决方案：

# ✅ 使用相对路径（相对于项目根目录）
files:
  file: ./data/users.xlsx

# ✅ 或使用绝对路径
files:
  file: /home/user/project/data/users.xlsx

# 不要使用以下方式
files:
  file: ./testcase/../data/users.xlsx  # ❌ 复杂的相对路径

---

八、性能优化与最佳实践

8.1 性能优化建议

1. 使用连接池

# common/sendrequest.py
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class SendRequest:
    def __init__(self):
        self.session = requests.Session()
        
        # 配置重试策略
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        
        # 使用连接池
        adapter = HTTPAdapter(
            max_retries=retry_strategy,
            pool_connections=10,
            pool_maxsize=20
        )
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

2. 并行执行测试

# 安装 pytest-xdist
pip install pytest-xdist

# 4 个进程并行执行
pytest -n 4 ./testcase

# 自动检测 CPU 核数
pytest -n auto ./testcase

3. 使用 Mock 数据

# 避免真实数据库操作
validation:
  # ❌ 不好的做法：每次都查询数据库
  - db: select * from users where id=1
  
  # ✅ 好的做法：Mock 响应数据
  - rv: {"user_id": 1, "name": "test"}

8.2 最佳实践

1. 合理组织测试用例

data/
├── login/               # 登录相关接口
│   ├── login.yaml
│   ├── logout.yaml
│   └── refresh_token.yaml
├── user/                # 用户相关接口
│   ├── create_user.yaml
│   ├── update_user.yaml
│   └── delete_user.yaml
└── order/               # 订单相关接口
    ├── create_order.yaml
    ├── query_orders.yaml
    └── cancel_order.yaml

2. 使用环境变量管理配置

# conf/setting.py
import os

# 从环境变量读取配置
API_HOST = os.getenv('API_HOST', 'http://127.0.0.1:8080')
DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
DB_PORT = int(os.getenv('DB_PORT', '3306'))

# 这样可以支持不同的环境
# 开发环境：export API_HOST=http://dev.example.com
# 测试环境：export API_HOST=http://test.example.com
# 生产环境：export API_HOST=http://prod.example.com

3. 添加测试标记

# conftest.py
import pytest

@pytest.mark.smoke         # 冒烟测试
@pytest.mark.regression    # 回归测试
@pytest.mark.performance   # 性能测试

# 运行指定标记的用例
pytest -m smoke ./testcase
pytest -m "not slow" ./testcase

4. 详细的日志记录

# common/recordlog.py
import logging

logger = logging.getLogger(__name__)

logger.info(f"发送请求：{method} {url}")
logger.info(f"请求头：{header}")
logger.info(f"请求体：{data}")
logger.info(f"响应状态码：{status_code}")
logger.info(f"响应体：{response}")
logger.info(f"提取参数：{extract_data}")

---

九、总结

框架核心优势

特性	优势	应用场景
配置化测试	无需代码，YAML 定义	降低学习成本
参数依赖管理	支持接口间参数传递	复杂业务流程测试
多种断言模式	灵活的断言方式	各种场景验证
数据库集成	原生支持多种数据库	数据一致性验证
智能参数替换	反射机制动态替换	动态测试数据生成
完整报告系统	Allure + 钉钉通知	测试结果追踪

适用场景

✅ API 功能测试 - 验证接口功能正确性
✅ 集成测试 - 多个接口的业务流程测试
✅ 回归测试 - 变更后的影响范围测试
✅ 性能测试 - 接口响应时间验证
✅ 数据验证 - 数据库数据一致性验证
✅ 异常测试 - 异常场景和边界值测试

与其他框架的对比

框架	优势	劣势
本框架（YAML+Pytest）	配置化、无代码、易维护	学习曲线，需要了解 Pytest
Postman	UI 友好、易上手	无参��依赖、难以维护大量用例
Robot Framework	关键词驱动、功能全	过于复杂、学习成本高
Rest Assured（Java）	类型安全、性能好	需要编程、维护成本高

---

十、快速开始模板

最小化示例

# data/api_example.yaml
baseInfo:
  api_name: 获取用户信息
  url: /api/users/1
  method: get
  header:
    Content-Type: application/json

testCase:
  - case_name: 获取用户信息成功
    validation:
      - contains: {status_code: 200}
      - contains: {'id': 1}
      - rv: {'name': 'admin'}

完整示例

# data/user_management.yaml
baseInfo:
  api_name: 用户管理
  url: /api/users
  method: post
  header:
    Content-Type: application/json
    Authorization: Bearer ${get_extract_data(token)}

testCase:
  # 创建用户
  - case_name: 创建新用户
    data:
      name: ${faker_name()}
      email: ${faker_email()}
      age: 18
    
    validation:
      - contains: {status_code: 201}
      - contains: {'message': 'success'}
    
    extract:
      user_id: $.data.id
  
  # 查询用户
  - case_name: 查询新创建的用户
    data:
      id: ${get_extract_data(user_id)}
    
    validation:
      - contains: {status_code: 200}
      - rv: {'id': ${get_extract_data(user_id)}}

---

参考资源

• 📚 项目 GitHub
• 📚 Pytest 官方文档
• 📚 JSONPath 在线测试
• 📚 Allure 官方文档

---

结语： API 自动化测试是保证 API 质量的重要手段。一个好的测试框架应该让写测试变得简单，这个框架正是为此而设计的。