Pytest 模拟 SQLalchmey

Question

当我为我创建的路线编写 UT 时，我遇到了以下错误

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

下面是我的路线方法，我根据一些参数注册用户

@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": "[email protected]",
                "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."

我期待在模拟实际的数据库会话时出现一些其他错误。

Answer 1

为了使其工作，您必须覆盖

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": "[email protected]",
                "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."

请注意，您必须在测试后恢复原始依赖关系，否则会影响其他测试，例如使用实际数据库的测试。

为此，请参阅https://stackoverflow.com/a/68687967和https://stackoverflow.com/a/72203335