1#
2# Copyright 2015 Google Inc.
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#     http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16"""Tests for stream_slice."""
17
18import string
19import unittest
20
21import six
22
23from apitools.base.py import exceptions
24from apitools.base.py import stream_slice
25
26
27class StreamSliceTest(unittest.TestCase):
28
29    def setUp(self):
30        self.stream = six.StringIO(string.ascii_letters)
31        self.value = self.stream.getvalue()
32        self.stream.seek(0)
33
34    def testSimpleSlice(self):
35        ss = stream_slice.StreamSlice(self.stream, 10)
36        self.assertEqual('', ss.read(0))
37        self.assertEqual(self.value[0:3], ss.read(3))
38        self.assertIn('7/10', str(ss))
39        self.assertEqual(self.value[3:10], ss.read())
40        self.assertEqual('', ss.read())
41        self.assertEqual('', ss.read(10))
42        self.assertEqual(10, self.stream.tell())
43
44    def testEmptySlice(self):
45        ss = stream_slice.StreamSlice(self.stream, 0)
46        self.assertEqual('', ss.read(5))
47        self.assertEqual('', ss.read())
48        self.assertEqual(0, self.stream.tell())
49
50    def testOffsetStream(self):
51        self.stream.seek(26)
52        ss = stream_slice.StreamSlice(self.stream, 26)
53        self.assertEqual(self.value[26:36], ss.read(10))
54        self.assertEqual(self.value[36:], ss.read())
55        self.assertEqual('', ss.read())
56
57    def testTooShortStream(self):
58        ss = stream_slice.StreamSlice(self.stream, 1000)
59        self.assertEqual(self.value, ss.read())
60        self.assertEqual('', ss.read(0))
61        with self.assertRaises(exceptions.StreamExhausted) as e:
62            ss.read()
63        with self.assertRaises(exceptions.StreamExhausted) as e:
64            ss.read(10)
65        self.assertIn('exhausted after %d' % len(self.value), str(e.exception))
66