Mocking service objects in FastAPI controllers using pytest.mock and patch

Viewed 1791

I'm working on a FastAPI application where I need to test the controller behavior without actually triggering the internal service call. My goal is to implement something similar to an instance_double in RSpec

My controller code looks like this

# module_name.api.routers.v1.py

@router.post("/notes")
async def create(note: Note):
    s3_persister = S3Persister()
    kafka_persister = KafkaPersister()
    
    service = Persist.create(
        note=note,
        persisters=[s3_persister, kafka_persister]        
    )
    await service.invoke()
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=success_response(payload=jsonable_encoder(service.note)),
    )

The corresponding unit test that I have managed to get working in a non Pythonic way is

@pytest.fixture(scope='session')
def client():    
    client = TestClient(app)
    yield client  # testing happens here

def test_create_success_response(client):
    """
    When the required parameters are passed
        it should return a success response        
    """
    # copies some sample request body data
    data = payload.copy() 
    
    # creates a note object using Pydantic models
    valid_note = Note(**valid_note_payload) 

    # creates an service object with no persisters
    persist_svc = Persist(note=note, persisters=[], event_publisher_version='1.0') 
  
    # mock the method that I intend to avoid calling in the controller flow
    persist_svc.invoke = AsyncMock()     
   
    # A objects that will reflect on the svc_object post processing
    persist_svc.note = valid_note

    # create a mock of on the factory method so it returns my modified object
    Persist.create = Mock(return_value=persist_svc)    

    response = client.post('/api/v1/notes', 
        headers=default_headers,
        data=json.dumps(data)
    )
    assert persist_svc.invoke.called
    assert response.status_code == 200

While this works I am unable to replicate this using the recommended @patch decorator.

I notice that the patched instance returned by the decorator isn't called and Fastapi ends up calling the actual service methods.

I was forced to add a create class method to the Persist service purely because I couldn't work with the default constructor object to obtain the same result.

@patch('module_name.api.services.persist.Persist')    
def test_create_success_response(client, mockPersist):
    data = payload.copy()          

    # I expect `persistInstance` to behave like an instance variable
    persistInstance = mockPersist.return_value    

    valid_note = Note(**valid_note_payload)

    # I attempt to mock the instance method
    persistInstance.invoke = AsyncMock(return_value=valid_note)    
    persistInstance.note.return_value = valid_note    
    

    response = client.post('/api/v1/notes', 
        headers=default_headers,
        data=json.dumps(data)
    )
    assert persistInstance.invoke.called
    assert response.status_code == 200

The idea here is that I don't create another controller entirely as recommended by the FastAPI docs but instead mock the service method call alone.

I would like help on how to get write the unitTest in the recommended Pythonic way.

0 Answers
Related