Index: tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/buffered_stream_test.py |
diff --git a/tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/buffered_stream_test.py b/tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/buffered_stream_test.py |
deleted file mode 100644 |
index 4dcf62f9b04ea35d98f0b358686c5a7627199286..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/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) |