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