xref: /aosp_15_r20/external/tink/python/tink/testing/bytes_io_test.py (revision e7b1675dde1b92d52ec075b0a92829627f2c52a5)
1# Copyright 2020 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Tests for tink.python.tink.testing.bytes_io."""
15
16import io
17
18from absl.testing import absltest
19
20from tink.testing import bytes_io
21
22
23class BytesIoTest(absltest.TestCase):
24
25  def test_close_success(self):
26    f = bytes_io.BytesIOWithValueAfterClose()
27    f.write(b'foobar')
28    f.close()
29    self.assertEqual(f.value_after_close(), b'foobar')
30
31  def test_with_text_wrapper_success(self):
32    f = bytes_io.BytesIOWithValueAfterClose()
33    with io.TextIOWrapper(f, encoding='utf8') as t:
34      t.write(u'foobar')
35    self.assertTrue(f.closed)
36    value = f.value_after_close()
37    self.assertEqual(value, u'foobar'.encode('utf8'))
38
39  def test_initial_bytes_success(self):
40    f = bytes_io.BytesIOWithValueAfterClose(b'abc')
41    f.write(b'foobar')
42    f.close()
43    self.assertEqual(f.value_after_close(), b'foobar')
44
45  def test_value_before_close_fails(self):
46    f = bytes_io.BytesIOWithValueAfterClose()
47    f.write(b'foobar')
48    with self.assertRaises(ValueError):
49      f.value_after_close()
50    f.close()
51
52  def test_close_twice_success(self):
53    f = bytes_io.BytesIOWithValueAfterClose(b'abc')
54    f.write(b'foobar')
55    f.close()
56    f.close()
57    self.assertEqual(f.value_after_close(), b'foobar')
58
59
60class SlowBytesIOTest(absltest.TestCase):
61
62  def test_read(self):
63    with bytes_io.SlowBytesIO(b'The quick brown fox') as f:
64      with self.assertRaises(io.BlockingIOError):
65        f.read(10)
66      self.assertEqual(b'The q', f.read(10))
67      self.assertEqual(b'uick ', f.read(10))
68      with self.assertRaises(io.BlockingIOError):
69        f.read(10)
70      self.assertEqual(b'brown', f.read(10))
71      self.assertEqual(b' fox', f.read(10))
72      with self.assertRaises(io.BlockingIOError):
73        f.read(10)
74      self.assertEqual(b'', f.read(10))
75    self.assertTrue(f.closed)
76
77  def test_read_no_argument(self):
78    with bytes_io.SlowBytesIO(b'The quick brown fox') as f:
79      self.assertEqual(b'The quick brown fox', f.read())
80      self.assertEqual(b'', f.read())
81
82  def test_read_minus_one(self):
83    with bytes_io.SlowBytesIO(b'The quick brown fox') as f:
84      self.assertEqual(b'The quick brown fox', f.read(-1))
85      self.assertEqual(b'', f.read(-1))
86
87  def test_not_seekable(self):
88    with bytes_io.SlowBytesIO(b'The quick brown fox', seekable=False) as f:
89      self.assertFalse(f.seekable())
90      with self.assertRaises(io.UnsupportedOperation):
91        f.seek(0)
92
93  def test_seekable(self):
94    with bytes_io.SlowBytesIO(b'The quick brown fox', seekable=True) as f:
95      self.assertTrue(f.seekable())
96      self.assertEqual(b'The quick brown fox', f.read())
97      f.seek(0)
98      self.assertEqual(b'The quick brown fox', f.read())
99
100
101class SlowReadableRawBytesTest(absltest.TestCase):
102
103  def test_read(self):
104    with bytes_io.SlowReadableRawBytes(b'The quick brown fox') as f:
105      self.assertIsNone(f.read(10))
106      self.assertEqual(b'The q', f.read(10))
107      self.assertEqual(b'uick ', f.read(10))
108      self.assertIsNone(f.read(10))
109      self.assertEqual(b'brown', f.read(10))
110      self.assertEqual(b' fox', f.read(10))
111      self.assertIsNone(f.read(10))
112      self.assertEqual(b'', f.read(10))
113    self.assertTrue(f.closed)
114
115  # Note that the documentation of readall(), read() and read(-1) is wrong:
116  # https://docs.python.org/2/library/io.html#io.RawIOBase.readall.
117  # readall does multiple reads, but it stops as soon as it can't read any data.
118  # The documentation says that it reads until it reaches EOF.
119  def test_readall(self):
120    with bytes_io.SlowReadableRawBytes(b'The quick brown fox') as f:
121      self.assertIsNone(None, f.readall())
122      self.assertEqual(b'The quick ', f.readall())
123      self.assertEqual(b'brown fox', f.readall())
124      self.assertEqual(b'', f.readall())
125
126  def test_read_no_argument(self):
127    with bytes_io.SlowReadableRawBytes(b'The quick brown fox') as f:
128      self.assertIsNone(None, f.read())
129      self.assertEqual(b'The quick ', f.read())
130      self.assertEqual(b'brown fox', f.read())
131      self.assertEqual(b'', f.read())
132
133  def test_read_minus_one(self):
134    with bytes_io.SlowReadableRawBytes(b'The quick brown fox') as f:
135      self.assertIsNone(None, f.read(-1))
136      self.assertEqual(b'The quick ', f.read(-1))
137      self.assertEqual(b'brown fox', f.read(-1))
138      self.assertEqual(b'', f.read(-1))
139
140  def test_not_seekable(self):
141    with bytes_io.SlowReadableRawBytes(
142        b'The quick brown fox', seekable=False) as f:
143      self.assertFalse(f.seekable())
144      with self.assertRaises(io.UnsupportedOperation):
145        f.seek(0)
146
147  def test_seekable(self):
148    with bytes_io.SlowReadableRawBytes(
149        b'The quick brown fox', seekable=True) as f:
150      self.assertTrue(f.seekable())
151      self.assertIsNone(None, f.read())
152      self.assertEqual(b'The quick ', f.read())
153      f.seek(0)
154      self.assertEqual(b'The quick ', f.read())
155
156if __name__ == '__main__':
157  absltest.main()
158