How to test python with threading and callback function?

Question

I want to test async_who function by pytest.

How do I test callback is called and the return value is 'Bob'

import threading


def async_who(callback):
    t = threading.Thread(target=_who, args=(callback,))
    t.start()


def _who(callback):
    return callback('Bob')


def callback(name):
    print(name)
    return name


async_who(callback)

Because the async_who didn't return value. I can't do this,

def test_async_who():
    res = async_who(callback)
    assert res == 'Bob'

Even that it looks very simple your test is too complex. Test every function separately and mock what you don't test. For example `Thread` is well tested already, test that you make the right call, not that the call works (creates a thread). — Klaus D., Jun 21 '19 at 10:15

score 3 · Accepted Answer · answered Jun 21 '19 at 11:52

ThreadPool from multiprocessing module or ThreadPoolExecutor (for python version >= 3.2) are ways to get the return value of a thread.

With concurrent.futures.ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor

def async_who(callback):
    executor = ThreadPoolExecutor(max_workers=2)
    res = executor.submit(_who, callback)

    return res.result()

def _who(callback):
    return callback('Bob')


def callback(name):
    print(name)
    return name

def test_async_who():
    res = async_who(callback)
    assert res == 'Bob'

With multiprocessing.pool.ThreadPool

from multiprocessing.pool import ThreadPool
pool = ThreadPool(processes=2)


def async_who(callback):
    res = pool.apply_async(_who, args=(callback,))
    return res.get()


def _who(callback):
    return callback('Bob')


def callback(name):
    print(name)
    return name


def test_async_who():
    res = async_who(callback)
    assert res == 'Bob'

How to test python with threading and callback function?

1 Answers1

With concurrent.futures.ThreadPoolExecutor

With multiprocessing.pool.ThreadPool