PythonYAML模块使用教程：接口测试参数存储与配置

Python YAML 模块在接口测试中的参数存储与配置教程

一、YAML 简介

1.1 什么是 YAML

YAML（YAML Ain't Markup Language）是一种人类可读的数据序列化格式，常用于配置文件和数据交换。

1.2 YAML 优势

简洁易读的语法
支持复杂数据结构
跨语言支持
支持注释

二、安装 PyYAML

pip install pyyaml

三、基础语法

3.1 基本数据类型

# 字符串
name: "API Test"

# 数字
port: 8080

# 布尔值
enabled: true

# 空值
value: null

3.2 复杂数据结构

# 列表
endpoints:
  - /api/login
  - /api/users
  - /api/products

# 字典
database:
  host: localhost
  port: 5432
  name: test_db

# 混合结构
test_suite:
  name: "用户管理接口测试"
  test_cases:
    - name: "创建用户"
      method: "POST"
    - name: "查询用户"
      method: "GET"

四、在接口测试中的应用

4.1 项目结构

api_test_project/
├── config/
│   ├── base.yaml
│   ├── test_env.yaml
│   └── prod_env.yaml
├── test_cases/
│   └── user_api.yaml
├── utils/
│   └── config_loader.py
└── test_main.py

4.2 配置文件示例

config/base.yaml

# 基础配置
project:
  name: "API自动化测试平台"
  version: "1.0.0"

# 请求默认配置
request:
  timeout: 30
  verify_ssl: false
  default_headers:
    Content-Type: "application/json"
    User-Agent: "API-Test-Framework"

# 日志配置
logging:
  level: "INFO"
  file_path: "./logs/api_test.log"

config/test_env.yaml

# 测试环境配置
environment: "test"

api:
  base_url: "https://api.test.example.com"

database:
  host: "test-db.example.com"
  username: "test_user"
  password: "test_pass123"

test_data:
  user:
    admin:
      username: "admin_test"
      password: "admin123"
    normal:
      username: "user_test"
      password: "user123"

4.3 测试用例配置

test_cases/user_api.yaml

test_suite: "用户接口测试套件"

test_cases:
  user_login:
    name: "用户登录接口测试"
    description: "验证登录接口功能"
    endpoint: "/api/v1/login"
    method: "POST"

    test_data:
      valid_login:
        username: "test_user"
        password: "password123"
        expected_status: 200

      invalid_password:
        username: "test_user"
        password: "wrong_pass"
        expected_status: 401
        expected_message: "密码错误"

    assertions:
      - field: "code"
        value: 0
        comparator: "equal"
      - field: "data.token"
        exists: true

  create_user:
    name: "创建用户接口测试"
    endpoint: "/api/v1/users"
    method: "POST"
    headers:
      Authorization: "Bearer ${token}"

    test_data:
      normal_user:
        username: "new_user_${timestamp}"
        email: "test${timestamp}@example.com"
        role: "user"

    preconditions:
      - type: "login"
        save_response_as: "token"

    assertions:
      - field: "code"
        value: 0
      - field: "data.id"
        exists: true

五、Python 实现

5.1 配置加载器

utils/config_loader.py

import yaml
import os
from pathlib import Path
from typing import Dict, Any, Optional
import logging

class ConfigLoader:
    """YAML配置加载器"""

    def __init__(self, config_dir: str = "config"):
        self.config_dir = Path(config_dir)
        self.config_cache = {}
        self.logger = logging.getLogger(__name__)

    def load_yaml(self, file_path: str) -> Dict[str, Any]:
        """加载YAML文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                return yaml.safe_load(file)
        except Exception as e:
            self.logger.error(f"加载配置文件失败: {file_path}, 错误: {e}")
            raise

    def load_config(self, env: str = "test") -> Dict[str, Any]:
        """加载完整配置"""
        config_key = f"config_{env}"

        if config_key in self.config_cache:
            return self.config_cache[config_key]

        # 加载基础配置
        base_config = self.load_yaml(self.config_dir / "base.yaml")

        # 加载环境配置
        env_file = self.config_dir / f"{env}_env.yaml"
        env_config = self.load_yaml(env_file)

        # 合并配置
        merged_config = self._merge_dicts(base_config, env_config)

        self.config_cache[config_key] = merged_config
        return merged_config

    def _merge_dicts(self, dict1: Dict, dict2: Dict) -> Dict:
        """深度合并两个字典"""
        result = dict1.copy()

        for key, value in dict2.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = self._merge_dicts(result[key], value)
            else:
                result[key] = value

        return result

    def load_test_cases(self, test_case_file: str) -> Dict[str, Any]:
        """加载测试用例"""
        test_case_path = Path("test_cases") / test_case_file
        return self.load_yaml(test_case_path)

class EnvironmentConfig:
    """环境配置管理"""

    def __init__(self, env: str = None):
        self.loader = ConfigLoader()
        self.env = env or self._detect_environment()
        self.config = self.loader.load_config(self.env)

    def _detect_environment(self) -> str:
        """检测运行环境"""
        env = os.getenv("TEST_ENV", "test")
        return env

    def get_api_url(self, endpoint: str = "") -> str:
        """获取完整的API URL"""
        base_url = self.config['api']['base_url']
        return f"{base_url}{endpoint}"

    def get_database_config(self) -> Dict[str, Any]:
        """获取数据库配置"""
        return self.config.get('database', {})

    def get_test_data(self, path: str) -> Any:
        """获取测试数据"""
        keys = path.split('.')
        data = self.config.get('test_data', {})

        for key in keys:
            if isinstance(data, dict) and key in data:
                data = data[key]
            else:
                return None

        return data

5.2 测试数据生成器

utils/test_data_generator.py

import yaml
import random
import string
import time
from typing import Dict, Any
import hashlib

class TestDataGenerator:
    """测试数据生成器"""

    def __init__(self, template_file: str = "templates/test_data_templates.yaml"):
        self.templates = self._load_templates(template_file)

    def _load_templates(self, file_path: str) -> Dict[str, Any]:
        """加载数据模板"""
        with open(file_path, 'r', encoding='utf-8') as file:
            return yaml.safe_load(file)

    def generate_data(self, template_name: str, **kwargs) -> Dict[str, Any]:
        """根据模板生成测试数据"""
        template = self.templates.get(template_name, {})
        data = template.copy()

        # 处理动态变量
        data = self._process_dynamic_values(data, **kwargs)

        return data

    def _process_dynamic_values(self, data: Any, **kwargs) -> Any:
        """处理动态值"""
        if isinstance(data, dict):
            return {k: self._process_dynamic_values(v, **kwargs) for k, v in data.items()}
        elif isinstance(data, list):
            return [self._process_dynamic_values(item, **kwargs) for item in data]
        elif isinstance(data, str):
            return self._replace_placeholders(data, **kwargs)
        else:
            return data

    def _replace_placeholders(self, text: str, **kwargs) -> str:
        """替换占位符"""
        placeholders = {
            '${timestamp}': str(int(time.time())),
            '${random_int}': str(random.randint(1000, 9999)),
            '${random_string}': ''.join(random.choices(string.ascii_letters, k=8)),
            '${random_email}': f"test_{int(time.time())}@example.com"
        }

        # 更新自定义参数
        placeholders.update(kwargs)

        for placeholder, value in placeholders.items():
            text = text.replace(placeholder, str(value))

        return text

# 数据模板示例
test_data_templates_content = """
user:
  basic:
    username: "user_${random_int}"
    email: "${random_email}"
    password: "Test@123456"
    phone: "138${random_int}"

  admin:
    username: "admin_${timestamp}"
    email: "admin_${timestamp}@example.com"
    password: "Admin@123456"
    role: "admin"

product:
  basic:
    name: "产品_${random_string}"
    price: ${random_int}
    category: "电子产品"
    stock: 100
"""

# 保存模板文件
with open("templates/test_data_templates.yaml", "w", encoding="utf-8") as f:
    f.write(test_data_templates_content)

5.3 测试用例执行器

test_main.py

import yaml
import requests
import logging
from typing import Dict, Any, List
from utils.config_loader import EnvironmentConfig
from utils.test_data_generator import TestDataGenerator

class APITestRunner:
    """API测试执行器"""

    def __init__(self, env: str = None):
        self.config = EnvironmentConfig(env)
        self.data_generator = TestDataGenerator()
        self.session = requests.Session()
        self.logger = logging.getLogger(__name__)

        # 配置会话
        request_config = self.config.config['request']
        self.session.headers.update(request_config.get('default_headers', {}))
        self.timeout = request_config.get('timeout', 30)

    def execute_test_case(self, test_case_file: str):
        """执行测试用例"""
        test_cases = self.config.loader.load_test_cases(test_case_file)

        self.logger.info(f"开始执行测试套件: {test_cases['test_suite']}")

        results = []
        for case_key, case_config in test_cases['test_cases'].items():
            try:
                result = self._run_single_case(case_key, case_config)
                results.append(result)
                self.logger.info(f"测试用例 '{case_config['name']}' 执行结果: {result['status']}")
            except Exception as e:
                self.logger.error(f"测试用例 '{case_key}' 执行失败: {e}")
                results.append({
                    'case': case_key,
                    'status': 'FAILED',
                    'error': str(e)
                })

        self._generate_report(results)
        return results

    def _run_single_case(self, case_key: str, case_config: Dict[str, Any]) -> Dict[str, Any]:
        """执行单个测试用例"""
        endpoint = case_config['endpoint']
        method = case_config['method'].lower()
        url = self.config.get_api_url(endpoint)

        # 处理测试数据
        test_data_sets = case_config.get('test_data', {})
        results = []

        for data_key, data_config in test_data_sets.items():
            # 生成请求数据
            request_data = self._prepare_request_data(data_config, case_config)

            # 发送请求
            response = self._send_request(method, url, request_data, case_config)

            # 验证结果
            assertions = case_config.get('assertions', [])
            passed = self._verify_response(response, assertions)

            results.append({
                'data_set': data_key,
                'status': 'PASSED' if passed else 'FAILED',
                'response_code': response.status_code
            })

        return {
            'case': case_key,
            'name': case_config['name'],
            'results': results
        }

    def _prepare_request_data(self, data_config: Dict, case_config: Dict) -> Dict:
        """准备请求数据"""
        # 使用模板生成数据
        if 'template' in data_config:
            template_name = data_config['template']
            data = self.data_generator.generate_data(template_name, **data_config)
        else:
            data = data_config.copy()

        # 移除测试专用字段
        data.pop('expected_status', None)
        data.pop('expected_message', None)

        return data

    def _send_request(self, method: str, url: str, data: Dict, case_config: Dict):
        """发送HTTP请求"""
        headers = case_config.get('headers', {})

        # 处理动态header（如token）
        processed_headers = {}
        for key, value in headers.items():
            if isinstance(value, str) and value.startswith('${') and value.endswith('}'):
                # 从缓存中获取动态值
                var_name = value[2:-1]
                processed_headers[key] = self._get_variable(var_name)
            else:
                processed_headers[key] = value

        self.logger.debug(f"发送请求: {method} {url}")
        self.logger.debug(f"请求数据: {data}")

        response = self.session.request(
            method=method,
            url=url,
            json=data,
            headers=processed_headers,
            timeout=self.timeout
        )

        return response

    def _get_variable(self, var_name: str) -> Any:
        """获取变量值"""
        # 这里可以实现变量存储和获取逻辑
        # 例如从之前的响应中提取的token
        return None

    def _verify_response(self, response: requests.Response, assertions: List[Dict]) -> bool:
        """验证响应"""
        try:
            response_data = response.json()
        except:
            response_data = {}

        for assertion in assertions:
            if not self._check_assertion(response_data, response.status_code, assertion):
                return False

        return True

    def _check_assertion(self, response_data: Dict, status_code: int, assertion: Dict) -> bool:
        """检查单个断言"""
        assertion_type = assertion.get('type', 'field')

        if assertion_type == 'status_code':
            expected = assertion.get('value')
            return status_code == expected

        elif assertion_type == 'field':
            field_path = assertion.get('field', '')
            expected_value = assertion.get('value')
            comparator = assertion.get('comparator', 'equal')

            # 获取字段值
            field_value = self._get_field_value(response_data, field_path)

            # 比较
            if comparator == 'equal':
                return field_value == expected_value
            elif comparator == 'exists':
                return field_value is not None

        return True

    def _get_field_value(self, data: Any, path: str) -> Any:
        """通过路径获取字段值"""
        if not path:
            return data

        keys = path.split('.')
        current = data

        for key in keys:
            if isinstance(current, dict) and key in current:
                current = current[key]
            else:
                return None

        return current

    def _generate_report(self, results: List[Dict]):
        """生成测试报告"""
        total = len(results)
        passed = sum(1 for r in results if r['status'] != 'FAILED')

        report = {
            'summary': {
                'total': total,
                'passed': passed,
                'failed': total - passed,
                'pass_rate': f"{(passed/total*100):.1f}%" if total > 0 else "0%"
            },
            'details': results
        }

        # 保存报告为YAML
        report_file = f"reports/test_report_{int(time.time())}.yaml"
        with open(report_file, 'w', encoding='utf-8') as f:
            yaml.dump(report, f, allow_unicode=True, default_flow_style=False)

        self.logger.info(f"测试报告已生成: {report_file}")

def main():
    """主函数"""
    # 配置日志
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    # 执行测试
    runner = APITestRunner()

    # 执行用户接口测试
    results = runner.execute_test_case("user_api.yaml")

    # 输出摘要
    print("\n测试执行完成!")
    for result in results:
        print(f"{result['name']}: {result['status']}")

if __name__ == "__main__":
    main()

六、高级用法

6.1 多环境配置管理

config/prod_env.yaml

environment: "production"

api:
  base_url: "https://api.example.com"

database:
  host: "prod-db.example.com"
  username: "${DB_USERNAME}"
  password: "${DB_PASSWORD}"

# 使用环境变量
test_data:
  user:
    admin:
      username: "${ADMIN_USERNAME}"
      password: "${ADMIN_PASSWORD}"

6.2 配置文件加密

import yaml
import base64
from cryptography.fernet import Fernet

class SecureConfigLoader:
    """加密配置加载器"""

    def __init__(self, key_file: str = "config/key.key"):
        self.key = self._load_key(key_file)
        self.cipher = Fernet(self.key)

    def _load_key(self, key_file: str) -> bytes:
        """加载加密密钥"""
        with open(key_file, 'rb') as f:
            return f.read()

    def load_encrypted_yaml(self, file_path: str) -> Dict:
        """加载加密的YAML文件"""
        with open(file_path, 'rb') as f:
            encrypted_data = f.read()

        decrypted_data = self.cipher.decrypt(encrypted_data)
        return yaml.safe_load(decrypted_data.decode('utf-8'))

6.3 配置验证

import yaml
import jsonschema
from typing import Dict

class ConfigValidator:
    """配置验证器"""

    def __init__(self, schema_file: str = "schemas/config_schema.yaml"):
        self.schema = self.load_yaml(schema_file)

    def validate_config(self, config: Dict, config_type: str = "base") -> bool:
        """验证配置"""
        try:
            jsonschema.validate(instance=config, schema=self.schema[config_type])
            return True
        except jsonschema.ValidationError as e:
            print(f"配置验证失败: {e}")
            return False

# 配置模式示例
config_schema_content = """
base:
  type: object
  required: ["project", "request"]
  properties:
    project:
      type: object
      properties:
        name:
          type: string
        version:
          type: string

    request:
      type: object
      properties:
        timeout:
          type: integer
        verify_ssl:
          type: boolean

test_case:
  type: object
  required: ["test_suite", "test_cases"]
  properties:
    test_suite:
      type: string
    test_cases:
      type: object
"""