Method test_send_buffer_timeout

Lib/test/test__interpchannels.py:942–977

(self)

            t.join()

    def test_send_buffer_timeout(self):
        try:
            self._has_run_once_timeout
        except AttributeError:
            # At the moment, this test leaks a few references.
            # It looks like the leak originates with the addition
            # of _channels.send_buffer() (gh-110246), whereas the
            # tests were added afterward.  We want this test even
            # if the refleak isn't fixed yet, so we skip here.
            raise unittest.SkipTest('temporarily skipped due to refleaks')
        else:
            self._has_run_once_timeout = True

        obj = bytearray(b'spam')

        with self.subTest('non-blocking with timeout'):
            cid = _channels.create(REPLACE)
            with self.assertRaises(ValueError):
                _channels.send_buffer(cid, obj, blocking=False, timeout=0.1)

        with self.subTest('timeout hit'):
            cid = _channels.create(REPLACE)
            with self.assertRaises(TimeoutError):
                _channels.send_buffer(cid, obj, blocking=True, timeout=0.1)
            with self.assertRaises(_channels.ChannelEmptyError):
                received = recv_nowait(cid)
                print(repr(received))

        with self.subTest('timeout not hit'):
            cid = _channels.create(REPLACE)
            def f():
                recv_wait(cid)
            t = threading.Thread(target=f)
            t.start()
            _channels.send_buffer(cid, obj, blocking=True, timeout=10)
            t.join()

    def test_send_closed_while_waiting(self):
        obj = b'spam'