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'
ThreadPool from multiprocessing module or ThreadPoolExecutor (for python version >= 3.2) are ways to get the return value of a thread.
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'
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'
Threadis well tested already, test that you make the right call, not that the call works (creates a thread).