Как протестировать Python с использованием потоков и функции обратного вызова?

Question

Я хочу проверить async_who функцию с помощью pytest.

Как проверить вызов обратного вызова и получить возвращаемое значение '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 didnне вернуть значение.Я не могу этого сделать,

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

Answer 1

ThreadPool из многопроцессорного модуля или ThreadPoolExecutor (для версии Python> = 3.2) способы получить возвращаемое значение потока.

С 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'

С мультипроцессором. 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'