Source

pypy / pypy / module / _io / test / test_textio.py

Full commit
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
class AppTestTextIO:
    spaceconfig = dict(usemodules=['_io', '_locale'])

    def test_constructor(self):
        import _io
        r = _io.BytesIO(b"\xc3\xa9\n\n")
        b = _io.BufferedReader(r, 1000)
        t = _io.TextIOWrapper(b)
        t.__init__(b, encoding="latin1", newline="\r\n")
        assert t.encoding == "latin1"
        assert t.line_buffering == False
        t.__init__(b, encoding="utf8", line_buffering=True)
        assert t.encoding == "utf8"
        assert t.line_buffering == True
        assert t.readline() == "\xe9\n"
        raises(TypeError, t.__init__, b, newline=42)
        raises(ValueError, t.__init__, b, newline='xyzzy')
        t = _io.TextIOWrapper(b)
        assert t.encoding

    def test_properties(self):
        import _io
        r = _io.BytesIO(b"\xc3\xa9\n\n")
        b = _io.BufferedReader(r, 1000)
        t = _io.TextIOWrapper(b)
        assert t.readable()
        assert t.seekable()

    def test_isatty(self):
        import _io
        class Tty(_io.BytesIO):
            def isatty(self):
                return True
        txt = _io.TextIOWrapper(Tty())
        assert txt.isatty()

    def test_unreadable(self):
        import _io
        class UnReadable(_io.BytesIO):
            def readable(self):
                return False
        txt = _io.TextIOWrapper(UnReadable())
        raises(IOError, txt.read)

    def test_unwritable(self):
        import _io
        class UnWritable(_io.BytesIO):
            def writable(self):
                return False
        txt = _io.TextIOWrapper(UnWritable())
        raises(_io.UnsupportedOperation, txt.write, "blah")
        raises(_io.UnsupportedOperation, txt.writelines, ["blah\n"])

    def test_invalid_seek(self):
        import _io
        t = _io.TextIOWrapper(_io.BytesIO(b"\xc3\xa9\n\n"))
        raises(_io.UnsupportedOperation, t.seek, 1, 1)
        raises(_io.UnsupportedOperation, t.seek, 1, 2)

    def test_unseekable(self):
        import _io
        class Unseekable(_io.BytesIO):
            def seekable(self):
                return False
        txt = _io.TextIOWrapper(Unseekable())
        raises(_io.UnsupportedOperation, txt.tell)
        raises(_io.UnsupportedOperation, txt.seek, 0)

    def test_detach(self):
        import _io
        b = _io.BytesIO()
        f = _io.TextIOWrapper(b)
        assert f.detach() is b
        raises(ValueError, f.fileno)
        raises(ValueError, f.close)
        raises(ValueError, f.detach)
        raises(ValueError, f.flush)
        assert not b.closed
        b.close()

    def test_newlinetranslate(self):
        import _io
        r = _io.BytesIO(b"abc\r\ndef\rg")
        b = _io.BufferedReader(r, 1000)
        t = _io.TextIOWrapper(b)
        assert t.read() == "abc\ndef\ng"

    def test_one_by_one(self):
        import _io
        r = _io.BytesIO(b"abc\r\ndef\rg")
        t = _io.TextIOWrapper(r)
        reads = []
        while True:
            c = t.read(1)
            assert len(c) <= 1
            if not c:
                break
            reads.append(c)
        assert ''.join(reads) == "abc\ndef\ng"

    def test_read_some_then_all(self):
        import _io
        r = _io.BytesIO(b"abc\ndef\n")
        t = _io.TextIOWrapper(r)
        reads = t.read(4)
        reads += t.read()
        assert reads == "abc\ndef\n"

    def test_read_some_then_readline(self):
        import _io
        r = _io.BytesIO(b"abc\ndef\n")
        t = _io.TextIOWrapper(r)
        reads = t.read(4)
        reads += t.readline()
        assert reads == "abc\ndef\n"

    def test_encoded_writes(self):
        import _io
        data = "1234567890"
        tests = ("utf-16",
                 "utf-16-le",
                 "utf-16-be",
                 "utf-32",
                 "utf-32-le",
                 "utf-32-be")
        for encoding in tests:
            buf = _io.BytesIO()
            f = _io.TextIOWrapper(buf, encoding=encoding)
            # Check if the BOM is written only once (see issue1753).
            f.write(data)
            f.write(data)
            f.seek(0)
            assert f.read() == data * 2
            f.seek(0)
            assert f.read() == data * 2
            assert buf.getvalue() == (data * 2).encode(encoding)

    def test_tell(self):
        import _io
        r = _io.BytesIO(b"abc\ndef\n")
        t = _io.TextIOWrapper(r)
        assert t.tell() == 0
        t.read(4)
        assert t.tell() == 4

    def test_destructor(self):
        import _io
        l = []
        class MyBytesIO(_io.BytesIO):
            def close(self):
                l.append(self.getvalue())
                _io.BytesIO.close(self)
        b = MyBytesIO()
        t = _io.TextIOWrapper(b, encoding="ascii")
        t.write("abc")
        del t
        import gc; gc.collect()
        assert l == [b"abc"]

    def test_newlines(self):
        import _io
        input_lines = [ "unix\n", "windows\r\n", "os9\r", "last\n", "nonl" ]

        tests = [
            [ None, [ 'unix\n', 'windows\n', 'os9\n', 'last\n', 'nonl' ] ],
            [ '', input_lines ],
            [ '\n', [ "unix\n", "windows\r\n", "os9\rlast\n", "nonl" ] ],
            [ '\r\n', [ "unix\nwindows\r\n", "os9\rlast\nnonl" ] ],
            [ '\r', [ "unix\nwindows\r", "\nos9\r", "last\nnonl" ] ],
        ]

        # Try a range of buffer sizes to test the case where \r is the last
        # character in TextIOWrapper._pending_line.
        encoding = "ascii"
        # XXX: str.encode() should return bytes
        data = bytes(''.join(input_lines).encode(encoding))
        for do_reads in (False, True):
            for bufsize in range(1, 10):
                for newline, exp_lines in tests:
                    bufio = _io.BufferedReader(_io.BytesIO(data), bufsize)
                    textio = _io.TextIOWrapper(bufio, newline=newline,
                                              encoding=encoding)
                    if do_reads:
                        got_lines = []
                        while True:
                            c2 = textio.read(2)
                            if c2 == '':
                                break
                            len(c2) == 2
                            got_lines.append(c2 + textio.readline())
                    else:
                        got_lines = list(textio)

                    for got_line, exp_line in zip(got_lines, exp_lines):
                        assert got_line == exp_line
                    assert len(got_lines) == len(exp_lines)

    def test_readline(self):
        import _io

        s = b"AAA\r\nBBB\rCCC\r\nDDD\nEEE\r\n"
        r = "AAA\nBBB\nCCC\nDDD\nEEE\n"
        txt = _io.TextIOWrapper(_io.BytesIO(s), encoding="ascii")
        txt._CHUNK_SIZE = 4

        reads = txt.read(4)
        reads += txt.read(4)
        reads += txt.readline()
        reads += txt.readline()
        reads += txt.readline()
        assert reads == r

    def test_name(self):
        import _io

        t = _io.TextIOWrapper(_io.BytesIO(b""))
        # CPython raises an AttributeError, we raise a TypeError.
        raises((AttributeError, TypeError), setattr, t, "name", "anything")

    def test_repr(self):
        import _io

        t = _io.TextIOWrapper(_io.BytesIO(b""), encoding="utf-8")
        assert repr(t) == "<_io.TextIOWrapper encoding='utf-8'>"
        t = _io.TextIOWrapper(_io.BytesIO(b""), encoding="ascii")
        assert repr(t) == "<_io.TextIOWrapper encoding='ascii'>"
        t = _io.TextIOWrapper(_io.BytesIO(b""), encoding="utf-8")
        assert repr(t) == "<_io.TextIOWrapper encoding='utf-8'>"
        b = _io.BytesIO(b"")
        t = _io.TextIOWrapper(b, encoding="utf-8")
        b.name = "dummy"
        assert repr(t) == "<_io.TextIOWrapper name='dummy' encoding='utf-8'>"
        t.mode = "r"
        assert repr(t) == "<_io.TextIOWrapper name='dummy' mode='r' encoding='utf-8'>"
        b.name = b"dummy"
        assert repr(t) == "<_io.TextIOWrapper name=b'dummy' mode='r' encoding='utf-8'>"

    def test_rawio(self):
        # Issue #12591: TextIOWrapper must work with raw I/O objects, so
        # that subprocess.Popen() can have the required unbuffered
        # semantics with universal_newlines=True.
        import _io
        raw = self.get_MockRawIO()([b'abc', b'def', b'ghi\njkl\nopq\n'])
        txt = _io.TextIOWrapper(raw, encoding='ascii', newline='\n')
        # Reads
        assert txt.read(4) == 'abcd'
        assert txt.readline() == 'efghi\n'
        assert list(txt) == ['jkl\n', 'opq\n']

    def test_rawio_write_through(self):
        # Issue #12591: with write_through=True, writes don't need a flush
        import _io
        raw = self.get_MockRawIO()([b'abc', b'def', b'ghi\njkl\nopq\n'])
        txt = _io.TextIOWrapper(raw, encoding='ascii', newline='\n',
                                write_through=True)
        txt.write('1')
        txt.write('23\n4')
        txt.write('5')
        assert b''.join(raw._write_stack) == b'123\n45'

    def w_get_MockRawIO(self):
        import _io
        class MockRawIO(_io._RawIOBase):
            def __init__(self, read_stack=()):
                self._read_stack = list(read_stack)
                self._write_stack = []
                self._reads = 0
                self._extraneous_reads = 0

            def write(self, b):
                self._write_stack.append(bytes(b))
                return len(b)

            def writable(self):
                return True

            def fileno(self):
                return 42

            def readable(self):
                return True

            def seekable(self):
                return True

            def seek(self, pos, whence):
                return 0   # wrong but we gotta return something

            def tell(self):
                return 0   # same comment as above

            def readinto(self, buf):
                self._reads += 1
                max_len = len(buf)
                try:
                    data = self._read_stack[0]
                except IndexError:
                    self._extraneous_reads += 1
                    return 0
                if data is None:
                    del self._read_stack[0]
                    return None
                n = len(data)
                if len(data) <= max_len:
                    del self._read_stack[0]
                    buf[:n] = data
                    return n
                else:
                    buf[:] = data[:max_len]
                    self._read_stack[0] = data[max_len:]
                    return max_len

            def truncate(self, pos=None):
                return pos

            def read(self, n=None):
                self._reads += 1
                try:
                    return self._read_stack.pop(0)
                except:
                    self._extraneous_reads += 1
                    return b""
        return MockRawIO


class AppTestIncrementalNewlineDecoder:

    def test_newline_decoder(self):
        import _io
        def check_newline_decoding_utf8(decoder):
            # UTF-8 specific tests for a newline decoder
            def _check_decode(b, s, **kwargs):
                # We exercise getstate() / setstate() as well as decode()
                state = decoder.getstate()
                assert decoder.decode(b, **kwargs) == s
                decoder.setstate(state)
                assert decoder.decode(b, **kwargs) == s

            _check_decode(b'\xe8\xa2\x88', "\u8888")

            _check_decode(b'\xe8', "")
            _check_decode(b'\xa2', "")
            _check_decode(b'\x88', "\u8888")

            _check_decode(b'\xe8', "")
            _check_decode(b'\xa2', "")
            _check_decode(b'\x88', "\u8888")

            _check_decode(b'\xe8', "")
            raises(UnicodeDecodeError, decoder.decode, b'', final=True)

            decoder.reset()
            _check_decode(b'\n', "\n")
            _check_decode(b'\r', "")
            _check_decode(b'', "\n", final=True)
            _check_decode(b'\r', "\n", final=True)

            _check_decode(b'\r', "")
            _check_decode(b'a', "\na")

            _check_decode(b'\r\r\n', "\n\n")
            _check_decode(b'\r', "")
            _check_decode(b'\r', "\n")
            _check_decode(b'\na', "\na")

            _check_decode(b'\xe8\xa2\x88\r\n', "\u8888\n")
            _check_decode(b'\xe8\xa2\x88', "\u8888")
            _check_decode(b'\n', "\n")
            _check_decode(b'\xe8\xa2\x88\r', "\u8888")
            _check_decode(b'\n', "\n")

        def check_newline_decoding(decoder, encoding):
            result = []
            if encoding is not None:
                encoder = codecs.getincrementalencoder(encoding)()
                def _decode_bytewise(s):
                    # Decode one byte at a time
                    for b in encoder.encode(s):
                        result.append(decoder.decode(bytes([b])))
            else:
                encoder = None
                def _decode_bytewise(s):
                    # Decode one char at a time
                    for c in s:
                        result.append(decoder.decode(c))
            assert decoder.newlines == None
            _decode_bytewise("abc\n\r")
            assert decoder.newlines == '\n'
            _decode_bytewise("\nabc")
            assert decoder.newlines == ('\n', '\r\n')
            _decode_bytewise("abc\r")
            assert decoder.newlines == ('\n', '\r\n')
            _decode_bytewise("abc")
            assert decoder.newlines == ('\r', '\n', '\r\n')
            _decode_bytewise("abc\r")
            assert "".join(result) == "abc\n\nabcabc\nabcabc"
            decoder.reset()
            input = "abc"
            if encoder is not None:
                encoder.reset()
                input = encoder.encode(input)
            assert decoder.decode(input) == "abc"
            assert decoder.newlines is None

        encodings = (
            # None meaning the IncrementalNewlineDecoder takes unicode input
            # rather than bytes input
            None, 'utf-8', 'latin-1',
            'utf-16', 'utf-16-le', 'utf-16-be',
            'utf-32', 'utf-32-le', 'utf-32-be',
        )
        import codecs
        for enc in encodings:
            decoder = enc and codecs.getincrementaldecoder(enc)()
            decoder = _io.IncrementalNewlineDecoder(decoder, translate=True)
            check_newline_decoding(decoder, enc)
        decoder = codecs.getincrementaldecoder("utf-8")()
        decoder = _io.IncrementalNewlineDecoder(decoder, translate=True)
        check_newline_decoding_utf8(decoder)

    def test_newline_bytes(self):
        import _io
        # Issue 5433: Excessive optimization in IncrementalNewlineDecoder
        def _check(dec):
            assert dec.newlines is None
            assert dec.decode("\u0D00") == "\u0D00"
            assert dec.newlines is None
            assert dec.decode("\u0A00") == "\u0A00"
            assert dec.newlines is None
        dec = _io.IncrementalNewlineDecoder(None, translate=False)
        _check(dec)
        dec = _io.IncrementalNewlineDecoder(None, translate=True)
        _check(dec)