CODEWITHSUNDEEP
pythonqueuepytestpython-multiprocessing
GlaceCelery
GlaceCelery

Reputation: 1043

Pytest + Multiprocessing Queue don't play well together

I'm pytesting a multi-queue. My test involves a simple class with a getter and setter, wrapping around a multi-queue.

My test alternates between pass and fail as I re-run i [edit: this occurs when using block=False. Setting it to True as recommended hangs the program].

How do I re-write this so as to

(1) clear the multi-queue between program executions and (2) Read all the values from the queue in the .get() method?

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"]

Upvotes: 6

Views: 7980

Answers (1)

Dane White
Dane White

Reputation: 3542

This isn't an issue with pytest. There's more going on behind the scenes with queue than you realize.

If you set block=True in your getter, it should fix the problem.

Upvotes: 3

Related Questions