mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 10:44:36 +08:00
youknowdot
7d15b50cd4
* Align weak-reference collection with Python 3.15 finalization Trace and retain owned callback edges independently of weak referent links. Clear callback-bearing weakrefs before finalizers, suppress callbacks from unreachable weakrefs, and close remaining weakref ingress after resurrection checks while the world is stopped. Publish and unlink intrusive nodes safely. Refresh weakref and canonical weakref/GC/finalization tests through update_lib from pinned CPython 3.15.0. Retire four obsolete guards only after fresh body passes, retaining applicable RustPython limitations. Assisted-by: OpenAI Codex:model version unavailable * Refresh canonical IO tests for weakref finalization Migrate the complete CPython 3.15.0 test_io package at d220409 with the fail-closed legacy-map updater, preserving all ten existing guards. Retire the seven consolidated standalone suites and update io/file dependency paths with mapping regression tests. The canonical GC tests ignore ResourceWarning without retaining its source, so callback-free weakrefs correctly reflect surviving owners. This closes the stale test dependency exposed by the lifecycle change. Assisted-by: OpenAI Codex:model version unavailable * Protect weakref callbacks with the original referent stripe Follow CPython's free-threading weakref locking model instead of keeping an independent callback RwLock. Retain the original referent stripe key so callback access stays synchronized after GC clears the referent. Unlink and steal callbacks under that stripe, then release ownership outside the lock. Preserve synchronization for running-world GC paths and gc.get_referents while keeping ordinary weakref upgrades unchanged. Assisted-by: OpenAI Codex:model version unavailable * Defer weakref GC tracking until referent publication Keep freshly allocated weakrefs out of GC's object lists until they are linked and their referent is published. Release the referent stripe before tracking, since tracking can collect and invoke finalizers. Generic reuse losers remain untracked. Exercise the allocator boundary directly for reference, proxy, callable proxy and subclass payloads, and ensure completed constructors track their objects. This prevents observers from retaining a half-built dead weakref that later becomes live. Assisted-by: OpenAI Codex:model version unavailable * Retain the original weakref stripe lock like CPython Store a safe handle to the selected static weakref stripe instead of retaining and rehashing the referent address. Keep the same lock after referent clearing, matching CPython free-threaded weakrefs_lock. The nonthreaded handle is zero-sized and introduces no locking. Preserve lock selection, acquire/release ordering, fork reset, deferred GC publication, and callback-owner release outside the stripe. No new locks or unsafe code are needed. Assisted-by: OpenAI Codex:model version unavailable
1705 lines
65 KiB
Python
Vendored
1705 lines
65 KiB
Python
Vendored
import array
|
|
import codecs
|
|
import locale
|
|
import os
|
|
import pickle
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import warnings
|
|
import weakref
|
|
from collections import UserList
|
|
from test import support
|
|
from test.support import os_helper, threading_helper
|
|
from test.support.script_helper import assert_python_ok
|
|
from .utils import CTestCase, PyTestCase
|
|
|
|
import io # C implementation of io
|
|
import _pyio as pyio # Python implementation of io
|
|
|
|
|
|
def _default_chunk_size():
|
|
"""Get the default TextIOWrapper chunk size"""
|
|
with open(__file__, "r", encoding="latin-1") as f:
|
|
return f._CHUNK_SIZE
|
|
|
|
|
|
class BadIndex:
|
|
def __index__(self):
|
|
1/0
|
|
|
|
|
|
# To fully exercise seek/tell, the StatefulIncrementalDecoder has these
|
|
# properties:
|
|
# - A single output character can correspond to many bytes of input.
|
|
# - The number of input bytes to complete the character can be
|
|
# undetermined until the last input byte is received.
|
|
# - The number of input bytes can vary depending on previous input.
|
|
# - A single input byte can correspond to many characters of output.
|
|
# - The number of output characters can be undetermined until the
|
|
# last input byte is received.
|
|
# - The number of output characters can vary depending on previous input.
|
|
|
|
class StatefulIncrementalDecoder(codecs.IncrementalDecoder):
|
|
"""
|
|
For testing seek/tell behavior with a stateful, buffering decoder.
|
|
|
|
Input is a sequence of words. Words may be fixed-length (length set
|
|
by input) or variable-length (period-terminated). In variable-length
|
|
mode, extra periods are ignored. Possible words are:
|
|
- 'i' followed by a number sets the input length, I (maximum 99).
|
|
When I is set to 0, words are space-terminated.
|
|
- 'o' followed by a number sets the output length, O (maximum 99).
|
|
- Any other word is converted into a word followed by a period on
|
|
the output. The output word consists of the input word truncated
|
|
or padded out with hyphens to make its length equal to O. If O
|
|
is 0, the word is output verbatim without truncating or padding.
|
|
I and O are initially set to 1. When I changes, any buffered input is
|
|
re-scanned according to the new I. EOF also terminates the last word.
|
|
"""
|
|
|
|
def __init__(self, errors='strict'):
|
|
codecs.IncrementalDecoder.__init__(self, errors)
|
|
self.reset()
|
|
|
|
def __repr__(self):
|
|
return '<SID %x>' % id(self)
|
|
|
|
def reset(self):
|
|
self.i = 1
|
|
self.o = 1
|
|
self.buffer = bytearray()
|
|
|
|
def getstate(self):
|
|
i, o = self.i ^ 1, self.o ^ 1 # so that flags = 0 after reset()
|
|
return bytes(self.buffer), i*100 + o
|
|
|
|
def setstate(self, state):
|
|
buffer, io = state
|
|
self.buffer = bytearray(buffer)
|
|
i, o = divmod(io, 100)
|
|
self.i, self.o = i ^ 1, o ^ 1
|
|
|
|
def decode(self, input, final=False):
|
|
output = ''
|
|
for b in input:
|
|
if self.i == 0: # variable-length, terminated with period
|
|
if b == ord('.'):
|
|
if self.buffer:
|
|
output += self.process_word()
|
|
else:
|
|
self.buffer.append(b)
|
|
else: # fixed-length, terminate after self.i bytes
|
|
self.buffer.append(b)
|
|
if len(self.buffer) == self.i:
|
|
output += self.process_word()
|
|
if final and self.buffer: # EOF terminates the last word
|
|
output += self.process_word()
|
|
return output
|
|
|
|
def process_word(self):
|
|
output = ''
|
|
if self.buffer[0] == ord('i'):
|
|
self.i = min(99, int(self.buffer[1:] or 0)) # set input length
|
|
elif self.buffer[0] == ord('o'):
|
|
self.o = min(99, int(self.buffer[1:] or 0)) # set output length
|
|
else:
|
|
output = self.buffer.decode('ascii')
|
|
if len(output) < self.o:
|
|
output += '-'*self.o # pad out with hyphens
|
|
if self.o:
|
|
output = output[:self.o] # truncate to output length
|
|
output += '.'
|
|
self.buffer = bytearray()
|
|
return output
|
|
|
|
codecEnabled = False
|
|
|
|
|
|
# bpo-41919: This method is separated from StatefulIncrementalDecoder to avoid a resource leak
|
|
# when registering codecs and cleanup functions.
|
|
def lookupTestDecoder(name):
|
|
if StatefulIncrementalDecoder.codecEnabled and name == 'test_decoder':
|
|
latin1 = codecs.lookup('latin-1')
|
|
return codecs.CodecInfo(
|
|
name='test_decoder', encode=latin1.encode, decode=None,
|
|
incrementalencoder=None,
|
|
streamreader=None, streamwriter=None,
|
|
incrementaldecoder=StatefulIncrementalDecoder)
|
|
|
|
|
|
class StatefulIncrementalDecoderTest(unittest.TestCase):
|
|
"""
|
|
Make sure the StatefulIncrementalDecoder actually works.
|
|
"""
|
|
|
|
test_cases = [
|
|
# I=1, O=1 (fixed-length input == fixed-length output)
|
|
(b'abcd', False, 'a.b.c.d.'),
|
|
# I=0, O=0 (variable-length input, variable-length output)
|
|
(b'oiabcd', True, 'abcd.'),
|
|
# I=0, O=0 (should ignore extra periods)
|
|
(b'oi...abcd...', True, 'abcd.'),
|
|
# I=0, O=6 (variable-length input, fixed-length output)
|
|
(b'i.o6.x.xyz.toolongtofit.', False, 'x-----.xyz---.toolon.'),
|
|
# I=2, O=6 (fixed-length input < fixed-length output)
|
|
(b'i.i2.o6xyz', True, 'xy----.z-----.'),
|
|
# I=6, O=3 (fixed-length input > fixed-length output)
|
|
(b'i.o3.i6.abcdefghijklmnop', True, 'abc.ghi.mno.'),
|
|
# I=0, then 3; O=29, then 15 (with longer output)
|
|
(b'i.o29.a.b.cde.o15.abcdefghijabcdefghij.i3.a.b.c.d.ei00k.l.m', True,
|
|
'a----------------------------.' +
|
|
'b----------------------------.' +
|
|
'cde--------------------------.' +
|
|
'abcdefghijabcde.' +
|
|
'a.b------------.' +
|
|
'.c.------------.' +
|
|
'd.e------------.' +
|
|
'k--------------.' +
|
|
'l--------------.' +
|
|
'm--------------.')
|
|
]
|
|
|
|
def test_decoder(self):
|
|
# Try a few one-shot test cases.
|
|
for input, eof, output in self.test_cases:
|
|
d = StatefulIncrementalDecoder()
|
|
self.assertEqual(d.decode(input, eof), output)
|
|
|
|
# Also test an unfinished decode, followed by forcing EOF.
|
|
d = StatefulIncrementalDecoder()
|
|
self.assertEqual(d.decode(b'oiabcd'), '')
|
|
self.assertEqual(d.decode(b'', 1), 'abcd.')
|
|
|
|
class TextIOWrapperTest:
|
|
|
|
def setUp(self):
|
|
self.testdata = b"AAA\r\nBBB\rCCC\r\nDDD\nEEE\r\n"
|
|
self.normalized = b"AAA\nBBB\nCCC\nDDD\nEEE\n".decode("ascii")
|
|
os_helper.unlink(os_helper.TESTFN)
|
|
codecs.register(lookupTestDecoder)
|
|
self.addCleanup(codecs.unregister, lookupTestDecoder)
|
|
|
|
def tearDown(self):
|
|
os_helper.unlink(os_helper.TESTFN)
|
|
|
|
def test_constructor(self):
|
|
r = self.BytesIO(b"\xc3\xa9\n\n")
|
|
b = self.BufferedReader(r, 1000)
|
|
t = self.TextIOWrapper(b, encoding="utf-8")
|
|
t.__init__(b, encoding="latin-1", newline="\r\n")
|
|
self.assertEqual(t.encoding, "latin-1")
|
|
self.assertEqual(t.line_buffering, False)
|
|
t.__init__(b, encoding="utf-8", line_buffering=True)
|
|
self.assertEqual(t.encoding, "utf-8")
|
|
self.assertEqual(t.line_buffering, True)
|
|
self.assertEqual("\xe9\n", t.readline())
|
|
invalid_type = TypeError if self.is_C else ValueError
|
|
with self.assertRaises(invalid_type):
|
|
t.__init__(b, encoding=42)
|
|
with self.assertRaises(UnicodeEncodeError):
|
|
t.__init__(b, encoding='\udcfe')
|
|
with self.assertRaises(ValueError):
|
|
t.__init__(b, encoding='utf-8\0')
|
|
with self.assertRaises(invalid_type):
|
|
t.__init__(b, encoding="utf-8", errors=42)
|
|
if support.Py_DEBUG or sys.flags.dev_mode or self.is_C:
|
|
with self.assertRaises(UnicodeEncodeError):
|
|
t.__init__(b, encoding="utf-8", errors='\udcfe')
|
|
if support.Py_DEBUG or sys.flags.dev_mode or self.is_C:
|
|
with self.assertRaises(ValueError):
|
|
t.__init__(b, encoding="utf-8", errors='replace\0')
|
|
with self.assertRaises(TypeError):
|
|
t.__init__(b, encoding="utf-8", newline=42)
|
|
with self.assertRaises(ValueError):
|
|
t.__init__(b, encoding="utf-8", newline='\udcfe')
|
|
with self.assertRaises(ValueError):
|
|
t.__init__(b, encoding="utf-8", newline='\n\0')
|
|
with self.assertRaises(ValueError):
|
|
t.__init__(b, encoding="utf-8", newline='xyzzy')
|
|
|
|
def test_uninitialized(self):
|
|
t = self.TextIOWrapper.__new__(self.TextIOWrapper)
|
|
del t
|
|
t = self.TextIOWrapper.__new__(self.TextIOWrapper)
|
|
self.assertRaises(Exception, repr, t)
|
|
self.assertRaisesRegex((ValueError, AttributeError),
|
|
'uninitialized|has no attribute',
|
|
t.read, 0)
|
|
t.__init__(self.MockRawIO(), encoding="utf-8")
|
|
self.assertEqual(t.read(0), '')
|
|
|
|
def test_non_text_encoding_codecs_are_rejected(self):
|
|
# Ensure the constructor complains if passed a codec that isn't
|
|
# marked as a text encoding
|
|
# http://bugs.python.org/issue20404
|
|
r = self.BytesIO()
|
|
b = self.BufferedWriter(r)
|
|
with self.assertRaisesRegex(LookupError, "is not a text encoding"):
|
|
self.TextIOWrapper(b, encoding="hex")
|
|
|
|
def test_detach(self):
|
|
r = self.BytesIO()
|
|
b = self.BufferedWriter(r)
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
self.assertIs(t.detach(), b)
|
|
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
t.write("howdy")
|
|
self.assertFalse(r.getvalue())
|
|
t.detach()
|
|
self.assertEqual(r.getvalue(), b"howdy")
|
|
self.assertRaises(ValueError, t.detach)
|
|
|
|
# Operations independent of the detached stream should still work
|
|
repr(t)
|
|
self.assertEqual(t.encoding, "ascii")
|
|
self.assertEqual(t.errors, "strict")
|
|
self.assertFalse(t.line_buffering)
|
|
self.assertFalse(t.write_through)
|
|
|
|
def test_repr(self):
|
|
raw = self.BytesIO("hello".encode("utf-8"))
|
|
b = self.BufferedReader(raw)
|
|
t = self.TextIOWrapper(b, encoding="utf-8")
|
|
modname = self.TextIOWrapper.__module__
|
|
self.assertRegex(repr(t),
|
|
r"<(%s\.)?TextIOWrapper encoding='utf-8'>" % modname)
|
|
raw.name = "dummy"
|
|
self.assertRegex(repr(t),
|
|
r"<(%s\.)?TextIOWrapper name='dummy' encoding='utf-8'>" % modname)
|
|
t.mode = "r"
|
|
self.assertRegex(repr(t),
|
|
r"<(%s\.)?TextIOWrapper name='dummy' mode='r' encoding='utf-8'>" % modname)
|
|
raw.name = b"dummy"
|
|
self.assertRegex(repr(t),
|
|
r"<(%s\.)?TextIOWrapper name=b'dummy' mode='r' encoding='utf-8'>" % modname)
|
|
|
|
t.buffer.detach()
|
|
repr(t) # Should not raise an exception
|
|
|
|
def test_recursive_repr(self):
|
|
# Issue #25455
|
|
raw = self.BytesIO()
|
|
t = self.TextIOWrapper(raw, encoding="utf-8")
|
|
with support.swap_attr(raw, 'name', t), support.infinite_recursion(25):
|
|
with self.assertRaises(RuntimeError):
|
|
repr(t) # Should not crash
|
|
|
|
def test_subclass_repr(self):
|
|
class TestSubclass(self.TextIOWrapper):
|
|
pass
|
|
|
|
f = TestSubclass(self.StringIO())
|
|
self.assertIn(TestSubclass.__name__, repr(f))
|
|
|
|
def test_line_buffering(self):
|
|
r = self.BytesIO()
|
|
b = self.BufferedWriter(r, 1000)
|
|
t = self.TextIOWrapper(b, encoding="utf-8", newline="\n", line_buffering=True)
|
|
t.write("X")
|
|
self.assertEqual(r.getvalue(), b"") # No flush happened
|
|
t.write("Y\nZ")
|
|
self.assertEqual(r.getvalue(), b"XY\nZ") # All got flushed
|
|
t.write("A\rB")
|
|
self.assertEqual(r.getvalue(), b"XY\nZA\rB")
|
|
|
|
def test_reconfigure_line_buffering(self):
|
|
r = self.BytesIO()
|
|
b = self.BufferedWriter(r, 1000)
|
|
t = self.TextIOWrapper(b, encoding="utf-8", newline="\n", line_buffering=False)
|
|
t.write("AB\nC")
|
|
self.assertEqual(r.getvalue(), b"")
|
|
|
|
t.reconfigure(line_buffering=True) # implicit flush
|
|
self.assertEqual(r.getvalue(), b"AB\nC")
|
|
t.write("DEF\nG")
|
|
self.assertEqual(r.getvalue(), b"AB\nCDEF\nG")
|
|
t.write("H")
|
|
self.assertEqual(r.getvalue(), b"AB\nCDEF\nG")
|
|
t.reconfigure(line_buffering=False) # implicit flush
|
|
self.assertEqual(r.getvalue(), b"AB\nCDEF\nGH")
|
|
t.write("IJ")
|
|
self.assertEqual(r.getvalue(), b"AB\nCDEF\nGH")
|
|
|
|
# Keeping default value
|
|
t.reconfigure()
|
|
t.reconfigure(line_buffering=None)
|
|
self.assertEqual(t.line_buffering, False)
|
|
t.reconfigure(line_buffering=True)
|
|
t.reconfigure()
|
|
t.reconfigure(line_buffering=None)
|
|
self.assertEqual(t.line_buffering, True)
|
|
|
|
@unittest.skipIf(sys.flags.utf8_mode, "utf-8 mode is enabled")
|
|
def test_default_encoding(self):
|
|
with os_helper.EnvironmentVarGuard() as env:
|
|
# try to get a user preferred encoding different than the current
|
|
# locale encoding to check that TextIOWrapper() uses the current
|
|
# locale encoding and not the user preferred encoding
|
|
env.unset('LC_ALL', 'LANG', 'LC_CTYPE')
|
|
|
|
current_locale_encoding = locale.getencoding()
|
|
b = self.BytesIO()
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", EncodingWarning)
|
|
t = self.TextIOWrapper(b)
|
|
self.assertEqual(t.encoding, current_locale_encoding)
|
|
|
|
def test_encoding(self):
|
|
# Check the encoding attribute is always set, and valid
|
|
b = self.BytesIO()
|
|
t = self.TextIOWrapper(b, encoding="utf-8")
|
|
self.assertEqual(t.encoding, "utf-8")
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", EncodingWarning)
|
|
t = self.TextIOWrapper(b)
|
|
self.assertIsNotNone(t.encoding)
|
|
codecs.lookup(t.encoding)
|
|
|
|
def test_encoding_errors_reading(self):
|
|
# (1) default
|
|
b = self.BytesIO(b"abc\n\xff\n")
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
self.assertRaises(UnicodeError, t.read)
|
|
# (2) explicit strict
|
|
b = self.BytesIO(b"abc\n\xff\n")
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="strict")
|
|
self.assertRaises(UnicodeError, t.read)
|
|
# (3) ignore
|
|
b = self.BytesIO(b"abc\n\xff\n")
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="ignore")
|
|
self.assertEqual(t.read(), "abc\n\n")
|
|
# (4) replace
|
|
b = self.BytesIO(b"abc\n\xff\n")
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="replace")
|
|
self.assertEqual(t.read(), "abc\n\ufffd\n")
|
|
|
|
def test_encoding_errors_writing(self):
|
|
# (1) default
|
|
b = self.BytesIO()
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
self.assertRaises(UnicodeError, t.write, "\xff")
|
|
# (2) explicit strict
|
|
b = self.BytesIO()
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="strict")
|
|
self.assertRaises(UnicodeError, t.write, "\xff")
|
|
# (3) ignore
|
|
b = self.BytesIO()
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="ignore",
|
|
newline="\n")
|
|
t.write("abc\xffdef\n")
|
|
t.flush()
|
|
self.assertEqual(b.getvalue(), b"abcdef\n")
|
|
# (4) replace
|
|
b = self.BytesIO()
|
|
t = self.TextIOWrapper(b, encoding="ascii", errors="replace",
|
|
newline="\n")
|
|
t.write("abc\xffdef\n")
|
|
t.flush()
|
|
self.assertEqual(b.getvalue(), b"abc?def\n")
|
|
|
|
def test_newlines(self):
|
|
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" ] ],
|
|
]
|
|
encodings = (
|
|
'utf-8', 'latin-1',
|
|
'utf-16', 'utf-16-le', 'utf-16-be',
|
|
'utf-32', 'utf-32-le', 'utf-32-be',
|
|
)
|
|
|
|
# Try a range of buffer sizes to test the case where \r is the last
|
|
# character in TextIOWrapper._pending_line.
|
|
for encoding in encodings:
|
|
# 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 = self.BufferedReader(self.BytesIO(data), bufsize)
|
|
textio = self.TextIOWrapper(bufio, newline=newline,
|
|
encoding=encoding)
|
|
if do_reads:
|
|
got_lines = []
|
|
while True:
|
|
c2 = textio.read(2)
|
|
if c2 == '':
|
|
break
|
|
self.assertEqual(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):
|
|
self.assertEqual(got_line, exp_line)
|
|
self.assertEqual(len(got_lines), len(exp_lines))
|
|
|
|
def test_newlines_input(self):
|
|
testdata = b"AAA\nBB\x00B\nCCC\rDDD\rEEE\r\nFFF\r\nGGG"
|
|
normalized = testdata.replace(b"\r\n", b"\n").replace(b"\r", b"\n")
|
|
for newline, expected in [
|
|
(None, normalized.decode("ascii").splitlines(keepends=True)),
|
|
("", testdata.decode("ascii").splitlines(keepends=True)),
|
|
("\n", ["AAA\n", "BB\x00B\n", "CCC\rDDD\rEEE\r\n", "FFF\r\n", "GGG"]),
|
|
("\r\n", ["AAA\nBB\x00B\nCCC\rDDD\rEEE\r\n", "FFF\r\n", "GGG"]),
|
|
("\r", ["AAA\nBB\x00B\nCCC\r", "DDD\r", "EEE\r", "\nFFF\r", "\nGGG"]),
|
|
]:
|
|
buf = self.BytesIO(testdata)
|
|
txt = self.TextIOWrapper(buf, encoding="ascii", newline=newline)
|
|
self.assertEqual(txt.readlines(), expected)
|
|
txt.seek(0)
|
|
self.assertEqual(txt.read(), "".join(expected))
|
|
|
|
def test_newlines_output(self):
|
|
testdict = {
|
|
"": b"AAA\nBBB\nCCC\nX\rY\r\nZ",
|
|
"\n": b"AAA\nBBB\nCCC\nX\rY\r\nZ",
|
|
"\r": b"AAA\rBBB\rCCC\rX\rY\r\rZ",
|
|
"\r\n": b"AAA\r\nBBB\r\nCCC\r\nX\rY\r\r\nZ",
|
|
}
|
|
tests = [(None, testdict[os.linesep])] + sorted(testdict.items())
|
|
for newline, expected in tests:
|
|
buf = self.BytesIO()
|
|
txt = self.TextIOWrapper(buf, encoding="ascii", newline=newline)
|
|
txt.write("AAA\nB")
|
|
txt.write("BB\nCCC\n")
|
|
txt.write("X\rY\r\nZ")
|
|
txt.flush()
|
|
self.assertEqual(buf.closed, False)
|
|
self.assertEqual(buf.getvalue(), expected)
|
|
|
|
def test_destructor(self):
|
|
l = []
|
|
base = self.BytesIO
|
|
class MyBytesIO(base):
|
|
def close(self):
|
|
l.append(self.getvalue())
|
|
base.close(self)
|
|
b = MyBytesIO()
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
t.write("abc")
|
|
del t
|
|
support.gc_collect()
|
|
self.assertEqual([b"abc"], l)
|
|
|
|
def test_override_destructor(self):
|
|
record = []
|
|
class MyTextIO(self.TextIOWrapper):
|
|
def __del__(self):
|
|
record.append(1)
|
|
try:
|
|
f = super().__del__
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
f()
|
|
def close(self):
|
|
record.append(2)
|
|
super().close()
|
|
def flush(self):
|
|
record.append(3)
|
|
super().flush()
|
|
b = self.BytesIO()
|
|
t = MyTextIO(b, encoding="ascii")
|
|
del t
|
|
support.gc_collect()
|
|
self.assertEqual(record, [1, 2, 3])
|
|
|
|
def test_error_through_destructor(self):
|
|
# Test that the exception state is not modified by a destructor,
|
|
# even if close() fails.
|
|
rawio = self.CloseFailureIO()
|
|
with support.catch_unraisable_exception() as cm:
|
|
with self.assertRaises(AttributeError):
|
|
self.TextIOWrapper(rawio, encoding="utf-8").xyzzy
|
|
|
|
self.assertEqual(cm.unraisable.exc_type, OSError)
|
|
|
|
# Systematic tests of the text I/O API
|
|
|
|
def test_basic_io(self):
|
|
for chunksize in (1, 2, 3, 4, 5, 15, 16, 17, 31, 32, 33, 63, 64, 65):
|
|
for enc in "ascii", "latin-1", "utf-8" :# , "utf-16-be", "utf-16-le":
|
|
f = self.open(os_helper.TESTFN, "w+", encoding=enc)
|
|
f._CHUNK_SIZE = chunksize
|
|
self.assertEqual(f.write("abc"), 3)
|
|
f.close()
|
|
f = self.open(os_helper.TESTFN, "r+", encoding=enc)
|
|
f._CHUNK_SIZE = chunksize
|
|
self.assertEqual(f.tell(), 0)
|
|
self.assertEqual(f.read(), "abc")
|
|
cookie = f.tell()
|
|
self.assertEqual(f.seek(0), 0)
|
|
self.assertEqual(f.read(None), "abc")
|
|
f.seek(0)
|
|
self.assertEqual(f.read(2), "ab")
|
|
self.assertEqual(f.read(1), "c")
|
|
self.assertEqual(f.read(1), "")
|
|
self.assertEqual(f.read(), "")
|
|
self.assertEqual(f.tell(), cookie)
|
|
self.assertEqual(f.seek(0), 0)
|
|
self.assertEqual(f.seek(0, 2), cookie)
|
|
self.assertEqual(f.write("def"), 3)
|
|
self.assertEqual(f.seek(cookie), cookie)
|
|
self.assertEqual(f.read(), "def")
|
|
if enc.startswith("utf"):
|
|
self.multi_line_test(f, enc)
|
|
f.close()
|
|
|
|
def multi_line_test(self, f, enc):
|
|
f.seek(0)
|
|
f.truncate()
|
|
sample = "s\xff\u0fff\uffff"
|
|
wlines = []
|
|
for size in (0, 1, 2, 3, 4, 5, 30, 31, 32, 33, 62, 63, 64, 65, 1000):
|
|
chars = []
|
|
for i in range(size):
|
|
chars.append(sample[i % len(sample)])
|
|
line = "".join(chars) + "\n"
|
|
wlines.append((f.tell(), line))
|
|
f.write(line)
|
|
f.seek(0)
|
|
rlines = []
|
|
while True:
|
|
pos = f.tell()
|
|
line = f.readline()
|
|
if not line:
|
|
break
|
|
rlines.append((pos, line))
|
|
self.assertEqual(rlines, wlines)
|
|
|
|
def test_telling(self):
|
|
f = self.open(os_helper.TESTFN, "w+", encoding="utf-8")
|
|
p0 = f.tell()
|
|
f.write("\xff\n")
|
|
p1 = f.tell()
|
|
f.write("\xff\n")
|
|
p2 = f.tell()
|
|
f.seek(0)
|
|
self.assertEqual(f.tell(), p0)
|
|
self.assertEqual(f.readline(), "\xff\n")
|
|
self.assertEqual(f.tell(), p1)
|
|
self.assertEqual(f.readline(), "\xff\n")
|
|
self.assertEqual(f.tell(), p2)
|
|
f.seek(0)
|
|
for line in f:
|
|
self.assertEqual(line, "\xff\n")
|
|
self.assertRaises(OSError, f.tell)
|
|
self.assertEqual(f.tell(), p2)
|
|
f.close()
|
|
|
|
def test_seeking(self):
|
|
chunk_size = _default_chunk_size()
|
|
prefix_size = chunk_size - 2
|
|
u_prefix = "a" * prefix_size
|
|
prefix = bytes(u_prefix.encode("utf-8"))
|
|
self.assertEqual(len(u_prefix), len(prefix))
|
|
u_suffix = "\u8888\n"
|
|
suffix = bytes(u_suffix.encode("utf-8"))
|
|
line = prefix + suffix
|
|
with self.open(os_helper.TESTFN, "wb") as f:
|
|
f.write(line*2)
|
|
with self.open(os_helper.TESTFN, "r", encoding="utf-8") as f:
|
|
s = f.read(prefix_size)
|
|
self.assertEqual(s, str(prefix, "ascii"))
|
|
self.assertEqual(f.tell(), prefix_size)
|
|
self.assertEqual(f.readline(), u_suffix)
|
|
|
|
def test_seeking_too(self):
|
|
# Regression test for a specific bug
|
|
data = b'\xe0\xbf\xbf\n'
|
|
with self.open(os_helper.TESTFN, "wb") as f:
|
|
f.write(data)
|
|
with self.open(os_helper.TESTFN, "r", encoding="utf-8") as f:
|
|
f._CHUNK_SIZE # Just test that it exists
|
|
f._CHUNK_SIZE = 2
|
|
f.readline()
|
|
f.tell()
|
|
|
|
def test_seek_and_tell(self):
|
|
#Test seek/tell using the StatefulIncrementalDecoder.
|
|
# Make test faster by doing smaller seeks
|
|
CHUNK_SIZE = 128
|
|
|
|
def test_seek_and_tell_with_data(data, min_pos=0):
|
|
"""Tell/seek to various points within a data stream and ensure
|
|
that the decoded data returned by read() is consistent."""
|
|
f = self.open(os_helper.TESTFN, 'wb')
|
|
f.write(data)
|
|
f.close()
|
|
f = self.open(os_helper.TESTFN, encoding='test_decoder')
|
|
f._CHUNK_SIZE = CHUNK_SIZE
|
|
decoded = f.read()
|
|
f.close()
|
|
|
|
for i in range(min_pos, len(decoded) + 1): # seek positions
|
|
for j in [1, 5, len(decoded) - i]: # read lengths
|
|
f = self.open(os_helper.TESTFN, encoding='test_decoder')
|
|
self.assertEqual(f.read(i), decoded[:i])
|
|
cookie = f.tell()
|
|
self.assertEqual(f.read(j), decoded[i:i + j])
|
|
f.seek(cookie)
|
|
self.assertEqual(f.read(), decoded[i:])
|
|
f.close()
|
|
|
|
# Enable the test decoder.
|
|
StatefulIncrementalDecoder.codecEnabled = 1
|
|
|
|
# Run the tests.
|
|
try:
|
|
# Try each test case.
|
|
for input, _, _ in StatefulIncrementalDecoderTest.test_cases:
|
|
test_seek_and_tell_with_data(input)
|
|
|
|
# Position each test case so that it crosses a chunk boundary.
|
|
for input, _, _ in StatefulIncrementalDecoderTest.test_cases:
|
|
offset = CHUNK_SIZE - len(input)//2
|
|
prefix = b'.'*offset
|
|
# Don't bother seeking into the prefix (takes too long).
|
|
min_pos = offset*2
|
|
test_seek_and_tell_with_data(prefix + input, min_pos)
|
|
|
|
# Ensure our test decoder won't interfere with subsequent tests.
|
|
finally:
|
|
StatefulIncrementalDecoder.codecEnabled = 0
|
|
|
|
def test_multibyte_seek_and_tell(self):
|
|
f = self.open(os_helper.TESTFN, "w", encoding="euc_jp")
|
|
f.write("AB\n\u3046\u3048\n")
|
|
f.close()
|
|
|
|
f = self.open(os_helper.TESTFN, "r", encoding="euc_jp")
|
|
self.assertEqual(f.readline(), "AB\n")
|
|
p0 = f.tell()
|
|
self.assertEqual(f.readline(), "\u3046\u3048\n")
|
|
p1 = f.tell()
|
|
f.seek(p0)
|
|
self.assertEqual(f.readline(), "\u3046\u3048\n")
|
|
self.assertEqual(f.tell(), p1)
|
|
f.close()
|
|
|
|
def test_tell_after_readline_with_cr(self):
|
|
# Test for gh-141314: TextIOWrapper.tell() assertion failure
|
|
# when dealing with standalone carriage returns
|
|
data = b'line1\r'
|
|
with self.open(os_helper.TESTFN, "wb") as f:
|
|
f.write(data)
|
|
|
|
with self.open(os_helper.TESTFN, "r") as f:
|
|
# Read line that ends with \r
|
|
line = f.readline()
|
|
self.assertEqual(line, "line1\n")
|
|
# This should not cause an assertion failure
|
|
pos = f.tell()
|
|
# Verify we can seek back to this position
|
|
f.seek(pos)
|
|
remaining = f.read()
|
|
self.assertEqual(remaining, "")
|
|
|
|
|
|
def test_seek_with_encoder_state(self):
|
|
f = self.open(os_helper.TESTFN, "w", encoding="euc_jis_2004")
|
|
f.write("\u00e6\u0300")
|
|
p0 = f.tell()
|
|
f.write("\u00e6")
|
|
f.seek(p0)
|
|
f.write("\u0300")
|
|
f.close()
|
|
|
|
f = self.open(os_helper.TESTFN, "r", encoding="euc_jis_2004")
|
|
self.assertEqual(f.readline(), "\u00e6\u0300\u0300")
|
|
f.close()
|
|
|
|
def test_encoded_writes(self):
|
|
data = "1234567890"
|
|
tests = ("utf-16",
|
|
"utf-16-le",
|
|
"utf-16-be",
|
|
"utf-32",
|
|
"utf-32-le",
|
|
"utf-32-be")
|
|
for encoding in tests:
|
|
buf = self.BytesIO()
|
|
f = self.TextIOWrapper(buf, encoding=encoding)
|
|
# Check if the BOM is written only once (see issue1753).
|
|
f.write(data)
|
|
f.write(data)
|
|
f.seek(0)
|
|
self.assertEqual(f.read(), data * 2)
|
|
f.seek(0)
|
|
self.assertEqual(f.read(), data * 2)
|
|
self.assertEqual(buf.getvalue(), (data * 2).encode(encoding))
|
|
|
|
def test_unreadable(self):
|
|
class UnReadable(self.BytesIO):
|
|
def readable(self):
|
|
return False
|
|
txt = self.TextIOWrapper(UnReadable(), encoding="utf-8")
|
|
self.assertRaises(OSError, txt.read)
|
|
|
|
def test_read_one_by_one(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(b"AA\r\nBB"), encoding="utf-8")
|
|
reads = ""
|
|
while True:
|
|
c = txt.read(1)
|
|
if not c:
|
|
break
|
|
reads += c
|
|
self.assertEqual(reads, "AA\nBB")
|
|
|
|
def test_readlines(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(b"AA\nBB\nCC"), encoding="utf-8")
|
|
self.assertEqual(txt.readlines(), ["AA\n", "BB\n", "CC"])
|
|
txt.seek(0)
|
|
self.assertEqual(txt.readlines(None), ["AA\n", "BB\n", "CC"])
|
|
txt.seek(0)
|
|
self.assertEqual(txt.readlines(5), ["AA\n", "BB\n"])
|
|
|
|
# read in amounts equal to TextIOWrapper._CHUNK_SIZE which is 128.
|
|
def test_read_by_chunk(self):
|
|
# make sure "\r\n" straddles 128 char boundary.
|
|
txt = self.TextIOWrapper(self.BytesIO(b"A" * 127 + b"\r\nB"), encoding="utf-8")
|
|
reads = ""
|
|
while True:
|
|
c = txt.read(128)
|
|
if not c:
|
|
break
|
|
reads += c
|
|
self.assertEqual(reads, "A"*127+"\nB")
|
|
|
|
def test_writelines(self):
|
|
l = ['ab', 'cd', 'ef']
|
|
buf = self.BytesIO()
|
|
txt = self.TextIOWrapper(buf, encoding="utf-8")
|
|
txt.writelines(l)
|
|
txt.flush()
|
|
self.assertEqual(buf.getvalue(), b'abcdef')
|
|
|
|
def test_writelines_userlist(self):
|
|
l = UserList(['ab', 'cd', 'ef'])
|
|
buf = self.BytesIO()
|
|
txt = self.TextIOWrapper(buf, encoding="utf-8")
|
|
txt.writelines(l)
|
|
txt.flush()
|
|
self.assertEqual(buf.getvalue(), b'abcdef')
|
|
|
|
def test_writelines_error(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(), encoding="utf-8")
|
|
self.assertRaises(TypeError, txt.writelines, [1, 2, 3])
|
|
self.assertRaises(TypeError, txt.writelines, None)
|
|
self.assertRaises(TypeError, txt.writelines, b'abc')
|
|
|
|
def test_issue1395_1(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
|
|
# read one char at a time
|
|
reads = ""
|
|
while True:
|
|
c = txt.read(1)
|
|
if not c:
|
|
break
|
|
reads += c
|
|
self.assertEqual(reads, self.normalized)
|
|
|
|
def test_issue1395_2(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
txt._CHUNK_SIZE = 4
|
|
|
|
reads = ""
|
|
while True:
|
|
c = txt.read(4)
|
|
if not c:
|
|
break
|
|
reads += c
|
|
self.assertEqual(reads, self.normalized)
|
|
|
|
def test_issue1395_3(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
txt._CHUNK_SIZE = 4
|
|
|
|
reads = txt.read(4)
|
|
reads += txt.read(4)
|
|
reads += txt.readline()
|
|
reads += txt.readline()
|
|
reads += txt.readline()
|
|
self.assertEqual(reads, self.normalized)
|
|
|
|
def test_issue1395_4(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
txt._CHUNK_SIZE = 4
|
|
|
|
reads = txt.read(4)
|
|
reads += txt.read()
|
|
self.assertEqual(reads, self.normalized)
|
|
|
|
def test_issue1395_5(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
txt._CHUNK_SIZE = 4
|
|
|
|
reads = txt.read(4)
|
|
pos = txt.tell()
|
|
txt.seek(0)
|
|
txt.seek(pos)
|
|
self.assertEqual(txt.read(4), "BBB\n")
|
|
|
|
def test_issue2282(self):
|
|
buffer = self.BytesIO(self.testdata)
|
|
txt = self.TextIOWrapper(buffer, encoding="ascii")
|
|
|
|
self.assertEqual(buffer.seekable(), txt.seekable())
|
|
|
|
def test_append_bom(self):
|
|
# The BOM is not written again when appending to a non-empty file
|
|
filename = os_helper.TESTFN
|
|
for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
|
|
with self.open(filename, 'w', encoding=charset) as f:
|
|
f.write('aaa')
|
|
pos = f.tell()
|
|
with self.open(filename, 'rb') as f:
|
|
self.assertEqual(f.read(), 'aaa'.encode(charset))
|
|
|
|
with self.open(filename, 'a', encoding=charset) as f:
|
|
f.write('xxx')
|
|
with self.open(filename, 'rb') as f:
|
|
self.assertEqual(f.read(), 'aaaxxx'.encode(charset))
|
|
|
|
def test_seek_bom(self):
|
|
# Same test, but when seeking manually
|
|
filename = os_helper.TESTFN
|
|
for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
|
|
with self.open(filename, 'w', encoding=charset) as f:
|
|
f.write('aaa')
|
|
pos = f.tell()
|
|
with self.open(filename, 'r+', encoding=charset) as f:
|
|
f.seek(pos)
|
|
f.write('zzz')
|
|
f.seek(0)
|
|
f.write('bbb')
|
|
with self.open(filename, 'rb') as f:
|
|
self.assertEqual(f.read(), 'bbbzzz'.encode(charset))
|
|
|
|
def test_seek_append_bom(self):
|
|
# Same test, but first seek to the start and then to the end
|
|
filename = os_helper.TESTFN
|
|
for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
|
|
with self.open(filename, 'w', encoding=charset) as f:
|
|
f.write('aaa')
|
|
with self.open(filename, 'a', encoding=charset) as f:
|
|
f.seek(0)
|
|
f.seek(0, self.SEEK_END)
|
|
f.write('xxx')
|
|
with self.open(filename, 'rb') as f:
|
|
self.assertEqual(f.read(), 'aaaxxx'.encode(charset))
|
|
|
|
def test_errors_property(self):
|
|
with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f:
|
|
self.assertEqual(f.errors, "strict")
|
|
with self.open(os_helper.TESTFN, "w", encoding="utf-8", errors="replace") as f:
|
|
self.assertEqual(f.errors, "replace")
|
|
|
|
@support.no_tracing
|
|
@threading_helper.requires_working_threading()
|
|
def test_threads_write(self):
|
|
# Issue6750: concurrent writes could duplicate data
|
|
event = threading.Event()
|
|
with self.open(os_helper.TESTFN, "w", encoding="utf-8", buffering=1) as f:
|
|
def run(n):
|
|
text = "Thread%03d\n" % n
|
|
event.wait()
|
|
f.write(text)
|
|
threads = [threading.Thread(target=run, args=(x,))
|
|
for x in range(20)]
|
|
with threading_helper.start_threads(threads, event.set):
|
|
time.sleep(0.02)
|
|
with self.open(os_helper.TESTFN, encoding="utf-8") as f:
|
|
content = f.read()
|
|
for n in range(20):
|
|
self.assertEqual(content.count("Thread%03d\n" % n), 1)
|
|
|
|
def test_flush_error_on_close(self):
|
|
# Test that text file is closed despite failed flush
|
|
# and that flush() is called before file closed.
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
closed = []
|
|
def bad_flush():
|
|
closed[:] = [txt.closed, txt.buffer.closed]
|
|
raise OSError()
|
|
txt.flush = bad_flush
|
|
self.assertRaises(OSError, txt.close) # exception not swallowed
|
|
self.assertTrue(txt.closed)
|
|
self.assertTrue(txt.buffer.closed)
|
|
self.assertTrue(closed) # flush() called
|
|
self.assertFalse(closed[0]) # flush() called before file closed
|
|
self.assertFalse(closed[1])
|
|
txt.flush = lambda: None # break reference loop
|
|
|
|
def test_close_error_on_close(self):
|
|
buffer = self.BytesIO(self.testdata)
|
|
def bad_flush():
|
|
raise OSError('flush')
|
|
def bad_close():
|
|
raise OSError('close')
|
|
buffer.close = bad_close
|
|
txt = self.TextIOWrapper(buffer, encoding="ascii")
|
|
txt.flush = bad_flush
|
|
with self.assertRaises(OSError) as err: # exception not swallowed
|
|
txt.close()
|
|
self.assertEqual(err.exception.args, ('close',))
|
|
self.assertIsInstance(err.exception.__context__, OSError)
|
|
self.assertEqual(err.exception.__context__.args, ('flush',))
|
|
self.assertFalse(txt.closed)
|
|
|
|
# Silence destructor error
|
|
buffer.close = lambda: None
|
|
txt.flush = lambda: None
|
|
|
|
def test_nonnormalized_close_error_on_close(self):
|
|
# Issue #21677
|
|
buffer = self.BytesIO(self.testdata)
|
|
def bad_flush():
|
|
raise non_existing_flush
|
|
def bad_close():
|
|
raise non_existing_close
|
|
buffer.close = bad_close
|
|
txt = self.TextIOWrapper(buffer, encoding="ascii")
|
|
txt.flush = bad_flush
|
|
with self.assertRaises(NameError) as err: # exception not swallowed
|
|
txt.close()
|
|
self.assertIn('non_existing_close', str(err.exception))
|
|
self.assertIsInstance(err.exception.__context__, NameError)
|
|
self.assertIn('non_existing_flush', str(err.exception.__context__))
|
|
self.assertFalse(txt.closed)
|
|
|
|
# Silence destructor error
|
|
buffer.close = lambda: None
|
|
txt.flush = lambda: None
|
|
|
|
def test_multi_close(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
txt.close()
|
|
txt.close()
|
|
txt.close()
|
|
self.assertRaises(ValueError, txt.flush)
|
|
|
|
def test_unseekable(self):
|
|
txt = self.TextIOWrapper(self.MockUnseekableIO(self.testdata), encoding="utf-8")
|
|
self.assertRaises(self.UnsupportedOperation, txt.tell)
|
|
self.assertRaises(self.UnsupportedOperation, txt.seek, 0)
|
|
|
|
def test_readonly_attributes(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
|
|
buf = self.BytesIO(self.testdata)
|
|
with self.assertRaises(AttributeError):
|
|
txt.buffer = buf
|
|
|
|
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.
|
|
raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
|
|
txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
|
|
# Reads
|
|
self.assertEqual(txt.read(4), 'abcd')
|
|
self.assertEqual(txt.readline(), 'efghi\n')
|
|
self.assertEqual(list(txt), ['jkl\n', 'opq\n'])
|
|
|
|
def test_rawio_write_through(self):
|
|
# Issue #12591: with write_through=True, writes don't need a flush
|
|
raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
|
|
txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n',
|
|
write_through=True)
|
|
txt.write('1')
|
|
txt.write('23\n4')
|
|
txt.write('5')
|
|
self.assertEqual(b''.join(raw._write_stack), b'123\n45')
|
|
|
|
def test_bufio_write_through(self):
|
|
# Issue #21396: write_through=True doesn't force a flush()
|
|
# on the underlying binary buffered object.
|
|
flush_called, write_called = [], []
|
|
class BufferedWriter(self.BufferedWriter):
|
|
def flush(self, *args, **kwargs):
|
|
flush_called.append(True)
|
|
return super().flush(*args, **kwargs)
|
|
def write(self, *args, **kwargs):
|
|
write_called.append(True)
|
|
return super().write(*args, **kwargs)
|
|
|
|
rawio = self.BytesIO()
|
|
data = b"a"
|
|
bufio = BufferedWriter(rawio, len(data)*2)
|
|
textio = self.TextIOWrapper(bufio, encoding='ascii',
|
|
write_through=True)
|
|
# write to the buffered io but don't overflow the buffer
|
|
text = data.decode('ascii')
|
|
textio.write(text)
|
|
|
|
# buffer.flush is not called with write_through=True
|
|
self.assertFalse(flush_called)
|
|
# buffer.write *is* called with write_through=True
|
|
self.assertTrue(write_called)
|
|
self.assertEqual(rawio.getvalue(), b"") # no flush
|
|
|
|
write_called = [] # reset
|
|
textio.write(text * 10) # total content is larger than bufio buffer
|
|
self.assertTrue(write_called)
|
|
self.assertEqual(rawio.getvalue(), data * 11) # all flushed
|
|
|
|
def test_reconfigure_write_through(self):
|
|
raw = self.MockRawIO([])
|
|
t = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
|
|
t.write('1')
|
|
t.reconfigure(write_through=True) # implied flush
|
|
self.assertEqual(t.write_through, True)
|
|
self.assertEqual(b''.join(raw._write_stack), b'1')
|
|
t.write('23')
|
|
self.assertEqual(b''.join(raw._write_stack), b'123')
|
|
t.reconfigure(write_through=False)
|
|
self.assertEqual(t.write_through, False)
|
|
t.write('45')
|
|
t.flush()
|
|
self.assertEqual(b''.join(raw._write_stack), b'12345')
|
|
# Keeping default value
|
|
t.reconfigure()
|
|
t.reconfigure(write_through=None)
|
|
self.assertEqual(t.write_through, False)
|
|
t.reconfigure(write_through=True)
|
|
t.reconfigure()
|
|
t.reconfigure(write_through=None)
|
|
self.assertEqual(t.write_through, True)
|
|
|
|
def test_read_nonbytes(self):
|
|
# Issue #17106
|
|
# Crash when underlying read() returns non-bytes
|
|
t = self.TextIOWrapper(self.StringIO('a'), encoding="utf-8")
|
|
self.assertRaises(TypeError, t.read, 1)
|
|
t = self.TextIOWrapper(self.StringIO('a'), encoding="utf-8")
|
|
self.assertRaises(TypeError, t.readline)
|
|
t = self.TextIOWrapper(self.StringIO('a'), encoding="utf-8")
|
|
self.assertRaises(TypeError, t.read)
|
|
|
|
def test_illegal_encoder(self):
|
|
# Issue 31271: Calling write() while the return value of encoder's
|
|
# encode() is invalid shouldn't cause an assertion failure.
|
|
rot13 = codecs.lookup("rot13")
|
|
with support.swap_attr(rot13, '_is_text_encoding', True):
|
|
t = self.TextIOWrapper(self.BytesIO(b'foo'), encoding="rot13")
|
|
self.assertRaises(TypeError, t.write, 'bar')
|
|
|
|
def test_illegal_decoder(self):
|
|
# Issue #17106
|
|
# Bypass the early encoding check added in issue 20404
|
|
def _make_illegal_wrapper():
|
|
quopri = codecs.lookup("quopri")
|
|
quopri._is_text_encoding = True
|
|
try:
|
|
t = self.TextIOWrapper(self.BytesIO(b'aaaaaa'),
|
|
newline='\n', encoding="quopri")
|
|
finally:
|
|
quopri._is_text_encoding = False
|
|
return t
|
|
# Crash when decoder returns non-string
|
|
t = _make_illegal_wrapper()
|
|
self.assertRaises(TypeError, t.read, 1)
|
|
t = _make_illegal_wrapper()
|
|
self.assertRaises(TypeError, t.readline)
|
|
t = _make_illegal_wrapper()
|
|
self.assertRaises(TypeError, t.read)
|
|
|
|
# Issue 31243: calling read() while the return value of decoder's
|
|
# getstate() is invalid should neither crash the interpreter nor
|
|
# raise a SystemError.
|
|
def _make_very_illegal_wrapper(getstate_ret_val):
|
|
class BadDecoder:
|
|
def getstate(self):
|
|
return getstate_ret_val
|
|
def _get_bad_decoder(dummy):
|
|
return BadDecoder()
|
|
quopri = codecs.lookup("quopri")
|
|
with support.swap_attr(quopri, 'incrementaldecoder',
|
|
_get_bad_decoder):
|
|
return _make_illegal_wrapper()
|
|
t = _make_very_illegal_wrapper(42)
|
|
self.assertRaises(TypeError, t.read, 42)
|
|
t = _make_very_illegal_wrapper(())
|
|
self.assertRaises(TypeError, t.read, 42)
|
|
t = _make_very_illegal_wrapper((1, 2))
|
|
self.assertRaises(TypeError, t.read, 42)
|
|
|
|
def _check_create_at_shutdown(self, **kwargs):
|
|
# Issue #20037: creating a TextIOWrapper at shutdown
|
|
# shouldn't crash the interpreter.
|
|
iomod = self.io.__name__
|
|
code = """if 1:
|
|
import codecs
|
|
import {iomod} as io
|
|
|
|
# Avoid looking up codecs at shutdown
|
|
codecs.lookup('utf-8')
|
|
|
|
class C:
|
|
def __del__(self):
|
|
io.TextIOWrapper(io.BytesIO(), **{kwargs})
|
|
print("ok")
|
|
c = C()
|
|
""".format(iomod=iomod, kwargs=kwargs)
|
|
return assert_python_ok("-c", code)
|
|
|
|
def test_create_at_shutdown_without_encoding(self):
|
|
rc, out, err = self._check_create_at_shutdown()
|
|
if err:
|
|
# Can error out with a RuntimeError if the module state
|
|
# isn't found.
|
|
self.assertIn(self.shutdown_error, err.decode())
|
|
else:
|
|
self.assertEqual("ok", out.decode().strip())
|
|
|
|
def test_create_at_shutdown_with_encoding(self):
|
|
rc, out, err = self._check_create_at_shutdown(encoding='utf-8',
|
|
errors='strict')
|
|
self.assertFalse(err)
|
|
self.assertEqual("ok", out.decode().strip())
|
|
|
|
def test_read_byteslike(self):
|
|
r = MemviewBytesIO(b'Just some random string\n')
|
|
t = self.TextIOWrapper(r, 'utf-8')
|
|
|
|
# TextIOwrapper will not read the full string, because
|
|
# we truncate it to a multiple of the native int size
|
|
# so that we can construct a more complex memoryview.
|
|
bytes_val = _to_memoryview(r.getvalue()).tobytes()
|
|
|
|
self.assertEqual(t.read(200), bytes_val.decode('utf-8'))
|
|
|
|
def test_issue22849(self):
|
|
class F(object):
|
|
def readable(self): return True
|
|
def writable(self): return True
|
|
def seekable(self): return True
|
|
|
|
for i in range(10):
|
|
try:
|
|
self.TextIOWrapper(F(), encoding='utf-8')
|
|
except Exception:
|
|
pass
|
|
|
|
F.tell = lambda x: 0
|
|
t = self.TextIOWrapper(F(), encoding='utf-8')
|
|
|
|
def test_reconfigure_locale(self):
|
|
wrapper = self.TextIOWrapper(self.BytesIO(b"test"))
|
|
wrapper.reconfigure(encoding="locale")
|
|
|
|
def test_reconfigure_encoding_read(self):
|
|
# latin1 -> utf8
|
|
# (latin1 can decode utf-8 encoded string)
|
|
data = 'abc\xe9\n'.encode('latin1') + 'd\xe9f\n'.encode('utf8')
|
|
raw = self.BytesIO(data)
|
|
txt = self.TextIOWrapper(raw, encoding='latin1', newline='\n')
|
|
self.assertEqual(txt.readline(), 'abc\xe9\n')
|
|
with self.assertRaises(self.UnsupportedOperation):
|
|
txt.reconfigure(encoding='utf-8')
|
|
with self.assertRaises(self.UnsupportedOperation):
|
|
txt.reconfigure(newline=None)
|
|
|
|
def test_reconfigure_write_fromascii(self):
|
|
# ascii has a specific encodefunc in the C implementation,
|
|
# but utf-8-sig has not. Make sure that we get rid of the
|
|
# cached encodefunc when we switch encoders.
|
|
raw = self.BytesIO()
|
|
txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
|
|
txt.write('foo\n')
|
|
txt.reconfigure(encoding='utf-8-sig')
|
|
txt.write('\xe9\n')
|
|
txt.flush()
|
|
self.assertEqual(raw.getvalue(), b'foo\n\xc3\xa9\n')
|
|
|
|
def test_reconfigure_write(self):
|
|
# latin -> utf8
|
|
raw = self.BytesIO()
|
|
txt = self.TextIOWrapper(raw, encoding='latin1', newline='\n')
|
|
txt.write('abc\xe9\n')
|
|
txt.reconfigure(encoding='utf-8')
|
|
self.assertEqual(raw.getvalue(), b'abc\xe9\n')
|
|
txt.write('d\xe9f\n')
|
|
txt.flush()
|
|
self.assertEqual(raw.getvalue(), b'abc\xe9\nd\xc3\xa9f\n')
|
|
|
|
# ascii -> utf-8-sig: ensure that no BOM is written in the middle of
|
|
# the file
|
|
raw = self.BytesIO()
|
|
txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
|
|
txt.write('abc\n')
|
|
txt.reconfigure(encoding='utf-8-sig')
|
|
txt.write('d\xe9f\n')
|
|
txt.flush()
|
|
self.assertEqual(raw.getvalue(), b'abc\nd\xc3\xa9f\n')
|
|
|
|
def test_reconfigure_write_non_seekable(self):
|
|
raw = self.BytesIO()
|
|
raw.seekable = lambda: False
|
|
raw.seek = None
|
|
txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
|
|
txt.write('abc\n')
|
|
txt.reconfigure(encoding='utf-8-sig')
|
|
txt.write('d\xe9f\n')
|
|
txt.flush()
|
|
|
|
# If the raw stream is not seekable, there'll be a BOM
|
|
self.assertEqual(raw.getvalue(), b'abc\n\xef\xbb\xbfd\xc3\xa9f\n')
|
|
|
|
def test_reconfigure_defaults(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(), 'ascii', 'replace', '\n')
|
|
txt.reconfigure(encoding=None)
|
|
self.assertEqual(txt.encoding, 'ascii')
|
|
self.assertEqual(txt.errors, 'replace')
|
|
txt.write('LF\n')
|
|
|
|
txt.reconfigure(newline='\r\n')
|
|
self.assertEqual(txt.encoding, 'ascii')
|
|
self.assertEqual(txt.errors, 'replace')
|
|
|
|
txt.reconfigure(errors='ignore')
|
|
self.assertEqual(txt.encoding, 'ascii')
|
|
self.assertEqual(txt.errors, 'ignore')
|
|
txt.write('CRLF\n')
|
|
|
|
txt.reconfigure(encoding='utf-8', newline=None)
|
|
self.assertEqual(txt.errors, 'strict')
|
|
txt.seek(0)
|
|
self.assertEqual(txt.read(), 'LF\nCRLF\n')
|
|
|
|
self.assertEqual(txt.detach().getvalue(), b'LF\nCRLF\r\n')
|
|
|
|
def test_reconfigure_errors(self):
|
|
txt = self.TextIOWrapper(self.BytesIO(), 'ascii', 'replace', '\r')
|
|
with self.assertRaises(TypeError): # there was a crash
|
|
txt.reconfigure(encoding=42)
|
|
if self.is_C:
|
|
with self.assertRaises(UnicodeEncodeError):
|
|
txt.reconfigure(encoding='\udcfe')
|
|
with self.assertRaises(LookupError):
|
|
txt.reconfigure(encoding='locale\0')
|
|
# TODO: txt.reconfigure(encoding='utf-8\0')
|
|
# TODO: txt.reconfigure(encoding='nonexisting')
|
|
with self.assertRaises(TypeError):
|
|
txt.reconfigure(errors=42)
|
|
if self.is_C:
|
|
with self.assertRaises(UnicodeEncodeError):
|
|
txt.reconfigure(errors='\udcfe')
|
|
# TODO: txt.reconfigure(errors='ignore\0')
|
|
# TODO: txt.reconfigure(errors='nonexisting')
|
|
with self.assertRaises(TypeError):
|
|
txt.reconfigure(newline=42)
|
|
with self.assertRaises(ValueError):
|
|
txt.reconfigure(newline='\udcfe')
|
|
with self.assertRaises(ValueError):
|
|
txt.reconfigure(newline='xyz')
|
|
if not self.is_C:
|
|
# TODO: Should fail in C too.
|
|
with self.assertRaises(ValueError):
|
|
txt.reconfigure(newline='\n\0')
|
|
if self.is_C:
|
|
# TODO: Use __bool__(), not __index__().
|
|
with self.assertRaises(ZeroDivisionError):
|
|
txt.reconfigure(line_buffering=BadIndex())
|
|
with self.assertRaises(OverflowError):
|
|
txt.reconfigure(line_buffering=2**1000)
|
|
with self.assertRaises(ZeroDivisionError):
|
|
txt.reconfigure(write_through=BadIndex())
|
|
with self.assertRaises(OverflowError):
|
|
txt.reconfigure(write_through=2**1000)
|
|
with self.assertRaises(ZeroDivisionError): # there was a crash
|
|
txt.reconfigure(line_buffering=BadIndex(),
|
|
write_through=BadIndex())
|
|
self.assertEqual(txt.encoding, 'ascii')
|
|
self.assertEqual(txt.errors, 'replace')
|
|
self.assertIs(txt.line_buffering, False)
|
|
self.assertIs(txt.write_through, False)
|
|
|
|
txt.reconfigure(encoding='latin1', errors='ignore', newline='\r\n',
|
|
line_buffering=True, write_through=True)
|
|
self.assertEqual(txt.encoding, 'latin1')
|
|
self.assertEqual(txt.errors, 'ignore')
|
|
self.assertIs(txt.line_buffering, True)
|
|
self.assertIs(txt.write_through, True)
|
|
|
|
def test_reconfigure_newline(self):
|
|
raw = self.BytesIO(b'CR\rEOF')
|
|
txt = self.TextIOWrapper(raw, 'ascii', newline='\n')
|
|
txt.reconfigure(newline=None)
|
|
self.assertEqual(txt.readline(), 'CR\n')
|
|
raw = self.BytesIO(b'CR\rEOF')
|
|
txt = self.TextIOWrapper(raw, 'ascii', newline='\n')
|
|
txt.reconfigure(newline='')
|
|
self.assertEqual(txt.readline(), 'CR\r')
|
|
raw = self.BytesIO(b'CR\rLF\nEOF')
|
|
txt = self.TextIOWrapper(raw, 'ascii', newline='\r')
|
|
txt.reconfigure(newline='\n')
|
|
self.assertEqual(txt.readline(), 'CR\rLF\n')
|
|
raw = self.BytesIO(b'LF\nCR\rEOF')
|
|
txt = self.TextIOWrapper(raw, 'ascii', newline='\n')
|
|
txt.reconfigure(newline='\r')
|
|
self.assertEqual(txt.readline(), 'LF\nCR\r')
|
|
raw = self.BytesIO(b'CR\rCRLF\r\nEOF')
|
|
txt = self.TextIOWrapper(raw, 'ascii', newline='\r')
|
|
txt.reconfigure(newline='\r\n')
|
|
self.assertEqual(txt.readline(), 'CR\rCRLF\r\n')
|
|
|
|
txt = self.TextIOWrapper(self.BytesIO(), 'ascii', newline='\r')
|
|
txt.reconfigure(newline=None)
|
|
txt.write('linesep\n')
|
|
txt.reconfigure(newline='')
|
|
txt.write('LF\n')
|
|
txt.reconfigure(newline='\n')
|
|
txt.write('LF\n')
|
|
txt.reconfigure(newline='\r')
|
|
txt.write('CR\n')
|
|
txt.reconfigure(newline='\r\n')
|
|
txt.write('CRLF\n')
|
|
expected = 'linesep' + os.linesep + 'LF\nLF\nCR\rCRLF\r\n'
|
|
self.assertEqual(txt.detach().getvalue().decode('ascii'), expected)
|
|
|
|
def test_issue25862(self):
|
|
# Assertion failures occurred in tell() after read() and write().
|
|
t = self.TextIOWrapper(self.BytesIO(b'test'), encoding='ascii')
|
|
t.read(1)
|
|
t.read()
|
|
t.tell()
|
|
t = self.TextIOWrapper(self.BytesIO(b'test'), encoding='ascii')
|
|
t.read(1)
|
|
t.write('x')
|
|
t.tell()
|
|
|
|
def test_issue35928(self):
|
|
p = self.BufferedRWPair(self.BytesIO(b'foo\nbar\n'), self.BytesIO())
|
|
f = self.TextIOWrapper(p)
|
|
res = f.readline()
|
|
self.assertEqual(res, 'foo\n')
|
|
f.write(res)
|
|
self.assertEqual(res + f.readline(), 'foo\nbar\n')
|
|
|
|
def test_pickling_subclass(self):
|
|
global MyTextIO
|
|
class MyTextIO(self.TextIOWrapper):
|
|
def __init__(self, raw, tag):
|
|
super().__init__(raw)
|
|
self.tag = tag
|
|
def __getstate__(self):
|
|
return self.tag, self.buffer.getvalue()
|
|
def __setstate__(slf, state):
|
|
tag, value = state
|
|
slf.__init__(self.BytesIO(value), tag)
|
|
|
|
raw = self.BytesIO(b'data')
|
|
txt = MyTextIO(raw, 'ham')
|
|
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
|
|
with self.subTest(protocol=proto):
|
|
pickled = pickle.dumps(txt, proto)
|
|
newtxt = pickle.loads(pickled)
|
|
self.assertEqual(newtxt.buffer.getvalue(), b'data')
|
|
self.assertEqual(newtxt.tag, 'ham')
|
|
del MyTextIO
|
|
|
|
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
|
|
def test_read_non_blocking(self):
|
|
import os
|
|
r, w = os.pipe()
|
|
try:
|
|
os.set_blocking(r, False)
|
|
with self.io.open(r, 'rt') as textfile:
|
|
r = None
|
|
# Nothing has been written so a non-blocking read raises a BlockingIOError exception.
|
|
with self.assertRaises(BlockingIOError):
|
|
textfile.read()
|
|
finally:
|
|
if r is not None:
|
|
os.close(r)
|
|
os.close(w)
|
|
|
|
|
|
class MemviewBytesIO(io.BytesIO):
|
|
'''A BytesIO object whose read method returns memoryviews
|
|
rather than bytes'''
|
|
|
|
def read1(self, len_):
|
|
return _to_memoryview(super().read1(len_))
|
|
|
|
def read(self, len_):
|
|
return _to_memoryview(super().read(len_))
|
|
|
|
def _to_memoryview(buf):
|
|
'''Convert bytes-object *buf* to a non-trivial memoryview'''
|
|
|
|
arr = array.array('i')
|
|
idx = len(buf) - len(buf) % arr.itemsize
|
|
arr.frombytes(buf[:idx])
|
|
return memoryview(arr)
|
|
|
|
|
|
class CTextIOWrapperTest(TextIOWrapperTest, CTestCase):
|
|
shutdown_error = "LookupError: unknown encoding: ascii"
|
|
|
|
def test_initialization(self):
|
|
r = self.BytesIO(b"\xc3\xa9\n\n")
|
|
b = self.BufferedReader(r, 1000)
|
|
t = self.TextIOWrapper(b, encoding="utf-8")
|
|
self.assertRaises(ValueError, t.__init__, b, encoding="utf-8", newline='xyzzy')
|
|
self.assertRaises(ValueError, t.read)
|
|
|
|
t = self.TextIOWrapper.__new__(self.TextIOWrapper)
|
|
self.assertRaises(Exception, repr, t)
|
|
|
|
@unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
|
|
def test_garbage_collection(self):
|
|
# C TextIOWrapper objects are collected, and collecting them flushes
|
|
# all data to disk.
|
|
# The Python version has __del__, so it ends in gc.garbage instead.
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", ResourceWarning)
|
|
rawio = self.FileIO(os_helper.TESTFN, "wb")
|
|
b = self.BufferedWriter(rawio)
|
|
t = self.TextIOWrapper(b, encoding="ascii")
|
|
t.write("456def")
|
|
t.x = t
|
|
wr = weakref.ref(t)
|
|
del t
|
|
support.gc_collect()
|
|
self.assertIsNone(wr(), wr)
|
|
with self.open(os_helper.TESTFN, "rb") as f:
|
|
self.assertEqual(f.read(), b"456def")
|
|
|
|
def test_rwpair_cleared_before_textio(self):
|
|
# Issue 13070: TextIOWrapper's finalization would crash when called
|
|
# after the reference to the underlying BufferedRWPair's writer got
|
|
# cleared by the GC.
|
|
for i in range(1000):
|
|
b1 = self.BufferedRWPair(self.MockRawIO(), self.MockRawIO())
|
|
t1 = self.TextIOWrapper(b1, encoding="ascii")
|
|
b2 = self.BufferedRWPair(self.MockRawIO(), self.MockRawIO())
|
|
t2 = self.TextIOWrapper(b2, encoding="ascii")
|
|
# circular references
|
|
t1.buddy = t2
|
|
t2.buddy = t1
|
|
support.gc_collect()
|
|
|
|
def test_del__CHUNK_SIZE_SystemError(self):
|
|
t = self.TextIOWrapper(self.BytesIO(), encoding='ascii')
|
|
with self.assertRaises(AttributeError):
|
|
del t._CHUNK_SIZE
|
|
|
|
def test_internal_buffer_size(self):
|
|
# bpo-43260: TextIOWrapper's internal buffer should not store
|
|
# data larger than chunk size.
|
|
chunk_size = 8192 # default chunk size, updated later
|
|
|
|
class MockIO(self.MockRawIO):
|
|
def write(self, data):
|
|
if len(data) > chunk_size:
|
|
raise RuntimeError
|
|
return super().write(data)
|
|
|
|
buf = MockIO()
|
|
t = self.TextIOWrapper(buf, encoding="ascii")
|
|
chunk_size = t._CHUNK_SIZE
|
|
t.write("abc")
|
|
t.write("def")
|
|
# default chunk size is 8192 bytes so t don't write data to buf.
|
|
self.assertEqual([], buf._write_stack)
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
t.write("x"*(chunk_size+1))
|
|
|
|
self.assertEqual([b"abcdef"], buf._write_stack)
|
|
t.write("ghi")
|
|
t.write("x"*chunk_size)
|
|
self.assertEqual([b"abcdef", b"ghi", b"x"*chunk_size], buf._write_stack)
|
|
|
|
def test_issue119506(self):
|
|
chunk_size = 8192
|
|
|
|
class MockIO(self.MockRawIO):
|
|
written = False
|
|
def write(self, data):
|
|
if not self.written:
|
|
self.written = True
|
|
t.write("middle")
|
|
return super().write(data)
|
|
|
|
buf = MockIO()
|
|
t = self.TextIOWrapper(buf)
|
|
t.write("abc")
|
|
t.write("def")
|
|
# writing data which size >= chunk_size cause flushing buffer before write.
|
|
t.write("g" * chunk_size)
|
|
t.flush()
|
|
|
|
self.assertEqual([b"abcdef", b"middle", b"g"*chunk_size],
|
|
buf._write_stack)
|
|
|
|
def test_issue142594(self):
|
|
wrapper = None
|
|
detached = False
|
|
class ReentrantRawIO(self.RawIOBase):
|
|
@property
|
|
def closed(self):
|
|
nonlocal detached
|
|
if wrapper is not None and not detached:
|
|
detached = True
|
|
wrapper.detach()
|
|
return False
|
|
|
|
raw = ReentrantRawIO()
|
|
wrapper = self.TextIOWrapper(raw)
|
|
wrapper.close() # should not crash
|
|
|
|
@unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: a bytes-like object is required, not 'NoneType'
|
|
def test_read_non_blocking(self):
|
|
return super().test_read_non_blocking()
|
|
|
|
|
|
class PyTextIOWrapperTest(TextIOWrapperTest, PyTestCase):
|
|
shutdown_error = "LookupError: unknown encoding: ascii"
|
|
|
|
|
|
class IncrementalNewlineDecoderTest:
|
|
|
|
def check_newline_decoding_utf8(self, 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()
|
|
self.assertEqual(decoder.decode(b, **kwargs), s)
|
|
decoder.setstate(state)
|
|
self.assertEqual(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', "")
|
|
self.assertRaises(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(self, 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))
|
|
self.assertEqual(decoder.newlines, None)
|
|
_decode_bytewise("abc\n\r")
|
|
self.assertEqual(decoder.newlines, '\n')
|
|
_decode_bytewise("\nabc")
|
|
self.assertEqual(decoder.newlines, ('\n', '\r\n'))
|
|
_decode_bytewise("abc\r")
|
|
self.assertEqual(decoder.newlines, ('\n', '\r\n'))
|
|
_decode_bytewise("abc")
|
|
self.assertEqual(decoder.newlines, ('\r', '\n', '\r\n'))
|
|
_decode_bytewise("abc\r")
|
|
self.assertEqual("".join(result), "abc\n\nabcabc\nabcabc")
|
|
decoder.reset()
|
|
input = "abc"
|
|
if encoder is not None:
|
|
encoder.reset()
|
|
input = encoder.encode(input)
|
|
self.assertEqual(decoder.decode(input), "abc")
|
|
self.assertEqual(decoder.newlines, None)
|
|
|
|
def test_newline_decoder(self):
|
|
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',
|
|
)
|
|
for enc in encodings:
|
|
decoder = enc and codecs.getincrementaldecoder(enc)()
|
|
decoder = self.IncrementalNewlineDecoder(decoder, translate=True)
|
|
self.check_newline_decoding(decoder, enc)
|
|
decoder = codecs.getincrementaldecoder("utf-8")()
|
|
decoder = self.IncrementalNewlineDecoder(decoder, translate=True)
|
|
self.check_newline_decoding_utf8(decoder)
|
|
self.assertRaises(TypeError, decoder.setstate, 42)
|
|
|
|
def test_newline_bytes(self):
|
|
# Issue 5433: Excessive optimization in IncrementalNewlineDecoder
|
|
def _check(dec):
|
|
self.assertEqual(dec.newlines, None)
|
|
self.assertEqual(dec.decode("\u0D00"), "\u0D00")
|
|
self.assertEqual(dec.newlines, None)
|
|
self.assertEqual(dec.decode("\u0A00"), "\u0A00")
|
|
self.assertEqual(dec.newlines, None)
|
|
dec = self.IncrementalNewlineDecoder(None, translate=False)
|
|
_check(dec)
|
|
dec = self.IncrementalNewlineDecoder(None, translate=True)
|
|
_check(dec)
|
|
|
|
def test_translate(self):
|
|
# issue 35062
|
|
for translate in (-2, -1, 1, 2):
|
|
decoder = codecs.getincrementaldecoder("utf-8")()
|
|
decoder = self.IncrementalNewlineDecoder(decoder, translate)
|
|
self.check_newline_decoding_utf8(decoder)
|
|
decoder = codecs.getincrementaldecoder("utf-8")()
|
|
decoder = self.IncrementalNewlineDecoder(decoder, translate=0)
|
|
self.assertEqual(decoder.decode(b"\r\r\n"), "\r\r\n")
|
|
|
|
class CIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest, unittest.TestCase):
|
|
IncrementalNewlineDecoder = io.IncrementalNewlineDecoder
|
|
|
|
@support.cpython_only
|
|
def test_uninitialized(self):
|
|
uninitialized = self.IncrementalNewlineDecoder.__new__(
|
|
self.IncrementalNewlineDecoder)
|
|
self.assertRaises(ValueError, uninitialized.decode, b'bar')
|
|
self.assertRaises(ValueError, uninitialized.getstate)
|
|
self.assertRaises(ValueError, uninitialized.setstate, (b'foo', 0))
|
|
self.assertRaises(ValueError, uninitialized.reset)
|
|
|
|
|
|
class PyIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest, unittest.TestCase):
|
|
IncrementalNewlineDecoder = pyio.IncrementalNewlineDecoder
|