Method test_poll_blocks_with_negative_ms

Lib/test/test_poll.py:214–232

(self)

    @unittest.skipUnless(threading, 'Threading required for this test.')
    @threading_helper.reap_threads
    def test_poll_blocks_with_negative_ms(self):
        for timeout_ms in [None, -1000, -1, -1.0, -0.1, -1e-100]:
            # Create two file descriptors. This will be used to unlock
            # the blocking call to poll.poll inside the thread
            r, w = os.pipe()
            pollster = select.poll()
            pollster.register(r, select.POLLIN)

            poll_thread = threading.Thread(target=pollster.poll, args=(timeout_ms,))
            poll_thread.start()
            poll_thread.join(timeout=0.1)
            self.assertTrue(poll_thread.is_alive())

            # Write to the pipe so pollster.poll unblocks and the thread ends.
            os.write(w, b'spam')
            poll_thread.join()
            self.assertFalse(poll_thread.is_alive())
            os.close(r)
            os.close(w)


if __name__ == '__main__':