15.pytest的yaml

作者

Pytest 与 YAML 详解

1. 什么是 YAML

1.1 YAML 简介

YAML（YAML Ain’t Markup Language）是一种人类可读的数据序列化标准，常用于配置文件、数据存储和配置文件。它的设计目标是让人类容易阅读和编写。

1.2 YAML 的特点

  • 简洁易读：语法简单，不需要复杂的符号
  • 层次清晰：使用缩进表示层级关系
  • 数据类型丰富：支持字符串、数字、布尔值、列表、字典等
  • 注释支持：可以使用 # 添加注释
  • 跨语言：多种编程语言都支持 YAML

1.3 YAML 在测试中的作用

在 pytest 测试中，YAML 主要用于：

  1. 存储测试数据：将测试用例数据与测试代码分离
  2. 配置文件：配置测试环境、数据库连接等
  3. 接口测试：存储 API 请求和响应数据
  4. 测试报告：生成测试报告的数据格式

2. YAML 基础语法

2.1 基本数据类型

2.1.1 字符串（String）

# 普通字符串
name: 张三
username: admin

# 包含特殊字符的字符串（使用引号）
message: "Hello, World!"
description: '这是一个测试'

# 多行字符串（使用 | 保留换行）
content: |
  这是第一行
  这是第二行
  这是第三行

# 多行字符串（使用 > 折叠换行）
summary: >
  这是一段很长的文本
  会被折叠成一行
  但保留空格

2.1.2 数字（Number）

# 整数
age: 25
count: 100

# 浮点数
price: 99.99
score: 98.5

# 科学计数法
large_number: 1.23e+10

2.1.3 布尔值（Boolean）

# 布尔值（多种写法）
is_active: true
is_deleted: false

# 也可以写成
enabled: True
disabled: False

# 或者
on: ON
off: OFF

2.1.4 空值（Null）

# 空值
middle_name: null
description: ~
empty_value:

2.2 集合类型

2.2.1 列表（List）

# 行内列表
fruits: [apple, banana, orange]

# 多行列表（使用 -）
fruits:
  - apple
  - banana
  - orange

# 嵌套列表
matrix:
  - [1, 2, 3]
  - [4, 5, 6]
  - [7, 8, 9]

# 混合类型列表
mixed_list:
  - 字符串
  - 123
  - true
  - null

2.2.2 字典（Dictionary/Map）

# 行内字典
person: {name: 张三, age: 25, city: 北京}

# 多行字典
person:
  name: 张三
  age: 25
  city: 北京
  email: zhangsan@example.com

# 嵌套字典
user:
  personal_info:
    name: 张三
    age: 25
  contact_info:
    email: zhangsan@example.com
    phone: 13800138000

2.3 缩进规则

重要提示：YAML 使用空格进行缩进，不能使用 Tab！

# ✅ 正确：使用空格缩进
user:
  name: 张三
  address:
    city: 北京
    street: 中关村大街

# ❌ 错误：使用 Tab 缩进（会导致解析错误）
user:
    name: 张三  # Tab 缩进，会报错

缩进规则：

  • 同一层级使用相同的缩进
  • 通常使用 2 个空格作为一级缩进
  • 子元素比父元素多一级缩进

2.4 注释

# 这是单行注释

user:
  name: 张三  # 行内注释
  age: 25

  # 这是多行注释
  # 可以写多行
  email: zhangsan@example.com

2.5 引用和锚点

# 定义锚点（使用 &）
defaults: &defaults
  timeout: 30
  retry: 3

# 引用锚点（使用 <<）
test_config:
  <<: *defaults
  url: https://api.example.com

# 结果等同于：
# test_config:
#   timeout: 30
#   retry: 3
#   url: https://api.example.com

3. 在 Python 中读取 YAML 文件

3.1 安装 PyYAML

在使用 YAML 之前，需要先安装 PyYAML 库：

pip install PyYAML

3.2 基本读取方法

3.2.1 使用 yaml.safe_load()

推荐使用 safe_load()，它只加载基本的 YAML 标签，更安全。

import yaml

# 读取 YAML 文件
with open('config.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)
    print(data)

示例 YAML 文件（config.yaml）：

database:
  host: localhost
  port: 3306
  username: root
  password: 123456

api:
  base_url: https://api.example.com
  timeout: 30

读取结果：

{
    'database': {
        'host': 'localhost',
        'port': 3306,
        'username': 'root',
        'password': '123456'
    },
    'api': {
        'base_url': 'https://api.example.com',
        'timeout': 30
    }
}

3.2.2 使用 yaml.load()

不推荐使用，因为它可能执行任意 Python 代码，存在安全风险。

import yaml

# 不推荐：可能存在安全风险
with open('config.yaml', 'r', encoding='utf-8') as f:
    data = yaml.load(f, Loader=yaml.FullLoader)

3.3 读取多个文档

YAML 文件可以包含多个文档，使用 --- 分隔：

# 第一个文档
---
name: 文档1
value: 100

# 第二个文档
---
name: 文档2
value: 200

读取多个文档：

import yaml

with open('multi_doc.yaml', 'r', encoding='utf-8') as f:
    documents = list(yaml.safe_load_all(f))
    for doc in documents:
        print(doc)

3.4 写入 YAML 文件

import yaml

data = {
    'name': '张三',
    'age': 25,
    'hobbies': ['读书', '游泳', '编程']
}

# 写入 YAML 文件
with open('output.yaml', 'w', encoding='utf-8') as f:
    yaml.dump(data, f, allow_unicode=True, default_flow_style=False)

生成的 YAML 文件：

age: 25
hobbies:
  - 读书
  - 游泳
  - 编程
name: 张三

参数说明：

  • allow_unicode=True：允许 Unicode 字符（中文等）
  • default_flow_style=False：使用块样式（多行），而不是流样式（单行）

4. Pytest 中使用 YAML 存储测试数据

4.1 为什么使用 YAML 存储测试数据

4.1.1 数据与代码分离

优点：

  • 测试代码更简洁
  • 测试数据易于维护
  • 非技术人员也可以修改测试数据
  • 数据可以复用

对比示例：

❌ 不使用 YAML（数据硬编码）：

def test_login():
    # 数据硬编码在代码中
    test_cases = [
        ("admin", "admin123", True),
        ("user", "user123", True),
        ("invalid", "wrong", False)
    ]
    for username, password, expected in test_cases:
        result = login(username, password)
        assert result == expected

✅ 使用 YAML（数据分离）：

import pytest
import yaml

# 从 YAML 文件读取数据
with open('test_data.yaml', 'r', encoding='utf-8') as f:
    test_data = yaml.safe_load(f)

@pytest.mark.parametrize("username, password, expected", 
                         test_data['login_cases'])
def test_login(username, password, expected):
    result = login(username, password)
    assert result == expected

4.2 基本使用示例

4.2.1 创建 YAML 数据文件

创建 test_data.yaml：

# 登录测试数据
login_cases:
  - username: admin
    password: admin123
    expected: true
    description: "管理员登录成功"

  - username: user
    password: user123
    expected: true
    description: "普通用户登录成功"

  - username: invalid
    password: wrong
    expected: false
    description: "无效用户名密码登录失败"

  - username: admin
    password: wrong
    expected: false
    description: "正确用户名错误密码登录失败"

# 用户注册测试数据
register_cases:
  - username: newuser1
    password: password123
    email: newuser1@example.com
    expected: true

  - username: admin  # 已存在的用户名
    password: password123
    email: admin@example.com
    expected: false

4.2.2 在测试中读取 YAML 数据

创建 test_login.py：

import pytest
import yaml
import os

def load_yaml_data(file_name, key=None):
    """
    加载 YAML 文件数据

    参数:
        file_name: YAML 文件名
        key: 要获取的键名，如果为 None 则返回全部数据

    返回:
        字典或列表数据
    """
    # 获取当前文件所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 构建文件路径
    file_path = os.path.join(current_dir, file_name)

    # 读取 YAML 文件
    with open(file_path, 'r', encoding='utf-8') as f:
        data = yaml.safe_load(f)

    # 如果指定了 key，返回对应的值
    if key:
        return data.get(key)
    return data

# 加载登录测试数据
login_data = load_yaml_data('test_data.yaml', 'login_cases')

# 使用参数化装饰器
@pytest.mark.parametrize("case", login_data)
def test_login(case):
    """
    测试登录功能

    参数:
        case: 测试用例字典，包含 username, password, expected, description
    """
    username = case['username']
    password = case['password']
    expected = case['expected']
    description = case.get('description', '')

    print(f"n测试用例：{description}")
    print(f"用户名：{username}, 密码：{password}")

    # 模拟登录函数（实际项目中替换为真实的登录函数）
    result = login(username, password)

    # 断言
    assert result == expected, f"登录结果不符合预期：期望 {expected}，实际 {result}"

def login(username, password):
    """
    模拟登录函数
    实际项目中应该调用真实的登录接口
    """
    # 模拟验证逻辑
    valid_users = {
        'admin': 'admin123',
        'user': 'user123'
    }
    return valid_users.get(username) == password

if __name__ == '__main__':
    pytest.main([__file__, '-v'])

4.2.3 运行测试

pytest test_login.py -v

输出示例：

test_login.py::test_login[case0] PASSED
test_login.py::test_login[case1] PASSED
test_login.py::test_login[case2] PASSED
test_login.py::test_login[case3] PASSED

4.3 使用 @pytest.mark.parametrize 的多种方式

4.3.1 方式一：直接传递字典列表

import pytest
import yaml

# 加载数据
with open('test_data.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)
    login_cases = data['login_cases']

# 参数化：传递整个字典
@pytest.mark.parametrize("case", login_cases)
def test_login(case):
    username = case['username']
    password = case['password']
    expected = case['expected']
    # 测试逻辑...

4.3.2 方式二：解构字典字段

import pytest
import yaml

# 加载数据
with open('test_data.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)
    login_cases = data['login_cases']

# 提取字段并转换为元组列表
test_data = [
    (case['username'], case['password'], case['expected'])
    for case in login_cases
]

# 参数化：传递多个参数
@pytest.mark.parametrize("username, password, expected", test_data)
def test_login(username, password, expected):
    result = login(username, password)
    assert result == expected

4.3.3 方式三：使用 ids 自定义测试名称

import pytest
import yaml

# 加载数据
with open('test_data.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)
    login_cases = data['login_cases']

# 提取测试 ID（用于显示在测试报告中）
test_ids = [case.get('description', f"case_{i}") 
            for i, case in enumerate(login_cases)]

@pytest.mark.parametrize("case", login_cases, ids=test_ids)
def test_login(case):
    username = case['username']
    password = case['password']
    expected = case['expected']
    result = login(username, password)
    assert result == expected

运行结果：

test_login.py::test_login[管理员登录成功] PASSED
test_login.py::test_login[普通用户登录成功] PASSED
test_login.py::test_login[无效用户名密码登录失败] PASSED
test_login.py::test_login[正确用户名错误密码登录失败] PASSED

4.4 复杂数据结构示例

4.4.1 嵌套数据结构

YAML 文件（complex_data.yaml）：

# 用户信息测试数据
user_test_cases:
  - user_id: 1
    personal_info:
      name: 张三
      age: 25
      gender: 男
    contact_info:
      email: zhangsan@example.com
      phone: 13800138000
      address:
        province: 北京
        city: 北京市
        district: 海淀区
        street: 中关村大街1号
    expected:
      status_code: 200
      message: "创建成功"

  - user_id: 2
    personal_info:
      name: 李四
      age: 30
      gender: 女
    contact_info:
      email: lisi@example.com
      phone: 13900139000
      address:
        province: 上海
        city: 上海市
        district: 浦东新区
        street: 陆家嘴环路1000号
    expected:
      status_code: 200
      message: "创建成功"

# API 测试数据
api_test_cases:
  - name: "获取用户信息"
    method: GET
    url: "/api/users/1"
    headers:
      Authorization: "Bearer token123"
      Content-Type: "application/json"
    params:
      include: ["profile", "contacts"]
    expected:
      status_code: 200
      data:
        user_id: 1
        name: "张三"

  - name: "创建用户"
    method: POST
    url: "/api/users"
    headers:
      Authorization: "Bearer token123"
      Content-Type: "application/json"
    body:
      name: "王五"
      age: 28
      email: "wangwu@example.com"
    expected:
      status_code: 201
      message: "创建成功"

测试代码：

import pytest
import yaml

# 加载复杂数据
with open('complex_data.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)

@pytest.mark.parametrize("user_case", data['user_test_cases'])
def test_create_user(user_case):
    """测试创建用户"""
    user_id = user_case['user_id']
    personal_info = user_case['personal_info']
    contact_info = user_case['contact_info']
    expected = user_case['expected']

    print(f"n测试用户 ID: {user_id}")
    print(f"姓名: {personal_info['name']}")
    print(f"邮箱: {contact_info['email']}")
    print(f"地址: {contact_info['address']['street']}")

    # 模拟创建用户
    result = create_user(personal_info, contact_info)

    assert result['status_code'] == expected['status_code']
    assert result['message'] == expected['message']

@pytest.mark.parametrize("api_case", data['api_test_cases'])
def test_api_request(api_case):
    """测试 API 请求"""
    method = api_case['method']
    url = api_case['url']
    headers = api_case.get('headers', {})
    params = api_case.get('params', {})
    body = api_case.get('body', {})
    expected = api_case['expected']

    print(f"n测试用例: {api_case['name']}")
    print(f"方法: {method}, URL: {url}")

    # 模拟 API 请求
    result = make_api_request(method, url, headers, params, body)

    assert result['status_code'] == expected['status_code']
    if 'data' in expected:
        assert result['data'] == expected['data']

def create_user(personal_info, contact_info):
    """模拟创建用户函数"""
    return {
        'status_code': 200,
        'message': '创建成功'
    }

def make_api_request(method, url, headers, params, body):
    """模拟 API 请求函数"""
    return {
        'status_code': 200,
        'data': {'user_id': 1, 'name': '张三'}
    }

4.5 使用 Fixture 加载 YAML 数据

4.5.1 在 conftest.py 中定义 Fixture

创建 conftest.py：

import pytest
import yaml
import os

@pytest.fixture(scope='session')
def yaml_data():
    """加载 YAML 测试数据，整个测试会话只加载一次"""
    file_path = os.path.join(os.path.dirname(__file__), 'test_data.yaml')
    with open(file_path, 'r', encoding='utf-8') as f:
        data = yaml.safe_load(f)
    return data

@pytest.fixture
def login_cases(yaml_data):
    """获取登录测试用例"""
    return yaml_data.get('login_cases', [])

@pytest.fixture
def register_cases(yaml_data):
    """获取注册测试用例"""
    return yaml_data.get('register_cases', [])

4.5.2 在测试中使用 Fixture

import pytest

def test_login_with_fixture(login_cases):
    """使用 Fixture 获取测试数据"""
    for case in login_cases:
        username = case['username']
        password = case['password']
        expected = case['expected']

        result = login(username, password)
        assert result == expected

@pytest.mark.parametrize("case", pytest.fixture(scope='function')(lambda: []))
def test_login_parametrize(login_cases):
    """结合参数化使用"""
    # 注意：这里需要特殊处理，因为 parametrize 不能直接使用 fixture
    pass

更好的方式：直接在测试中加载数据

import pytest
import yaml

@pytest.fixture(scope='module')
def test_data():
    """模块级别的 Fixture，加载测试数据"""
    with open('test_data.yaml', 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

def test_login(test_data):
    """使用 Fixture 获取数据"""
    login_cases = test_data['login_cases']

    for case in login_cases:
        username = case['username']
        password = case['password']
        expected = case['expected']

        result = login(username, password)
        assert result == expected

5. 使用 YAML 配置 Pytest

5.1 pytest.ini 配置文件

虽然 pytest.ini 通常使用 INI 格式，但我们可以使用 YAML 来存储测试配置数据。

5.1.1 创建配置文件（config.yaml）

# Pytest 测试配置
pytest_config:
  # 测试路径
  testpaths:
    - tests
    - integration_tests

  # 测试文件匹配模式
  python_files:
    - test_*.py
    - *_test.py

  # 测试类匹配模式
  python_classes:
    - Test*

  # 测试函数匹配模式
  python_functions:
    - test_*

  # 命令行选项
  addopts:
    - -v
    - --tb=short
    - --strict-markers
    - --disable-warnings

  # 标记定义
  markers:
    smoke: "冒烟测试"
    regression: "回归测试"
    api: "API 测试"
    ui: "UI 测试"
    slow: "慢速测试"

  # 日志配置
  log_cli: true
  log_cli_level: INFO
  log_cli_format: "%(asctime)s [%(levelname)s] %(message)s"
  log_cli_date_format: "%Y-%m-%d %H:%M:%S"

# 测试环境配置
test_environments:
  dev:
    base_url: "http://dev.example.com"
    database:
      host: "localhost"
      port: 3306
      name: "test_db"
    timeout: 30

  test:
    base_url: "http://test.example.com"
    database:
      host: "test-db.example.com"
      port: 3306
      name: "test_db"
    timeout: 60

  prod:
    base_url: "https://api.example.com"
    database:
      host: "prod-db.example.com"
      port: 3306
      name: "prod_db"
    timeout: 120

5.1.2 在 conftest.py 中加载配置

import pytest
import yaml
import os

def load_config():
    """加载 YAML 配置文件"""
    config_path = os.path.join(os.path.dirname(__file__), 'config.yaml')
    with open(config_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

@pytest.fixture(scope='session')
def test_config():
    """获取测试配置"""
    config = load_config()
    return config['pytest_config']

@pytest.fixture(scope='session')
def test_env():
    """获取测试环境配置"""
    import os
    env = os.getenv('TEST_ENV', 'dev')  # 默认使用 dev 环境
    config = load_config()
    return config['test_environments'][env]

@pytest.fixture(scope='session')
def base_url(test_env):
    """获取基础 URL"""
    return test_env['base_url']

@pytest.fixture(scope='session')
def db_config(test_env):
    """获取数据库配置"""
    return test_env['database']

5.1.3 在测试中使用配置

import pytest
import requests

def test_api_request(base_url):
    """使用配置的 base_url"""
    url = f"{base_url}/api/users/1"
    response = requests.get(url)
    assert response.status_code == 200

def test_database_connection(db_config):
    """使用配置的数据库信息"""
    host = db_config['host']
    port = db_config['port']
    db_name = db_config['name']

    print(f"连接数据库: {host}:{port}/{db_name}")
    # 实际项目中连接数据库...

5.2 环境变量配置

# environments.yaml
environments:
  dev:
    name: "开发环境"
    base_url: "http://dev.example.com"
    api_key: "dev_api_key_123"
    database_url: "mysql://localhost:3306/test_db"

  staging:
    name: "预发布环境"
    base_url: "http://staging.example.com"
    api_key: "staging_api_key_456"
    database_url: "mysql://staging-db:3306/staging_db"

  production:
    name: "生产环境"
    base_url: "https://api.example.com"
    api_key: "${PROD_API_KEY}"  # 从环境变量读取
    database_url: "${PROD_DB_URL}"  # 从环境变量读取

加载并处理环境变量：

import pytest
import yaml
import os
import re

def load_env_config(env_name='dev'):
    """加载环境配置，支持环境变量替换"""
    with open('environments.yaml', 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)

    env_config = config['environments'][env_name]

    # 替换环境变量
    def replace_env_vars(value):
        if isinstance(value, str):
            # 匹配 ${VAR_NAME} 格式
            pattern = r'${([^}]+)}'
            matches = re.findall(pattern, value)
            for var_name in matches:
                env_value = os.getenv(var_name)
                if env_value:
                    value = value.replace(f'${{{var_name}}}', env_value)
        return value

    # 递归处理所有值
    def process_dict(d):
        if isinstance(d, dict):
            return {k: process_dict(v) for k, v in d.items()}
        elif isinstance(d, list):
            return [process_dict(item) for item in d]
        else:
            return replace_env_vars(d)

    return process_dict(env_config)

@pytest.fixture(scope='session')
def env_config():
    """获取环境配置"""
    env = os.getenv('TEST_ENV', 'dev')
    return load_env_config(env)

6. 使用 YAML 进行接口测试

6.1 接口测试数据结构设计

6.1.1 API 测试用例 YAML 结构

创建 api_test_cases.yaml：

# API 测试用例配置
api_base_url: "https://api.example.com"

# 测试用例列表
test_cases:
  # 测试用例 1：获取用户信息
  - name: "获取用户信息 - 成功"
    description: "测试获取指定用户信息的接口"
    method: GET
    endpoint: "/api/users/{user_id}"
    path_params:
      user_id: 1
    query_params:
      include: ["profile", "contacts"]
    headers:
      Authorization: "Bearer {token}"
      Content-Type: "application/json"
    expected:
      status_code: 200
      response_time: 1000  # 毫秒
      schema:
        type: object
        properties:
          user_id:
            type: integer
          name:
            type: string
          email:
            type: string
      data:
        user_id: 1
        name: "张三"
    validate:
      - type: "status_code"
        expected: 200
      - type: "json_path"
        path: "$.user_id"
        expected: 1
      - type: "response_time"
        max: 1000

  # 测试用例 2：创建用户
  - name: "创建用户 - 成功"
    description: "测试创建新用户的接口"
    method: POST
    endpoint: "/api/users"
    headers:
      Authorization: "Bearer {token}"
      Content-Type: "application/json"
    body:
      name: "李四"
      age: 30
      email: "lisi@example.com"
      password: "password123"
    expected:
      status_code: 201
      response_time: 2000
      data:
        user_id: 2
        name: "李四"
        message: "创建成功"
    validate:
      - type: "status_code"
        expected: 201
      - type: "json_path"
        path: "$.user_id"
        expected_type: "integer"
      - type: "contains"
        field: "message"
        value: "创建成功"

  # 测试用例 3：更新用户信息
  - name: "更新用户信息 - 成功"
    description: "测试更新用户信息的接口"
    method: PUT
    endpoint: "/api/users/{user_id}"
    path_params:
      user_id: 1
    headers:
      Authorization: "Bearer {token}"
      Content-Type: "application/json"
    body:
      name: "张三（已更新）"
      age: 26
    expected:
      status_code: 200
      data:
        user_id: 1
        name: "张三（已更新）"
        age: 26
    validate:
      - type: "status_code"
        expected: 200
      - type: "json_path"
        path: "$.name"
        expected: "张三（已更新）"

  # 测试用例 4：删除用户
  - name: "删除用户 - 成功"
    description: "测试删除用户的接口"
    method: DELETE
    endpoint: "/api/users/{user_id}"
    path_params:
      user_id: 1
    headers:
      Authorization: "Bearer {token}"
    expected:
      status_code: 204
    validate:
      - type: "status_code"
        expected: 204

# 公共配置
common_config:
  timeout: 30
  retry_times: 3
  retry_delay: 1  # 秒
  default_headers:
    User-Agent: "pytest-api-test/1.0"
    Accept: "application/json"

6.2 接口测试框架实现

6.2.1 创建测试工具类

创建 api_test_utils.py：

import requests
import json
import time
from typing import Dict, Any, Optional
import jsonpath

class APITestClient:
    """API 测试客户端"""

    def __init__(self, base_url: str, default_headers: Optional[Dict] = None):
        """
        初始化 API 测试客户端

        参数:
            base_url: API 基础 URL
            default_headers: 默认请求头
        """
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        if default_headers:
            self.session.headers.update(default_headers)

    def request(self, method: str, endpoint: str, 
                path_params: Optional[Dict] = None,
                query_params: Optional[Dict] = None,
                headers: Optional[Dict] = None,
                body: Optional[Dict] = None,
                timeout: int = 30) -> requests.Response:
        """
        发送 HTTP 请求

        参数:
            method: HTTP 方法（GET, POST, PUT, DELETE 等）
            endpoint: API 端点
            path_params: 路径参数（用于替换 URL 中的 {param}）
            query_params: 查询参数
            headers: 请求头
            body: 请求体
            timeout: 超时时间（秒）

        返回:
            Response 对象
        """
        # 替换路径参数
        url = endpoint
        if path_params:
            for key, value in path_params.items():
                url = url.replace(f'{{{key}}}', str(value))

        # 构建完整 URL
        full_url = f"{self.base_url}{url}"

        # 准备请求头
        request_headers = {}
        if headers:
            request_headers.update(headers)

        # 发送请求
        start_time = time.time()
        try:
            response = self.session.request(
                method=method.upper(),
                url=full_url,
                params=query_params,
                headers=request_headers,
                json=body,
                timeout=timeout
            )
            response.elapsed_time = (time.time() - start_time) * 1000  # 转换为毫秒
            return response
        except requests.exceptions.RequestException as e:
            raise Exception(f"请求失败: {e}")

    def validate_response(self, response: requests.Response, 
                         validations: list) -> tuple[bool, list]:
        """
        验证响应

        参数:
            response: Response 对象
            validations: 验证规则列表

        返回:
            (是否通过, 错误信息列表)
        """
        errors = []

        for validation in validations:
            v_type = validation.get('type')

            if v_type == 'status_code':
                expected = validation.get('expected')
                if response.status_code != expected:
                    errors.append(
                        f"状态码验证失败: 期望 {expected}, 实际 {response.status_code}"
                    )

            elif v_type == 'json_path':
                path = validation.get('path')
                expected = validation.get('expected')
                expected_type = validation.get('expected_type')

                try:
                    json_data = response.json()
                    actual_value = jsonpath.jsonpath(json_data, path)

                    if actual_value:
                        actual_value = actual_value[0]
                        if expected is not None:
                            if actual_value != expected:
                                errors.append(
                                    f"JSON 路径 {path} 验证失败: "
                                    f"期望 {expected}, 实际 {actual_value}"
                                )
                        if expected_type:
                            if not isinstance(actual_value, eval(expected_type)):
                                errors.append(
                                    f"JSON 路径 {path} 类型验证失败: "
                                    f"期望 {expected_type}, 实际 {type(actual_value).__name__}"
                                )
                    else:
                        errors.append(f"JSON 路径 {path} 不存在")
                except json.JSONDecodeError:
                    errors.append("响应不是有效的 JSON 格式")

            elif v_type == 'contains':
                field = validation.get('field')
                value = validation.get('value')
                try:
                    json_data = response.json()
                    if field in json_data:
                        if value not in str(json_data[field]):
                            errors.append(
                                f"字段 {field} 不包含值 {value}"
                            )
                    else:
                        errors.append(f"字段 {field} 不存在")
                except json.JSONDecodeError:
                    errors.append("响应不是有效的 JSON 格式")

            elif v_type == 'response_time':
                max_time = validation.get('max')
                if hasattr(response, 'elapsed_time'):
                    if response.elapsed_time > max_time:
                        errors.append(
                            f"响应时间超时: 期望 <= {max_time}ms, "
                            f"实际 {response.elapsed_time:.2f}ms"
                        )

        return len(errors) == 0, errors

6.2.2 创建测试用例

创建 test_api.py：

import pytest
import yaml
import os
from api_test_utils import APITestClient

def load_api_test_cases():
    """加载 API 测试用例"""
    file_path = os.path.join(os.path.dirname(__file__), 'api_test_cases.yaml')
    with open(file_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

@pytest.fixture(scope='session')
def api_config():
    """获取 API 配置"""
    config = load_api_test_cases()
    return config

@pytest.fixture(scope='session')
def api_client(api_config):
    """创建 API 客户端"""
    base_url = api_config['api_base_url']
    default_headers = api_config.get('common_config', {}).get('default_headers', {})
    return APITestClient(base_url, default_headers)

@pytest.fixture(scope='session')
def auth_token():
    """获取认证 Token（示例）"""
    # 实际项目中应该调用登录接口获取 token
    return "mock_token_12345"

@pytest.mark.parametrize("test_case", 
                         load_api_test_cases()['test_cases'],
                         ids=lambda case: case['name'])
def test_api(api_client, api_config, auth_token, test_case):
    """
    执行 API 测试用例

    参数:
        api_client: API 客户端
        api_config: API 配置
        auth_token: 认证 Token
        test_case: 测试用例数据
    """
    # 替换 Token
    headers = test_case.get('headers', {})
    if headers:
        headers = {k: v.replace('{token}', auth_token) 
                   for k, v in headers.items()}

    # 发送请求
    response = api_client.request(
        method=test_case['method'],
        endpoint=test_case['endpoint'],
        path_params=test_case.get('path_params'),
        query_params=test_case.get('query_params'),
        headers=headers,
        body=test_case.get('body'),
        timeout=api_config.get('common_config', {}).get('timeout', 30)
    )

    # 验证响应
    validations = test_case.get('validate', [])
    passed, errors = api_client.validate_response(response, validations)

    # 打印响应信息（用于调试）
    print(f"n测试用例: {test_case['name']}")
    print(f"请求 URL: {response.url}")
    print(f"状态码: {response.status_code}")
    print(f"响应时间: {response.elapsed_time:.2f}ms")
    if response.text:
        try:
            print(f"响应内容: {response.json()}")
        except:
            print(f"响应内容: {response.text}")

    # 断言
    assert passed, f"验证失败:n" + "n".join(errors)

    # 验证状态码（如果 expected 中有定义）
    if 'expected' in test_case:
        expected_status = test_case['expected'].get('status_code')
        if expected_status:
            assert response.status_code == expected_status, 
                f"状态码不符合预期: 期望 {expected_status}, 实际 {response.status_code}"

6.3 接口关联测试

6.3.1 接口关联 YAML 配置

创建 api_chain_test.yaml：

# 接口关联测试用例
test_chains:
  # 测试链 1：用户注册 -> 登录 -> 获取用户信息
  - name: "用户注册登录流程"
    description: "测试完整的用户注册和登录流程"
    steps:
      # 步骤 1：注册用户
      - step_name: "注册用户"
        method: POST
        endpoint: "/api/users/register"
        body:
          username: "testuser_{timestamp}"
          password: "password123"
          email: "testuser_{timestamp}@example.com"
        extract:
          # 提取响应中的数据，供后续步骤使用
          user_id: "$.user_id"
          username: "$.username"
        validate:
          - type: "status_code"
            expected: 201

      # 步骤 2：用户登录
      - step_name: "用户登录"
        method: POST
        endpoint: "/api/users/login"
        body:
          username: "{username}"  # 使用上一步提取的 username
          password: "password123"
        extract:
          token: "$.token"
          user_id: "$.user_id"
        validate:
          - type: "status_code"
            expected: 200

      # 步骤 3：获取用户信息
      - step_name: "获取用户信息"
        method: GET
        endpoint: "/api/users/{user_id}"  # 使用上一步提取的 user_id
        headers:
          Authorization: "Bearer {token}"  # 使用上一步提取的 token
        validate:
          - type: "status_code"
            expected: 200
          - type: "json_path"
            path: "$.user_id"
            expected: "{user_id}"  # 验证 user_id 匹配

  # 测试链 2：创建订单 -> 支付 -> 查询订单
  - name: "订单创建支付流程"
    description: "测试订单创建和支付流程"
    steps:
      - step_name: "创建订单"
        method: POST
        endpoint: "/api/orders"
        headers:
          Authorization: "Bearer {token}"
        body:
          product_id: 1
          quantity: 2
          price: 99.99
        extract:
          order_id: "$.order_id"
          total_amount: "$.total_amount"
        validate:
          - type: "status_code"
            expected: 201

      - step_name: "支付订单"
        method: POST
        endpoint: "/api/orders/{order_id}/pay"
        headers:
          Authorization: "Bearer {token}"
        body:
          payment_method: "credit_card"
          amount: "{total_amount}"
        extract:
          payment_id: "$.payment_id"
          payment_status: "$.status"
        validate:
          - type: "status_code"
            expected: 200
          - type: "json_path"
            path: "$.status"
            expected: "paid"

      - step_name: "查询订单状态"
        method: GET
        endpoint: "/api/orders/{order_id}"
        headers:
          Authorization: "Bearer {token}"
        validate:
          - type: "status_code"
            expected: 200
          - type: "json_path"
            path: "$.status"
            expected: "paid"

6.2.2 接口关联测试实现

创建 test_api_chain.py：

import pytest
import yaml
import os
import re
import time
from api_test_utils import APITestClient

def load_api_chain_tests():
    """加载接口关联测试用例"""
    file_path = os.path.join(os.path.dirname(__file__), 'api_chain_test.yaml')
    with open(file_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

def extract_json_path(data, path):
    """从 JSON 数据中提取路径值"""
    import jsonpath
    result = jsonpath.jsonpath(data, path)
    return result[0] if result else None

def replace_variables(text, variables):
    """替换文本中的变量"""
    pattern = r'{([^}]+)}'

    def replace(match):
        var_name = match.group(1)
        # 特殊处理 timestamp
        if var_name == 'timestamp':
            return str(int(time.time()))
        return str(variables.get(var_name, match.group(0)))

    return re.sub(pattern, replace, text)

@pytest.fixture(scope='session')
def api_client():
    """创建 API 客户端"""
    base_url = "https://api.example.com"
    return APITestClient(base_url)

@pytest.mark.parametrize("test_chain",
                         load_api_chain_tests()['test_chains'],
                         ids=lambda chain: chain['name'])
def test_api_chain(api_client, test_chain):
    """
    执行接口关联测试

    参数:
        api_client: API 客户端
        test_chain: 测试链数据
    """
    # 存储提取的变量
    variables = {}

    print(f"n{'='*60}")
    print(f"测试链: {test_chain['name']}")
    print(f"描述: {test_chain['description']}")
    print(f"{'='*60}")

    # 执行每个步骤
    for step in test_chain['steps']:
        step_name = step['step_name']
        print(f"n执行步骤: {step_name}")

        # 替换变量
        endpoint = replace_variables(step['endpoint'], variables)
        headers = step.get('headers', {})
        if headers:
            headers = {k: replace_variables(v, variables) 
                      for k, v in headers.items()}

        body = step.get('body', {})
        if body:
            body = {k: replace_variables(str(v), variables) 
                   if isinstance(v, str) else v
                   for k, v in body.items()}

        # 发送请求
        response = api_client.request(
            method=step['method'],
            endpoint=endpoint,
            headers=headers,
            body=body
        )

        print(f"  请求 URL: {response.url}")
        print(f"  状态码: {response.status_code}")

        # 提取数据
        extract_rules = step.get('extract', {})
        if extract_rules:
            try:
                response_json = response.json()
                for var_name, json_path in extract_rules.items():
                    value = extract_json_path(response_json, json_path)
                    variables[var_name] = value
                    print(f"  提取变量 {var_name} = {value}")
            except:
                pass

        # 验证响应
        validations = step.get('validate', [])
        # 替换验证规则中的变量
        for validation in validations:
            if 'expected' in validation:
                expected = validation['expected']
                if isinstance(expected, str) and expected.startswith('{'):
                    validation['expected'] = variables.get(
                        expected.strip('{}'), expected
                    )

        passed, errors = api_client.validate_response(response, validations)

        # 断言
        assert passed, f"步骤 '{step_name}' 验证失败:n" + "n".join(errors)

        print(f"  ✓ 步骤 '{step_name}' 执行成功")

    print(f"n{'='*60}")
    print(f"测试链 '{test_chain['name']}' 全部通过")
    print(f"{'='*60}")

7. 高级用法和最佳实践

7.1 数据驱动测试的完整示例

7.1.1 完整的测试数据文件结构

创建 complete_test_data.yaml：

# 完整的测试数据文件示例

# 元数据
metadata:
  version: "1.0"
  author: "测试团队"
  created_date: "2024-01-01"
  description: "完整的测试数据文件"

# 环境配置
environments:
  dev:
    base_url: "http://dev.example.com"
    database: "test_db_dev"

  test:
    base_url: "http://test.example.com"
    database: "test_db_test"

# 用户测试数据
users:
  admin:
    username: "admin"
    password: "admin123"
    role: "administrator"
    permissions:
      - "read"
      - "write"
      - "delete"

  normal_user:
    username: "user"
    password: "user123"
    role: "user"
    permissions:
      - "read"

# 登录测试用例
login_test_cases:
  - name: "管理员登录成功"
    user_type: "admin"
    expected:
      status_code: 200
      role: "administrator"
      has_permissions: ["read", "write", "delete"]

  - name: "普通用户登录成功"
    user_type: "normal_user"
    expected:
      status_code: 200
      role: "user"
      has_permissions: ["read"]

  - name: "错误密码登录失败"
    username: "admin"
    password: "wrong_password"
    expected:
      status_code: 401
      error_message: "用户名或密码错误"

# API 测试数据
api_tests:
  get_user:
    - user_id: 1
      expected_name: "张三"
      expected_email: "zhangsan@example.com"

    - user_id: 2
      expected_name: "李四"
      expected_email: "lisi@example.com"

  create_user:
    - data:
        name: "王五"
        age: 28
        email: "wangwu@example.com"
      expected_status: 201

    - data:
        name: ""  # 空名称，应该失败
        age: 28
        email: "invalid@example.com"
      expected_status: 400
      expected_error: "名称不能为空"

7.1.2 使用数据驱动测试

import pytest
import yaml
import os

class TestDataManager:
    """测试数据管理器"""

    def __init__(self, yaml_file):
        """初始化数据管理器"""
        self.yaml_file = yaml_file
        self._data = None
        self._load_data()

    def _load_data(self):
        """加载 YAML 数据"""
        file_path = os.path.join(os.path.dirname(__file__), self.yaml_file)
        with open(file_path, 'r', encoding='utf-8') as f:
            self._data = yaml.safe_load(f)

    def get(self, *keys):
        """
        获取嵌套的数据

        参数:
            *keys: 键的路径，例如 get('users', 'admin')

        返回:
            对应的值
        """
        value = self._data
        for key in keys:
            if isinstance(value, dict):
                value = value.get(key)
            else:
                return None
        return value

    def get_user(self, user_type):
        """获取用户数据"""
        return self.get('users', user_type)

    def get_login_cases(self):
        """获取登录测试用例"""
        return self.get('login_test_cases') or []

    def get_api_tests(self, api_name):
        """获取 API 测试数据"""
        return self.get('api_tests', api_name) or []

# 创建全局数据管理器
test_data = TestDataManager('complete_test_data.yaml')

@pytest.fixture(scope='session')
def data_manager():
    """提供数据管理器 Fixture"""
    return test_data

def test_login_with_data_manager(data_manager):
    """使用数据管理器进行登录测试"""
    login_cases = data_manager.get_login_cases()

    for case in login_cases:
        # 获取用户数据
        if 'user_type' in case:
            user = data_manager.get_user(case['user_type'])
            username = user['username']
            password = user['password']
        else:
            username = case['username']
            password = case['password']

        expected = case['expected']

        # 执行登录（模拟）
        result = login(username, password)

        # 验证
        assert result['status_code'] == expected['status_code']
        if 'role' in expected:
            assert result['role'] == expected['role']

def login(username, password):
    """模拟登录函数"""
    users_db = {
        'admin': {'password': 'admin123', 'role': 'administrator'},
        'user': {'password': 'user123', 'role': 'user'}
    }

    if username in users_db and users_db[username]['password'] == password:
        return {
            'status_code': 200,
            'role': users_db[username]['role']
        }
    else:
        return {
            'status_code': 401,
            'error_message': '用户名或密码错误'
        }

7.2 错误处理和调试

7.2.1 YAML 文件错误处理

import pytest
import yaml
import os

def safe_load_yaml(file_path, default=None):
    """
    安全加载 YAML 文件，包含错误处理

    参数:
        file_path: YAML 文件路径
        default: 加载失败时的默认值

    返回:
        加载的数据或默认值
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
            if data is None:
                return default
            return data
    except FileNotFoundError:
        pytest.fail(f"YAML 文件不存在: {file_path}")
    except yaml.YAMLError as e:
        pytest.fail(f"YAML 文件格式错误: {e}")
    except Exception as e:
        pytest.fail(f"加载 YAML 文件时发生错误: {e}")

def validate_yaml_structure(data, required_keys):
    """
    验证 YAML 数据结构

    参数:
        data: YAML 数据
        required_keys: 必需的键列表

    返回:
        是否有效
    """
    if not isinstance(data, dict):
        return False

    for key in required_keys:
        if key not in data:
            return False

    return True

# 使用示例
def test_with_error_handling():
    """带错误处理的测试"""
    file_path = 'test_data.yaml'
    data = safe_load_yaml(file_path, default={})

    # 验证数据结构
    required_keys = ['login_cases', 'register_cases']
    assert validate_yaml_structure(data, required_keys), 
        "YAML 文件缺少必需的键"

    # 使用数据...

7.2.2 调试技巧

import pytest
import yaml

@pytest.fixture(autouse=True)
def debug_yaml_data(request):
    """自动打印 YAML 数据（用于调试）"""
    # 只在 verbose 模式下打印
    if request.config.getoption("-v"):
        # 可以在这里添加调试代码
        pass

def test_with_debug():
    """带调试信息的测试"""
    with open('test_data.yaml', 'r', encoding='utf-8') as f:
        data = yaml.safe_load(f)

    # 打印数据结构（用于调试）
    import json
    print("nYAML 数据结构:")
    print(json.dumps(data, indent=2, ensure_ascii=False))

    # 测试逻辑...

7.3 性能优化

7.3.1 缓存 YAML 数据

import pytest
import yaml
import os
from functools import lru_cache

@lru_cache(maxsize=10)
def load_yaml_cached(file_path):
    """缓存 YAML 文件加载结果"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

@pytest.fixture(scope='session')
def cached_test_data():
    """缓存的测试数据"""
    file_path = os.path.join(os.path.dirname(__file__), 'test_data.yaml')
    return load_yaml_cached(file_path)

7.3.2 按需加载数据

import pytest
import yaml

class LazyYAMLLoader:
    """延迟加载 YAML 数据"""

    def __init__(self, file_path):
        self.file_path = file_path
        self._data = None

    @property
    def data(self):
        """延迟加载数据"""
        if self._data is None:
            with open(self.file_path, 'r', encoding='utf-8') as f:
                self._data = yaml.safe_load(f)
        return self._data

    def get(self, key, default=None):
        """获取指定键的数据"""
        return self.data.get(key, default)

# 使用示例
loader = LazyYAMLLoader('test_data.yaml')

def test_with_lazy_loading():
    """使用延迟加载"""
    # 只有在访问时才加载数据
    login_cases = loader.get('login_cases', [])
    # 测试逻辑...

8. 常见问题和解决方案

8.1 YAML 文件编码问题

问题：YAML 文件包含中文时出现乱码

解决方案：

# ✅ 正确：指定 UTF-8 编码
with open('test_data.yaml', 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)

# ❌ 错误：不指定编码（Windows 上可能出错）
with open('test_data.yaml', 'r') as f:
    data = yaml.safe_load(f)

8.2 YAML 缩进问题

问题：YAML 解析错误，提示缩进问题

解决方案：

# ✅ 正确：使用空格缩进（2 个或 4 个空格）
user:
  name: 张三
  age: 25

# ❌ 错误：使用 Tab 缩进
user:
    name: 张三  # Tab 缩进会导致错误
    age: 25

检查缩进的方法：

# 在编辑器中显示空格和 Tab
# VSCode: 设置 "editor.renderWhitespace": "all"
# PyCharm: 设置 "Show whitespaces"

8.3 数据类型问题

问题：YAML 中的数字被解析为字符串

解决方案：

# YAML 会自动识别类型
age: 25        # 整数
price: 99.99   # 浮点数
count: "100"   # 字符串（使用引号）

# 在 Python 中转换类型
import yaml

data = yaml.safe_load("""
age: 25
count: "100"
""")

age = int(data['age'])  # 已经是整数，不需要转换
count = int(data['count'])  # 字符串需要转换

8.4 空值处理

问题：YAML 中的 null 值处理

解决方案：

# YAML 中的空值
middle_name: null
description: ~
empty_value: 

# 在 Python 中检查
if data.get('middle_name') is None:
    print("middle_name 是空值")

8.5 特殊字符处理

问题：YAML 中包含特殊字符（如冒号、引号等）

解决方案：

# 使用引号包裹包含特殊字符的字符串
message: "Hello: World"
description: 'It's a test'
url: "https://example.com?param=value&other=123"

# 多行字符串
content: |
  这是多行内容
  可以包含: 冒号
  可以包含"引号"
  可以包含'单引号'

9. 实战项目示例

9.1 完整的测试项目结构

project/
├── tests/
│   ├── conftest.py              # Pytest 配置和 Fixture
│   ├── test_data/
│   │   ├── login_data.yaml      # 登录测试数据
│   │   ├── user_data.yaml       # 用户测试数据
│   │   └── api_data.yaml        # API 测试数据
│   ├── test_login.py            # 登录测试
│   ├── test_user.py             # 用户测试
│   └── test_api.py              # API 测试
├── config/
│   ├── config.yaml              # 配置文件
│   └── environments.yaml        # 环境配置
├── utils/
│   ├── yaml_loader.py           # YAML 加载工具
│   └── api_client.py            # API 客户端
└── pytest.ini                   # Pytest 配置

9.2 conftest.py 完整示例

import pytest
import yaml
import os
from pathlib import Path

# 项目根目录
PROJECT_ROOT = Path(__file__).parent.parent

@pytest.fixture(scope='session')
def project_root():
    """返回项目根目录"""
    return PROJECT_ROOT

@pytest.fixture(scope='session')
def load_yaml():
    """YAML 加载函数 Fixture"""
    def _load(file_path, key=None):
        """加载 YAML 文件"""
        full_path = PROJECT_ROOT / file_path
        with open(full_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
        return data.get(key) if key else data
    return _load

@pytest.fixture(scope='session')
def test_config(load_yaml):
    """加载测试配置"""
    return load_yaml('config/config.yaml')

@pytest.fixture(scope='session')
def env_config(load_yaml):
    """加载环境配置"""
    import os
    env = os.getenv('TEST_ENV', 'dev')
    config = load_yaml('config/environments.yaml')
    return config['environments'][env]

@pytest.fixture(scope='session')
def login_data(load_yaml):
    """加载登录测试数据"""
    return load_yaml('tests/test_data/login_data.yaml')

@pytest.fixture(scope='session')
def user_data(load_yaml):
    """加载用户测试数据"""
    return load_yaml('tests/test_data/user_data.yaml')

@pytest.fixture(scope='session')
def api_data(load_yaml):
    """加载 API 测试数据"""
    return load_yaml('tests/test_data/api_data.yaml')

9.3 测试用例示例

import pytest

def test_login_with_yaml_data(login_data):
    """使用 YAML 数据的登录测试"""
    login_cases = login_data['login_cases']

    for case in login_cases:
        username = case['username']
        password = case['password']
        expected = case['expected']

        # 执行登录
        result = perform_login(username, password)

        # 验证
        assert result == expected

def test_user_operations(user_data):
    """使用 YAML 数据的用户操作测试"""
    user_cases = user_data['user_cases']

    for case in user_cases:
        # 测试逻辑...
        pass

def perform_login(username, password):
    """执行登录（示例函数）"""
    # 实际项目中应该调用真实的登录接口
    return True

10. 总结

10.1 关键要点

  1. YAML 语法：

    • 使用空格缩进，不要使用 Tab
    • 字符串、数字、布尔值、列表、字典的基本用法
    • 注释和多行字符串的使用

  2. 在 Pytest 中使用 YAML：

    • 使用 yaml.safe_load() 安全加载 YAML 文件
    • 使用 @pytest.mark.parametrize 进行数据驱动测试
    • 使用 Fixture 管理 YAML 数据的加载和共享

  3. 最佳实践：

    • 数据与代码分离
    • 使用有意义的文件名和数据结构
    • 添加错误处理和验证
    • 使用缓存优化性能

10.2 学习路径建议

  1. 初级阶段：

    • 学习 YAML 基本语法
    • 掌握在 Python 中读取 YAML 文件
    • 使用 YAML 存储简单的测试数据

  2. 中级阶段：

    • 使用 YAML 进行数据驱动测试
    • 使用 Fixture 管理 YAML 数据
    • 处理复杂的数据结构

  3. 高级阶段：

    • 使用 YAML 配置测试环境
    • 实现接口关联测试
    • 优化性能和错误处理

10.3 参考资料

祝你在使用 Pytest 和 YAML 进行测试时顺利！ 🎉

发表评论