如何使用pytest正确mock psycopg2实现无真实DB连接测试

你之前尝试的pytest-pgsql、pytest-postgresql都属于集成测试工具，运行时会拉起真实的PostgreSQL进程（要么本地安装的实例，要么Docker容器），天然不符合「不建立真实数据库连接」的需求。
纯mock方案不需要依赖任何第三方数据库测试插件，用Python标准库自带的unittest.mock就能实现，核心逻辑是拦截psycopg2的连接调用，替换为模拟对象，完全不触发真实网络连接。

先修正两个会影响正常运行和mock逻辑的问题：

1. add_event方法里SQL传参的单元素元组漏了逗号，(event)是单个变量不是元组，会触发参数类型错误，要改成(event,)
2. events方法里的表名写错，建表时表名是history，查询时写成了event
3. FastAPI代码里漏了os和secrets的导入，启动会报错

修正后的对应代码段：

# db.py 修正部分
def add_event(self, *, event):
    insert_event_table = """
        INSERT INTO history (event) VALUES (%s);
    """
    self.cursor.execute(insert_event_table, (event,)) # 单元素元组补逗号

def events(self):
    select_event_table = """SELECT * FROM history;""" # 表名修正为history
    self.cursor.execute(select_event_table)
    return self.cursor.fetchall()

# main.py 头部补导入
import os
import secrets
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from decimal import Decimal, getcontext
from db import PostgresqlApi

核心是用patch拦截你代码里导入的psycopg2.connect方法，替换为返回预设的模拟连接、游标对象，所有数据库操作都不会真的发起连接。
注意patch路径必须是被测模块内引用psycopg2的路径，也就是db.psycopg2.connect，不能直接patch全局的psycopg2.connect，否则会拦截失败。

1. 数据库层单元测试示例 #

from unittest.mock import MagicMock, patch
import pytest
import os
# 先设置测试用环境变量，避免初始化读值报错
os.environ["DATABASE_URL"] = "postgresql://test:test@localhost:5432/testdb"
from db import PostgresqlApi

@patch("db.psycopg2.connect")
def test_db_operation(mock_connect):
    # 构造模拟连接、模拟游标
    mock_conn = MagicMock()
    mock_cursor = MagicMock()
    mock_connect.return_value = mock_conn
    mock_conn.cursor.return_value = mock_cursor

    # 初始化数据库实例，不会触发真实连接
    db_api = PostgresqlApi()

    # 校验连接参数是否正确
    connect_kwargs = mock_connect.call_args.kwargs
    assert connect_kwargs.get("sslmode") == "require"

    # 校验建表逻辑是否执行
    create_table_executed = False
    for call in mock_cursor.execute.call_args_list:
        sql = call[0][0]
        if "CREATE TABLE IF NOT EXISTS history" in sql:
            create_table_executed = True
            break
    assert create_table_executed
    mock_conn.commit.assert_called_once()

    # 测试新增事件逻辑
    mock_cursor.reset_mock()
    db_api.add_event(event="test_record")
    execute_sql, execute_params = mock_cursor.execute.call_args[0]
    assert "INSERT INTO history (event) VALUES (%s)" in execute_sql
    assert execute_params == ("test_record",)

    # 测试查询事件逻辑，预设返回值
    mock_cursor.reset_mock()
    mock_cursor.fetchall.return_value = [(1, "test_record", "2024-01-01 00:00:00+00")]
    result = db_api.events()
    assert result == [(1, "test_record", "2024-01-01 00:00:00+00")]
    assert "SELECT * FROM history" in mock_cursor.execute.call_args[0][0]

    # 测试关闭连接逻辑
    db_api.close()
    mock_cursor.close.assert_called_once()
    mock_conn.close.assert_called_once()

2. FastAPI接口测试示例 #

测试接口时通过FastAPI自带的依赖覆盖机制，把数据库连接替换为mock对象即可：

from unittest.mock import MagicMock
from fastapi.testclient import TestClient
import os

os.environ["DATABASE_URL"] = "postgresql://test:test@localhost:5432/testdb"
os.environ["APP_USERNAME"] = "test_user"
os.environ["APP_PASSWORD"] = "test_pass"

from main import app, db_connection
import main

# 测试前自动重置全局数据库对象，避免用例间状态污染
@pytest.fixture(autouse=True)
def reset_db_state():
    main.database = None
    yield
    main.database = None

client = TestClient(app)

def test_currencies接口正常返回():
    # 构造mock数据库对象
    mock_db = MagicMock()
    # 覆盖数据库连接依赖
    app.dependency_overrides[db_connection] = lambda: mock_db

    # 带正确认证信息请求
    resp = client.get("/currencies", auth=("test_user", "test_pass"))
    assert resp.status_code == 200
    assert resp.json() == {"get events": "ok"}

    # 认证失败场景
    resp = client.get("/currencies", auth=("wrong", "wrong"))
    assert resp.status_code == 401

• 纯mock方案仅适合业务逻辑单元测试，覆盖不到SQL语法校验、表约束、事务、复杂查询正确性等数据库特性，如果需要验证这部分逻辑，还是要通过临时PostgreSQL实例做集成测试。
• 不要随意扩大mock范围，只需要拦截真实连接的入口即可，业务逻辑本身要走真实代码路径，否则测试就失去了意义。
• 如果后续要扩展mock的游标方法，比如fetchone、fetchmany、事务回滚等，只需要给MagicMock实例添加对应的返回值或副作用即可。

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