Coverage for 'gcloud._apitools.buffered_stream'

gcloud/_apitools/buffered_stream.py

@@ -13,32 +13,32 @@ class BufferedStream(object):
     """Buffers a stream, reading ahead to determine if we're at the end."""
 
     def __init__(self, stream, start, size):
-        self.__stream = stream
-        self.__start_pos = start
-        self.__buffer_pos = 0
-        self.__buffered_data = self.__stream.read(size)
-        self.__stream_at_end = len(self.__buffered_data) < size
-        self.__end_pos = self.__start_pos + len(self.__buffered_data)
-
-    def __str__(self):
+        self._stream = stream
+        self._start_pos = start
+        self._buffer_pos = 0
+        self._buffered_data = self._stream.read(size)
+        self._stream_at_end = len(self._buffered_data) < size
+        self._end_pos = self._start_pos + len(self._buffered_data)
+
+    def __repr__(self):
         return ('Buffered stream %s from position %s-%s with %s '
-                'bytes remaining' % (self.__stream, self.__start_pos,
-                                     self.__end_pos, self._bytes_remaining))
+                'bytes remaining' % (self._stream, self._start_pos,
+                                     self._end_pos, self._bytes_remaining))
 
     def __len__(self):
-        return len(self.__buffered_data)
+        return len(self._buffered_data)
 
     @property
     def stream_exhausted(self):
-        return self.__stream_at_end
+        return self._stream_at_end
 
     @property
     def stream_end_position(self):
-        return self.__end_pos
+        return self._end_pos
 
     @property
     def _bytes_remaining(self):
-        return len(self.__buffered_data) - self.__buffer_pos
+        return len(self._buffered_data) - self._buffer_pos
 
     def read(self, size=None):  # pylint: disable=invalid-name
         """Reads from the buffer."""
@@ -47,13 +47,13 @@ def read(self, size=None):  # pylint: disable=invalid-name
                 'Illegal read of size %s requested on BufferedStream. '
                 'Wrapped stream %s is at position %s-%s, '
                 '%s bytes remaining.' %
-                (size, self.__stream, self.__start_pos, self.__end_pos,
+                (size, self._stream, self._start_pos, self._end_pos,
                  self._bytes_remaining))
 
-        data = ''
+        data = b''
         if self._bytes_remaining:
             size = min(size, self._bytes_remaining)
-            data = self.__buffered_data[
-                self.__buffer_pos:self.__buffer_pos + size]
-            self.__buffer_pos += size
+            data = self._buffered_data[
+                self._buffer_pos:self._buffer_pos + size]
+            self._buffer_pos += size
         return data

gcloud/_apitools/test_buffered_stream.py

@@ -0,0 +1,106 @@
+# pylint: skip-file
+import unittest2
+
+
+class Test_BufferedStream(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud._apitools.buffered_stream import BufferedStream
+        return BufferedStream
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_start_zero_longer_than_buffer(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = 0
+        BUFSIZE = 4
+        stream = BytesIO(CONTENT)
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertTrue(bufstream._stream is stream)
+        self.assertEqual(bufstream._start_pos, START)
+        self.assertEqual(bufstream._buffer_pos, 0)
+        self.assertEqual(bufstream._buffered_data, CONTENT[:BUFSIZE])
+        self.assertEqual(len(bufstream), BUFSIZE)
+        self.assertFalse(bufstream.stream_exhausted)
+        self.assertEqual(bufstream.stream_end_position, BUFSIZE)
+
+    def test_ctor_start_nonzero_shorter_than_buffer(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = 8
+        BUFSIZE = 10
+        stream = BytesIO(CONTENT)
+        stream.read(START)  # already consumed
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertTrue(bufstream._stream is stream)
+        self.assertEqual(bufstream._start_pos, START)
+        self.assertEqual(bufstream._buffer_pos, 0)
+        self.assertEqual(bufstream._buffered_data, CONTENT[START:])
+        self.assertEqual(len(bufstream), len(CONTENT) - START)
+        self.assertTrue(bufstream.stream_exhausted)
+        self.assertEqual(bufstream.stream_end_position, len(CONTENT))
+
+    def test__bytes_remaining_start_zero_longer_than_buffer(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = 0
+        BUFSIZE = 4
+        stream = BytesIO(CONTENT)
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertEqual(bufstream._bytes_remaining, BUFSIZE)
+
+    def test__bytes_remaining_start_zero_shorter_than_buffer(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = 8
+        BUFSIZE = 10
+        stream = BytesIO(CONTENT)
+        stream.read(START)  # already consumed
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertEqual(bufstream._bytes_remaining, len(CONTENT) - START)
+
+    def test_read_w_none(self):
+        from io import BytesIO
+        from gcloud._apitools.exceptions import NotYetImplementedError
+        CONTENT = b'CONTENT GOES HERE'
+        START = 0
+        BUFSIZE = 4
+        stream = BytesIO(CONTENT)
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        with self.assertRaises(NotYetImplementedError):
+            bufstream.read(None)
+
+    def test_read_w_negative_size(self):
+        from io import BytesIO
+        from gcloud._apitools.exceptions import NotYetImplementedError
+        CONTENT = b'CONTENT GOES HERE'
+        START = 0
+        BUFSIZE = 4
+        stream = BytesIO(CONTENT)
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        with self.assertRaises(NotYetImplementedError):
+            bufstream.read(-2)
+
+    def test_read_from_start(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = 0
+        BUFSIZE = 4
+        stream = BytesIO(CONTENT)
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertEqual(bufstream.read(4), CONTENT[:4])
+
+    def test_read_exhausted(self):
+        from io import BytesIO
+        CONTENT = b'CONTENT GOES HERE'
+        START = len(CONTENT)
+        BUFSIZE = 10
+        stream = BytesIO(CONTENT)
+        stream.read(START)  # already consumed
+        bufstream = self._makeOne(stream, START, BUFSIZE)
+        self.assertTrue(bufstream.stream_exhausted)
+        self.assertEqual(bufstream.stream_end_position, len(CONTENT))
+        self.assertEqual(bufstream._bytes_remaining, 0)
+        self.assertEqual(bufstream.read(10), b'')