How I can mock an async function with pytest

Question

I need to test a function with a async function inside, but I do not know how to mock the async function.

matching_ingr_zingr.py

def first_table(ingredient_raw, ingredient_extracted, ingredient_processed):
    ingredient_body, table_rows_I = [], []
    for idx, ing_extr in enumerate(ingredient_extracted):

        ingr_extr = " ".join(list(set(ing_extr.split())))

        ext_ing_st = stemization(ingredient_extracted[idx])

        try:
            # # ======== MULTI-SEARCH
            ingredient_body = format_for_batch_search(ingr_extr, ingredient_raw[idx], ext_ing_st)
            res = asyncio.run(batch_search_v2(ingredient_body))
            
        except BaseException:
            res = 'Não retornou nada do Banco de Dados'
            continue

        result_search_clean, score, zingr_id = eliminate_duplicates_search(res) 

        for l in range(len(result_search_clean)):

            proc_zing = text_normalization(result_search_clean[l])

            table_rows_I.append({'Raw_Ingred': ingredient_raw[idx],
                                 'Zapl_Ingre': result_search_clean[l],
                                 'Proc_Ingre': ingredient_processed[idx],
                                 'Extr_Ingre': ingredient_extracted[idx],
                                 'Score_Elas': score[l],
                                 'Ext_Ing_St': ext_ing_st,
                                 'Proc_Zingr': proc_zing,
                                 'Stem_Zingr': stemization(proc_zing),
                                 'Zingred_id': zingr_id[l]})
    return table_rows_I

The line to be mocked is asyncio.run(batch_search_v2(ingredient_body)) in the above code. To test the function first_table() I write the test below:

test_matching_ingr_zingr.py

@pytest.mark.asyncio
def test_first_table_entrada_correta(mocker):
    ingredient_raw = ['Colher de açaí 2colher(es) de sopa']
    ingredient_extracted = ('acai',)
    ingredient_processed = ('colher acai  colheres sopa',)
    result_expected = table_rows_result

    # mock mocking uma função assincrona (asynchronous function)

    mocker.patch('src.services.matching_ingr_zingr.batch_search_v2', return_value=async_res)

    # mocker.patch('src.services.matching_ingr_zingr.batch_search_v2', return_value=async_res)
    result = first_table(ingredient_raw, ingredient_extracted, ingredient_processed)

    

    assert result == result_expected

The variable table_rows_result is imported from another file to test. Can anyone help me to learn how to do mock this async function, I want to test the first_table() but I dont want to acess the DB by batch_search_v2() async function.

The estructure of async_res is a list of tuples with lenght 3:

async_res = [('polpa de açaí', 9.626554, 2779),
             ('açaí', 8.914546, 1764),
             ('sopa de cebola', 8.442016, 388405)]

What is the structure of res when it completes without error? mock.patch async.io.run with return_value=that. — Amos Baker
– Amos Baker, Commented Aug 11, 2022 at 19:21
@AmosBaker I put (now) the estructure of async_res in the final of the question. — Atilio
– Atilio, Commented Aug 11, 2022 at 19:28

Amos Baker · Accepted Answer · 2022-08-11 20:25:24Z

0

As described

@pytest.mark.asyncio
def test_first_table_entrada_correta(mocker):
    async_res = [('polpa de açaí', 9.626554, 2779),
        ('açaí', 8.914546, 1764),
        ('sopa de cebola', 8.442016, 388405)]
    with mock.patch(asyncio.run) as mock_async:
        mock_async.configure_mock(return_value=async_res)
        ingredient_raw = ['Colher de açaí 2colher(es) de sopa']
        ingredient_extracted = ('acai',)
        ingredient_processed = ('colher acai  colheres sopa',)
        result_expected = table_rows_result

        result = first_table(ingredient_raw, ingredient_extracted, ingredient_processed)

        

        assert result == result_expected

answered Aug 11, 2022 at 20:25

Amos Baker

8497 silver badges13 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Atilio Over a year ago

Hi, thanks for help, but it raise an error: AttributeError: 'function' object has no attribute 'rsplit'

Collectives™ on Stack Overflow

How I can mock an async function with pytest

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related