From d757aaf9dd767d13205bf9917e520ebf43e7f6e5 Mon Sep 17 00:00:00 2001 From: Yury Selivanov Date: Mon, 18 Dec 2017 17:03:23 -0500 Subject: bpo-32356: idempotent pause_/resume_reading; new is_reading method. (#4914) --- Lib/test/test_asyncio/test_selector_events.py | 28 ++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) (limited to 'Lib/test/test_asyncio/test_selector_events.py') diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py index b1ca3fcf0b1..89c3d5e16c1 100644 --- a/Lib/test/test_asyncio/test_selector_events.py +++ b/Lib/test/test_asyncio/test_selector_events.py @@ -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() -- cgit v1.2.3