1"""Tests for stream_slice."""
2
3import string
4
5import six
6import unittest2
7
8from apitools.base.py import exceptions
9from apitools.base.py import stream_slice
10
11
12class StreamSliceTest(unittest2.TestCase):
13
14    def setUp(self):
15        self.stream = six.StringIO(string.ascii_letters)
16        self.value = self.stream.getvalue()
17        self.stream.seek(0)
18
19    def testSimpleSlice(self):
20        ss = stream_slice.StreamSlice(self.stream, 10)
21        self.assertEqual('', ss.read(0))
22        self.assertEqual(self.value[0:3], ss.read(3))
23        self.assertIn('7/10', str(ss))
24        self.assertEqual(self.value[3:10], ss.read())
25        self.assertEqual('', ss.read())
26        self.assertEqual('', ss.read(10))
27        self.assertEqual(10, self.stream.tell())
28
29    def testEmptySlice(self):
30        ss = stream_slice.StreamSlice(self.stream, 0)
31        self.assertEqual('', ss.read(5))
32        self.assertEqual('', ss.read())
33        self.assertEqual(0, self.stream.tell())
34
35    def testOffsetStream(self):
36        self.stream.seek(26)
37        ss = stream_slice.StreamSlice(self.stream, 26)
38        self.assertEqual(self.value[26:36], ss.read(10))
39        self.assertEqual(self.value[36:], ss.read())
40        self.assertEqual('', ss.read())
41
42    def testTooShortStream(self):
43        ss = stream_slice.StreamSlice(self.stream, 1000)
44        self.assertEqual(self.value, ss.read())
45        self.assertEqual('', ss.read(0))
46        with self.assertRaises(exceptions.StreamExhausted) as e:
47            ss.read()
48        with self.assertRaises(exceptions.StreamExhausted) as e:
49            ss.read(10)
50        self.assertIn('exhausted after %d' % len(self.value), str(e.exception))
51