mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

asyncio.streams.StreamReader: Add 'at_eof()' method

Browse source
This commit is contained in:
Yury Selivanov 2014-02-06 00:14:30 -05:00
parent e694c9745f
commit f0020f5d77
2 changed files with 19 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

4
Lib/asyncio/streams.py
View file

@ -293,6 +293,10 @@ class StreamReader:
if not waiter.cancelled(): if not waiter.cancelled():
waiter.set_result(True) waiter.set_result(True)
def at_eof(self):
"""Return True if the buffer is empty and 'feed_eof' was called."""
return self._eof and not self._buffer
def feed_data(self, data): def feed_data(self, data):
assert not self._eof, 'feed_data after feed_eof' assert not self._eof, 'feed_data after feed_eof'

15
Lib/test/test_asyncio/test_streams.py
View file

@ -204,6 +204,21 @@ class StreamReaderTests(unittest.TestCase):
# expected to be empty now. # expected to be empty now.
self.assertEqual(b'', stream._buffer) self.assertEqual(b'', stream._buffer)
def test_at_eof(self):
stream = asyncio.StreamReader(loop=self.loop)
self.assertFalse(stream.at_eof())
stream.feed_data(b'some data\n')
self.assertFalse(stream.at_eof())
self.loop.run_until_complete(stream.readline())
self.assertFalse(stream.at_eof())
stream.feed_data(b'some data\n')
stream.feed_eof()
self.loop.run_until_complete(stream.readline())
self.assertTrue(stream.at_eof())
def test_readline_limit(self): def test_readline_limit(self):
# Read one line. StreamReaders are fed with data after # Read one line. StreamReaders are fed with data after
# their 'readline' methods are called. # their 'readline' methods are called.