Python接口模拟案例如何模拟接口

本文目录导读：

1. 使用unittest.mock模拟接口
2. 使用Flask创建模拟服务
3. 使用第三方库：responses
4. 使用pytest-mock
5. 完整示例：模拟RESTful API
6. 高级模拟技巧

我来详细介绍几种Python模拟接口的方法,从简单到复杂：

使用unittest.mock模拟接口

基础Mock对象

from unittest.mock import Mock, patch
import requests
# 创建Mock对象
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"name": "张三", "age": 25}
# 模拟接口调用
def get_user_info(user_id):
    response = requests.get(f"http://api.example.com/users/{user_id}")
    return response.json()
# 使用patch替换requests.get
with patch('requests.get') as mock_get:
    mock_get.return_value = mock_response
    result = get_user_info(1)
    print(result)  # {"name": "张三", "age": 25}

模拟不同场景

from unittest.mock import Mock, patch
# 模拟成功返回
def test_success_response():
    with patch('requests.get') as mock_get:
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {"status": "success", "data": []}
        mock_get.return_value = mock_response
        # 执行测试
        result = your_api_function()
        assert result["status"] == "success"
# 模拟异常情况
def test_error_response():
    with patch('requests.get') as mock_get:
        mock_response = Mock()
        mock_response.status_code = 404
        mock_response.json.return_value = {"error": "Not found"}
        mock_get.return_value = mock_response
        # 执行测试
        result = your_api_function()
        assert "error" in result
# 模拟网络超时
def test_timeout():
    with patch('requests.get') as mock_get:
        mock_get.side_effect = requests.exceptions.Timeout
        try:
            your_api_function()
        except requests.exceptions.Timeout:
            print("超时处理成功")

使用Flask创建模拟服务

from flask import Flask, jsonify, request
import threading
import requests
import time
app = Flask(__name__)
# 模拟GET接口
@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    # 模拟用户数据
    users = {
        1: {"id": 1, "name": "张三", "email": "zhang@example.com"},
        2: {"id": 2, "name": "李四", "email": "li@example.com"}
    }
    if user_id in users:
        return jsonify({"code": 200, "data": users[user_id]})
    else:
        return jsonify({"code": 404, "message": "用户不存在"}), 404
# 模拟POST接口
@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data or 'name' not in data:
        return jsonify({"code": 400, "message": "参数错误"}), 400
    # 模拟创建用户
    new_user = {
        "id": 3,
        "name": data['name'],
        "email": data.get('email', '')
    }
    return jsonify({"code": 200, "data": new_user}), 201
# 启动模拟服务器
def run_mock_server():
    app.run(host='127.0.0.1', port=5001, debug=False, use_reloader=False)
# 使用模拟服务器
def test_with_mock_server():
    # 启动模拟服务器
    server_thread = threading.Thread(target=run_mock_server)
    server_thread.daemon = True
    server_thread.start()
    time.sleep(1)  # 等待服务器启动
    # 测试接口
    response = requests.get('http://127.0.0.1:5001/api/users/1')
    print(response.json())
    response = requests.post('http://127.0.0.1:5001/api/users', 
                            json={"name": "王五"})
    print(response.json())

使用第三方库：responses

import responses
import requests
# 安装：pip install responses
@responses.activate
def test_api_with_responses():
    # 模拟GET请求
    responses.add(
        responses.GET,
        'http://api.example.com/users/1',
        json={"id": 1, "name": "张三"},
        status=200
    )
    # 模拟POST请求
    responses.add(
        responses.POST,
        'http://api.example.com/users',
        json={"id": 2, "name": "新用户"},
        status=201
    )
    # 模拟错误响应
    responses.add(
        responses.GET,
        'http://api.example.com/users/999',
        json={"error": "用户不存在"},
        status=404
    )
    # 测试
    response = requests.get('http://api.example.com/users/1')
    assert response.json()['name'] == '张三'
    response = requests.post('http://api.example.com/users', 
                            json={"name": "新用户"})
    assert response.status_code == 201

使用pytest-mock

import pytest
import requests
# 安装：pip install pytest-mock
def test_user_api(mocker):
    # 模拟请求返回
    mock_response = mocker.Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "id": 1,
        "name": "张三"
    }
    # 替换requests.get
    mocker.patch('requests.get', return_value=mock_response)
    # 调用真实的API函数
    result = get_user_info(1)
    assert result['name'] == '张三'
# 参数化测试
@pytest.mark.parametrize("user_id,expected_status,expected_data", [
    (1, 200, {"id": 1, "name": "张三"}),
    (2, 200, {"id": 2, "name": "李四"}),
    (999, 404, {"error": "Not Found"}),
])
def test_multiple_scenarios(mocker, user_id, expected_status, expected_data):
    mock_response = mocker.Mock()
    mock_response.status_code = expected_status
    mock_response.json.return_value = expected_data
    mocker.patch('requests.get', return_value=mock_response)
    response = requests.get(f'http://api.example.com/users/{user_id}')
    assert response.status_code == expected_status

完整示例：模拟RESTful API

import requests
from unittest.mock import Mock, patch
import json
class API模拟器:
    """API模拟器类"""
    def __init__(self):
        self.mock_responses = {}
        self.setup_mock_responses()
    def setup_mock_responses(self):
        """设置模拟响应"""
        # 用户接口
        self.mock_responses['get_users'] = {
            "code": 200,
            "data": [
                {"id": 1, "name": "张三", "email": "zhang@example.com"},
                {"id": 2, "name": "李四", "email": "li@example.com"}
            ],
            "total": 2
        }
        # 用户详情接口
        self.mock_responses['get_user_detail'] = {
            "code": 200,
            "data": {
                "id": 1,
                "name": "张三",
                "email": "zhang@example.com",
                "phone": "13800138000",
                "address": "北京市海淀区"
            }
        }
        # 创建用户接口
        self.mock_responses['create_user'] = {
            "code": 201,
            "message": "创建成功",
            "data": {
                "id": 3,
                "name": "新用户"
            }
        }
    @patch('requests.get')
    @patch('requests.post')
    def run_mock_test(self, mock_post, mock_get):
        """运行模拟测试"""
        # 配置模拟响应
        mock_get_response = Mock()
        mock_get_response.status_code = 200
        mock_get_response.json.return_value = self.mock_responses['get_users']
        mock_get.return_value = mock_get_response
        mock_post_response = Mock()
        mock_post_response.status_code = 201
        mock_post_response.json.return_value = self.mock_responses['create_user']
        mock_post.return_value = mock_post_response
        # 测试GET请求
        print("测试GET请求:")
        get_response = requests.get('http://api.example.com/users')
        print(f"状态码: {get_response.status_code}")
        print(f"返回数据: {json.dumps(get_response.json(), ensure_ascii=False, indent=2)}")
        # 测试POST请求
        print("\n测试POST请求:")
        post_response = requests.post('http://api.example.com/users', 
                                     json={"name": "新用户", "email": "new@example.com"})
        print(f"状态码: {post_response.status_code}")
        print(f"返回数据: {json.dumps(post_response.json(), ensure_ascii=False, indent=2)}")
# 使用示例
if __name__ == "__main__":
    simulator = API模拟器()
    simulator.run_mock_test()

高级模拟技巧

模拟延迟和异步

import asyncio
from unittest.mock import AsyncMock
# 模拟异步接口
async def mock_async_api():
    await asyncio.sleep(1)  # 模拟网络延迟
    return {"status": "success", "data": "异步数据"}
# 使用AsyncMock
async def test_async_mock():
    mock_api = AsyncMock()
    mock_api.return_value = {"status": "success"}
    result = await mock_api()
    print(result)  # {"status": "success"}
# 模拟循环调用
def mock_with_side_effects():
    mock_response = Mock()
    # 第一次调用返回成功，第二次返回失败
    mock_response.json.side_effect = [
        {"status": "pending"},
        {"status": "processing"}, 
        {"status": "completed"}
    ]
    for i in range(3):
        print(mock_response.json())

模拟认证和权限

def test_authentication_mock():
    # 模拟需要认证的接口
    mock_response = Mock()
    # 模拟token验证
    def check_token(headers):
        token = headers.get('Authorization', '')
        if token == 'Bearer valid_token':
            return {"status": "success", "data": "授权数据"}
        else:
            return {"status": "error", "message": "未授权"}
    # 测试有效token
    mock_response.status_code = 200
    mock_response.json.return_value = {"status": "success"}
    # 测试无效token
    mock_response.status_code = 401
    mock_response.json.return_value = {"status": "error", "message": "未授权"}

这些方法覆盖了从简单到复杂的接口模拟需求,选择哪种方法取决于你的具体场景：

• 单元测试：使用unittest.mock或pytest-mock
• 集成测试：使用Flask创建模拟服务器
• 简单模拟：使用responses库
• 异步测试：使用AsyncMock