如何用线程和回调函数测试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

来自多处理模块的ThreadPool或ThreadPoolExecutor（对于python版本>；=3.2） 是获取线程返回值的方法。你知道吗

与并发.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'

与多处理池线程池

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'