Method test_closed

Lib/test/_test_multiprocessing.py:6076–6080

(self)

    # Test specific to CPython since it tests private attributes
    @test.support.cpython_only
    def test_closed(self):
        queue = multiprocessing.SimpleQueue()
        queue.close()
        self.assertTrue(queue._reader.closed)
        self.assertTrue(queue._writer.closed)


class TestPoolNotLeakOnFailure(unittest.TestCase):