How to properly mock functions inside FastAPI endpoint handler

Question

I'm having trouble mocking a function that resides inside a FastAPI endpoint handler. No matter what I've tried it doesn't seem to work. A brief explanation of what I'm trying to test. I have a single endpoint that receives a file and does a few things with it including uploading to an S3 bucket using Boto3. The call to boto is in it's own function which is the one I need to mock. I can easily mock it when calling the function directly but when using a FastAPI TestClient to call the endpoint it never works as expected.

# a_module/s3_upload.py

def s3_upload(bucket, filename, file):
    client = boto3.client('s3')
    client.put_object(Bucket="a-bucket" Key=filename, body=file)

# endpoints/route_file.py

from a_module import s3_upload

@app.put("/")
def route_file(file, filename):
    check_valid_file()
    s3_upload()

@pytest.fixture
def client():
    yield TestClient(app)

# tests/test_routes.py
def test_route_file(client, mocker):
    test_file = <whatever>
    mocker.patch("endpoints.s3_upload")
    response = client.put("/", file=test_file)

When I run this test I will basically see it fail trying to legitimately upload to the s3. I cannot get it to either use my mocked s3 client and bucket, or to simply mock the s3_upload function in the first place. I'm assuming this has something to do with the call to FastAPI, because this works just find if testing the handler directly without the TestClient.

Can someone point me in the right direction on how to properly mock objects or functions that are going to be called inside a handler?

Answer 1

I was facing an issue like this as well, also in the context of wanting to mock boto3. The problem, as other commenters have indicated, is that you are not patching the correct path.

This might be as simple as a typo, or if you're like me maybe you were trying to patch the original source value of the s3 helper (aka a_module.s3_upload.s3_upload) instead of the handler's value of the s3 helper (aka endpoints.route_file.s3_upload).

My understanding is that in Python variables are references, and at the point a module is imported Python will populate the module tree / ALL of the import references get locked in across various modules.

This means that at the point you define your TestClient(app) fixture, a_module.s3_upload.s3_upload is pointing to a memory address, and endpoints.route_file.s3_upload has also been populated to point to that same memory space (due to the line from a_module import s3_upload).

When you invoke the patch on a_module.s3_upload.s3_upload, the mocker changes the reference of a_module.s3_upload.s3_upload to point to your mocked value. However, endpoints.route_file.s3_upload does NOT get updated, because why would it? Its value was populated long ago, it's not being constantly synced to match the original import's value.

However, if you patch endpoints.route_file.s3_upload then you'll be in the clear -- the handler will use the mocked value.

---

In case another example helps, in my case I had a layout like this:

service/core/s3.py

from service.core.settings import settings
import boto3

s3_client = boto3.client(
    "s3",
    region_name=settings.S3_REGION,
    endpoint_url=settings.S3_ENDPOINT,
    aws_access_key_id=settings.S3_ACCESS_KEY_ID,
    aws_secret_access_key=settings.S3_SECRET_ACCESS_KEY,
)

service/api/handlers/presigned_posts.py

import uuid
from service.core.s3 import s3_client
from service.core.settings import settings


async def create_presigned_post():
    key = uuid.uuid4()
    print(s3_client)

    presigned_post = s3_client.generate_presigned_post(
        Bucket=settings.S3_BUCKET, Key=key
    )
    return presigned_post

service/api/routers/presigned_posts.py

from fastapi import APIRouter
from service.api.handlers.presigned_posts import create_presigned_post


router = APIRouter()

router.post("/")(create_presigned_post)

tests/test_presigned_post.py

import pytest
from fastapi.testclient import TestClient
from service.main import app

client = TestClient(app)


@pytest.mark.integration
def test_create_presigned_post(mocker):
    mock_s3_client = mocker.Mock()
    mock_s3_client.generate_presigned_post.return_value = {
        "url": "https://mock-s3-url.com",
        "fields": {
            "key": "my-key",
            "acl": "public-read",
            "policy": "mock-policy",
            "signature": "mock-signature",
        },
    }
    mocker.patch("service.core.s3.s3_client", mock_s3_client)

    response = client.post("/presignedPosts")

    assert response.json() == {
        "url": "https://mock-s3-url.com",
        "fields": {
            "key": "my-key",
            "acl": "public-read",
            "policy": "mock-policy",
            "signature": "mock-signature",
        },
    }

---

The problem with my setup was the line:

mocker.patch("service.core.s3.s3_client", mock_s3_client)

As part of my exploration I did get it to work by moving the import below the mock:

mocker.patch("service.core.s3.s3_client", mock_s3_client)
from service.main import app
client = TestClient(app)

This works because service.core.s3.s3_client is replaced with the mock, and then when I import the app it holds that mocked value AT THE POINT that service.api.handlers.presigned_posts.s3_client is populated, and so the mock is also used by the handler, and my test succeeds.

But importing the app and defining a full blown TestClient inside of an individual test is ugly, bad form, and no way to live.

The real solution is to simply change that mocker line to:

mocker.patch("service.api.handlers.presigned_posts.s3_client", mock_s3_client)