test_proactor_events.py 文件源码

def test_pause_writing_2write(self):
        tr = self.pause_writing_transport(high=4)

        # first short write, the buffer is not full (3 <= 4)
        fut1 = asyncio.Future(loop=self.loop)
        self.loop._proactor.send.return_value = fut1
        tr.write(b'123')
        self.loop._run_once()
        self.assertEqual(tr.get_write_buffer_size(), 3)
        self.assertFalse(self.protocol.pause_writing.called)

        # fill the buffer, must pause writing (6 > 4)
        tr.write(b'abc')
        self.loop._run_once()
        self.assertEqual(tr.get_write_buffer_size(), 6)
        self.assertTrue(self.protocol.pause_writing.called)