mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-27 20:25:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-38019: correctly handle pause/resume reading of closed asyncio unix pipe (GH-16472)

Browse source
This commit is contained in:
Andrew Svetlov 2019-09-29 15:00:35 +03:00 • committed by GitHub
parent 9a7d951950
commit 58498bc717
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 39 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

27
Lib/test/test_asyncio/test_unix_events.py
View file

@ -736,6 +736,7 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
@mock.patch('os.read')
def test_resume_reading(self, m_read):
tr = self.read_pipe_transport()
tr.pause_reading()
tr.resume_reading()
self.loop.assert_reader(5, tr._read_ready)
@ -790,6 +791,32 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.assertIsNone(tr._protocol)
self.assertIsNone(tr._loop)
def test_pause_reading_on_closed_pipe(self):
tr = self.read_pipe_transport()
tr.close()
test_utils.run_briefly(self.loop)
self.assertIsNone(tr._loop)
tr.pause_reading()
def test_pause_reading_on_paused_pipe(self):
tr = self.read_pipe_transport()
tr.pause_reading()
# the second call should do nothing
tr.pause_reading()
def test_resume_reading_on_closed_pipe(self):
tr = self.read_pipe_transport()
tr.close()
test_utils.run_briefly(self.loop)
self.assertIsNone(tr._loop)
tr.resume_reading()
def test_resume_reading_on_paused_pipe(self):
tr = self.read_pipe_transport()
# the pipe is not paused
# resuming should do nothing
tr.resume_reading()
class UnixWritePipeTransportTests(test_utils.TestCase):