asyncio.streams: Use bytebuffer in StreamReader; Add assertion in feed_data

1 files changed, 52 insertions, 27 deletions

diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py
index 01d565c..83474a8 100644
--- a/Lib/test/test_asyncio/test_streams.py
+++ b/Lib/test/test_asyncio/test_streams.py
@@ -79,13 +79,13 @@ class StreamReaderTests(unittest.TestCase):
         stream = asyncio.StreamReader(loop=self.loop)
 
         stream.feed_data(b'')
-        self.assertEqual(0, stream._byte_count)
+        self.assertEqual(b'', stream._buffer)
 
-    def test_feed_data_byte_count(self):
+    def test_feed_nonempty_data(self):
         stream = asyncio.StreamReader(loop=self.loop)
 
         stream.feed_data(self.DATA)
-        self.assertEqual(len(self.DATA), stream._byte_count)
+        self.assertEqual(self.DATA, stream._buffer)
 
     def test_read_zero(self):
         # Read zero bytes.
@@ -94,7 +94,7 @@ class StreamReaderTests(unittest.TestCase):
 
         data = self.loop.run_until_complete(stream.read(0))
         self.assertEqual(b'', data)
-        self.assertEqual(len(self.DATA), stream._byte_count)
+        self.assertEqual(self.DATA, stream._buffer)
 
     def test_read(self):
         # Read bytes.
@@ -107,7 +107,7 @@ class StreamReaderTests(unittest.TestCase):
 
         data = self.loop.run_until_complete(read_task)
         self.assertEqual(self.DATA, data)
-        self.assertFalse(stream._byte_count)
+        self.assertEqual(b'', stream._buffer)
 
     def test_read_line_breaks(self):
         # Read bytes without line breaks.
@@ -118,7 +118,7 @@ class StreamReaderTests(unittest.TestCase):
         data = self.loop.run_until_complete(stream.read(5))
 
         self.assertEqual(b'line1', data)
-        self.assertEqual(5, stream._byte_count)
+        self.assertEqual(b'line2', stream._buffer)
 
     def test_read_eof(self):
         # Read bytes, stop at eof.
@@ -131,7 +131,7 @@ class StreamReaderTests(unittest.TestCase):
 
         data = self.loop.run_until_complete(read_task)
         self.assertEqual(b'', data)
-        self.assertFalse(stream._byte_count)
+        self.assertEqual(b'', stream._buffer)
 
     def test_read_until_eof(self):
         # Read all bytes until eof.
@@ -147,7 +147,7 @@ class StreamReaderTests(unittest.TestCase):
         data = self.loop.run_until_complete(read_task)
 
         self.assertEqual(b'chunk1\nchunk2', data)
-        self.assertFalse(stream._byte_count)
+        self.assertEqual(b'', stream._buffer)
 
     def test_read_exception(self):
         stream = asyncio.StreamReader(loop=self.loop)
@@ -161,7 +161,8 @@ class StreamReaderTests(unittest.TestCase):
             ValueError, self.loop.run_until_complete, stream.read(2))
 
     def test_readline(self):
-        # Read one line.
+        # Read one line. 'readline' will need to wait for the data
+        # to come from 'cb'
         stream = asyncio.StreamReader(loop=self.loop)
         stream.feed_data(b'chunk1 ')
         read_task = asyncio.Task(stream.readline(), loop=self.loop)
@@ -174,30 +175,40 @@ class StreamReaderTests(unittest.TestCase):
 
         line = self.loop.run_until_complete(read_task)
         self.assertEqual(b'chunk1 chunk2 chunk3 \n', line)
-        self.assertEqual(len(b'\n chunk4')-1, stream._byte_count)
+        self.assertEqual(b' chunk4', stream._buffer)
 
     def test_readline_limit_with_existing_data(self):
-        stream = asyncio.StreamReader(3, loop=self.loop)
+        # Read one line. The data is in StreamReader's buffer
+        # before the event loop is run.
+
+        stream = asyncio.StreamReader(limit=3, loop=self.loop)
         stream.feed_data(b'li')
         stream.feed_data(b'ne1\nline2\n')
 
         self.assertRaises(
             ValueError, self.loop.run_until_complete, stream.readline())
-        self.assertEqual([b'line2\n'], list(stream._buffer))
+        # The buffer should contain the remaining data after exception
+        self.assertEqual(b'line2\n', stream._buffer)
 
-        stream = asyncio.StreamReader(3, loop=self.loop)
+        stream = asyncio.StreamReader(limit=3, loop=self.loop)
         stream.feed_data(b'li')
         stream.feed_data(b'ne1')
         stream.feed_data(b'li')
 
         self.assertRaises(
             ValueError, self.loop.run_until_complete, stream.readline())
-        self.assertEqual([b'li'], list(stream._buffer))
-        self.assertEqual(2, stream._byte_count)
+        # No b'\n' at the end. The 'limit' is set to 3. So before
+        # waiting for the new data in buffer, 'readline' will consume
+        # the entire buffer, and since the length of the consumed data
+        # is more than 3, it will raise a ValudError. The buffer is
+        # expected to be empty now.
+        self.assertEqual(b'', stream._buffer)
 
     def test_readline_limit(self):
-        stream = asyncio.StreamReader(7, loop=self.loop)
+        # Read one line. StreamReaders are fed with data after
+        # their 'readline' methods are called.
 
+        stream = asyncio.StreamReader(limit=7, loop=self.loop)
         def cb():
             stream.feed_data(b'chunk1')
             stream.feed_data(b'chunk2')
@@ -207,10 +218,25 @@ class StreamReaderTests(unittest.TestCase):
 
         self.assertRaises(
             ValueError, self.loop.run_until_complete, stream.readline())
-        self.assertEqual([b'chunk3\n'], list(stream._buffer))
-        self.assertEqual(7, stream._byte_count)
+        # The buffer had just one line of data, and after raising
+        # a ValueError it should be empty.
+        self.assertEqual(b'', stream._buffer)
+
+        stream = asyncio.StreamReader(limit=7, loop=self.loop)
+        def cb():
+            stream.feed_data(b'chunk1')
+            stream.feed_data(b'chunk2\n')
+            stream.feed_data(b'chunk3\n')
+            stream.feed_eof()
+        self.loop.call_soon(cb)
+
+        self.assertRaises(
+            ValueError, self.loop.run_until_complete, stream.readline())
+        self.assertEqual(b'chunk3\n', stream._buffer)
 
-    def test_readline_line_byte_count(self):
+    def test_readline_nolimit_nowait(self):
+        # All needed data for the first 'readline' call will be
+        # in the buffer.
         stream = asyncio.StreamReader(loop=self.loop)
         stream.feed_data(self.DATA[:6])
         stream.feed_data(self.DATA[6:])
@@ -218,7 +244,7 @@ class StreamReaderTests(unittest.TestCase):
         line = self.loop.run_until_complete(stream.readline())
 
         self.assertEqual(b'line1\n', line)
-        self.assertEqual(len(self.DATA) - len(b'line1\n'), stream._byte_count)
+        self.assertEqual(b'line2\nline3\n', stream._buffer)
 
     def test_readline_eof(self):
         stream = asyncio.StreamReader(loop=self.loop)
@@ -244,9 +270,7 @@ class StreamReaderTests(unittest.TestCase):
         data = self.loop.run_until_complete(stream.read(7))
 
         self.assertEqual(b'line2\nl', data)
-        self.assertEqual(
-            len(self.DATA) - len(b'line1\n') - len(b'line2\nl'),
-            stream._byte_count)
+        self.assertEqual(b'ine3\n', stream._buffer)
 
     def test_readline_exception(self):
         stream = asyncio.StreamReader(loop=self.loop)
@@ -258,6 +282,7 @@ class StreamReaderTests(unittest.TestCase):
         stream.set_exception(ValueError())
         self.assertRaises(
             ValueError, self.loop.run_until_complete, stream.readline())
+        self.assertEqual(b'', stream._buffer)
 
     def test_readexactly_zero_or_less(self):
         # Read exact number of bytes (zero or less).
@@ -266,11 +291,11 @@ class StreamReaderTests(unittest.TestCase):
 
         data = self.loop.run_until_complete(stream.readexactly(0))
         self.assertEqual(b'', data)
-        self.assertEqual(len(self.DATA), stream._byte_count)
+        self.assertEqual(self.DATA, stream._buffer)
 
         data = self.loop.run_until_complete(stream.readexactly(-1))
         self.assertEqual(b'', data)
-        self.assertEqual(len(self.DATA), stream._byte_count)
+        self.assertEqual(self.DATA, stream._buffer)
 
     def test_readexactly(self):
         # Read exact number of bytes.
@@ -287,7 +312,7 @@ class StreamReaderTests(unittest.TestCase):
 
         data = self.loop.run_until_complete(read_task)
         self.assertEqual(self.DATA + self.DATA, data)
-        self.assertEqual(len(self.DATA), stream._byte_count)
+        self.assertEqual(self.DATA, stream._buffer)
 
     def test_readexactly_eof(self):
         # Read exact number of bytes (eof).
@@ -306,7 +331,7 @@ class StreamReaderTests(unittest.TestCase):
         self.assertEqual(cm.exception.expected, n)
         self.assertEqual(str(cm.exception),
                          '18 bytes read on a total of 36 expected bytes')
-        self.assertFalse(stream._byte_count)
+        self.assertEqual(b'', stream._buffer)
 
     def test_readexactly_exception(self):
         stream = asyncio.StreamReader(loop=self.loop)