Pytest 模拟 SQLAlchemy

0 投票
1 回答
24 浏览
提问于 2025-04-12 13:21

当我在为我创建的路由写单元测试时,遇到了下面这个错误:

tests/routes/test_users_routes.py - sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) 连接到服务器 "localhost" (::1),端口 5439 失败:连接被拒绝

下面是我注册用户的路由方法,基于一些参数进行注册:

@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."

我本来期待能得到其他的错误信息,同时在模拟实际的数据库会话时遇到问题。

1 个回答

0

为了让它正常工作,你需要重写一下 get_db 这个依赖。

from unittest import mock

@mock.patch("app.routes.users.create_user_method", new_callable=mock.MagicMock)
def test_create_user(create_user_mock):
    """
    Test case to create a user when API status is True
    """
    def fake_get_db():
        return mock.Mock()

    app.dependency_overrides[get_db] = fake_get_db  # Here!
    create_user_mock.return_value = True, {}, "User Registered successfully.", 200

    payload = {
        "data": {
            "user": {
                "first_name": "Shaheer",
                "last_name": "Muhammad",
                "email": "shaheer@yl.com",
                "mobile": "1234567890",
                "password": "Shaheer@88888",
            }
        }
    }
    response = client.post("/api/v1/users/register", json=payload)

    assert response.status_code == 200
    assert response.json()["status"] is True
    assert response.json()["message"] == "User Registered successfully."

请注意,测试结束后你需要把原来的依赖恢复回来,不然会影响其他使用真实数据库的测试。

有关详细信息,可以参考 这个链接这个链接

撰写回答