Как протестировать python с функцией потоковой передачи и обратного вызова?

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

Как проверить вызов обратного вызова и возвращаемое значение «Боб»

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'

Даже если это выглядит очень просто, ваш тест слишком сложен. Тестируйте каждую функцию отдельно и имитируйте то, что вы не тестируете. Например, Thread уже хорошо протестирован, проверьте, что вы делаете правильный вызов, а не то, что вызов работает (создает поток).

— 21.06.2019 12:15

python pytest

21.06.2019 12:03

Почему в Python есть оператор "pass"?

Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.

Некоторые методы, о которых вы не знали, что они существуют в Python

Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...

Основы Python Часть I

Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?

LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа

Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:

Оптимизация кода с помощью тернарного оператора Python

И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это

Советы по эффективной веб-разработке с помощью Python

Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.

1 313

Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

пул потоков из многопроцессорного модуля или 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'

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

21.06.2019 13:52