如何使用线程和回调函数测试 python

Question

我想通过pytest测试async_who函数。

如何测试调用回调并且返回值为"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)

因为async_who没有返回值。我做不到，

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

Answer 1

来自多处理模块或 ThreadPoolExecutor 的 ThreadPool(对于 python 版本>= 3.2(是获取线程返回值的方法。

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'

使用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'