基于Python的RESTful API自动化测试框架设计与实现

一、框架设计目标

1. 核心需求：

2. 技术栈：

二、框架架构设计

```
└── api_test_framework/
    ├── config/               # 配置管理
    │   ├── env_config.ini    
    │   └── schema/          
    ├── core/                # 核心模块
    │   ├── rest_client.py    
    │   ├── assertion.py     
    │   └── logger.py        
    ├── test_data/           # 测试数据
    │   └── user_api.json    
    ├── test_cases/          # 测试用例
    │   └── test_user_api.py
    ├── reports/             # 测试报告
    ├── conftest.py          # Pytest Fixtures
    └── requirements.txt
```

三、核心模块实现

1. HTTP客户端封装（`core/rest_client.py`）

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class RESTClient:
    def __init__(self, base_url, timeout=10, max_retries=3):
        self.session = requests.Session()
        self.base_url = base_url
        
        # 配置重试策略
        retry = Retry(
            total=max_retries,
            backoff_factor=0.3,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)

    def request(self, method, endpoint, params=None, data=None, headers=None):
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(
            method=method.upper(),
            url=url,
            params=params,
            json=data,
            headers=headers,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response

    def get(self, endpoint, params=None, headers=None):
        return self.request('GET', endpoint, params=params, headers=headers)

    def post(self, endpoint, data=None, headers=None):
        return self.request('POST', endpoint, data=data, headers=headers)

2. 数据驱动测试（`test_data/user_api.json`）

{
  "create_user": {
    "test_data": [
      {
        "name": "valid_user",
        "data": {
          "username": "testuser1",
          "email": "test1@example.com"
        },
        "expected_status": 201
      },
      {
        "name": "invalid_email",
        "data": {
          "username": "testuser2",
          "email": "invalid_email"
        },
        "expected_status": 400
      }
    ]
  }
}

3. 增强断言机制（`core/assertion.py`）

import json
from jsonschema import validate

class APIResponseValidator:
    
    @staticmethod
    def assert_status_code(response, expected_code):
        assert response.status_code == expected_code, \
            f"Expected {expected_code}, Actual: {response.status_code}"
    
    @staticmethod
    def assert_json_schema(response, schema_file):
        with open(f"config/schema/{schema_file}") as f:
            schema = json.load(f)
        validate(instance=response.json(), schema=schema)
    
    @staticmethod
    def assert_response_time(response, max_time):
        assert response.elapsed.total_seconds() <= max_time, \
            f"Response time {response.elapsed} exceeds {max_time}s"
    
    @staticmethod
    def assert_header(response, header_key, expected_value):
        actual_value = response.headers.get(header_key)
        assert actual_value == expected_value, \
            f"Header {header_key}: Expected {expected_value}, Actual {actual_value}"

4. 测试用例示例（`test_cases/test_user_api.py`）

import pytest
import json
from core.rest_client import RESTClient
from core.assertion import APIResponseValidator

@pytest.fixture(scope="module")
def api_client():
    return RESTClient(base_url="https://api.example.com/v1")

def load_test_data(file_name, test_case):
    with open(f'test_data/{file_name}') as f:
        data = json.load(f)
        return data[test_case]['test_data']

@pytest.mark.parametrize("test_case", load_test_data("user_api.json", "create_user"))
def test_create_user(api_client, test_case):
    # 执行请求
    response = api_client.post(
        "/users",
        data=test_case['data'],
        headers={"Content-Type": "application/json"}
    )
    
    # 断言验证
    APIResponseValidator.assert_status_code(response, test_case['expected_status'])
    APIResponseValidator.assert_json_schema(response, "user_schema.json")
    APIResponseValidator.assert_response_time(response, 2)

—

5. 配置管理（`config/env_config.ini`）

; 环境配置管理文件
; 使用方式：通过环境变量 API_ENV 指定运行环境（默认DEV）

[DEFAULT]
log_level = INFO
request_timeout = 10
max_retries = 3

[DEV]
base_url = https://dev.api.example.com/v1
auth_type = JWT
db_connection = mysql://dev_user:dev_pass@dev-db:3306/test_db
cache_enabled = false

[TEST]
base_url = https://test.api.example.com/v1
auth_type = Basic
db_connection = postgresql://test_user:test_pass@test-db:5432/test_db
cache_enabled = true

[PROD]
base_url = https://api.example.com/v1
auth_type = OAuth2
db_connection = mysql://prod_user:${DB_PASSWORD}@prod-db:3306/prod_db  ; 使用环境变量注入密码
cache_enabled = true

配置加载器（core/config_loader.py）

import os
import configparser
from dotenv import load_dotenv

class ConfigHandler:
    def __init__(self, env=None):
        self.config = configparser.ConfigParser()
        self.config.read('config/env_config.ini')
        
        # 加载环境变量
        load_dotenv()  # 从.env文件加载
        
        # 确定运行环境
        self.env = env or os.getenv("API_ENV", "DEV")
        
        # 处理环境变量替换
        self._replace_env_vars()

    def _replace_env_vars(self):
        """替换配置中的${ENV_VAR}格式环境变量"""
        for section in self.config.sections():
            for key, value in self.config[section].items():
                if value.startswith('${') and value.endswith('}'):
                    env_var = value[2:-1]
                    self.config[section][key] = os.getenv(env_var, '')

    def get(self, key, section=None, fallback=None):
        """获取配置项（优先使用指定section，其次DEFAULT）"""
        section = section or self.env
        return self.config.get(
            section=section,
            option=key,
            fallback=self.config.get(
                section='DEFAULT',
                option=key,
                fallback=fallback
            )
        )

    @property
    def base_url(self):
        return self.get('base_url')

    @property
    def db_connection(self):
        return self.get('db_connection')

# 单例配置对象
config = ConfigHandler()

四、进阶功能实现示例

1. 鉴权处理扩展（core/auth_handler.py）

from typing import Optional
from core.rest_client import RESTClient

class JWTAuthHandler:
    def __init__(self, client: RESTClient, auth_endpoint: str):
        self.client = client
        self.auth_endpoint = auth_endpoint
        self._token: Optional[str] = None

    def get_token(self, username: str, password: str) -> str:
        """获取并缓存JWT令牌"""
        if not self._token:
            response = self.client.post(
                self.auth_endpoint,
                data={"username": username, "password": password}
            )
            self._token = response.json()["access_token"]
        return self._token

    def auth_header(self) -> dict:
        """生成认证头部"""
        return {"Authorization": f"Bearer {self._token}"}

# 使用示例（在conftest.py中）
@pytest.fixture(scope="session")
def auth_client(api_client):
    auth = JWTAuthHandler(api_client, "/auth/login")
    auth.get_token("admin", "securepassword")
    return auth

2. Pydantic数据建模（config/schema/user_schema.py）

from pydantic import BaseModel, EmailStr
from typing import List

class UserCreateRequest(BaseModel):
    username: str
    email: EmailStr
    roles: List[str] = ["user"]

class UserResponseSchema(BaseModel):
    id: int
    username: str
    email: str
    created_at: str
    updated_at: str

# 在断言中使用
def validate_user_response(response):
    validated_data = UserResponseSchema.parse_obj(response.json())
    return validated_data

3. Allure报告增强（test_cases/test_user_api.py扩展）

import allure

@allure.feature("用户管理API")
@allure.story("用户基础功能")
class TestUserAPI:

    @allure.title("创建用户 - 正常流测试")
    @allure.severity(allure.severity_level.CRITICAL)
    @pytest.mark.parametrize("test_case", load_test_data(...))
    def test_create_user(self, api_client, test_case):
        with allure.step("准备测试数据"):
            test_data = test_case["data"]
            
        with allure.step("发送创建用户请求"):
            response = api_client.post(...)
            
        with allure.step("验证响应结果"):
            APIResponseValidator.assert_status_code(...)
            validate_user_response(response)  # 使用Pydantic验证
            
        allure.attach(json.dumps(response.json(), indent=2), 
                     name="响应数据", 
                     attachment_type=allure.attachment_type.JSON)

五、框架运行与报告生成

1. 安装依赖（requirements.txt）

requests==2.31.0
pytest==8.0.0
pydantic==2.6.0
allure-pytest==2.13.2
python-dotenv==1.0.0
jsonschema==4.20.0

2. 执行测试并生成报告

# 运行测试
pytest test_cases/ -v --alluredir=./reports/allure-results

# 生成HTML报告（需要提前安装Allure命令行工具）
allure serve ./reports/allure-results

六、典型测试场景示例

场景1：带权限验证的API测试

def test_get_sensitive_data(api_client, auth_client):
    """验证带JWT鉴权的敏感数据访问"""
    # 获取认证头部
    headers = auth_client.auth_header()
    
    # 发送请求
    response = api_client.get(
        endpoint="/admin/data",
        headers=headers
    )
    
    # 断言验证
    APIResponseValidator.assert_status_code(response, 200)
    APIResponseValidator.assert_json_schema(response, "admin_data_schema.json")
    assert len(response.json()["results"]) > 0, "返回数据列表不应为空"
    
    # 验证数据时效性
    assert "2023" in response.json()["update_time"], "数据应包含当前年份的更新时间"

场景2：参数化数据驱动测试

import pytest

@pytest.mark.parametrize("user_id, expected_code", [
    (123, 200),    # 正常用户
    (999, 404),    # 不存在用户
    ("abc", 400)   # 错误参数类型
])
def test_get_user_by_id(api_client, user_id, expected_code):
    """验证不同用户ID的查询响应"""
    response = api_client.get(f"/users/{user_id}")
    
    # 基础断言
    APIResponseValidator.assert_status_code(response, expected_code)
    
    # 条件断言
    if expected_code == 200:
        assert response.json()["id"] == user_id, "返回用户ID应与请求一致"
        APIResponseValidator.assert_header(response, "Cache-Control", "max-age=3600")
    elif expected_code == 404:
        assert response.json()["error"] == "user_not_found", "应返回标准错误码"

场景3：多步骤事务测试

def test_user_lifecycle(api_client, db_connection):
    """验证用户全生命周期操作：创建->查询->更新->删除"""
    # 创建用户
    create_res = api_client.post(
        "/users",
        data={
            "username": "lifecycle_test",
            "email": "lifecycle@test.com"
        }
    )
    user_id = create_res.json()["id"]
    
    # 查询用户
    get_res = api_client.get(f"/users/{user_id}")
    assert get_res.json()["status"] == "active", "新用户应处于激活状态"
    
    # 更新用户
    update_res = api_client.put(
        f"/users/{user_id}",
        data={"status": "suspended"}
    )
    assert update_res.json()["updated"] is True, "应返回更新成功标识"
    
    # 验证数据库状态
    with db_connection.cursor() as cursor:
        cursor.execute(f"SELECT status FROM users WHERE id = {user_id}")
        assert cursor.fetchone()[0] == "suspended", "数据库状态应与更新一致"
    
    # 删除用户
    delete_res = api_client.delete(f"/users/{user_id}")
    assert delete_res.status_code == 204, "应返回无内容响应"
    
    # 验证删除结果
    check_res = api_client.get(f"/users/{user_id}")
    assert check_res.status_code == 404, "用户应不再存在"

场景4：异常流测试

def test_create_user_edge_cases(api_client):
    """验证用户创建异常场景"""
    # 测试数据准备
    invalid_cases = [
        (
            {"username": "a"*65, "email": "valid@test.com"},  # 用户名超长
            {"error_code": "invalid_username"}
        ),
        (
            {"username": "valid", "email": "invalid-email"},  # 错误邮箱格式
            {"error_code": "invalid_email"}
        ),
        (
            {},  # 空请求体
            {"error_code": "missing_required_fields"}
        )
    ]
    
    for data, expected in invalid_cases:
        response = api_client.post("/users", data=data)
        APIResponseValidator.assert_status_code(response, 400)
        assert response.json()["error_code"] == expected["error_code"],

作者：fengyun4623