1. 为什么需要封装请求

1.1 直接使用requests的问题

# 未封装的请求示例 - 问题明显
import requests

def test_create_user():
    url = "https://api.example.com/users"
    headers = {"Content-Type": "application/json"}
    data = {"name": "John", "email": "john@example.com"}
    
    # 问题1：硬编码配置
    # 问题2：无统一错误处理
    # 问题3：缺乏日志记录
    response = requests.post(url, json=data, headers=headers)
    
    assert response.status_code == 201

1.2 封装请求的核心价值

  1. 统一管理：集中处理headers、超时、认证等配置
  2. 减少重复：避免每个测试用例重复编写请求代码
  3. 增强健壮性：统一错误处理和重试机制
  4. 日志追踪：自动记录请求/响应详情
  5. 易于维护：修改请求逻辑只需调整一处

2. 请求封装设计原则

2.1 分层架构设计

测试框架
├── tests/              # 测试用例层
├── utils/              # 工具层
│   └── request_util.py # 请求封装
└── conftest.py         # 全局配置

2.2 核心功能规划

功能模块实现要点
基础请求方法GET/POST/PUT/DELETE统一封装
请求头管理全局headers+用例级headers
超时控制连接超时+读取超时配置
日志记录自动记录请求/响应关键信息
错误处理网络异常重试+统一异常封装
响应处理JSON自动解析+状态码检查

3. 请求工具层实现

3.1 基础封装代码

# utils/request_util.py
import requests
import json
import logging

class RequestUtil:
    def __init__(self):
        # 基础配置
        self.base_url = "https://api.example.com"  # 从配置读取
        self.timeout = 10  # 默认超时时间
        self.session = requests.Session()
        
        # 日志配置
        self.logger = logging.getLogger(__name__)
        
        # 全局headers
        self.session.headers.update({
            "Content-Type": "application/json",
            "User-Agent": "PytestAPI/1.0"
        })
    
    def _request(self, method, endpoint, **kwargs):
        """统一请求方法"""
        url = f"{self.base_url}{endpoint}"
        
        # 处理超时
        kwargs.setdefault("timeout", self.timeout)
        
        # 记录请求日志
        self.log_request(method, url, kwargs)
        
        try:
            response = self.session.request(method, url, **kwargs)
            # 记录响应日志
            self.log_response(response)
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error(f"请求异常: {str(e)}")
            raise ConnectionError(f"网络请求失败: {str(e)}")
    
    def log_request(self, method, url, kwargs):
        """记录请求日志"""
        self.logger.info(f"请求方法: {method}")
        self.logger.info(f"请求URL: {url}")
        
        if 'json' in kwargs:
            self.logger.debug(f"请求体: {json.dumps(kwargs['json'], indent=2)}")
        if 'params' in kwargs:
            self.logger.debug(f"请求参数: {kwargs['params']}")
        if 'headers' in kwargs:
            self.logger.debug(f"请求头: {kwargs['headers']}")
    
    def log_response(self, response):
        """记录响应日志"""
        self.logger.info(f"响应状态: {response.status_code}")
        try:
            self.logger.debug(f"响应体: {json.dumps(response.json(), indent=2)}")
        except json.JSONDecodeError:
            self.logger.debug(f"响应体: {response.text[:500]}...")
    
    # 简化方法
    def get(self, endpoint, params=None, **kwargs):
        return self._request('GET', endpoint, params=params, **kwargs)
    
    def post(self, endpoint, json=None, **kwargs):
        return self._request('POST', endpoint, json=json, **kwargs)
    
    def put(self, endpoint, json=None, **kwargs):
        return self._request('PUT', endpoint, json=json, **kwargs)
    
    def delete(self, endpoint, **kwargs):
        return self._request('DELETE', endpoint, **kwargs)

3.2 响应处理增强

# 在RequestUtil类中添加

def request_with_check(self, method, endpoint, expected_status=200, **kwargs):
    """带状态检查的请求"""
    response = self._request(method, endpoint, **kwargs)
    
    # 状态码检查
    if response.status_code != expected_status:
        error_msg = (f"状态码异常! 预期: {expected_status}, "
                    f"实际: {response.status_code}, "
                    f"URL: {response.url}")
        self.logger.error(error_msg)
        raise AssertionError(error_msg)
    
    return response

def get_json(self, method, endpoint, **kwargs):
    """获取JSON响应"""
    response = self._request(method, endpoint, **kwargs)
    try:
        return response.json()
    except json.JSONDecodeError:
        self.logger.error(f"JSON解析失败: {response.text[:200]}")
        raise ValueError("响应不是有效的JSON格式")

4. 集成到测试框架

4.1 创建conftest.py全局工具

# conftest.py
import pytest
from utils.request_util import RequestUtil

@pytest.fixture(scope="session")
def api_client():
    """创建全局API客户端"""
    client = RequestUtil()
    
    # 可在此添加全局认证
    # client.session.headers["Authorization"] = "Bearer token"
    
    yield client
    
    # 测试结束后清理
    client.session.close()

4.2 测试用例中使用封装

# tests/test_user_api.py

def test_create_user(api_client):
    """测试创建用户"""
    # 准备测试数据
    user_data = {
        "name": "测试用户",
        "email": "test@example.com",
        "password": "P@ssw0rd"
    }
    
    # 发送请求（自动记录日志）
    response = api_client.post("/users", json=user_data)
    
    # 验证响应
    assert response.status_code == 201
    response_data = response.json()
    assert "id" in response_data
    assert response_data["email"] == user_data["email"]

def test_get_user(api_client):
    """测试获取用户信息"""
    # 使用带状态检查的方法
    user_data = api_client.get_json("GET", "/users/123", expected_status=200)
    
    # 验证数据结构
    assert isinstance(user_data, dict)
    assert "id" in user_data
    assert "name" in user_data
    assert "email" in user_data

5. 高级功能扩展

5.1 环境配置管理

# config.py
class Config:
    ENV = "test"  # dev/test/prod
    
    @property
    def base_url(self):
        return {
            "dev": "https://dev-api.example.com",
            "test": "https://test-api.example.com",
            "prod": "https://api.example.com"
        }[self.ENV]

# 在RequestUtil中修改
from config import Config

class RequestUtil:
    def __init__(self):
        self.config = Config()
        self.base_url = self.config.base_url
        # ...

5.2 请求重试机制

# 安装依赖: pip install tenacity
from tenacity import retry, stop_after_attempt, wait_exponential

class RequestUtil:
    # ...
    
    @retry(stop=stop_after_attempt(3), 
           wait=wait_exponential(multiplier=1, min=2, max=10))
    def _request(self, method, endpoint, **kwargs):
        # 原有代码...

5.3 响应结果解析器

class RequestUtil:
    # ...
    
    def extract_value(self, response, json_path):
        """使用JSONPath提取值"""
        from jsonpath_ng import parse  # pip install jsonpath-ng
        
        try:
            json_data = response.json()
            expr = parse(json_path)
            matches = [match.value for match in expr.find(json_data)]
            return matches[0] if matches else None
        except Exception as e:
            self.logger.error(f"JSONPath解析失败: {str(e)}")
            raise

# 使用示例
def test_extract_data(api_client):
    response = api_client.get("/users/123")
    user_id = api_client.extract_value(response, "$.data.id")
    assert user_id == 123

6. 避坑指南

6.1 常见错误及解决方案

  1. 硬编码环境配置

    • ❌ 错误：在测试用例中直接写URL
    • ✅ 解决：通过配置文件管理多环境

  2. 忽略网络异常

    • ❌ 错误：未处理requests异常导致测试失败
    • ✅ 解决：封装层统一捕获并重试

  3. 敏感信息泄露

    • ❌ 错误：日志中记录密码等敏感信息

    • ✅ 解决：添加敏感信息过滤

      def log_request(self, method, url, kwargs):
    if 'json' in kwargs and 'password' in kwargs['json']:
        sanitized = kwargs['json'].copy()
        sanitized['password'] = '***'
        self.logger.debug(f"请求体: {json.dumps(sanitized)}")
    # ...

  4. 过度封装

    • ❌ 错误：在工具层添加业务断言
    • ✅ 解决：工具层只负责请求，断言留在测试用例

  5. 未复用Session

    • ❌ 错误：每次请求创建新Session
    • ✅ 解决：使用Session保持连接提高性能
# pytest
Logo
乐奇 Rokid 开放社区

更多推荐

cover

TestStar | AIUI 数据库客户端测试Case执行实录

avatar 乐奇 Rokid 开放社区
cover

AR巡检哪家好

avatar 乐奇 Rokid 开放社区
cover

具身智能 | Ego数采设备与选择标准一文理清，避开无效采集

avatar 乐奇 Rokid 开放社区