基于requests的POST请求重试函数单元测试编写方案咨询

首先先对你的原函数做三处可测试性优化：

• 原函数直接引用外部作用域的url、data、ref、api_name变量，建议改为函数入参，提升复用性和可测试性
• 原配置仅挂载了http://协议的重试适配器，https请求不会触发重试，建议补充https协议的挂载配置
• 默认Retry配置仅在网络异常（连接失败、超时等）时重试，不会对5xx类服务端错误重试，如需覆盖这类场景可新增status_forcelist参数指定要重试的状态码

优化后的参考代码：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import json
import time
import logging

logger = logging.getLogger(__name__)

def request_retry_session(url: str, data: dict, ref: str, api_name: str):
    session = requests.Session()
    retry = Retry(
        total=3, 
        backoff_factor=0.1, 
        method_whitelist=frozenset(['POST']),
        status_forcelist=[429, 500, 502, 503, 504] # 可选，指定触发重试的服务端状态码
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter) # 补充https协议适配
    try:
        start_time = time.time()
        response = session.post(
            url=url,
            data=json.dumps(data),
            headers={"Content-Type": "application/json"},
            verify=False,
        )
        time_cost = round(time.time() - start_time, 2)
        response_data = response.json()  
        logger.info(f"{ref}processing time cost for {api_name}: {time_cost}s")
        return response_data
    except Exception as e:
        raise e

测试依赖安装 #

你需要用到两个测试工具：

• pytest：Python主流单元测试框架
• requests-mock：用于拦截模拟requests请求，无需请求真实接口

执行pip install pytest requests-mock即可安装。

核心测试用例设计 #

你需要覆盖四类核心场景：

• 首次请求直接成功
• 重试N次后请求成功
• 所有重试全部失败后抛出异常
• 响应解析异常场景

测试代码参考 #

import pytest
import requests
import json
import requests_mock
from your_module import request_retry_session # 替换为你实际的函数导入路径

# 测试常量定义
TEST_URL = "http://test-api.example.com/test"
TEST_REQ_DATA = {"user_id": 123, "content": "test"}
TEST_REF = "test_task_"
TEST_API_NAME = "user_submit"

def test_first_request_success():
    """测试首次请求正常返回的场景"""
    with requests_mock.Mocker() as m:
        mock_resp = {"code": 0, "message": "ok", "data": {}}
        m.post(TEST_URL, json=mock_resp, status_code=200)
        result = request_retry_session(TEST_URL, TEST_REQ_DATA, TEST_REF, TEST_API_NAME)
        # 断言返回值符合预期
        assert result == mock_resp
        # 断言仅发起1次请求，没有触发重试
        assert m.call_count == 1

def test_retry_success_after_multiple_failures():
    """测试前两次请求失败，第三次重试成功的场景"""
    with requests_mock.Mocker() as m:
        mock_resp = {"code": 0, "message": "ok", "data": {}}
        # 配置前两次返回500，第三次返回正常响应
        m.post(TEST_URL, [
            {"status_code": 500},
            {"status_code": 500},
            {"json": mock_resp, "status_code": 200}
        ])
        result = request_retry_session(TEST_URL, TEST_REQ_DATA, TEST_REF, TEST_API_NAME)
        assert result == mock_resp
        # 断言总共发起3次请求：首次+2次重试
        assert m.call_count == 3

def test_all_retries_failed_raise_exception():
    """测试3次重试全部失败的场景，验证会抛出异常且请求次数符合预期"""
    with requests_mock.Mocker() as m:
        # 所有请求都返回500错误
        m.post(TEST_URL, status_code=500)
        # 断言会抛出HTTP异常
        with pytest.raises(requests.exceptions.HTTPError):
            request_retry_session(TEST_URL, TEST_REQ_DATA, TEST_REF, TEST_API_NAME)
        # 断言总共发起4次请求：首次+3次重试
        assert m.call_count == 4

def test_response_not_json_raise_exception():
    """测试接口返回非JSON格式的场景，验证解析异常会正常抛出"""
    with requests_mock.Mocker() as m:
        m.post(TEST_URL, text="system error", status_code=200)
        with pytest.raises(json.JSONDecodeError):
            request_retry_session(TEST_URL, TEST_REQ_DATA, TEST_REF, TEST_API_NAME)

额外测试建议 #

• 如果需要验证日志打印是否正确，可以使用pytest内置的caplog fixture捕获日志输出，对日志内容和级别做断言
• 可以新增超时场景的测试，模拟接口超时验证重试逻辑是否生效
• 测试执行前可以关闭requests的SSL告警，避免无关日志干扰测试输出

内容的提问来源于stack exchange，提问作者Sam