From 3be40bd7d53369fb1895b9664fe4b91b900cff31 Mon Sep 17 00:00:00 2001 From: azibom Date: Sat, 23 May 2026 15:04:53 +0000 Subject: [PATCH] Add tests for computed flow control write limits --- Lib/test/test_asyncio/test_transports.py | 28 ++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/Lib/test/test_asyncio/test_transports.py b/Lib/test/test_asyncio/test_transports.py index dbb572e2e1536b..2b98f09c937e6e 100644 --- a/Lib/test/test_asyncio/test_transports.py +++ b/Lib/test/test_asyncio/test_transports.py @@ -98,6 +98,34 @@ def get_write_buffer_size(self): self.assertTrue(transport._protocol_paused) self.assertEqual(transport.get_write_buffer_limits(), (128, 256)) + def test_flowcontrol_mixin_compute_write_limits(self): + + class MyTransport(transports._FlowControlMixin, + transports.Transport): + + def get_write_buffer_size(self): + return 0 + + loop = mock.Mock() + transport = MyTransport(loop=loop) + + self.assertEqual( + transport.get_write_buffer_limits(), + (16 * 1024, 64 * 1024) + ) + + transport.set_write_buffer_limits(low=100) + self.assertEqual( + transport.get_write_buffer_limits(), + (100, 400) + ) + + transport.set_write_buffer_limits(high=200) + self.assertEqual( + transport.get_write_buffer_limits(), + (50, 200) + ) + if __name__ == '__main__': unittest.main()