| Index: tools/telemetry/third_party/gsutilz/third_party/apitools/apitools/base/py/buffered_stream_test.py
|
| diff --git a/tools/telemetry/third_party/gsutilz/third_party/apitools/apitools/base/py/buffered_stream_test.py b/tools/telemetry/third_party/gsutilz/third_party/apitools/apitools/base/py/buffered_stream_test.py
|
| deleted file mode 100644
|
| index 4dcf62f9b04ea35d98f0b358686c5a7627199286..0000000000000000000000000000000000000000
|
| --- a/tools/telemetry/third_party/gsutilz/third_party/apitools/apitools/base/py/buffered_stream_test.py
|
| +++ /dev/null
|
| @@ -1,52 +0,0 @@
|
| -"""Tests for buffered_stream."""
|
| -
|
| -import string
|
| -
|
| -import six
|
| -import unittest2
|
| -
|
| -from apitools.base.py import buffered_stream
|
| -from apitools.base.py import exceptions
|
| -
|
| -
|
| -class BufferedStreamTest(unittest2.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.stream = six.StringIO(string.ascii_letters)
|
| - self.value = self.stream.getvalue()
|
| - self.stream.seek(0)
|
| -
|
| - def testEmptyBuffer(self):
|
| - bs = buffered_stream.BufferedStream(self.stream, 0, 0)
|
| - self.assertEqual('', bs.read(0))
|
| - self.assertEqual(0, bs.stream_end_position)
|
| -
|
| - def testOffsetStream(self):
|
| - bs = buffered_stream.BufferedStream(self.stream, 50, 100)
|
| - self.assertEqual(len(self.value), len(bs))
|
| - self.assertEqual(self.value, bs.read(len(self.value)))
|
| - self.assertEqual(50 + len(self.value), bs.stream_end_position)
|
| -
|
| - def testUnexhaustedStream(self):
|
| - bs = buffered_stream.BufferedStream(self.stream, 0, 50)
|
| - self.assertEqual(50, bs.stream_end_position)
|
| - self.assertEqual(False, bs.stream_exhausted)
|
| - self.assertEqual(self.value[0:50], bs.read(50))
|
| - self.assertEqual(False, bs.stream_exhausted)
|
| - self.assertEqual('', bs.read(0))
|
| - self.assertEqual('', bs.read(100))
|
| -
|
| - def testExhaustedStream(self):
|
| - bs = buffered_stream.BufferedStream(self.stream, 0, 100)
|
| - self.assertEqual(len(self.value), bs.stream_end_position)
|
| - self.assertEqual(True, bs.stream_exhausted)
|
| - self.assertEqual(self.value, bs.read(100))
|
| - self.assertEqual('', bs.read(0))
|
| - self.assertEqual('', bs.read(100))
|
| -
|
| - def testArbitraryLengthRead(self):
|
| - bs = buffered_stream.BufferedStream(self.stream, 0, 20)
|
| - with self.assertRaises(exceptions.NotYetImplementedError):
|
| - bs.read()
|
| - with self.assertRaises(exceptions.NotYetImplementedError):
|
| - bs.read(size=-1)
|
|
|