Pytest + 多处理队列不能很好地协同工作

Question

我正在对多队列进行pytest。我的测试涉及一个简单的类，其中包含一个getter和setter，围绕着一个多队列。

当我重新运行 i 时，我的测试在通过和失败之间交替 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的问题。排队的幕后事情比你意识到的要多。

如果您在 getter 中设置block=True，它应该可以解决问题。