编写路由单元测试遇SQLAlchemy连接拒绝错误，求Mock数据库方案

在为自研路由编写单元测试时，遇到以下错误：

tests/routes/test_users_routes.py - sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "localhost" (::1), port 5439 failed: Connection refused

期望Mock实际数据库会话，但仍触发了真实数据库连接请求，请求解决。

路由方法代码 #

@router.post("/register")
def create_user(user: User, db: Session = Depends(get_db)):
    logger.info("Initiating the router to create user")
    request = user.dict()
    try:
        api_status, response, message, status_code = create_user_method(db, request)
        if api_status:
            return JSONResponse(
                status_code=status_code,
                content={"message": message, "status": api_status, "data": response},
            )
        else:
            return JSONResponse(
                status_code=status_code,
                content={"message": message, "status": api_status, "data": response},
            )
    except Exception as err:
        logger.error(err)
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"message": str(err), "status": False, "data": None},
        )

测试函数代码 #

@patch("app.routes.users.create_user_method", new_callable=MagicMock)
def test_create_user_1(create_user_mock,get_db_mock):
    """
    Test case to create a user when API status is True
    """
    # def get_db():
    #     return mock_session_local
    
    # app.dependency_overrides[get_db] = get_db
    mock_session_local = MagicMock()
    get_db_mock.return_value = mock_session_local

    payload = {
        "data": {
            "user": {
                "first_name": "Shaheer",
                "last_name": "Muhammad",
                "email": "shaheer@yl.com",
                "mobile": "1234567890",
                "password": "Shaheer@88888",
            }
        }
    }

    create_user_mock.return_value = True, {}, "User Registered successfully.", 200

    response = client.post("/api/v1/user/register", json=payload)
    assert response.status_code == 200
    assert response.json()["status"] is True
    assert response.json()["message"] == "User Registered successfully."

解决方法 #

问题根源 #

测试代码存在两个关键问题：

• get_db_mock参数未被正确注入：没有为get_db添加patch装饰器，该参数实际无效，导致get_db仍调用真实数据库会话生成逻辑，触发了真实的PostgreSQL连接请求。
• 注释的依赖覆盖代码未生效，没有替换FastAPI中get_db的默认实现。

修正后的测试代码 #

from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock
from app.main import app  # 导入你的FastAPI实例
from app.routes.users import get_db  # 导入get_db依赖

client = TestClient(app)

@patch("app.routes.users.create_user_method", new_callable=MagicMock)
def test_create_user_1(create_user_mock):
    """
    Test case to create a user when API status is True
    """
    # 创建Mock的数据库会话
    mock_session_local = MagicMock()
    
    # 覆盖get_db依赖，返回Mock会话
    app.dependency_overrides[get_db] = lambda: mock_session_local

    payload = {
        "first_name": "Shaheer",
        "last_name": "Muhammad",
        "email": "shaheer@yl.com",
        "mobile": "1234567890",
        "password": "Shaheer@88888",
    }

    create_user_mock.return_value = True, {}, "User Registered successfully.", 200

    response = client.post("/api/v1/user/register", json=payload)
    
    # 断言结果
    assert response.status_code == 200
    assert response.json()["status"] is True
    assert response.json()["message"] == "User Registered successfully."

    # 测试结束后清除依赖覆盖，避免影响其他测试
    app.dependency_overrides.clear()

额外说明 #

1. Payload格式修正：原测试中的payload嵌套了data.user，但路由接收的是User模型，直接传用户字段即可，否则会触发请求体验证失败。
2. 依赖覆盖清理：测试结束后调用app.dependency_overrides.clear()，确保其他测试不受当前覆盖的影响。
3. Mock有效性：因为已经patch了create_user_method，即使Mock的会话没有任何真实数据库操作逻辑，也不会影响测试结果——create_user_method的返回值已经被固定，不会执行真实的数据库代码。

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