mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-30 13:38:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-32356: idempotent pause_/resume_reading; new is_reading method. (#4914)

Browse source
This commit is contained in:
Yury Selivanov 2017-12-18 17:03:23 -05:00 • committed by GitHub
parent 2d8f06382e
commit d757aaf9dd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 87 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

28
Lib/test/test_asyncio/test_selector_events.py
View file

@ -80,10 +80,23 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
with test_utils.disable_logger():
transport = self.loop._make_ssl_transport(
m, asyncio.Protocol(), m, waiter)
with self.assertRaisesRegex(RuntimeError,
r'SSL transport.*not.*initialized'):
transport.is_reading()
# execute the handshake while the logger is disabled
# to ignore SSL handshake failure
test_utils.run_briefly(self.loop)
self.assertTrue(transport.is_reading())
transport.pause_reading()
transport.pause_reading()
self.assertFalse(transport.is_reading())
transport.resume_reading()
transport.resume_reading()
self.assertTrue(transport.is_reading())
# Sanity check
class_name = transport.__class__.__name__
self.assertIn("ssl", class_name.lower())
@ -894,15 +907,24 @@ class SelectorSocketTransportTests(test_utils.TestCase):
tr = self.socket_transport()
test_utils.run_briefly(self.loop)
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())
self.loop.assert_reader(7, tr._read_ready)
tr.pause_reading()
tr.pause_reading()
self.assertTrue(tr._paused)
self.assertFalse(7 in self.loop.readers)
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)
tr.resume_reading()
tr.resume_reading()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())
self.loop.assert_reader(7, tr._read_ready)
with self.assertRaises(RuntimeError):
tr.resume_reading()
tr.close()
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)
def test_read_ready(self):
transport = self.socket_transport()