Method test_send_timeout

Lib/test/test__interpchannels.py:917–940

(self)

        self.assertEqual(received, obj)

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

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

        with self.subTest('timeout hit'):
            cid = _channels.create(REPLACE)
            with self.assertRaises(TimeoutError):
                _channels.send(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(cid, obj, blocking=True, timeout=10)
            t.join()

    def test_send_buffer_timeout(self):
        try: