OXIESEC PANEL
- Current Dir:
/
/
opt
/
gsutil
/
third_party
/
apitools
/
apitools
/
base
/
py
Server IP: 2a02:4780:11:1594:0:ef5:22d7:a
Upload:
Create Dir:
Name
Size
Modified
Perms
📁
..
-
02/11/2025 08:19:48 AM
rwxr-xr-x
📄
__init__.py
1.22 KB
10/23/2024 10:32:22 PM
rw-r--r--
📁
__pycache__
-
02/11/2025 08:19:48 AM
rwxr-xr-x
📄
base_api.py
29.29 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
base_api_test.py
14.12 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
batch.py
17.93 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
batch_test.py
24.73 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
buffered_stream.py
2.52 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
buffered_stream_test.py
2.4 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
compression.py
5.02 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
compression_test.py
5.19 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
credentials_lib.py
29.42 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
credentials_lib_test.py
7.58 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
encoding.py
1.16 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
encoding_helper.py
30 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
encoding_test.py
31.97 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
exceptions.py
4.98 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
exceptions_test.py
2.88 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
extra_types.py
9.74 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
extra_types_test.py
7.81 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
gzip.py
21.39 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
gzip_test.py
17.17 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
http_wrapper.py
15.09 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
http_wrapper_test.py
5.09 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
list_pager.py
5.49 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
list_pager_test.py
13.08 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
stream_slice.py
2.67 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
stream_slice_test.py
2.2 KB
10/23/2024 10:32:22 PM
rw-r--r--
📁
testing
-
10/23/2024 10:32:22 PM
rwxr-xr-x
📄
transfer.py
45.42 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
transfer_test.py
26.06 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
util.py
8.28 KB
10/23/2024 10:32:22 PM
rw-r--r--
📄
util_test.py
7.01 KB
10/23/2024 10:32:22 PM
rw-r--r--
Editing: buffered_stream_test.py
Close
# # Copyright 2015 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for buffered_stream.""" import string import unittest import six from apitools.base.py import buffered_stream from apitools.base.py import exceptions class BufferedStreamTest(unittest.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)