You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何为Flask应用的POST/PUT/DELETE API编写无真实数据库操作的单元测试?

Python API单元测试:Mock数据库避免真实数据写入

假设我们有一个接收前端输入、验证后存入数据库的Flask API(示例代码如下),要写单元测试避免操作真实数据库,核心思路是mock掉所有直接操作数据库的函数/方法,用模拟对象替代真实的数据库交互,同时验证API的响应逻辑和mock对象的调用是否符合预期。

待测试的API代码(app.py)

from flask import Flask, request, jsonify
import psycopg2

app = Flask(__name__)

def save_user_to_db(name, email):
    """真实的数据库写入函数,测试时要mock掉它"""
    conn = psycopg2.connect("dbname=test user=postgres")
    cur = conn.cursor()
    cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)", (name, email))
    conn.commit()
    cur.close()
    conn.close()
    return True

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    # 基础参数验证
    if not data.get('name') or not data.get('email'):
        return jsonify({'error': '缺少必填字段'}), 400
    # 写入数据库
    try:
        save_user_to_db(data['name'], data['email'])
        return jsonify({'message': '用户创建成功'}), 201
    except Exception as e:
        return jsonify({'error': str(e)}), 500

方法一:用unittest.mock.patch(Python内置)

直接用Python标准库的unittest.mock.patch装饰器,替换掉数据库写入函数,模拟它的行为。

测试代码(test_app.py)

import unittest
from unittest.mock import patch, MagicMock
from app import app

class TestCreateUserAPI(unittest.TestCase):
    def setUp(self):
        # 启动Flask测试客户端
        app.testing = True
        self.client = app.test_client()

    @patch('app.save_user_to_db')  # 关键:patch的是app模块中实际使用的save_user_to_db引用
    def test_create_user_success(self, mock_save):
        # 配置mock函数的返回值
        mock_save.return_value = True

        # 发送测试请求
        response = self.client.post('/users', json={
            'name': '张三',
            'email': 'zhangsan@example.com'
        })

        # 验证API响应
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.get_json(), {'message': '用户创建成功'})

        # 验证mock函数是否被正确调用(参数是否匹配)
        mock_save.assert_called_once_with('张三', 'zhangsan@example.com')

    @patch('app.save_user_to_db')
    def test_create_user_missing_fields(self, mock_save):
        # 发送缺少必填字段的请求
        response = self.client.post('/users', json={'name': '张三'})

        # 验证响应
        self.assertEqual(response.status_code, 400)
        self.assertEqual(response.get_json(), {'error': '缺少必填字段'})

        # 验证mock函数根本没被调用(因为参数验证失败)
        mock_save.assert_not_called()

    @patch('app.save_user_to_db')
    def test_create_user_db_error(self, mock_save):
        # 让mock函数抛出异常,模拟数据库故障
        mock_save.side_effect = Exception('数据库连接失败')

        response = self.client.post('/users', json={
            'name': '李四',
            'email': 'lisi@example.com'
        })

        self.assertEqual(response.status_code, 500)
        self.assertEqual(response.get_json(), {'error': '数据库连接失败'})
        mock_save.assert_called_once_with('李四', 'lisi@example.com')

if __name__ == '__main__':
    unittest.main()

关键说明

  • patch路径的正确性:必须patch被测试代码中实际导入并使用的目标对象路径。比如如果save_user_to_db定义在db.py,且app.py中用from db import save_user_to_db导入,那就要patch'app.save_user_to_db'而非'db.save_user_to_db'。
  • mock的核心作用:既避免了真实数据库操作,又能验证业务逻辑是否正确触发了数据写入动作。

方法二:用pytest的monkeypatch

如果用pytest框架,可通过monkeypatch fixture替换数据库操作函数。

测试代码(test_app_pytest.py)

import pytest
from app import app, save_user_to_db

@pytest.fixture
def client():
    # 定义Flask测试客户端fixture
    app.testing = True
    with app.test_client() as client:
        yield client

def test_create_user_success(client, monkeypatch):
    # 定义模拟的数据库写入函数
    def mock_save(name, email):
        return True

    # 用monkeypatch替换原函数
    monkeypatch.setattr('app.save_user_to_db', mock_save)

    # 发送请求并验证
    response = client.post('/users', json={
        'name': '张三',
        'email': 'zhangsan@example.com'
    })
    assert response.status_code == 201
    assert response.get_json() == {'message': '用户创建成功'}

def test_create_user_db_error(client, monkeypatch):
    # 模拟数据库异常
    def mock_save(name, email):
        raise Exception('数据库连接失败')

    monkeypatch.setattr('app.save_user_to_db', mock_save)

    response = client.post('/users', json={
        'name': '李四',
        'email': 'lisi@example.com'
    })
    assert response.status_code == 500
    assert response.get_json() == {'error': '数据库连接失败'}

扩展:ORM场景的mock(以SQLAlchemy为例)

如果用SQLAlchemy,无需mock底层连接,直接mock会话的add和commit方法即可:

@patch('app.db.session.commit')
@patch('app.db.session.add')
def test_create_user_orm(mock_add, mock_commit, client):
    response = client.post('/users', json={'name': '王五', 'email': 'wangwu@example.com'})
    assert response.status_code == 201
    # 验证是否调用了add方法(传入的用户对象是否符合预期)
    mock_add.assert_called_once()
    added_user = mock_add.call_args[0][0]
    assert added_user.name == '王五'
    assert added_user.email == 'wangwu@example.com'
    # 验证commit是否被调用
    mock_commit.assert_called_once()

内容的提问来源于stack exchange,提问作者Sai Naveen

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.18 13:23:08