Pytest + Multiprocessing Queue плохо играют вместе

Question

Я тестирую несколько очередей.Мой тест включает в себя простой класс с геттером и сеттером, охватывающий несколько очередей.

Мой тест чередуется между проходом и неудачей при повторном запуске i [править: это происходит при использовании block = False.Установка в True, как рекомендуется, приводит к зависанию программы.

Как мне переписать это так, чтобы

(1) очистить мульти-очередь между выполнениями программы и (2) прочитать все значения из очереди в .get ()метод?

import multiprocessing
import queue

class MyClass:
    def __init__(self):
        self.q = multiprocessing.Queue()
        self.results = []

    def put(self, x):
        self.q.put(x)

    def get(self):
        while True:
            try:
                self.results.append(self.q.get(block=True))
            except queue.Empty:
                break
        return self.results

    @pytest.fixture
    def wrapped_queue():
        yield MyClass()

    def test_multiprocessing_queue(wrapped_queue):
        wrapped_queue.put("a")
        wrapped_queue.put("b")
        result = wrapped_queue.get()
        assert result == ["a", "b"]

Answer 1

Это не проблема с pytest.За кулисами с очередью происходит больше, чем вы думаете.

Если вы установите block=True в своем геттере, это должно решить проблему.