Python code coverage for Lib/test/test_io.py

#	count	content
1	n/a	"""Unit tests for the io module."""
2	n/a
3	n/a	# Tests of io are scattered over the test suite:
4	n/a	# * test_bufio - tests file buffering
5	n/a	# * test_memoryio - tests BytesIO and StringIO
6	n/a	# * test_fileio - tests FileIO
7	n/a	# * test_file - tests the file interface
8	n/a	# * test_io - tests everything else in the io module
9	n/a	# * test_univnewlines - tests universal newline support
10	n/a	# * test_largefile - tests operations on a file greater than 2**32 bytes
11	n/a	# (only enabled with -ulargefile)
12	n/a
13	n/a	################################################################################
14	n/a	# ATTENTION TEST WRITERS!!!
15	n/a	################################################################################
16	n/a	# When writing tests for io, it's important to test both the C and Python
17	n/a	# implementations. This is usually done by writing a base test that refers to
18	n/a	# the type it is testing as an attribute. Then it provides custom subclasses to
19	n/a	# test both implementations. This file has lots of examples.
20	n/a	################################################################################
21	n/a
22	n/a	import abc
23	n/a	import array
24	n/a	import errno
25	n/a	import locale
26	n/a	import os
27	n/a	import pickle
28	n/a	import random
29	n/a	import signal
30	n/a	import sys
31	n/a	import time
32	n/a	import unittest
33	n/a	import warnings
34	n/a	import weakref
35	n/a	from collections import deque, UserList
36	n/a	from itertools import cycle, count
37	n/a	from test import support
38	n/a	from test.support.script_helper import assert_python_ok, run_python_until_end
39	n/a
40	n/a	import codecs
41	n/a	import io # C implementation of io
42	n/a	import _pyio as pyio # Python implementation of io
43	n/a	try:
44	n/a	import threading
45	n/a	except ImportError:
46	n/a	threading = None
47	n/a
48	n/a	try:
49	n/a	import ctypes
50	n/a	except ImportError:
51	n/a	def byteslike(pos, *kw):
52	n/a	return array.array("b", bytes(pos, *kw))
53	n/a	else:
54	n/a	def byteslike(pos, *kw):
55	n/a	"""Create a bytes-like object having no string or sequence methods"""
56	n/a	data = bytes(pos, *kw)
57	n/a	obj = EmptyStruct()
58	n/a	ctypes.resize(obj, len(data))
59	n/a	memoryview(obj).cast("B")[:] = data
60	n/a	return obj
61	n/a	class EmptyStruct(ctypes.Structure):
62	n/a	pass
63	n/a
64	n/a	def _default_chunk_size():
65	n/a	"""Get the default TextIOWrapper chunk size"""
66	n/a	with open(__file__, "r", encoding="latin-1") as f:
67	n/a	return f._CHUNK_SIZE
68	n/a
69	n/a
70	n/a	class MockRawIOWithoutRead:
71	n/a	"""A RawIO implementation without read(), so as to exercise the default
72	n/a	RawIO.read() which calls readinto()."""
73	n/a
74	n/a	def __init__(self, read_stack=()):
75	n/a	self._read_stack = list(read_stack)
76	n/a	self._write_stack = []
77	n/a	self._reads = 0
78	n/a	self._extraneous_reads = 0
79	n/a
80	n/a	def write(self, b):
81	n/a	self._write_stack.append(bytes(b))
82	n/a	return len(b)
83	n/a
84	n/a	def writable(self):
85	n/a	return True
86	n/a
87	n/a	def fileno(self):
88	n/a	return 42
89	n/a
90	n/a	def readable(self):
91	n/a	return True
92	n/a
93	n/a	def seekable(self):
94	n/a	return True
95	n/a
96	n/a	def seek(self, pos, whence):
97	n/a	return 0 # wrong but we gotta return something
98	n/a
99	n/a	def tell(self):
100	n/a	return 0 # same comment as above
101	n/a
102	n/a	def readinto(self, buf):
103	n/a	self._reads += 1
104	n/a	max_len = len(buf)
105	n/a	try:
106	n/a	data = self._read_stack[0]
107	n/a	except IndexError:
108	n/a	self._extraneous_reads += 1
109	n/a	return 0
110	n/a	if data is None:
111	n/a	del self._read_stack[0]
112	n/a	return None
113	n/a	n = len(data)
114	n/a	if len(data) <= max_len:
115	n/a	del self._read_stack[0]
116	n/a	buf[:n] = data
117	n/a	return n
118	n/a	else:
119	n/a	buf[:] = data[:max_len]
120	n/a	self._read_stack[0] = data[max_len:]
121	n/a	return max_len
122	n/a
123	n/a	def truncate(self, pos=None):
124	n/a	return pos
125	n/a
126	n/a	class CMockRawIOWithoutRead(MockRawIOWithoutRead, io.RawIOBase):
127	n/a	pass
128	n/a
129	n/a	class PyMockRawIOWithoutRead(MockRawIOWithoutRead, pyio.RawIOBase):
130	n/a	pass
131	n/a
132	n/a
133	n/a	class MockRawIO(MockRawIOWithoutRead):
134	n/a
135	n/a	def read(self, n=None):
136	n/a	self._reads += 1
137	n/a	try:
138	n/a	return self._read_stack.pop(0)
139	n/a	except:
140	n/a	self._extraneous_reads += 1
141	n/a	return b""
142	n/a
143	n/a	class CMockRawIO(MockRawIO, io.RawIOBase):
144	n/a	pass
145	n/a
146	n/a	class PyMockRawIO(MockRawIO, pyio.RawIOBase):
147	n/a	pass
148	n/a
149	n/a
150	n/a	class MisbehavedRawIO(MockRawIO):
151	n/a	def write(self, b):
152	n/a	return super().write(b) * 2
153	n/a
154	n/a	def read(self, n=None):
155	n/a	return super().read(n) * 2
156	n/a
157	n/a	def seek(self, pos, whence):
158	n/a	return -123
159	n/a
160	n/a	def tell(self):
161	n/a	return -456
162	n/a
163	n/a	def readinto(self, buf):
164	n/a	super().readinto(buf)
165	n/a	return len(buf) * 5
166	n/a
167	n/a	class CMisbehavedRawIO(MisbehavedRawIO, io.RawIOBase):
168	n/a	pass
169	n/a
170	n/a	class PyMisbehavedRawIO(MisbehavedRawIO, pyio.RawIOBase):
171	n/a	pass
172	n/a
173	n/a
174	n/a	class CloseFailureIO(MockRawIO):
175	n/a	closed = 0
176	n/a
177	n/a	def close(self):
178	n/a	if not self.closed:
179	n/a	self.closed = 1
180	n/a	raise OSError
181	n/a
182	n/a	class CCloseFailureIO(CloseFailureIO, io.RawIOBase):
183	n/a	pass
184	n/a
185	n/a	class PyCloseFailureIO(CloseFailureIO, pyio.RawIOBase):
186	n/a	pass
187	n/a
188	n/a
189	n/a	class MockFileIO:
190	n/a
191	n/a	def __init__(self, data):
192	n/a	self.read_history = []
193	n/a	super().__init__(data)
194	n/a
195	n/a	def read(self, n=None):
196	n/a	res = super().read(n)
197	n/a	self.read_history.append(None if res is None else len(res))
198	n/a	return res
199	n/a
200	n/a	def readinto(self, b):
201	n/a	res = super().readinto(b)
202	n/a	self.read_history.append(res)
203	n/a	return res
204	n/a
205	n/a	class CMockFileIO(MockFileIO, io.BytesIO):
206	n/a	pass
207	n/a
208	n/a	class PyMockFileIO(MockFileIO, pyio.BytesIO):
209	n/a	pass
210	n/a
211	n/a
212	n/a	class MockUnseekableIO:
213	n/a	def seekable(self):
214	n/a	return False
215	n/a
216	n/a	def seek(self, *args):
217	n/a	raise self.UnsupportedOperation("not seekable")
218	n/a
219	n/a	def tell(self, *args):
220	n/a	raise self.UnsupportedOperation("not seekable")
221	n/a
222	n/a	def truncate(self, *args):
223	n/a	raise self.UnsupportedOperation("not seekable")
224	n/a
225	n/a	class CMockUnseekableIO(MockUnseekableIO, io.BytesIO):
226	n/a	UnsupportedOperation = io.UnsupportedOperation
227	n/a
228	n/a	class PyMockUnseekableIO(MockUnseekableIO, pyio.BytesIO):
229	n/a	UnsupportedOperation = pyio.UnsupportedOperation
230	n/a
231	n/a
232	n/a	class MockNonBlockWriterIO:
233	n/a
234	n/a	def __init__(self):
235	n/a	self._write_stack = []
236	n/a	self._blocker_char = None
237	n/a
238	n/a	def pop_written(self):
239	n/a	s = b"".join(self._write_stack)
240	n/a	self._write_stack[:] = []
241	n/a	return s
242	n/a
243	n/a	def block_on(self, char):
244	n/a	"""Block when a given char is encountered."""
245	n/a	self._blocker_char = char
246	n/a
247	n/a	def readable(self):
248	n/a	return True
249	n/a
250	n/a	def seekable(self):
251	n/a	return True
252	n/a
253	n/a	def writable(self):
254	n/a	return True
255	n/a
256	n/a	def write(self, b):
257	n/a	b = bytes(b)
258	n/a	n = -1
259	n/a	if self._blocker_char:
260	n/a	try:
261	n/a	n = b.index(self._blocker_char)
262	n/a	except ValueError:
263	n/a	pass
264	n/a	else:
265	n/a	if n > 0:
266	n/a	# write data up to the first blocker
267	n/a	self._write_stack.append(b[:n])
268	n/a	return n
269	n/a	else:
270	n/a	# cancel blocker and indicate would block
271	n/a	self._blocker_char = None
272	n/a	return None
273	n/a	self._write_stack.append(b)
274	n/a	return len(b)
275	n/a
276	n/a	class CMockNonBlockWriterIO(MockNonBlockWriterIO, io.RawIOBase):
277	n/a	BlockingIOError = io.BlockingIOError
278	n/a
279	n/a	class PyMockNonBlockWriterIO(MockNonBlockWriterIO, pyio.RawIOBase):
280	n/a	BlockingIOError = pyio.BlockingIOError
281	n/a
282	n/a
283	n/a	class IOTest(unittest.TestCase):
284	n/a
285	n/a	def setUp(self):
286	n/a	support.unlink(support.TESTFN)
287	n/a
288	n/a	def tearDown(self):
289	n/a	support.unlink(support.TESTFN)
290	n/a
291	n/a	def write_ops(self, f):
292	n/a	self.assertEqual(f.write(b"blah."), 5)
293	n/a	f.truncate(0)
294	n/a	self.assertEqual(f.tell(), 5)
295	n/a	f.seek(0)
296	n/a
297	n/a	self.assertEqual(f.write(b"blah."), 5)
298	n/a	self.assertEqual(f.seek(0), 0)
299	n/a	self.assertEqual(f.write(b"Hello."), 6)
300	n/a	self.assertEqual(f.tell(), 6)
301	n/a	self.assertEqual(f.seek(-1, 1), 5)
302	n/a	self.assertEqual(f.tell(), 5)
303	n/a	buffer = bytearray(b" world\n\n\n")
304	n/a	self.assertEqual(f.write(buffer), 9)
305	n/a	buffer[:] = b"" 9 # Overwrite our copy of the data
306	n/a	self.assertEqual(f.seek(0), 0)
307	n/a	self.assertEqual(f.write(b"h"), 1)
308	n/a	self.assertEqual(f.seek(-1, 2), 13)
309	n/a	self.assertEqual(f.tell(), 13)
310	n/a
311	n/a	self.assertEqual(f.truncate(12), 12)
312	n/a	self.assertEqual(f.tell(), 13)
313	n/a	self.assertRaises(TypeError, f.seek, 0.0)
314	n/a
315	n/a	def read_ops(self, f, buffered=False):
316	n/a	data = f.read(5)
317	n/a	self.assertEqual(data, b"hello")
318	n/a	data = byteslike(data)
319	n/a	self.assertEqual(f.readinto(data), 5)
320	n/a	self.assertEqual(bytes(data), b" worl")
321	n/a	data = bytearray(5)
322	n/a	self.assertEqual(f.readinto(data), 2)
323	n/a	self.assertEqual(len(data), 5)
324	n/a	self.assertEqual(data[:2], b"d\n")
325	n/a	self.assertEqual(f.seek(0), 0)
326	n/a	self.assertEqual(f.read(20), b"hello world\n")
327	n/a	self.assertEqual(f.read(1), b"")
328	n/a	self.assertEqual(f.readinto(byteslike(b"x")), 0)
329	n/a	self.assertEqual(f.seek(-6, 2), 6)
330	n/a	self.assertEqual(f.read(5), b"world")
331	n/a	self.assertEqual(f.read(0), b"")
332	n/a	self.assertEqual(f.readinto(byteslike()), 0)
333	n/a	self.assertEqual(f.seek(-6, 1), 5)
334	n/a	self.assertEqual(f.read(5), b" worl")
335	n/a	self.assertEqual(f.tell(), 10)
336	n/a	self.assertRaises(TypeError, f.seek, 0.0)
337	n/a	if buffered:
338	n/a	f.seek(0)
339	n/a	self.assertEqual(f.read(), b"hello world\n")
340	n/a	f.seek(6)
341	n/a	self.assertEqual(f.read(), b"world\n")
342	n/a	self.assertEqual(f.read(), b"")
343	n/a	f.seek(0)
344	n/a	data = byteslike(5)
345	n/a	self.assertEqual(f.readinto1(data), 5)
346	n/a	self.assertEqual(bytes(data), b"hello")
347	n/a
348	n/a	LARGE = 2**31
349	n/a
350	n/a	def large_file_ops(self, f):
351	n/a	assert f.readable()
352	n/a	assert f.writable()
353	n/a	try:
354	n/a	self.assertEqual(f.seek(self.LARGE), self.LARGE)
355	n/a	except (OverflowError, ValueError):
356	n/a	self.skipTest("no largefile support")
357	n/a	self.assertEqual(f.tell(), self.LARGE)
358	n/a	self.assertEqual(f.write(b"xxx"), 3)
359	n/a	self.assertEqual(f.tell(), self.LARGE + 3)
360	n/a	self.assertEqual(f.seek(-1, 1), self.LARGE + 2)
361	n/a	self.assertEqual(f.truncate(), self.LARGE + 2)
362	n/a	self.assertEqual(f.tell(), self.LARGE + 2)
363	n/a	self.assertEqual(f.seek(0, 2), self.LARGE + 2)
364	n/a	self.assertEqual(f.truncate(self.LARGE + 1), self.LARGE + 1)
365	n/a	self.assertEqual(f.tell(), self.LARGE + 2)
366	n/a	self.assertEqual(f.seek(0, 2), self.LARGE + 1)
367	n/a	self.assertEqual(f.seek(-1, 2), self.LARGE)
368	n/a	self.assertEqual(f.read(2), b"x")
369	n/a
370	n/a	def test_invalid_operations(self):
371	n/a	# Try writing on a file opened in read mode and vice-versa.
372	n/a	exc = self.UnsupportedOperation
373	n/a	for mode in ("w", "wb"):
374	n/a	with self.open(support.TESTFN, mode) as fp:
375	n/a	self.assertRaises(exc, fp.read)
376	n/a	self.assertRaises(exc, fp.readline)
377	n/a	with self.open(support.TESTFN, "wb", buffering=0) as fp:
378	n/a	self.assertRaises(exc, fp.read)
379	n/a	self.assertRaises(exc, fp.readline)
380	n/a	with self.open(support.TESTFN, "rb", buffering=0) as fp:
381	n/a	self.assertRaises(exc, fp.write, b"blah")
382	n/a	self.assertRaises(exc, fp.writelines, [b"blah\n"])
383	n/a	with self.open(support.TESTFN, "rb") as fp:
384	n/a	self.assertRaises(exc, fp.write, b"blah")
385	n/a	self.assertRaises(exc, fp.writelines, [b"blah\n"])
386	n/a	with self.open(support.TESTFN, "r") as fp:
387	n/a	self.assertRaises(exc, fp.write, "blah")
388	n/a	self.assertRaises(exc, fp.writelines, ["blah\n"])
389	n/a	# Non-zero seeking from current or end pos
390	n/a	self.assertRaises(exc, fp.seek, 1, self.SEEK_CUR)
391	n/a	self.assertRaises(exc, fp.seek, -1, self.SEEK_END)
392	n/a
393	n/a	def test_optional_abilities(self):
394	n/a	# Test for OSError when optional APIs are not supported
395	n/a	# The purpose of this test is to try fileno(), reading, writing and
396	n/a	# seeking operations with various objects that indicate they do not
397	n/a	# support these operations.
398	n/a
399	n/a	def pipe_reader():
400	n/a	[r, w] = os.pipe()
401	n/a	os.close(w) # So that read() is harmless
402	n/a	return self.FileIO(r, "r")
403	n/a
404	n/a	def pipe_writer():
405	n/a	[r, w] = os.pipe()
406	n/a	self.addCleanup(os.close, r)
407	n/a	# Guarantee that we can write into the pipe without blocking
408	n/a	thread = threading.Thread(target=os.read, args=(r, 100))
409	n/a	thread.start()
410	n/a	self.addCleanup(thread.join)
411	n/a	return self.FileIO(w, "w")
412	n/a
413	n/a	def buffered_reader():
414	n/a	return self.BufferedReader(self.MockUnseekableIO())
415	n/a
416	n/a	def buffered_writer():
417	n/a	return self.BufferedWriter(self.MockUnseekableIO())
418	n/a
419	n/a	def buffered_random():
420	n/a	return self.BufferedRandom(self.BytesIO())
421	n/a
422	n/a	def buffered_rw_pair():
423	n/a	return self.BufferedRWPair(self.MockUnseekableIO(),
424	n/a	self.MockUnseekableIO())
425	n/a
426	n/a	def text_reader():
427	n/a	class UnseekableReader(self.MockUnseekableIO):
428	n/a	writable = self.BufferedIOBase.writable
429	n/a	write = self.BufferedIOBase.write
430	n/a	return self.TextIOWrapper(UnseekableReader(), "ascii")
431	n/a
432	n/a	def text_writer():
433	n/a	class UnseekableWriter(self.MockUnseekableIO):
434	n/a	readable = self.BufferedIOBase.readable
435	n/a	read = self.BufferedIOBase.read
436	n/a	return self.TextIOWrapper(UnseekableWriter(), "ascii")
437	n/a
438	n/a	tests = (
439	n/a	(pipe_reader, "fr"), (pipe_writer, "fw"),
440	n/a	(buffered_reader, "r"), (buffered_writer, "w"),
441	n/a	(buffered_random, "rws"), (buffered_rw_pair, "rw"),
442	n/a	(text_reader, "r"), (text_writer, "w"),
443	n/a	(self.BytesIO, "rws"), (self.StringIO, "rws"),
444	n/a	)
445	n/a	for [test, abilities] in tests:
446	n/a	if test is pipe_writer and not threading:
447	n/a	continue # Skip subtest that uses a background thread
448	n/a	with self.subTest(test), test() as obj:
449	n/a	readable = "r" in abilities
450	n/a	self.assertEqual(obj.readable(), readable)
451	n/a	writable = "w" in abilities
452	n/a	self.assertEqual(obj.writable(), writable)
453	n/a
454	n/a	if isinstance(obj, self.TextIOBase):
455	n/a	data = "3"
456	n/a	elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)):
457	n/a	data = b"3"
458	n/a	else:
459	n/a	self.fail("Unknown base class")
460	n/a
461	n/a	if "f" in abilities:
462	n/a	obj.fileno()
463	n/a	else:
464	n/a	self.assertRaises(OSError, obj.fileno)
465	n/a
466	n/a	if readable:
467	n/a	obj.read(1)
468	n/a	obj.read()
469	n/a	else:
470	n/a	self.assertRaises(OSError, obj.read, 1)
471	n/a	self.assertRaises(OSError, obj.read)
472	n/a
473	n/a	if writable:
474	n/a	obj.write(data)
475	n/a	else:
476	n/a	self.assertRaises(OSError, obj.write, data)
477	n/a
478	n/a	if sys.platform.startswith("win") and test in (
479	n/a	pipe_reader, pipe_writer):
480	n/a	# Pipes seem to appear as seekable on Windows
481	n/a	continue
482	n/a	seekable = "s" in abilities
483	n/a	self.assertEqual(obj.seekable(), seekable)
484	n/a
485	n/a	if seekable:
486	n/a	obj.tell()
487	n/a	obj.seek(0)
488	n/a	else:
489	n/a	self.assertRaises(OSError, obj.tell)
490	n/a	self.assertRaises(OSError, obj.seek, 0)
491	n/a
492	n/a	if writable and seekable:
493	n/a	obj.truncate()
494	n/a	obj.truncate(0)
495	n/a	else:
496	n/a	self.assertRaises(OSError, obj.truncate)
497	n/a	self.assertRaises(OSError, obj.truncate, 0)
498	n/a
499	n/a	def test_open_handles_NUL_chars(self):
500	n/a	fn_with_NUL = 'foo\0bar'
501	n/a	self.assertRaises(ValueError, self.open, fn_with_NUL, 'w')
502	n/a
503	n/a	bytes_fn = bytes(fn_with_NUL, 'ascii')
504	n/a	with warnings.catch_warnings():
505	n/a	warnings.simplefilter("ignore", DeprecationWarning)
506	n/a	self.assertRaises(ValueError, self.open, bytes_fn, 'w')
507	n/a
508	n/a	def test_raw_file_io(self):
509	n/a	with self.open(support.TESTFN, "wb", buffering=0) as f:
510	n/a	self.assertEqual(f.readable(), False)
511	n/a	self.assertEqual(f.writable(), True)
512	n/a	self.assertEqual(f.seekable(), True)
513	n/a	self.write_ops(f)
514	n/a	with self.open(support.TESTFN, "rb", buffering=0) as f:
515	n/a	self.assertEqual(f.readable(), True)
516	n/a	self.assertEqual(f.writable(), False)
517	n/a	self.assertEqual(f.seekable(), True)
518	n/a	self.read_ops(f)
519	n/a
520	n/a	def test_buffered_file_io(self):
521	n/a	with self.open(support.TESTFN, "wb") as f:
522	n/a	self.assertEqual(f.readable(), False)
523	n/a	self.assertEqual(f.writable(), True)
524	n/a	self.assertEqual(f.seekable(), True)
525	n/a	self.write_ops(f)
526	n/a	with self.open(support.TESTFN, "rb") as f:
527	n/a	self.assertEqual(f.readable(), True)
528	n/a	self.assertEqual(f.writable(), False)
529	n/a	self.assertEqual(f.seekable(), True)
530	n/a	self.read_ops(f, True)
531	n/a
532	n/a	def test_readline(self):
533	n/a	with self.open(support.TESTFN, "wb") as f:
534	n/a	f.write(b"abc\ndef\nxyzzy\nfoo\x00bar\nanother line")
535	n/a	with self.open(support.TESTFN, "rb") as f:
536	n/a	self.assertEqual(f.readline(), b"abc\n")
537	n/a	self.assertEqual(f.readline(10), b"def\n")
538	n/a	self.assertEqual(f.readline(2), b"xy")
539	n/a	self.assertEqual(f.readline(4), b"zzy\n")
540	n/a	self.assertEqual(f.readline(), b"foo\x00bar\n")
541	n/a	self.assertEqual(f.readline(None), b"another line")
542	n/a	self.assertRaises(TypeError, f.readline, 5.3)
543	n/a	with self.open(support.TESTFN, "r") as f:
544	n/a	self.assertRaises(TypeError, f.readline, 5.3)
545	n/a
546	n/a	def test_raw_bytes_io(self):
547	n/a	f = self.BytesIO()
548	n/a	self.write_ops(f)
549	n/a	data = f.getvalue()
550	n/a	self.assertEqual(data, b"hello world\n")
551	n/a	f = self.BytesIO(data)
552	n/a	self.read_ops(f, True)
553	n/a
554	n/a	def test_large_file_ops(self):
555	n/a	# On Windows and Mac OSX this test comsumes large resources; It takes
556	n/a	# a long time to build the >2GB file and takes >2GB of disk space
557	n/a	# therefore the resource must be enabled to run this test.
558	n/a	if sys.platform[:3] == 'win' or sys.platform == 'darwin':
559	n/a	support.requires(
560	n/a	'largefile',
561	n/a	'test requires %s bytes and a long time to run' % self.LARGE)
562	n/a	with self.open(support.TESTFN, "w+b", 0) as f:
563	n/a	self.large_file_ops(f)
564	n/a	with self.open(support.TESTFN, "w+b") as f:
565	n/a	self.large_file_ops(f)
566	n/a
567	n/a	def test_with_open(self):
568	n/a	for bufsize in (0, 1, 100):
569	n/a	f = None
570	n/a	with self.open(support.TESTFN, "wb", bufsize) as f:
571	n/a	f.write(b"xxx")
572	n/a	self.assertEqual(f.closed, True)
573	n/a	f = None
574	n/a	try:
575	n/a	with self.open(support.TESTFN, "wb", bufsize) as f:
576	n/a	1/0
577	n/a	except ZeroDivisionError:
578	n/a	self.assertEqual(f.closed, True)
579	n/a	else:
580	n/a	self.fail("1/0 didn't raise an exception")
581	n/a
582	n/a	# issue 5008
583	n/a	def test_append_mode_tell(self):
584	n/a	with self.open(support.TESTFN, "wb") as f:
585	n/a	f.write(b"xxx")
586	n/a	with self.open(support.TESTFN, "ab", buffering=0) as f:
587	n/a	self.assertEqual(f.tell(), 3)
588	n/a	with self.open(support.TESTFN, "ab") as f:
589	n/a	self.assertEqual(f.tell(), 3)
590	n/a	with self.open(support.TESTFN, "a") as f:
591	n/a	self.assertGreater(f.tell(), 0)
592	n/a
593	n/a	def test_destructor(self):
594	n/a	record = []
595	n/a	class MyFileIO(self.FileIO):
596	n/a	def __del__(self):
597	n/a	record.append(1)
598	n/a	try:
599	n/a	f = super().__del__
600	n/a	except AttributeError:
601	n/a	pass
602	n/a	else:
603	n/a	f()
604	n/a	def close(self):
605	n/a	record.append(2)
606	n/a	super().close()
607	n/a	def flush(self):
608	n/a	record.append(3)
609	n/a	super().flush()
610	n/a	with support.check_warnings(('', ResourceWarning)):
611	n/a	f = MyFileIO(support.TESTFN, "wb")
612	n/a	f.write(b"xxx")
613	n/a	del f
614	n/a	support.gc_collect()
615	n/a	self.assertEqual(record, [1, 2, 3])
616	n/a	with self.open(support.TESTFN, "rb") as f:
617	n/a	self.assertEqual(f.read(), b"xxx")
618	n/a
619	n/a	def _check_base_destructor(self, base):
620	n/a	record = []
621	n/a	class MyIO(base):
622	n/a	def __init__(self):
623	n/a	# This exercises the availability of attributes on object
624	n/a	# destruction.
625	n/a	# (in the C version, close() is called by the tp_dealloc
626	n/a	# function, not by __del__)
627	n/a	self.on_del = 1
628	n/a	self.on_close = 2
629	n/a	self.on_flush = 3
630	n/a	def __del__(self):
631	n/a	record.append(self.on_del)
632	n/a	try:
633	n/a	f = super().__del__
634	n/a	except AttributeError:
635	n/a	pass
636	n/a	else:
637	n/a	f()
638	n/a	def close(self):
639	n/a	record.append(self.on_close)
640	n/a	super().close()
641	n/a	def flush(self):
642	n/a	record.append(self.on_flush)
643	n/a	super().flush()
644	n/a	f = MyIO()
645	n/a	del f
646	n/a	support.gc_collect()
647	n/a	self.assertEqual(record, [1, 2, 3])
648	n/a
649	n/a	def test_IOBase_destructor(self):
650	n/a	self._check_base_destructor(self.IOBase)
651	n/a
652	n/a	def test_RawIOBase_destructor(self):
653	n/a	self._check_base_destructor(self.RawIOBase)
654	n/a
655	n/a	def test_BufferedIOBase_destructor(self):
656	n/a	self._check_base_destructor(self.BufferedIOBase)
657	n/a
658	n/a	def test_TextIOBase_destructor(self):
659	n/a	self._check_base_destructor(self.TextIOBase)
660	n/a
661	n/a	def test_close_flushes(self):
662	n/a	with self.open(support.TESTFN, "wb") as f:
663	n/a	f.write(b"xxx")
664	n/a	with self.open(support.TESTFN, "rb") as f:
665	n/a	self.assertEqual(f.read(), b"xxx")
666	n/a
667	n/a	def test_array_writes(self):
668	n/a	a = array.array('i', range(10))
669	n/a	n = len(a.tobytes())
670	n/a	def check(f):
671	n/a	with f:
672	n/a	self.assertEqual(f.write(a), n)
673	n/a	f.writelines((a,))
674	n/a	check(self.BytesIO())
675	n/a	check(self.FileIO(support.TESTFN, "w"))
676	n/a	check(self.BufferedWriter(self.MockRawIO()))
677	n/a	check(self.BufferedRandom(self.MockRawIO()))
678	n/a	check(self.BufferedRWPair(self.MockRawIO(), self.MockRawIO()))
679	n/a
680	n/a	def test_closefd(self):
681	n/a	self.assertRaises(ValueError, self.open, support.TESTFN, 'w',
682	n/a	closefd=False)
683	n/a
684	n/a	def test_read_closed(self):
685	n/a	with self.open(support.TESTFN, "w") as f:
686	n/a	f.write("egg\n")
687	n/a	with self.open(support.TESTFN, "r") as f:
688	n/a	file = self.open(f.fileno(), "r", closefd=False)
689	n/a	self.assertEqual(file.read(), "egg\n")
690	n/a	file.seek(0)
691	n/a	file.close()
692	n/a	self.assertRaises(ValueError, file.read)
693	n/a
694	n/a	def test_no_closefd_with_filename(self):
695	n/a	# can't use closefd in combination with a file name
696	n/a	self.assertRaises(ValueError, self.open, support.TESTFN, "r", closefd=False)
697	n/a
698	n/a	def test_closefd_attr(self):
699	n/a	with self.open(support.TESTFN, "wb") as f:
700	n/a	f.write(b"egg\n")
701	n/a	with self.open(support.TESTFN, "r") as f:
702	n/a	self.assertEqual(f.buffer.raw.closefd, True)
703	n/a	file = self.open(f.fileno(), "r", closefd=False)
704	n/a	self.assertEqual(file.buffer.raw.closefd, False)
705	n/a
706	n/a	def test_garbage_collection(self):
707	n/a	# FileIO objects are collected, and collecting them flushes
708	n/a	# all data to disk.
709	n/a	with support.check_warnings(('', ResourceWarning)):
710	n/a	f = self.FileIO(support.TESTFN, "wb")
711	n/a	f.write(b"abcxxx")
712	n/a	f.f = f
713	n/a	wr = weakref.ref(f)
714	n/a	del f
715	n/a	support.gc_collect()
716	n/a	self.assertIsNone(wr(), wr)
717	n/a	with self.open(support.TESTFN, "rb") as f:
718	n/a	self.assertEqual(f.read(), b"abcxxx")
719	n/a
720	n/a	def test_unbounded_file(self):
721	n/a	# Issue #1174606: reading from an unbounded stream such as /dev/zero.
722	n/a	zero = "/dev/zero"
723	n/a	if not os.path.exists(zero):
724	n/a	self.skipTest("{0} does not exist".format(zero))
725	n/a	if sys.maxsize > 0x7FFFFFFF:
726	n/a	self.skipTest("test can only run in a 32-bit address space")
727	n/a	if support.real_max_memuse < support._2G:
728	n/a	self.skipTest("test requires at least 2GB of memory")
729	n/a	with self.open(zero, "rb", buffering=0) as f:
730	n/a	self.assertRaises(OverflowError, f.read)
731	n/a	with self.open(zero, "rb") as f:
732	n/a	self.assertRaises(OverflowError, f.read)
733	n/a	with self.open(zero, "r") as f:
734	n/a	self.assertRaises(OverflowError, f.read)
735	n/a
736	n/a	def check_flush_error_on_close(self, args, *kwargs):
737	n/a	# Test that the file is closed despite failed flush
738	n/a	# and that flush() is called before file closed.
739	n/a	f = self.open(args, *kwargs)
740	n/a	closed = []
741	n/a	def bad_flush():
742	n/a	closed[:] = [f.closed]
743	n/a	raise OSError()
744	n/a	f.flush = bad_flush
745	n/a	self.assertRaises(OSError, f.close) # exception not swallowed
746	n/a	self.assertTrue(f.closed)
747	n/a	self.assertTrue(closed) # flush() called
748	n/a	self.assertFalse(closed[0]) # flush() called before file closed
749	n/a	f.flush = lambda: None # break reference loop
750	n/a
751	n/a	def test_flush_error_on_close(self):
752	n/a	# raw file
753	n/a	# Issue #5700: io.FileIO calls flush() after file closed
754	n/a	self.check_flush_error_on_close(support.TESTFN, 'wb', buffering=0)
755	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
756	n/a	self.check_flush_error_on_close(fd, 'wb', buffering=0)
757	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
758	n/a	self.check_flush_error_on_close(fd, 'wb', buffering=0, closefd=False)
759	n/a	os.close(fd)
760	n/a	# buffered io
761	n/a	self.check_flush_error_on_close(support.TESTFN, 'wb')
762	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
763	n/a	self.check_flush_error_on_close(fd, 'wb')
764	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
765	n/a	self.check_flush_error_on_close(fd, 'wb', closefd=False)
766	n/a	os.close(fd)
767	n/a	# text io
768	n/a	self.check_flush_error_on_close(support.TESTFN, 'w')
769	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
770	n/a	self.check_flush_error_on_close(fd, 'w')
771	n/a	fd = os.open(support.TESTFN, os.O_WRONLY\|os.O_CREAT)
772	n/a	self.check_flush_error_on_close(fd, 'w', closefd=False)
773	n/a	os.close(fd)
774	n/a
775	n/a	def test_multi_close(self):
776	n/a	f = self.open(support.TESTFN, "wb", buffering=0)
777	n/a	f.close()
778	n/a	f.close()
779	n/a	f.close()
780	n/a	self.assertRaises(ValueError, f.flush)
781	n/a
782	n/a	def test_RawIOBase_read(self):
783	n/a	# Exercise the default RawIOBase.read() implementation (which calls
784	n/a	# readinto() internally).
785	n/a	rawio = self.MockRawIOWithoutRead((b"abc", b"d", None, b"efg", None))
786	n/a	self.assertEqual(rawio.read(2), b"ab")
787	n/a	self.assertEqual(rawio.read(2), b"c")
788	n/a	self.assertEqual(rawio.read(2), b"d")
789	n/a	self.assertEqual(rawio.read(2), None)
790	n/a	self.assertEqual(rawio.read(2), b"ef")
791	n/a	self.assertEqual(rawio.read(2), b"g")
792	n/a	self.assertEqual(rawio.read(2), None)
793	n/a	self.assertEqual(rawio.read(2), b"")
794	n/a
795	n/a	def test_types_have_dict(self):
796	n/a	test = (
797	n/a	self.IOBase(),
798	n/a	self.RawIOBase(),
799	n/a	self.TextIOBase(),
800	n/a	self.StringIO(),
801	n/a	self.BytesIO()
802	n/a	)
803	n/a	for obj in test:
804	n/a	self.assertTrue(hasattr(obj, "__dict__"))
805	n/a
806	n/a	def test_opener(self):
807	n/a	with self.open(support.TESTFN, "w") as f:
808	n/a	f.write("egg\n")
809	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
810	n/a	def opener(path, flags):
811	n/a	return fd
812	n/a	with self.open("non-existent", "r", opener=opener) as f:
813	n/a	self.assertEqual(f.read(), "egg\n")
814	n/a
815	n/a	def test_bad_opener_negative_1(self):
816	n/a	# Issue #27066.
817	n/a	def badopener(fname, flags):
818	n/a	return -1
819	n/a	with self.assertRaises(ValueError) as cm:
820	n/a	open('non-existent', 'r', opener=badopener)
821	n/a	self.assertEqual(str(cm.exception), 'opener returned -1')
822	n/a
823	n/a	def test_bad_opener_other_negative(self):
824	n/a	# Issue #27066.
825	n/a	def badopener(fname, flags):
826	n/a	return -2
827	n/a	with self.assertRaises(ValueError) as cm:
828	n/a	open('non-existent', 'r', opener=badopener)
829	n/a	self.assertEqual(str(cm.exception), 'opener returned -2')
830	n/a
831	n/a	def test_fileio_closefd(self):
832	n/a	# Issue #4841
833	n/a	with self.open(__file__, 'rb') as f1, \
834	n/a	self.open(__file__, 'rb') as f2:
835	n/a	fileio = self.FileIO(f1.fileno(), closefd=False)
836	n/a	# .__init__() must not close f1
837	n/a	fileio.__init__(f2.fileno(), closefd=False)
838	n/a	f1.readline()
839	n/a	# .close() must not close f2
840	n/a	fileio.close()
841	n/a	f2.readline()
842	n/a
843	n/a	def test_nonbuffered_textio(self):
844	n/a	with support.check_no_resource_warning(self):
845	n/a	with self.assertRaises(ValueError):
846	n/a	self.open(support.TESTFN, 'w', buffering=0)
847	n/a
848	n/a	def test_invalid_newline(self):
849	n/a	with support.check_no_resource_warning(self):
850	n/a	with self.assertRaises(ValueError):
851	n/a	self.open(support.TESTFN, 'w', newline='invalid')
852	n/a
853	n/a	def test_buffered_readinto_mixin(self):
854	n/a	# Test the implementation provided by BufferedIOBase
855	n/a	class Stream(self.BufferedIOBase):
856	n/a	def read(self, size):
857	n/a	return b"12345"
858	n/a	read1 = read
859	n/a	stream = Stream()
860	n/a	for method in ("readinto", "readinto1"):
861	n/a	with self.subTest(method):
862	n/a	buffer = byteslike(5)
863	n/a	self.assertEqual(getattr(stream, method)(buffer), 5)
864	n/a	self.assertEqual(bytes(buffer), b"12345")
865	n/a
866	n/a	def test_fspath_support(self):
867	n/a	class PathLike:
868	n/a	def __init__(self, path):
869	n/a	self.path = path
870	n/a
871	n/a	def __fspath__(self):
872	n/a	return self.path
873	n/a
874	n/a	def check_path_succeeds(path):
875	n/a	with self.open(path, "w") as f:
876	n/a	f.write("egg\n")
877	n/a
878	n/a	with self.open(path, "r") as f:
879	n/a	self.assertEqual(f.read(), "egg\n")
880	n/a
881	n/a	check_path_succeeds(PathLike(support.TESTFN))
882	n/a	check_path_succeeds(PathLike(support.TESTFN.encode('utf-8')))
883	n/a
884	n/a	bad_path = PathLike(TypeError)
885	n/a	with self.assertRaises(TypeError):
886	n/a	self.open(bad_path, 'w')
887	n/a
888	n/a	# ensure that refcounting is correct with some error conditions
889	n/a	with self.assertRaisesRegex(ValueError, 'read/write/append mode'):
890	n/a	self.open(PathLike(support.TESTFN), 'rwxa')
891	n/a
892	n/a
893	n/a	class CIOTest(IOTest):
894	n/a
895	n/a	def test_IOBase_finalize(self):
896	n/a	# Issue #12149: segmentation fault on _PyIOBase_finalize when both a
897	n/a	# class which inherits IOBase and an object of this class are caught
898	n/a	# in a reference cycle and close() is already in the method cache.
899	n/a	class MyIO(self.IOBase):
900	n/a	def close(self):
901	n/a	pass
902	n/a
903	n/a	# create an instance to populate the method cache
904	n/a	MyIO()
905	n/a	obj = MyIO()
906	n/a	obj.obj = obj
907	n/a	wr = weakref.ref(obj)
908	n/a	del MyIO
909	n/a	del obj
910	n/a	support.gc_collect()
911	n/a	self.assertIsNone(wr(), wr)
912	n/a
913	n/a	class PyIOTest(IOTest):
914	n/a	pass
915	n/a
916	n/a
917	n/a	@support.cpython_only
918	n/a	class APIMismatchTest(unittest.TestCase):
919	n/a
920	n/a	def test_RawIOBase_io_in_pyio_match(self):
921	n/a	"""Test that pyio RawIOBase class has all c RawIOBase methods"""
922	n/a	mismatch = support.detect_api_mismatch(pyio.RawIOBase, io.RawIOBase,
923	n/a	ignore=('__weakref__',))
924	n/a	self.assertEqual(mismatch, set(), msg='Python RawIOBase does not have all C RawIOBase methods')
925	n/a
926	n/a	def test_RawIOBase_pyio_in_io_match(self):
927	n/a	"""Test that c RawIOBase class has all pyio RawIOBase methods"""
928	n/a	mismatch = support.detect_api_mismatch(io.RawIOBase, pyio.RawIOBase)
929	n/a	self.assertEqual(mismatch, set(), msg='C RawIOBase does not have all Python RawIOBase methods')
930	n/a
931	n/a
932	n/a	class CommonBufferedTests:
933	n/a	# Tests common to BufferedReader, BufferedWriter and BufferedRandom
934	n/a
935	n/a	def test_detach(self):
936	n/a	raw = self.MockRawIO()
937	n/a	buf = self.tp(raw)
938	n/a	self.assertIs(buf.detach(), raw)
939	n/a	self.assertRaises(ValueError, buf.detach)
940	n/a
941	n/a	repr(buf) # Should still work
942	n/a
943	n/a	def test_fileno(self):
944	n/a	rawio = self.MockRawIO()
945	n/a	bufio = self.tp(rawio)
946	n/a
947	n/a	self.assertEqual(42, bufio.fileno())
948	n/a
949	n/a	def test_invalid_args(self):
950	n/a	rawio = self.MockRawIO()
951	n/a	bufio = self.tp(rawio)
952	n/a	# Invalid whence
953	n/a	self.assertRaises(ValueError, bufio.seek, 0, -1)
954	n/a	self.assertRaises(ValueError, bufio.seek, 0, 9)
955	n/a
956	n/a	def test_override_destructor(self):
957	n/a	tp = self.tp
958	n/a	record = []
959	n/a	class MyBufferedIO(tp):
960	n/a	def __del__(self):
961	n/a	record.append(1)
962	n/a	try:
963	n/a	f = super().__del__
964	n/a	except AttributeError:
965	n/a	pass
966	n/a	else:
967	n/a	f()
968	n/a	def close(self):
969	n/a	record.append(2)
970	n/a	super().close()
971	n/a	def flush(self):
972	n/a	record.append(3)
973	n/a	super().flush()
974	n/a	rawio = self.MockRawIO()
975	n/a	bufio = MyBufferedIO(rawio)
976	n/a	del bufio
977	n/a	support.gc_collect()
978	n/a	self.assertEqual(record, [1, 2, 3])
979	n/a
980	n/a	def test_context_manager(self):
981	n/a	# Test usability as a context manager
982	n/a	rawio = self.MockRawIO()
983	n/a	bufio = self.tp(rawio)
984	n/a	def _with():
985	n/a	with bufio:
986	n/a	pass
987	n/a	_with()
988	n/a	# bufio should now be closed, and using it a second time should raise
989	n/a	# a ValueError.
990	n/a	self.assertRaises(ValueError, _with)
991	n/a
992	n/a	def test_error_through_destructor(self):
993	n/a	# Test that the exception state is not modified by a destructor,
994	n/a	# even if close() fails.
995	n/a	rawio = self.CloseFailureIO()
996	n/a	def f():
997	n/a	self.tp(rawio).xyzzy
998	n/a	with support.captured_output("stderr") as s:
999	n/a	self.assertRaises(AttributeError, f)
1000	n/a	s = s.getvalue().strip()
1001	n/a	if s:
1002	n/a	# The destructor may have printed an unraisable error, check it
1003	n/a	self.assertEqual(len(s.splitlines()), 1)
1004	n/a	self.assertTrue(s.startswith("Exception OSError: "), s)
1005	n/a	self.assertTrue(s.endswith(" ignored"), s)
1006	n/a
1007	n/a	def test_repr(self):
1008	n/a	raw = self.MockRawIO()
1009	n/a	b = self.tp(raw)
1010	n/a	clsname = "%s.%s" % (self.tp.__module__, self.tp.__qualname__)
1011	n/a	self.assertEqual(repr(b), "<%s>" % clsname)
1012	n/a	raw.name = "dummy"
1013	n/a	self.assertEqual(repr(b), "<%s name='dummy'>" % clsname)
1014	n/a	raw.name = b"dummy"
1015	n/a	self.assertEqual(repr(b), "<%s name=b'dummy'>" % clsname)
1016	n/a
1017	n/a	def test_flush_error_on_close(self):
1018	n/a	# Test that buffered file is closed despite failed flush
1019	n/a	# and that flush() is called before file closed.
1020	n/a	raw = self.MockRawIO()
1021	n/a	closed = []
1022	n/a	def bad_flush():
1023	n/a	closed[:] = [b.closed, raw.closed]
1024	n/a	raise OSError()
1025	n/a	raw.flush = bad_flush
1026	n/a	b = self.tp(raw)
1027	n/a	self.assertRaises(OSError, b.close) # exception not swallowed
1028	n/a	self.assertTrue(b.closed)
1029	n/a	self.assertTrue(raw.closed)
1030	n/a	self.assertTrue(closed) # flush() called
1031	n/a	self.assertFalse(closed[0]) # flush() called before file closed
1032	n/a	self.assertFalse(closed[1])
1033	n/a	raw.flush = lambda: None # break reference loop
1034	n/a
1035	n/a	def test_close_error_on_close(self):
1036	n/a	raw = self.MockRawIO()
1037	n/a	def bad_flush():
1038	n/a	raise OSError('flush')
1039	n/a	def bad_close():
1040	n/a	raise OSError('close')
1041	n/a	raw.close = bad_close
1042	n/a	b = self.tp(raw)
1043	n/a	b.flush = bad_flush
1044	n/a	with self.assertRaises(OSError) as err: # exception not swallowed
1045	n/a	b.close()
1046	n/a	self.assertEqual(err.exception.args, ('close',))
1047	n/a	self.assertIsInstance(err.exception.__context__, OSError)
1048	n/a	self.assertEqual(err.exception.__context__.args, ('flush',))
1049	n/a	self.assertFalse(b.closed)
1050	n/a
1051	n/a	def test_nonnormalized_close_error_on_close(self):
1052	n/a	# Issue #21677
1053	n/a	raw = self.MockRawIO()
1054	n/a	def bad_flush():
1055	n/a	raise non_existing_flush
1056	n/a	def bad_close():
1057	n/a	raise non_existing_close
1058	n/a	raw.close = bad_close
1059	n/a	b = self.tp(raw)
1060	n/a	b.flush = bad_flush
1061	n/a	with self.assertRaises(NameError) as err: # exception not swallowed
1062	n/a	b.close()
1063	n/a	self.assertIn('non_existing_close', str(err.exception))
1064	n/a	self.assertIsInstance(err.exception.__context__, NameError)
1065	n/a	self.assertIn('non_existing_flush', str(err.exception.__context__))
1066	n/a	self.assertFalse(b.closed)
1067	n/a
1068	n/a	def test_multi_close(self):
1069	n/a	raw = self.MockRawIO()
1070	n/a	b = self.tp(raw)
1071	n/a	b.close()
1072	n/a	b.close()
1073	n/a	b.close()
1074	n/a	self.assertRaises(ValueError, b.flush)
1075	n/a
1076	n/a	def test_unseekable(self):
1077	n/a	bufio = self.tp(self.MockUnseekableIO(b"A" * 10))
1078	n/a	self.assertRaises(self.UnsupportedOperation, bufio.tell)
1079	n/a	self.assertRaises(self.UnsupportedOperation, bufio.seek, 0)
1080	n/a
1081	n/a	def test_readonly_attributes(self):
1082	n/a	raw = self.MockRawIO()
1083	n/a	buf = self.tp(raw)
1084	n/a	x = self.MockRawIO()
1085	n/a	with self.assertRaises(AttributeError):
1086	n/a	buf.raw = x
1087	n/a
1088	n/a
1089	n/a	class SizeofTest:
1090	n/a
1091	n/a	@support.cpython_only
1092	n/a	def test_sizeof(self):
1093	n/a	bufsize1 = 4096
1094	n/a	bufsize2 = 8192
1095	n/a	rawio = self.MockRawIO()
1096	n/a	bufio = self.tp(rawio, buffer_size=bufsize1)
1097	n/a	size = sys.getsizeof(bufio) - bufsize1
1098	n/a	rawio = self.MockRawIO()
1099	n/a	bufio = self.tp(rawio, buffer_size=bufsize2)
1100	n/a	self.assertEqual(sys.getsizeof(bufio), size + bufsize2)
1101	n/a
1102	n/a	@support.cpython_only
1103	n/a	def test_buffer_freeing(self) :
1104	n/a	bufsize = 4096
1105	n/a	rawio = self.MockRawIO()
1106	n/a	bufio = self.tp(rawio, buffer_size=bufsize)
1107	n/a	size = sys.getsizeof(bufio) - bufsize
1108	n/a	bufio.close()
1109	n/a	self.assertEqual(sys.getsizeof(bufio), size)
1110	n/a
1111	n/a	class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
1112	n/a	read_mode = "rb"
1113	n/a
1114	n/a	def test_constructor(self):
1115	n/a	rawio = self.MockRawIO([b"abc"])
1116	n/a	bufio = self.tp(rawio)
1117	n/a	bufio.__init__(rawio)
1118	n/a	bufio.__init__(rawio, buffer_size=1024)
1119	n/a	bufio.__init__(rawio, buffer_size=16)
1120	n/a	self.assertEqual(b"abc", bufio.read())
1121	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
1122	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
1123	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
1124	n/a	rawio = self.MockRawIO([b"abc"])
1125	n/a	bufio.__init__(rawio)
1126	n/a	self.assertEqual(b"abc", bufio.read())
1127	n/a
1128	n/a	def test_uninitialized(self):
1129	n/a	bufio = self.tp.__new__(self.tp)
1130	n/a	del bufio
1131	n/a	bufio = self.tp.__new__(self.tp)
1132	n/a	self.assertRaisesRegex((ValueError, AttributeError),
1133	n/a	'uninitialized\|has no attribute',
1134	n/a	bufio.read, 0)
1135	n/a	bufio.__init__(self.MockRawIO())
1136	n/a	self.assertEqual(bufio.read(0), b'')
1137	n/a
1138	n/a	def test_read(self):
1139	n/a	for arg in (None, 7):
1140	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1141	n/a	bufio = self.tp(rawio)
1142	n/a	self.assertEqual(b"abcdefg", bufio.read(arg))
1143	n/a	# Invalid args
1144	n/a	self.assertRaises(ValueError, bufio.read, -2)
1145	n/a
1146	n/a	def test_read1(self):
1147	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1148	n/a	bufio = self.tp(rawio)
1149	n/a	self.assertEqual(b"a", bufio.read(1))
1150	n/a	self.assertEqual(b"b", bufio.read1(1))
1151	n/a	self.assertEqual(rawio._reads, 1)
1152	n/a	self.assertEqual(b"", bufio.read1(0))
1153	n/a	self.assertEqual(b"c", bufio.read1(100))
1154	n/a	self.assertEqual(rawio._reads, 1)
1155	n/a	self.assertEqual(b"d", bufio.read1(100))
1156	n/a	self.assertEqual(rawio._reads, 2)
1157	n/a	self.assertEqual(b"efg", bufio.read1(100))
1158	n/a	self.assertEqual(rawio._reads, 3)
1159	n/a	self.assertEqual(b"", bufio.read1(100))
1160	n/a	self.assertEqual(rawio._reads, 4)
1161	n/a
1162	n/a	def test_read1_arbitrary(self):
1163	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1164	n/a	bufio = self.tp(rawio)
1165	n/a	self.assertEqual(b"a", bufio.read(1))
1166	n/a	self.assertEqual(b"bc", bufio.read1())
1167	n/a	self.assertEqual(b"d", bufio.read1())
1168	n/a	self.assertEqual(b"efg", bufio.read1(-1))
1169	n/a	self.assertEqual(rawio._reads, 3)
1170	n/a	self.assertEqual(b"", bufio.read1())
1171	n/a	self.assertEqual(rawio._reads, 4)
1172	n/a
1173	n/a	def test_readinto(self):
1174	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1175	n/a	bufio = self.tp(rawio)
1176	n/a	b = bytearray(2)
1177	n/a	self.assertEqual(bufio.readinto(b), 2)
1178	n/a	self.assertEqual(b, b"ab")
1179	n/a	self.assertEqual(bufio.readinto(b), 2)
1180	n/a	self.assertEqual(b, b"cd")
1181	n/a	self.assertEqual(bufio.readinto(b), 2)
1182	n/a	self.assertEqual(b, b"ef")
1183	n/a	self.assertEqual(bufio.readinto(b), 1)
1184	n/a	self.assertEqual(b, b"gf")
1185	n/a	self.assertEqual(bufio.readinto(b), 0)
1186	n/a	self.assertEqual(b, b"gf")
1187	n/a	rawio = self.MockRawIO((b"abc", None))
1188	n/a	bufio = self.tp(rawio)
1189	n/a	self.assertEqual(bufio.readinto(b), 2)
1190	n/a	self.assertEqual(b, b"ab")
1191	n/a	self.assertEqual(bufio.readinto(b), 1)
1192	n/a	self.assertEqual(b, b"cb")
1193	n/a
1194	n/a	def test_readinto1(self):
1195	n/a	buffer_size = 10
1196	n/a	rawio = self.MockRawIO((b"abc", b"de", b"fgh", b"jkl"))
1197	n/a	bufio = self.tp(rawio, buffer_size=buffer_size)
1198	n/a	b = bytearray(2)
1199	n/a	self.assertEqual(bufio.peek(3), b'abc')
1200	n/a	self.assertEqual(rawio._reads, 1)
1201	n/a	self.assertEqual(bufio.readinto1(b), 2)
1202	n/a	self.assertEqual(b, b"ab")
1203	n/a	self.assertEqual(rawio._reads, 1)
1204	n/a	self.assertEqual(bufio.readinto1(b), 1)
1205	n/a	self.assertEqual(b[:1], b"c")
1206	n/a	self.assertEqual(rawio._reads, 1)
1207	n/a	self.assertEqual(bufio.readinto1(b), 2)
1208	n/a	self.assertEqual(b, b"de")
1209	n/a	self.assertEqual(rawio._reads, 2)
1210	n/a	b = bytearray(2*buffer_size)
1211	n/a	self.assertEqual(bufio.peek(3), b'fgh')
1212	n/a	self.assertEqual(rawio._reads, 3)
1213	n/a	self.assertEqual(bufio.readinto1(b), 6)
1214	n/a	self.assertEqual(b[:6], b"fghjkl")
1215	n/a	self.assertEqual(rawio._reads, 4)
1216	n/a
1217	n/a	def test_readinto_array(self):
1218	n/a	buffer_size = 60
1219	n/a	data = b"a" * 26
1220	n/a	rawio = self.MockRawIO((data,))
1221	n/a	bufio = self.tp(rawio, buffer_size=buffer_size)
1222	n/a
1223	n/a	# Create an array with element size > 1 byte
1224	n/a	b = array.array('i', b'x' * 32)
1225	n/a	assert len(b) != 16
1226	n/a
1227	n/a	# Read into it. We should get as many bytes as we can fit into b
1228	n/a	# (which is more than the number of elements)
1229	n/a	n = bufio.readinto(b)
1230	n/a	self.assertGreater(n, len(b))
1231	n/a
1232	n/a	# Check that old contents of b are preserved
1233	n/a	bm = memoryview(b).cast('B')
1234	n/a	self.assertLess(n, len(bm))
1235	n/a	self.assertEqual(bm[:n], data[:n])
1236	n/a	self.assertEqual(bm[n:], b'x' * (len(bm[n:])))
1237	n/a
1238	n/a	def test_readinto1_array(self):
1239	n/a	buffer_size = 60
1240	n/a	data = b"a" * 26
1241	n/a	rawio = self.MockRawIO((data,))
1242	n/a	bufio = self.tp(rawio, buffer_size=buffer_size)
1243	n/a
1244	n/a	# Create an array with element size > 1 byte
1245	n/a	b = array.array('i', b'x' * 32)
1246	n/a	assert len(b) != 16
1247	n/a
1248	n/a	# Read into it. We should get as many bytes as we can fit into b
1249	n/a	# (which is more than the number of elements)
1250	n/a	n = bufio.readinto1(b)
1251	n/a	self.assertGreater(n, len(b))
1252	n/a
1253	n/a	# Check that old contents of b are preserved
1254	n/a	bm = memoryview(b).cast('B')
1255	n/a	self.assertLess(n, len(bm))
1256	n/a	self.assertEqual(bm[:n], data[:n])
1257	n/a	self.assertEqual(bm[n:], b'x' * (len(bm[n:])))
1258	n/a
1259	n/a	def test_readlines(self):
1260	n/a	def bufio():
1261	n/a	rawio = self.MockRawIO((b"abc\n", b"d\n", b"ef"))
1262	n/a	return self.tp(rawio)
1263	n/a	self.assertEqual(bufio().readlines(), [b"abc\n", b"d\n", b"ef"])
1264	n/a	self.assertEqual(bufio().readlines(5), [b"abc\n", b"d\n"])
1265	n/a	self.assertEqual(bufio().readlines(None), [b"abc\n", b"d\n", b"ef"])
1266	n/a
1267	n/a	def test_buffering(self):
1268	n/a	data = b"abcdefghi"
1269	n/a	dlen = len(data)
1270	n/a
1271	n/a	tests = [
1272	n/a	[ 100, [ 3, 1, 4, 8 ], [ dlen, 0 ] ],
1273	n/a	[ 100, [ 3, 3, 3], [ dlen ] ],
1274	n/a	[ 4, [ 1, 2, 4, 2 ], [ 4, 4, 1 ] ],
1275	n/a	]
1276	n/a
1277	n/a	for bufsize, buf_read_sizes, raw_read_sizes in tests:
1278	n/a	rawio = self.MockFileIO(data)
1279	n/a	bufio = self.tp(rawio, buffer_size=bufsize)
1280	n/a	pos = 0
1281	n/a	for nbytes in buf_read_sizes:
1282	n/a	self.assertEqual(bufio.read(nbytes), data[pos:pos+nbytes])
1283	n/a	pos += nbytes
1284	n/a	# this is mildly implementation-dependent
1285	n/a	self.assertEqual(rawio.read_history, raw_read_sizes)
1286	n/a
1287	n/a	def test_read_non_blocking(self):
1288	n/a	# Inject some None's in there to simulate EWOULDBLOCK
1289	n/a	rawio = self.MockRawIO((b"abc", b"d", None, b"efg", None, None, None))
1290	n/a	bufio = self.tp(rawio)
1291	n/a	self.assertEqual(b"abcd", bufio.read(6))
1292	n/a	self.assertEqual(b"e", bufio.read(1))
1293	n/a	self.assertEqual(b"fg", bufio.read())
1294	n/a	self.assertEqual(b"", bufio.peek(1))
1295	n/a	self.assertIsNone(bufio.read())
1296	n/a	self.assertEqual(b"", bufio.read())
1297	n/a
1298	n/a	rawio = self.MockRawIO((b"a", None, None))
1299	n/a	self.assertEqual(b"a", rawio.readall())
1300	n/a	self.assertIsNone(rawio.readall())
1301	n/a
1302	n/a	def test_read_past_eof(self):
1303	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1304	n/a	bufio = self.tp(rawio)
1305	n/a
1306	n/a	self.assertEqual(b"abcdefg", bufio.read(9000))
1307	n/a
1308	n/a	def test_read_all(self):
1309	n/a	rawio = self.MockRawIO((b"abc", b"d", b"efg"))
1310	n/a	bufio = self.tp(rawio)
1311	n/a
1312	n/a	self.assertEqual(b"abcdefg", bufio.read())
1313	n/a
1314	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
1315	n/a	@support.requires_resource('cpu')
1316	n/a	def test_threads(self):
1317	n/a	try:
1318	n/a	# Write out many bytes with exactly the same number of 0's,
1319	n/a	# 1's... 255's. This will help us check that concurrent reading
1320	n/a	# doesn't duplicate or forget contents.
1321	n/a	N = 1000
1322	n/a	l = list(range(256)) * N
1323	n/a	random.shuffle(l)
1324	n/a	s = bytes(bytearray(l))
1325	n/a	with self.open(support.TESTFN, "wb") as f:
1326	n/a	f.write(s)
1327	n/a	with self.open(support.TESTFN, self.read_mode, buffering=0) as raw:
1328	n/a	bufio = self.tp(raw, 8)
1329	n/a	errors = []
1330	n/a	results = []
1331	n/a	def f():
1332	n/a	try:
1333	n/a	# Intra-buffer read then buffer-flushing read
1334	n/a	for n in cycle([1, 19]):
1335	n/a	s = bufio.read(n)
1336	n/a	if not s:
1337	n/a	break
1338	n/a	# list.append() is atomic
1339	n/a	results.append(s)
1340	n/a	except Exception as e:
1341	n/a	errors.append(e)
1342	n/a	raise
1343	n/a	threads = [threading.Thread(target=f) for x in range(20)]
1344	n/a	with support.start_threads(threads):
1345	n/a	time.sleep(0.02) # yield
1346	n/a	self.assertFalse(errors,
1347	n/a	"the following exceptions were caught: %r" % errors)
1348	n/a	s = b''.join(results)
1349	n/a	for i in range(256):
1350	n/a	c = bytes(bytearray([i]))
1351	n/a	self.assertEqual(s.count(c), N)
1352	n/a	finally:
1353	n/a	support.unlink(support.TESTFN)
1354	n/a
1355	n/a	def test_unseekable(self):
1356	n/a	bufio = self.tp(self.MockUnseekableIO(b"A" * 10))
1357	n/a	self.assertRaises(self.UnsupportedOperation, bufio.tell)
1358	n/a	self.assertRaises(self.UnsupportedOperation, bufio.seek, 0)
1359	n/a	bufio.read(1)
1360	n/a	self.assertRaises(self.UnsupportedOperation, bufio.seek, 0)
1361	n/a	self.assertRaises(self.UnsupportedOperation, bufio.tell)
1362	n/a
1363	n/a	def test_misbehaved_io(self):
1364	n/a	rawio = self.MisbehavedRawIO((b"abc", b"d", b"efg"))
1365	n/a	bufio = self.tp(rawio)
1366	n/a	self.assertRaises(OSError, bufio.seek, 0)
1367	n/a	self.assertRaises(OSError, bufio.tell)
1368	n/a
1369	n/a	def test_no_extraneous_read(self):
1370	n/a	# Issue #9550; when the raw IO object has satisfied the read request,
1371	n/a	# we should not issue any additional reads, otherwise it may block
1372	n/a	# (e.g. socket).
1373	n/a	bufsize = 16
1374	n/a	for n in (2, bufsize - 1, bufsize, bufsize + 1, bufsize * 2):
1375	n/a	rawio = self.MockRawIO([b"x" * n])
1376	n/a	bufio = self.tp(rawio, bufsize)
1377	n/a	self.assertEqual(bufio.read(n), b"x" * n)
1378	n/a	# Simple case: one raw read is enough to satisfy the request.
1379	n/a	self.assertEqual(rawio._extraneous_reads, 0,
1380	n/a	"failed for {}: {} != 0".format(n, rawio._extraneous_reads))
1381	n/a	# A more complex case where two raw reads are needed to satisfy
1382	n/a	# the request.
1383	n/a	rawio = self.MockRawIO([b"x" * (n - 1), b"x"])
1384	n/a	bufio = self.tp(rawio, bufsize)
1385	n/a	self.assertEqual(bufio.read(n), b"x" * n)
1386	n/a	self.assertEqual(rawio._extraneous_reads, 0,
1387	n/a	"failed for {}: {} != 0".format(n, rawio._extraneous_reads))
1388	n/a
1389	n/a	def test_read_on_closed(self):
1390	n/a	# Issue #23796
1391	n/a	b = io.BufferedReader(io.BytesIO(b"12"))
1392	n/a	b.read(1)
1393	n/a	b.close()
1394	n/a	self.assertRaises(ValueError, b.peek)
1395	n/a	self.assertRaises(ValueError, b.read1, 1)
1396	n/a
1397	n/a
1398	n/a	class CBufferedReaderTest(BufferedReaderTest, SizeofTest):
1399	n/a	tp = io.BufferedReader
1400	n/a
1401	n/a	def test_constructor(self):
1402	n/a	BufferedReaderTest.test_constructor(self)
1403	n/a	# The allocation can succeed on 32-bit builds, e.g. with more
1404	n/a	# than 2GB RAM and a 64-bit kernel.
1405	n/a	if sys.maxsize > 0x7FFFFFFF:
1406	n/a	rawio = self.MockRawIO()
1407	n/a	bufio = self.tp(rawio)
1408	n/a	self.assertRaises((OverflowError, MemoryError, ValueError),
1409	n/a	bufio.__init__, rawio, sys.maxsize)
1410	n/a
1411	n/a	def test_initialization(self):
1412	n/a	rawio = self.MockRawIO([b"abc"])
1413	n/a	bufio = self.tp(rawio)
1414	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
1415	n/a	self.assertRaises(ValueError, bufio.read)
1416	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
1417	n/a	self.assertRaises(ValueError, bufio.read)
1418	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
1419	n/a	self.assertRaises(ValueError, bufio.read)
1420	n/a
1421	n/a	def test_misbehaved_io_read(self):
1422	n/a	rawio = self.MisbehavedRawIO((b"abc", b"d", b"efg"))
1423	n/a	bufio = self.tp(rawio)
1424	n/a	# _pyio.BufferedReader seems to implement reading different, so that
1425	n/a	# checking this is not so easy.
1426	n/a	self.assertRaises(OSError, bufio.read, 10)
1427	n/a
1428	n/a	def test_garbage_collection(self):
1429	n/a	# C BufferedReader objects are collected.
1430	n/a	# The Python version has __del__, so it ends into gc.garbage instead
1431	n/a	with support.check_warnings(('', ResourceWarning)):
1432	n/a	rawio = self.FileIO(support.TESTFN, "w+b")
1433	n/a	f = self.tp(rawio)
1434	n/a	f.f = f
1435	n/a	wr = weakref.ref(f)
1436	n/a	del f
1437	n/a	support.gc_collect()
1438	n/a	self.assertIsNone(wr(), wr)
1439	n/a
1440	n/a	def test_args_error(self):
1441	n/a	# Issue #17275
1442	n/a	with self.assertRaisesRegex(TypeError, "BufferedReader"):
1443	n/a	self.tp(io.BytesIO(), 1024, 1024, 1024)
1444	n/a
1445	n/a
1446	n/a	class PyBufferedReaderTest(BufferedReaderTest):
1447	n/a	tp = pyio.BufferedReader
1448	n/a
1449	n/a
1450	n/a	class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
1451	n/a	write_mode = "wb"
1452	n/a
1453	n/a	def test_constructor(self):
1454	n/a	rawio = self.MockRawIO()
1455	n/a	bufio = self.tp(rawio)
1456	n/a	bufio.__init__(rawio)
1457	n/a	bufio.__init__(rawio, buffer_size=1024)
1458	n/a	bufio.__init__(rawio, buffer_size=16)
1459	n/a	self.assertEqual(3, bufio.write(b"abc"))
1460	n/a	bufio.flush()
1461	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
1462	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
1463	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
1464	n/a	bufio.__init__(rawio)
1465	n/a	self.assertEqual(3, bufio.write(b"ghi"))
1466	n/a	bufio.flush()
1467	n/a	self.assertEqual(b"".join(rawio._write_stack), b"abcghi")
1468	n/a
1469	n/a	def test_uninitialized(self):
1470	n/a	bufio = self.tp.__new__(self.tp)
1471	n/a	del bufio
1472	n/a	bufio = self.tp.__new__(self.tp)
1473	n/a	self.assertRaisesRegex((ValueError, AttributeError),
1474	n/a	'uninitialized\|has no attribute',
1475	n/a	bufio.write, b'')
1476	n/a	bufio.__init__(self.MockRawIO())
1477	n/a	self.assertEqual(bufio.write(b''), 0)
1478	n/a
1479	n/a	def test_detach_flush(self):
1480	n/a	raw = self.MockRawIO()
1481	n/a	buf = self.tp(raw)
1482	n/a	buf.write(b"howdy!")
1483	n/a	self.assertFalse(raw._write_stack)
1484	n/a	buf.detach()
1485	n/a	self.assertEqual(raw._write_stack, [b"howdy!"])
1486	n/a
1487	n/a	def test_write(self):
1488	n/a	# Write to the buffered IO but don't overflow the buffer.
1489	n/a	writer = self.MockRawIO()
1490	n/a	bufio = self.tp(writer, 8)
1491	n/a	bufio.write(b"abc")
1492	n/a	self.assertFalse(writer._write_stack)
1493	n/a	buffer = bytearray(b"def")
1494	n/a	bufio.write(buffer)
1495	n/a	buffer[:] = b"***" # Overwrite our copy of the data
1496	n/a	bufio.flush()
1497	n/a	self.assertEqual(b"".join(writer._write_stack), b"abcdef")
1498	n/a
1499	n/a	def test_write_overflow(self):
1500	n/a	writer = self.MockRawIO()
1501	n/a	bufio = self.tp(writer, 8)
1502	n/a	contents = b"abcdefghijklmnop"
1503	n/a	for n in range(0, len(contents), 3):
1504	n/a	bufio.write(contents[n:n+3])
1505	n/a	flushed = b"".join(writer._write_stack)
1506	n/a	# At least (total - 8) bytes were implicitly flushed, perhaps more
1507	n/a	# depending on the implementation.
1508	n/a	self.assertTrue(flushed.startswith(contents[:-8]), flushed)
1509	n/a
1510	n/a	def check_writes(self, intermediate_func):
1511	n/a	# Lots of writes, test the flushed output is as expected.
1512	n/a	contents = bytes(range(256)) * 1000
1513	n/a	n = 0
1514	n/a	writer = self.MockRawIO()
1515	n/a	bufio = self.tp(writer, 13)
1516	n/a	# Generator of write sizes: repeat each N 15 times then proceed to N+1
1517	n/a	def gen_sizes():
1518	n/a	for size in count(1):
1519	n/a	for i in range(15):
1520	n/a	yield size
1521	n/a	sizes = gen_sizes()
1522	n/a	while n < len(contents):
1523	n/a	size = min(next(sizes), len(contents) - n)
1524	n/a	self.assertEqual(bufio.write(contents[n:n+size]), size)
1525	n/a	intermediate_func(bufio)
1526	n/a	n += size
1527	n/a	bufio.flush()
1528	n/a	self.assertEqual(contents, b"".join(writer._write_stack))
1529	n/a
1530	n/a	def test_writes(self):
1531	n/a	self.check_writes(lambda bufio: None)
1532	n/a
1533	n/a	def test_writes_and_flushes(self):
1534	n/a	self.check_writes(lambda bufio: bufio.flush())
1535	n/a
1536	n/a	def test_writes_and_seeks(self):
1537	n/a	def _seekabs(bufio):
1538	n/a	pos = bufio.tell()
1539	n/a	bufio.seek(pos + 1, 0)
1540	n/a	bufio.seek(pos - 1, 0)
1541	n/a	bufio.seek(pos, 0)
1542	n/a	self.check_writes(_seekabs)
1543	n/a	def _seekrel(bufio):
1544	n/a	pos = bufio.seek(0, 1)
1545	n/a	bufio.seek(+1, 1)
1546	n/a	bufio.seek(-1, 1)
1547	n/a	bufio.seek(pos, 0)
1548	n/a	self.check_writes(_seekrel)
1549	n/a
1550	n/a	def test_writes_and_truncates(self):
1551	n/a	self.check_writes(lambda bufio: bufio.truncate(bufio.tell()))
1552	n/a
1553	n/a	def test_write_non_blocking(self):
1554	n/a	raw = self.MockNonBlockWriterIO()
1555	n/a	bufio = self.tp(raw, 8)
1556	n/a
1557	n/a	self.assertEqual(bufio.write(b"abcd"), 4)
1558	n/a	self.assertEqual(bufio.write(b"efghi"), 5)
1559	n/a	# 1 byte will be written, the rest will be buffered
1560	n/a	raw.block_on(b"k")
1561	n/a	self.assertEqual(bufio.write(b"jklmn"), 5)
1562	n/a
1563	n/a	# 8 bytes will be written, 8 will be buffered and the rest will be lost
1564	n/a	raw.block_on(b"0")
1565	n/a	try:
1566	n/a	bufio.write(b"opqrwxyz0123456789")
1567	n/a	except self.BlockingIOError as e:
1568	n/a	written = e.characters_written
1569	n/a	else:
1570	n/a	self.fail("BlockingIOError should have been raised")
1571	n/a	self.assertEqual(written, 16)
1572	n/a	self.assertEqual(raw.pop_written(),
1573	n/a	b"abcdefghijklmnopqrwxyz")
1574	n/a
1575	n/a	self.assertEqual(bufio.write(b"ABCDEFGHI"), 9)
1576	n/a	s = raw.pop_written()
1577	n/a	# Previously buffered bytes were flushed
1578	n/a	self.assertTrue(s.startswith(b"01234567A"), s)
1579	n/a
1580	n/a	def test_write_and_rewind(self):
1581	n/a	raw = io.BytesIO()
1582	n/a	bufio = self.tp(raw, 4)
1583	n/a	self.assertEqual(bufio.write(b"abcdef"), 6)
1584	n/a	self.assertEqual(bufio.tell(), 6)
1585	n/a	bufio.seek(0, 0)
1586	n/a	self.assertEqual(bufio.write(b"XY"), 2)
1587	n/a	bufio.seek(6, 0)
1588	n/a	self.assertEqual(raw.getvalue(), b"XYcdef")
1589	n/a	self.assertEqual(bufio.write(b"123456"), 6)
1590	n/a	bufio.flush()
1591	n/a	self.assertEqual(raw.getvalue(), b"XYcdef123456")
1592	n/a
1593	n/a	def test_flush(self):
1594	n/a	writer = self.MockRawIO()
1595	n/a	bufio = self.tp(writer, 8)
1596	n/a	bufio.write(b"abc")
1597	n/a	bufio.flush()
1598	n/a	self.assertEqual(b"abc", writer._write_stack[0])
1599	n/a
1600	n/a	def test_writelines(self):
1601	n/a	l = [b'ab', b'cd', b'ef']
1602	n/a	writer = self.MockRawIO()
1603	n/a	bufio = self.tp(writer, 8)
1604	n/a	bufio.writelines(l)
1605	n/a	bufio.flush()
1606	n/a	self.assertEqual(b''.join(writer._write_stack), b'abcdef')
1607	n/a
1608	n/a	def test_writelines_userlist(self):
1609	n/a	l = UserList([b'ab', b'cd', b'ef'])
1610	n/a	writer = self.MockRawIO()
1611	n/a	bufio = self.tp(writer, 8)
1612	n/a	bufio.writelines(l)
1613	n/a	bufio.flush()
1614	n/a	self.assertEqual(b''.join(writer._write_stack), b'abcdef')
1615	n/a
1616	n/a	def test_writelines_error(self):
1617	n/a	writer = self.MockRawIO()
1618	n/a	bufio = self.tp(writer, 8)
1619	n/a	self.assertRaises(TypeError, bufio.writelines, [1, 2, 3])
1620	n/a	self.assertRaises(TypeError, bufio.writelines, None)
1621	n/a	self.assertRaises(TypeError, bufio.writelines, 'abc')
1622	n/a
1623	n/a	def test_destructor(self):
1624	n/a	writer = self.MockRawIO()
1625	n/a	bufio = self.tp(writer, 8)
1626	n/a	bufio.write(b"abc")
1627	n/a	del bufio
1628	n/a	support.gc_collect()
1629	n/a	self.assertEqual(b"abc", writer._write_stack[0])
1630	n/a
1631	n/a	def test_truncate(self):
1632	n/a	# Truncate implicitly flushes the buffer.
1633	n/a	with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
1634	n/a	bufio = self.tp(raw, 8)
1635	n/a	bufio.write(b"abcdef")
1636	n/a	self.assertEqual(bufio.truncate(3), 3)
1637	n/a	self.assertEqual(bufio.tell(), 6)
1638	n/a	with self.open(support.TESTFN, "rb", buffering=0) as f:
1639	n/a	self.assertEqual(f.read(), b"abc")
1640	n/a
1641	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
1642	n/a	@support.requires_resource('cpu')
1643	n/a	def test_threads(self):
1644	n/a	try:
1645	n/a	# Write out many bytes from many threads and test they were
1646	n/a	# all flushed.
1647	n/a	N = 1000
1648	n/a	contents = bytes(range(256)) * N
1649	n/a	sizes = cycle([1, 19])
1650	n/a	n = 0
1651	n/a	queue = deque()
1652	n/a	while n < len(contents):
1653	n/a	size = next(sizes)
1654	n/a	queue.append(contents[n:n+size])
1655	n/a	n += size
1656	n/a	del contents
1657	n/a	# We use a real file object because it allows us to
1658	n/a	# exercise situations where the GIL is released before
1659	n/a	# writing the buffer to the raw streams. This is in addition
1660	n/a	# to concurrency issues due to switching threads in the middle
1661	n/a	# of Python code.
1662	n/a	with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
1663	n/a	bufio = self.tp(raw, 8)
1664	n/a	errors = []
1665	n/a	def f():
1666	n/a	try:
1667	n/a	while True:
1668	n/a	try:
1669	n/a	s = queue.popleft()
1670	n/a	except IndexError:
1671	n/a	return
1672	n/a	bufio.write(s)
1673	n/a	except Exception as e:
1674	n/a	errors.append(e)
1675	n/a	raise
1676	n/a	threads = [threading.Thread(target=f) for x in range(20)]
1677	n/a	with support.start_threads(threads):
1678	n/a	time.sleep(0.02) # yield
1679	n/a	self.assertFalse(errors,
1680	n/a	"the following exceptions were caught: %r" % errors)
1681	n/a	bufio.close()
1682	n/a	with self.open(support.TESTFN, "rb") as f:
1683	n/a	s = f.read()
1684	n/a	for i in range(256):
1685	n/a	self.assertEqual(s.count(bytes([i])), N)
1686	n/a	finally:
1687	n/a	support.unlink(support.TESTFN)
1688	n/a
1689	n/a	def test_misbehaved_io(self):
1690	n/a	rawio = self.MisbehavedRawIO()
1691	n/a	bufio = self.tp(rawio, 5)
1692	n/a	self.assertRaises(OSError, bufio.seek, 0)
1693	n/a	self.assertRaises(OSError, bufio.tell)
1694	n/a	self.assertRaises(OSError, bufio.write, b"abcdef")
1695	n/a
1696	n/a	def test_max_buffer_size_removal(self):
1697	n/a	with self.assertRaises(TypeError):
1698	n/a	self.tp(self.MockRawIO(), 8, 12)
1699	n/a
1700	n/a	def test_write_error_on_close(self):
1701	n/a	raw = self.MockRawIO()
1702	n/a	def bad_write(b):
1703	n/a	raise OSError()
1704	n/a	raw.write = bad_write
1705	n/a	b = self.tp(raw)
1706	n/a	b.write(b'spam')
1707	n/a	self.assertRaises(OSError, b.close) # exception not swallowed
1708	n/a	self.assertTrue(b.closed)
1709	n/a
1710	n/a
1711	n/a	class CBufferedWriterTest(BufferedWriterTest, SizeofTest):
1712	n/a	tp = io.BufferedWriter
1713	n/a
1714	n/a	def test_constructor(self):
1715	n/a	BufferedWriterTest.test_constructor(self)
1716	n/a	# The allocation can succeed on 32-bit builds, e.g. with more
1717	n/a	# than 2GB RAM and a 64-bit kernel.
1718	n/a	if sys.maxsize > 0x7FFFFFFF:
1719	n/a	rawio = self.MockRawIO()
1720	n/a	bufio = self.tp(rawio)
1721	n/a	self.assertRaises((OverflowError, MemoryError, ValueError),
1722	n/a	bufio.__init__, rawio, sys.maxsize)
1723	n/a
1724	n/a	def test_initialization(self):
1725	n/a	rawio = self.MockRawIO()
1726	n/a	bufio = self.tp(rawio)
1727	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
1728	n/a	self.assertRaises(ValueError, bufio.write, b"def")
1729	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
1730	n/a	self.assertRaises(ValueError, bufio.write, b"def")
1731	n/a	self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
1732	n/a	self.assertRaises(ValueError, bufio.write, b"def")
1733	n/a
1734	n/a	def test_garbage_collection(self):
1735	n/a	# C BufferedWriter objects are collected, and collecting them flushes
1736	n/a	# all data to disk.
1737	n/a	# The Python version has __del__, so it ends into gc.garbage instead
1738	n/a	with support.check_warnings(('', ResourceWarning)):
1739	n/a	rawio = self.FileIO(support.TESTFN, "w+b")
1740	n/a	f = self.tp(rawio)
1741	n/a	f.write(b"123xxx")
1742	n/a	f.x = f
1743	n/a	wr = weakref.ref(f)
1744	n/a	del f
1745	n/a	support.gc_collect()
1746	n/a	self.assertIsNone(wr(), wr)
1747	n/a	with self.open(support.TESTFN, "rb") as f:
1748	n/a	self.assertEqual(f.read(), b"123xxx")
1749	n/a
1750	n/a	def test_args_error(self):
1751	n/a	# Issue #17275
1752	n/a	with self.assertRaisesRegex(TypeError, "BufferedWriter"):
1753	n/a	self.tp(io.BytesIO(), 1024, 1024, 1024)
1754	n/a
1755	n/a
1756	n/a	class PyBufferedWriterTest(BufferedWriterTest):
1757	n/a	tp = pyio.BufferedWriter
1758	n/a
1759	n/a	class BufferedRWPairTest(unittest.TestCase):
1760	n/a
1761	n/a	def test_constructor(self):
1762	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1763	n/a	self.assertFalse(pair.closed)
1764	n/a
1765	n/a	def test_uninitialized(self):
1766	n/a	pair = self.tp.__new__(self.tp)
1767	n/a	del pair
1768	n/a	pair = self.tp.__new__(self.tp)
1769	n/a	self.assertRaisesRegex((ValueError, AttributeError),
1770	n/a	'uninitialized\|has no attribute',
1771	n/a	pair.read, 0)
1772	n/a	self.assertRaisesRegex((ValueError, AttributeError),
1773	n/a	'uninitialized\|has no attribute',
1774	n/a	pair.write, b'')
1775	n/a	pair.__init__(self.MockRawIO(), self.MockRawIO())
1776	n/a	self.assertEqual(pair.read(0), b'')
1777	n/a	self.assertEqual(pair.write(b''), 0)
1778	n/a
1779	n/a	def test_detach(self):
1780	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1781	n/a	self.assertRaises(self.UnsupportedOperation, pair.detach)
1782	n/a
1783	n/a	def test_constructor_max_buffer_size_removal(self):
1784	n/a	with self.assertRaises(TypeError):
1785	n/a	self.tp(self.MockRawIO(), self.MockRawIO(), 8, 12)
1786	n/a
1787	n/a	def test_constructor_with_not_readable(self):
1788	n/a	class NotReadable(MockRawIO):
1789	n/a	def readable(self):
1790	n/a	return False
1791	n/a
1792	n/a	self.assertRaises(OSError, self.tp, NotReadable(), self.MockRawIO())
1793	n/a
1794	n/a	def test_constructor_with_not_writeable(self):
1795	n/a	class NotWriteable(MockRawIO):
1796	n/a	def writable(self):
1797	n/a	return False
1798	n/a
1799	n/a	self.assertRaises(OSError, self.tp, self.MockRawIO(), NotWriteable())
1800	n/a
1801	n/a	def test_read(self):
1802	n/a	pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
1803	n/a
1804	n/a	self.assertEqual(pair.read(3), b"abc")
1805	n/a	self.assertEqual(pair.read(1), b"d")
1806	n/a	self.assertEqual(pair.read(), b"ef")
1807	n/a	pair = self.tp(self.BytesIO(b"abc"), self.MockRawIO())
1808	n/a	self.assertEqual(pair.read(None), b"abc")
1809	n/a
1810	n/a	def test_readlines(self):
1811	n/a	pair = lambda: self.tp(self.BytesIO(b"abc\ndef\nh"), self.MockRawIO())
1812	n/a	self.assertEqual(pair().readlines(), [b"abc\n", b"def\n", b"h"])
1813	n/a	self.assertEqual(pair().readlines(), [b"abc\n", b"def\n", b"h"])
1814	n/a	self.assertEqual(pair().readlines(5), [b"abc\n", b"def\n"])
1815	n/a
1816	n/a	def test_read1(self):
1817	n/a	# .read1() is delegated to the underlying reader object, so this test
1818	n/a	# can be shallow.
1819	n/a	pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
1820	n/a
1821	n/a	self.assertEqual(pair.read1(3), b"abc")
1822	n/a	self.assertEqual(pair.read1(), b"def")
1823	n/a
1824	n/a	def test_readinto(self):
1825	n/a	for method in ("readinto", "readinto1"):
1826	n/a	with self.subTest(method):
1827	n/a	pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
1828	n/a
1829	n/a	data = byteslike(b'\0' * 5)
1830	n/a	self.assertEqual(getattr(pair, method)(data), 5)
1831	n/a	self.assertEqual(bytes(data), b"abcde")
1832	n/a
1833	n/a	def test_write(self):
1834	n/a	w = self.MockRawIO()
1835	n/a	pair = self.tp(self.MockRawIO(), w)
1836	n/a
1837	n/a	pair.write(b"abc")
1838	n/a	pair.flush()
1839	n/a	buffer = bytearray(b"def")
1840	n/a	pair.write(buffer)
1841	n/a	buffer[:] = b"***" # Overwrite our copy of the data
1842	n/a	pair.flush()
1843	n/a	self.assertEqual(w._write_stack, [b"abc", b"def"])
1844	n/a
1845	n/a	def test_peek(self):
1846	n/a	pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
1847	n/a
1848	n/a	self.assertTrue(pair.peek(3).startswith(b"abc"))
1849	n/a	self.assertEqual(pair.read(3), b"abc")
1850	n/a
1851	n/a	def test_readable(self):
1852	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1853	n/a	self.assertTrue(pair.readable())
1854	n/a
1855	n/a	def test_writeable(self):
1856	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1857	n/a	self.assertTrue(pair.writable())
1858	n/a
1859	n/a	def test_seekable(self):
1860	n/a	# BufferedRWPairs are never seekable, even if their readers and writers
1861	n/a	# are.
1862	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1863	n/a	self.assertFalse(pair.seekable())
1864	n/a
1865	n/a	# .flush() is delegated to the underlying writer object and has been
1866	n/a	# tested in the test_write method.
1867	n/a
1868	n/a	def test_close_and_closed(self):
1869	n/a	pair = self.tp(self.MockRawIO(), self.MockRawIO())
1870	n/a	self.assertFalse(pair.closed)
1871	n/a	pair.close()
1872	n/a	self.assertTrue(pair.closed)
1873	n/a
1874	n/a	def test_reader_close_error_on_close(self):
1875	n/a	def reader_close():
1876	n/a	reader_non_existing
1877	n/a	reader = self.MockRawIO()
1878	n/a	reader.close = reader_close
1879	n/a	writer = self.MockRawIO()
1880	n/a	pair = self.tp(reader, writer)
1881	n/a	with self.assertRaises(NameError) as err:
1882	n/a	pair.close()
1883	n/a	self.assertIn('reader_non_existing', str(err.exception))
1884	n/a	self.assertTrue(pair.closed)
1885	n/a	self.assertFalse(reader.closed)
1886	n/a	self.assertTrue(writer.closed)
1887	n/a
1888	n/a	def test_writer_close_error_on_close(self):
1889	n/a	def writer_close():
1890	n/a	writer_non_existing
1891	n/a	reader = self.MockRawIO()
1892	n/a	writer = self.MockRawIO()
1893	n/a	writer.close = writer_close
1894	n/a	pair = self.tp(reader, writer)
1895	n/a	with self.assertRaises(NameError) as err:
1896	n/a	pair.close()
1897	n/a	self.assertIn('writer_non_existing', str(err.exception))
1898	n/a	self.assertFalse(pair.closed)
1899	n/a	self.assertTrue(reader.closed)
1900	n/a	self.assertFalse(writer.closed)
1901	n/a
1902	n/a	def test_reader_writer_close_error_on_close(self):
1903	n/a	def reader_close():
1904	n/a	reader_non_existing
1905	n/a	def writer_close():
1906	n/a	writer_non_existing
1907	n/a	reader = self.MockRawIO()
1908	n/a	reader.close = reader_close
1909	n/a	writer = self.MockRawIO()
1910	n/a	writer.close = writer_close
1911	n/a	pair = self.tp(reader, writer)
1912	n/a	with self.assertRaises(NameError) as err:
1913	n/a	pair.close()
1914	n/a	self.assertIn('reader_non_existing', str(err.exception))
1915	n/a	self.assertIsInstance(err.exception.__context__, NameError)
1916	n/a	self.assertIn('writer_non_existing', str(err.exception.__context__))
1917	n/a	self.assertFalse(pair.closed)
1918	n/a	self.assertFalse(reader.closed)
1919	n/a	self.assertFalse(writer.closed)
1920	n/a
1921	n/a	def test_isatty(self):
1922	n/a	class SelectableIsAtty(MockRawIO):
1923	n/a	def __init__(self, isatty):
1924	n/a	MockRawIO.__init__(self)
1925	n/a	self._isatty = isatty
1926	n/a
1927	n/a	def isatty(self):
1928	n/a	return self._isatty
1929	n/a
1930	n/a	pair = self.tp(SelectableIsAtty(False), SelectableIsAtty(False))
1931	n/a	self.assertFalse(pair.isatty())
1932	n/a
1933	n/a	pair = self.tp(SelectableIsAtty(True), SelectableIsAtty(False))
1934	n/a	self.assertTrue(pair.isatty())
1935	n/a
1936	n/a	pair = self.tp(SelectableIsAtty(False), SelectableIsAtty(True))
1937	n/a	self.assertTrue(pair.isatty())
1938	n/a
1939	n/a	pair = self.tp(SelectableIsAtty(True), SelectableIsAtty(True))
1940	n/a	self.assertTrue(pair.isatty())
1941	n/a
1942	n/a	def test_weakref_clearing(self):
1943	n/a	brw = self.tp(self.MockRawIO(), self.MockRawIO())
1944	n/a	ref = weakref.ref(brw)
1945	n/a	brw = None
1946	n/a	ref = None # Shouldn't segfault.
1947	n/a
1948	n/a	class CBufferedRWPairTest(BufferedRWPairTest):
1949	n/a	tp = io.BufferedRWPair
1950	n/a
1951	n/a	class PyBufferedRWPairTest(BufferedRWPairTest):
1952	n/a	tp = pyio.BufferedRWPair
1953	n/a
1954	n/a
1955	n/a	class BufferedRandomTest(BufferedReaderTest, BufferedWriterTest):
1956	n/a	read_mode = "rb+"
1957	n/a	write_mode = "wb+"
1958	n/a
1959	n/a	def test_constructor(self):
1960	n/a	BufferedReaderTest.test_constructor(self)
1961	n/a	BufferedWriterTest.test_constructor(self)
1962	n/a
1963	n/a	def test_uninitialized(self):
1964	n/a	BufferedReaderTest.test_uninitialized(self)
1965	n/a	BufferedWriterTest.test_uninitialized(self)
1966	n/a
1967	n/a	def test_read_and_write(self):
1968	n/a	raw = self.MockRawIO((b"asdf", b"ghjk"))
1969	n/a	rw = self.tp(raw, 8)
1970	n/a
1971	n/a	self.assertEqual(b"as", rw.read(2))
1972	n/a	rw.write(b"ddd")
1973	n/a	rw.write(b"eee")
1974	n/a	self.assertFalse(raw._write_stack) # Buffer writes
1975	n/a	self.assertEqual(b"ghjk", rw.read())
1976	n/a	self.assertEqual(b"dddeee", raw._write_stack[0])
1977	n/a
1978	n/a	def test_seek_and_tell(self):
1979	n/a	raw = self.BytesIO(b"asdfghjkl")
1980	n/a	rw = self.tp(raw)
1981	n/a
1982	n/a	self.assertEqual(b"as", rw.read(2))
1983	n/a	self.assertEqual(2, rw.tell())
1984	n/a	rw.seek(0, 0)
1985	n/a	self.assertEqual(b"asdf", rw.read(4))
1986	n/a
1987	n/a	rw.write(b"123f")
1988	n/a	rw.seek(0, 0)
1989	n/a	self.assertEqual(b"asdf123fl", rw.read())
1990	n/a	self.assertEqual(9, rw.tell())
1991	n/a	rw.seek(-4, 2)
1992	n/a	self.assertEqual(5, rw.tell())
1993	n/a	rw.seek(2, 1)
1994	n/a	self.assertEqual(7, rw.tell())
1995	n/a	self.assertEqual(b"fl", rw.read(11))
1996	n/a	rw.flush()
1997	n/a	self.assertEqual(b"asdf123fl", raw.getvalue())
1998	n/a
1999	n/a	self.assertRaises(TypeError, rw.seek, 0.0)
2000	n/a
2001	n/a	def check_flush_and_read(self, read_func):
2002	n/a	raw = self.BytesIO(b"abcdefghi")
2003	n/a	bufio = self.tp(raw)
2004	n/a
2005	n/a	self.assertEqual(b"ab", read_func(bufio, 2))
2006	n/a	bufio.write(b"12")
2007	n/a	self.assertEqual(b"ef", read_func(bufio, 2))
2008	n/a	self.assertEqual(6, bufio.tell())
2009	n/a	bufio.flush()
2010	n/a	self.assertEqual(6, bufio.tell())
2011	n/a	self.assertEqual(b"ghi", read_func(bufio))
2012	n/a	raw.seek(0, 0)
2013	n/a	raw.write(b"XYZ")
2014	n/a	# flush() resets the read buffer
2015	n/a	bufio.flush()
2016	n/a	bufio.seek(0, 0)
2017	n/a	self.assertEqual(b"XYZ", read_func(bufio, 3))
2018	n/a
2019	n/a	def test_flush_and_read(self):
2020	n/a	self.check_flush_and_read(lambda bufio, args: bufio.read(args))
2021	n/a
2022	n/a	def test_flush_and_readinto(self):
2023	n/a	def _readinto(bufio, n=-1):
2024	n/a	b = bytearray(n if n >= 0 else 9999)
2025	n/a	n = bufio.readinto(b)
2026	n/a	return bytes(b[:n])
2027	n/a	self.check_flush_and_read(_readinto)
2028	n/a
2029	n/a	def test_flush_and_peek(self):
2030	n/a	def _peek(bufio, n=-1):
2031	n/a	# This relies on the fact that the buffer can contain the whole
2032	n/a	# raw stream, otherwise peek() can return less.
2033	n/a	b = bufio.peek(n)
2034	n/a	if n != -1:
2035	n/a	b = b[:n]
2036	n/a	bufio.seek(len(b), 1)
2037	n/a	return b
2038	n/a	self.check_flush_and_read(_peek)
2039	n/a
2040	n/a	def test_flush_and_write(self):
2041	n/a	raw = self.BytesIO(b"abcdefghi")
2042	n/a	bufio = self.tp(raw)
2043	n/a
2044	n/a	bufio.write(b"123")
2045	n/a	bufio.flush()
2046	n/a	bufio.write(b"45")
2047	n/a	bufio.flush()
2048	n/a	bufio.seek(0, 0)
2049	n/a	self.assertEqual(b"12345fghi", raw.getvalue())
2050	n/a	self.assertEqual(b"12345fghi", bufio.read())
2051	n/a
2052	n/a	def test_threads(self):
2053	n/a	BufferedReaderTest.test_threads(self)
2054	n/a	BufferedWriterTest.test_threads(self)
2055	n/a
2056	n/a	def test_writes_and_peek(self):
2057	n/a	def _peek(bufio):
2058	n/a	bufio.peek(1)
2059	n/a	self.check_writes(_peek)
2060	n/a	def _peek(bufio):
2061	n/a	pos = bufio.tell()
2062	n/a	bufio.seek(-1, 1)
2063	n/a	bufio.peek(1)
2064	n/a	bufio.seek(pos, 0)
2065	n/a	self.check_writes(_peek)
2066	n/a
2067	n/a	def test_writes_and_reads(self):
2068	n/a	def _read(bufio):
2069	n/a	bufio.seek(-1, 1)
2070	n/a	bufio.read(1)
2071	n/a	self.check_writes(_read)
2072	n/a
2073	n/a	def test_writes_and_read1s(self):
2074	n/a	def _read1(bufio):
2075	n/a	bufio.seek(-1, 1)
2076	n/a	bufio.read1(1)
2077	n/a	self.check_writes(_read1)
2078	n/a
2079	n/a	def test_writes_and_readintos(self):
2080	n/a	def _read(bufio):
2081	n/a	bufio.seek(-1, 1)
2082	n/a	bufio.readinto(bytearray(1))
2083	n/a	self.check_writes(_read)
2084	n/a
2085	n/a	def test_write_after_readahead(self):
2086	n/a	# Issue #6629: writing after the buffer was filled by readahead should
2087	n/a	# first rewind the raw stream.
2088	n/a	for overwrite_size in [1, 5]:
2089	n/a	raw = self.BytesIO(b"A" * 10)
2090	n/a	bufio = self.tp(raw, 4)
2091	n/a	# Trigger readahead
2092	n/a	self.assertEqual(bufio.read(1), b"A")
2093	n/a	self.assertEqual(bufio.tell(), 1)
2094	n/a	# Overwriting should rewind the raw stream if it needs so
2095	n/a	bufio.write(b"B" * overwrite_size)
2096	n/a	self.assertEqual(bufio.tell(), overwrite_size + 1)
2097	n/a	# If the write size was smaller than the buffer size, flush() and
2098	n/a	# check that rewind happens.
2099	n/a	bufio.flush()
2100	n/a	self.assertEqual(bufio.tell(), overwrite_size + 1)
2101	n/a	s = raw.getvalue()
2102	n/a	self.assertEqual(s,
2103	n/a	b"A" + b"B" * overwrite_size + b"A" * (9 - overwrite_size))
2104	n/a
2105	n/a	def test_write_rewind_write(self):
2106	n/a	# Various combinations of reading / writing / seeking backwards / writing again
2107	n/a	def mutate(bufio, pos1, pos2):
2108	n/a	assert pos2 >= pos1
2109	n/a	# Fill the buffer
2110	n/a	bufio.seek(pos1)
2111	n/a	bufio.read(pos2 - pos1)
2112	n/a	bufio.write(b'\x02')
2113	n/a	# This writes earlier than the previous write, but still inside
2114	n/a	# the buffer.
2115	n/a	bufio.seek(pos1)
2116	n/a	bufio.write(b'\x01')
2117	n/a
2118	n/a	b = b"\x80\x81\x82\x83\x84"
2119	n/a	for i in range(0, len(b)):
2120	n/a	for j in range(i, len(b)):
2121	n/a	raw = self.BytesIO(b)
2122	n/a	bufio = self.tp(raw, 100)
2123	n/a	mutate(bufio, i, j)
2124	n/a	bufio.flush()
2125	n/a	expected = bytearray(b)
2126	n/a	expected[j] = 2
2127	n/a	expected[i] = 1
2128	n/a	self.assertEqual(raw.getvalue(), expected,
2129	n/a	"failed result for i=%d, j=%d" % (i, j))
2130	n/a
2131	n/a	def test_truncate_after_read_or_write(self):
2132	n/a	raw = self.BytesIO(b"A" * 10)
2133	n/a	bufio = self.tp(raw, 100)
2134	n/a	self.assertEqual(bufio.read(2), b"AA") # the read buffer gets filled
2135	n/a	self.assertEqual(bufio.truncate(), 2)
2136	n/a	self.assertEqual(bufio.write(b"BB"), 2) # the write buffer increases
2137	n/a	self.assertEqual(bufio.truncate(), 4)
2138	n/a
2139	n/a	def test_misbehaved_io(self):
2140	n/a	BufferedReaderTest.test_misbehaved_io(self)
2141	n/a	BufferedWriterTest.test_misbehaved_io(self)
2142	n/a
2143	n/a	def test_interleaved_read_write(self):
2144	n/a	# Test for issue #12213
2145	n/a	with self.BytesIO(b'abcdefgh') as raw:
2146	n/a	with self.tp(raw, 100) as f:
2147	n/a	f.write(b"1")
2148	n/a	self.assertEqual(f.read(1), b'b')
2149	n/a	f.write(b'2')
2150	n/a	self.assertEqual(f.read1(1), b'd')
2151	n/a	f.write(b'3')
2152	n/a	buf = bytearray(1)
2153	n/a	f.readinto(buf)
2154	n/a	self.assertEqual(buf, b'f')
2155	n/a	f.write(b'4')
2156	n/a	self.assertEqual(f.peek(1), b'h')
2157	n/a	f.flush()
2158	n/a	self.assertEqual(raw.getvalue(), b'1b2d3f4h')
2159	n/a
2160	n/a	with self.BytesIO(b'abc') as raw:
2161	n/a	with self.tp(raw, 100) as f:
2162	n/a	self.assertEqual(f.read(1), b'a')
2163	n/a	f.write(b"2")
2164	n/a	self.assertEqual(f.read(1), b'c')
2165	n/a	f.flush()
2166	n/a	self.assertEqual(raw.getvalue(), b'a2c')
2167	n/a
2168	n/a	def test_interleaved_readline_write(self):
2169	n/a	with self.BytesIO(b'ab\ncdef\ng\n') as raw:
2170	n/a	with self.tp(raw) as f:
2171	n/a	f.write(b'1')
2172	n/a	self.assertEqual(f.readline(), b'b\n')
2173	n/a	f.write(b'2')
2174	n/a	self.assertEqual(f.readline(), b'def\n')
2175	n/a	f.write(b'3')
2176	n/a	self.assertEqual(f.readline(), b'\n')
2177	n/a	f.flush()
2178	n/a	self.assertEqual(raw.getvalue(), b'1b\n2def\n3\n')
2179	n/a
2180	n/a	# You can't construct a BufferedRandom over a non-seekable stream.
2181	n/a	test_unseekable = None
2182	n/a
2183	n/a
2184	n/a	class CBufferedRandomTest(BufferedRandomTest, SizeofTest):
2185	n/a	tp = io.BufferedRandom
2186	n/a
2187	n/a	def test_constructor(self):
2188	n/a	BufferedRandomTest.test_constructor(self)
2189	n/a	# The allocation can succeed on 32-bit builds, e.g. with more
2190	n/a	# than 2GB RAM and a 64-bit kernel.
2191	n/a	if sys.maxsize > 0x7FFFFFFF:
2192	n/a	rawio = self.MockRawIO()
2193	n/a	bufio = self.tp(rawio)
2194	n/a	self.assertRaises((OverflowError, MemoryError, ValueError),
2195	n/a	bufio.__init__, rawio, sys.maxsize)
2196	n/a
2197	n/a	def test_garbage_collection(self):
2198	n/a	CBufferedReaderTest.test_garbage_collection(self)
2199	n/a	CBufferedWriterTest.test_garbage_collection(self)
2200	n/a
2201	n/a	def test_args_error(self):
2202	n/a	# Issue #17275
2203	n/a	with self.assertRaisesRegex(TypeError, "BufferedRandom"):
2204	n/a	self.tp(io.BytesIO(), 1024, 1024, 1024)
2205	n/a
2206	n/a
2207	n/a	class PyBufferedRandomTest(BufferedRandomTest):
2208	n/a	tp = pyio.BufferedRandom
2209	n/a
2210	n/a
2211	n/a	# To fully exercise seek/tell, the StatefulIncrementalDecoder has these
2212	n/a	# properties:
2213	n/a	# - A single output character can correspond to many bytes of input.
2214	n/a	# - The number of input bytes to complete the character can be
2215	n/a	# undetermined until the last input byte is received.
2216	n/a	# - The number of input bytes can vary depending on previous input.
2217	n/a	# - A single input byte can correspond to many characters of output.
2218	n/a	# - The number of output characters can be undetermined until the
2219	n/a	# last input byte is received.
2220	n/a	# - The number of output characters can vary depending on previous input.
2221	n/a
2222	n/a	class StatefulIncrementalDecoder(codecs.IncrementalDecoder):
2223	n/a	"""
2224	n/a	For testing seek/tell behavior with a stateful, buffering decoder.
2225	n/a
2226	n/a	Input is a sequence of words. Words may be fixed-length (length set
2227	n/a	by input) or variable-length (period-terminated). In variable-length
2228	n/a	mode, extra periods are ignored. Possible words are:
2229	n/a	- 'i' followed by a number sets the input length, I (maximum 99).
2230	n/a	When I is set to 0, words are space-terminated.
2231	n/a	- 'o' followed by a number sets the output length, O (maximum 99).
2232	n/a	- Any other word is converted into a word followed by a period on
2233	n/a	the output. The output word consists of the input word truncated
2234	n/a	or padded out with hyphens to make its length equal to O. If O
2235	n/a	is 0, the word is output verbatim without truncating or padding.
2236	n/a	I and O are initially set to 1. When I changes, any buffered input is
2237	n/a	re-scanned according to the new I. EOF also terminates the last word.
2238	n/a	"""
2239	n/a
2240	n/a	def __init__(self, errors='strict'):
2241	n/a	codecs.IncrementalDecoder.__init__(self, errors)
2242	n/a	self.reset()
2243	n/a
2244	n/a	def __repr__(self):
2245	n/a	return '<SID %x>' % id(self)
2246	n/a
2247	n/a	def reset(self):
2248	n/a	self.i = 1
2249	n/a	self.o = 1
2250	n/a	self.buffer = bytearray()
2251	n/a
2252	n/a	def getstate(self):
2253	n/a	i, o = self.i ^ 1, self.o ^ 1 # so that flags = 0 after reset()
2254	n/a	return bytes(self.buffer), i*100 + o
2255	n/a
2256	n/a	def setstate(self, state):
2257	n/a	buffer, io = state
2258	n/a	self.buffer = bytearray(buffer)
2259	n/a	i, o = divmod(io, 100)
2260	n/a	self.i, self.o = i ^ 1, o ^ 1
2261	n/a
2262	n/a	def decode(self, input, final=False):
2263	n/a	output = ''
2264	n/a	for b in input:
2265	n/a	if self.i == 0: # variable-length, terminated with period
2266	n/a	if b == ord('.'):
2267	n/a	if self.buffer:
2268	n/a	output += self.process_word()
2269	n/a	else:
2270	n/a	self.buffer.append(b)
2271	n/a	else: # fixed-length, terminate after self.i bytes
2272	n/a	self.buffer.append(b)
2273	n/a	if len(self.buffer) == self.i:
2274	n/a	output += self.process_word()
2275	n/a	if final and self.buffer: # EOF terminates the last word
2276	n/a	output += self.process_word()
2277	n/a	return output
2278	n/a
2279	n/a	def process_word(self):
2280	n/a	output = ''
2281	n/a	if self.buffer[0] == ord('i'):
2282	n/a	self.i = min(99, int(self.buffer[1:] or 0)) # set input length
2283	n/a	elif self.buffer[0] == ord('o'):
2284	n/a	self.o = min(99, int(self.buffer[1:] or 0)) # set output length
2285	n/a	else:
2286	n/a	output = self.buffer.decode('ascii')
2287	n/a	if len(output) < self.o:
2288	n/a	output += '-'*self.o # pad out with hyphens
2289	n/a	if self.o:
2290	n/a	output = output[:self.o] # truncate to output length
2291	n/a	output += '.'
2292	n/a	self.buffer = bytearray()
2293	n/a	return output
2294	n/a
2295	n/a	codecEnabled = False
2296	n/a
2297	n/a	@classmethod
2298	n/a	def lookupTestDecoder(cls, name):
2299	n/a	if cls.codecEnabled and name == 'test_decoder':
2300	n/a	latin1 = codecs.lookup('latin-1')
2301	n/a	return codecs.CodecInfo(
2302	n/a	name='test_decoder', encode=latin1.encode, decode=None,
2303	n/a	incrementalencoder=None,
2304	n/a	streamreader=None, streamwriter=None,
2305	n/a	incrementaldecoder=cls)
2306	n/a
2307	n/a	# Register the previous decoder for testing.
2308	n/a	# Disabled by default, tests will enable it.
2309	n/a	codecs.register(StatefulIncrementalDecoder.lookupTestDecoder)
2310	n/a
2311	n/a
2312	n/a	class StatefulIncrementalDecoderTest(unittest.TestCase):
2313	n/a	"""
2314	n/a	Make sure the StatefulIncrementalDecoder actually works.
2315	n/a	"""
2316	n/a
2317	n/a	test_cases = [
2318	n/a	# I=1, O=1 (fixed-length input == fixed-length output)
2319	n/a	(b'abcd', False, 'a.b.c.d.'),
2320	n/a	# I=0, O=0 (variable-length input, variable-length output)
2321	n/a	(b'oiabcd', True, 'abcd.'),
2322	n/a	# I=0, O=0 (should ignore extra periods)
2323	n/a	(b'oi...abcd...', True, 'abcd.'),
2324	n/a	# I=0, O=6 (variable-length input, fixed-length output)
2325	n/a	(b'i.o6.x.xyz.toolongtofit.', False, 'x-----.xyz---.toolon.'),
2326	n/a	# I=2, O=6 (fixed-length input < fixed-length output)
2327	n/a	(b'i.i2.o6xyz', True, 'xy----.z-----.'),
2328	n/a	# I=6, O=3 (fixed-length input > fixed-length output)
2329	n/a	(b'i.o3.i6.abcdefghijklmnop', True, 'abc.ghi.mno.'),
2330	n/a	# I=0, then 3; O=29, then 15 (with longer output)
2331	n/a	(b'i.o29.a.b.cde.o15.abcdefghijabcdefghij.i3.a.b.c.d.ei00k.l.m', True,
2332	n/a	'a----------------------------.' +
2333	n/a	'b----------------------------.' +
2334	n/a	'cde--------------------------.' +
2335	n/a	'abcdefghijabcde.' +
2336	n/a	'a.b------------.' +
2337	n/a	'.c.------------.' +
2338	n/a	'd.e------------.' +
2339	n/a	'k--------------.' +
2340	n/a	'l--------------.' +
2341	n/a	'm--------------.')
2342	n/a	]
2343	n/a
2344	n/a	def test_decoder(self):
2345	n/a	# Try a few one-shot test cases.
2346	n/a	for input, eof, output in self.test_cases:
2347	n/a	d = StatefulIncrementalDecoder()
2348	n/a	self.assertEqual(d.decode(input, eof), output)
2349	n/a
2350	n/a	# Also test an unfinished decode, followed by forcing EOF.
2351	n/a	d = StatefulIncrementalDecoder()
2352	n/a	self.assertEqual(d.decode(b'oiabcd'), '')
2353	n/a	self.assertEqual(d.decode(b'', 1), 'abcd.')
2354	n/a
2355	n/a	class TextIOWrapperTest(unittest.TestCase):
2356	n/a
2357	n/a	def setUp(self):
2358	n/a	self.testdata = b"AAA\r\nBBB\rCCC\r\nDDD\nEEE\r\n"
2359	n/a	self.normalized = b"AAA\nBBB\nCCC\nDDD\nEEE\n".decode("ascii")
2360	n/a	support.unlink(support.TESTFN)
2361	n/a
2362	n/a	def tearDown(self):
2363	n/a	support.unlink(support.TESTFN)
2364	n/a
2365	n/a	def test_constructor(self):
2366	n/a	r = self.BytesIO(b"\xc3\xa9\n\n")
2367	n/a	b = self.BufferedReader(r, 1000)
2368	n/a	t = self.TextIOWrapper(b)
2369	n/a	t.__init__(b, encoding="latin-1", newline="\r\n")
2370	n/a	self.assertEqual(t.encoding, "latin-1")
2371	n/a	self.assertEqual(t.line_buffering, False)
2372	n/a	t.__init__(b, encoding="utf-8", line_buffering=True)
2373	n/a	self.assertEqual(t.encoding, "utf-8")
2374	n/a	self.assertEqual(t.line_buffering, True)
2375	n/a	self.assertEqual("\xe9\n", t.readline())
2376	n/a	self.assertRaises(TypeError, t.__init__, b, newline=42)
2377	n/a	self.assertRaises(ValueError, t.__init__, b, newline='xyzzy')
2378	n/a
2379	n/a	def test_uninitialized(self):
2380	n/a	t = self.TextIOWrapper.__new__(self.TextIOWrapper)
2381	n/a	del t
2382	n/a	t = self.TextIOWrapper.__new__(self.TextIOWrapper)
2383	n/a	self.assertRaises(Exception, repr, t)
2384	n/a	self.assertRaisesRegex((ValueError, AttributeError),
2385	n/a	'uninitialized\|has no attribute',
2386	n/a	t.read, 0)
2387	n/a	t.__init__(self.MockRawIO())
2388	n/a	self.assertEqual(t.read(0), '')
2389	n/a
2390	n/a	def test_non_text_encoding_codecs_are_rejected(self):
2391	n/a	# Ensure the constructor complains if passed a codec that isn't
2392	n/a	# marked as a text encoding
2393	n/a	# http://bugs.python.org/issue20404
2394	n/a	r = self.BytesIO()
2395	n/a	b = self.BufferedWriter(r)
2396	n/a	with self.assertRaisesRegex(LookupError, "is not a text encoding"):
2397	n/a	self.TextIOWrapper(b, encoding="hex")
2398	n/a
2399	n/a	def test_detach(self):
2400	n/a	r = self.BytesIO()
2401	n/a	b = self.BufferedWriter(r)
2402	n/a	t = self.TextIOWrapper(b)
2403	n/a	self.assertIs(t.detach(), b)
2404	n/a
2405	n/a	t = self.TextIOWrapper(b, encoding="ascii")
2406	n/a	t.write("howdy")
2407	n/a	self.assertFalse(r.getvalue())
2408	n/a	t.detach()
2409	n/a	self.assertEqual(r.getvalue(), b"howdy")
2410	n/a	self.assertRaises(ValueError, t.detach)
2411	n/a
2412	n/a	# Operations independent of the detached stream should still work
2413	n/a	repr(t)
2414	n/a	self.assertEqual(t.encoding, "ascii")
2415	n/a	self.assertEqual(t.errors, "strict")
2416	n/a	self.assertFalse(t.line_buffering)
2417	n/a
2418	n/a	def test_repr(self):
2419	n/a	raw = self.BytesIO("hello".encode("utf-8"))
2420	n/a	b = self.BufferedReader(raw)
2421	n/a	t = self.TextIOWrapper(b, encoding="utf-8")
2422	n/a	modname = self.TextIOWrapper.__module__
2423	n/a	self.assertEqual(repr(t),
2424	n/a	"<%s.TextIOWrapper encoding='utf-8'>" % modname)
2425	n/a	raw.name = "dummy"
2426	n/a	self.assertEqual(repr(t),
2427	n/a	"<%s.TextIOWrapper name='dummy' encoding='utf-8'>" % modname)
2428	n/a	t.mode = "r"
2429	n/a	self.assertEqual(repr(t),
2430	n/a	"<%s.TextIOWrapper name='dummy' mode='r' encoding='utf-8'>" % modname)
2431	n/a	raw.name = b"dummy"
2432	n/a	self.assertEqual(repr(t),
2433	n/a	"<%s.TextIOWrapper name=b'dummy' mode='r' encoding='utf-8'>" % modname)
2434	n/a
2435	n/a	t.buffer.detach()
2436	n/a	repr(t) # Should not raise an exception
2437	n/a
2438	n/a	def test_line_buffering(self):
2439	n/a	r = self.BytesIO()
2440	n/a	b = self.BufferedWriter(r, 1000)
2441	n/a	t = self.TextIOWrapper(b, newline="\n", line_buffering=True)
2442	n/a	t.write("X")
2443	n/a	self.assertEqual(r.getvalue(), b"") # No flush happened
2444	n/a	t.write("Y\nZ")
2445	n/a	self.assertEqual(r.getvalue(), b"XY\nZ") # All got flushed
2446	n/a	t.write("A\rB")
2447	n/a	self.assertEqual(r.getvalue(), b"XY\nZA\rB")
2448	n/a
2449	n/a	def test_default_encoding(self):
2450	n/a	old_environ = dict(os.environ)
2451	n/a	try:
2452	n/a	# try to get a user preferred encoding different than the current
2453	n/a	# locale encoding to check that TextIOWrapper() uses the current
2454	n/a	# locale encoding and not the user preferred encoding
2455	n/a	for key in ('LC_ALL', 'LANG', 'LC_CTYPE'):
2456	n/a	if key in os.environ:
2457	n/a	del os.environ[key]
2458	n/a
2459	n/a	current_locale_encoding = locale.getpreferredencoding(False)
2460	n/a	b = self.BytesIO()
2461	n/a	t = self.TextIOWrapper(b)
2462	n/a	self.assertEqual(t.encoding, current_locale_encoding)
2463	n/a	finally:
2464	n/a	os.environ.clear()
2465	n/a	os.environ.update(old_environ)
2466	n/a
2467	n/a	@support.cpython_only
2468	n/a	def test_device_encoding(self):
2469	n/a	# Issue 15989
2470	n/a	import _testcapi
2471	n/a	b = self.BytesIO()
2472	n/a	b.fileno = lambda: _testcapi.INT_MAX + 1
2473	n/a	self.assertRaises(OverflowError, self.TextIOWrapper, b)
2474	n/a	b.fileno = lambda: _testcapi.UINT_MAX + 1
2475	n/a	self.assertRaises(OverflowError, self.TextIOWrapper, b)
2476	n/a
2477	n/a	def test_encoding(self):
2478	n/a	# Check the encoding attribute is always set, and valid
2479	n/a	b = self.BytesIO()
2480	n/a	t = self.TextIOWrapper(b, encoding="utf-8")
2481	n/a	self.assertEqual(t.encoding, "utf-8")
2482	n/a	t = self.TextIOWrapper(b)
2483	n/a	self.assertIsNotNone(t.encoding)
2484	n/a	codecs.lookup(t.encoding)
2485	n/a
2486	n/a	def test_encoding_errors_reading(self):
2487	n/a	# (1) default
2488	n/a	b = self.BytesIO(b"abc\n\xff\n")
2489	n/a	t = self.TextIOWrapper(b, encoding="ascii")
2490	n/a	self.assertRaises(UnicodeError, t.read)
2491	n/a	# (2) explicit strict
2492	n/a	b = self.BytesIO(b"abc\n\xff\n")
2493	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="strict")
2494	n/a	self.assertRaises(UnicodeError, t.read)
2495	n/a	# (3) ignore
2496	n/a	b = self.BytesIO(b"abc\n\xff\n")
2497	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="ignore")
2498	n/a	self.assertEqual(t.read(), "abc\n\n")
2499	n/a	# (4) replace
2500	n/a	b = self.BytesIO(b"abc\n\xff\n")
2501	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="replace")
2502	n/a	self.assertEqual(t.read(), "abc\n\ufffd\n")
2503	n/a
2504	n/a	def test_encoding_errors_writing(self):
2505	n/a	# (1) default
2506	n/a	b = self.BytesIO()
2507	n/a	t = self.TextIOWrapper(b, encoding="ascii")
2508	n/a	self.assertRaises(UnicodeError, t.write, "\xff")
2509	n/a	# (2) explicit strict
2510	n/a	b = self.BytesIO()
2511	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="strict")
2512	n/a	self.assertRaises(UnicodeError, t.write, "\xff")
2513	n/a	# (3) ignore
2514	n/a	b = self.BytesIO()
2515	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="ignore",
2516	n/a	newline="\n")
2517	n/a	t.write("abc\xffdef\n")
2518	n/a	t.flush()
2519	n/a	self.assertEqual(b.getvalue(), b"abcdef\n")
2520	n/a	# (4) replace
2521	n/a	b = self.BytesIO()
2522	n/a	t = self.TextIOWrapper(b, encoding="ascii", errors="replace",
2523	n/a	newline="\n")
2524	n/a	t.write("abc\xffdef\n")
2525	n/a	t.flush()
2526	n/a	self.assertEqual(b.getvalue(), b"abc?def\n")
2527	n/a
2528	n/a	def test_newlines(self):
2529	n/a	input_lines = [ "unix\n", "windows\r\n", "os9\r", "last\n", "nonl" ]
2530	n/a
2531	n/a	tests = [
2532	n/a	[ None, [ 'unix\n', 'windows\n', 'os9\n', 'last\n', 'nonl' ] ],
2533	n/a	[ '', input_lines ],
2534	n/a	[ '\n', [ "unix\n", "windows\r\n", "os9\rlast\n", "nonl" ] ],
2535	n/a	[ '\r\n', [ "unix\nwindows\r\n", "os9\rlast\nnonl" ] ],
2536	n/a	[ '\r', [ "unix\nwindows\r", "\nos9\r", "last\nnonl" ] ],
2537	n/a	]
2538	n/a	encodings = (
2539	n/a	'utf-8', 'latin-1',
2540	n/a	'utf-16', 'utf-16-le', 'utf-16-be',
2541	n/a	'utf-32', 'utf-32-le', 'utf-32-be',
2542	n/a	)
2543	n/a
2544	n/a	# Try a range of buffer sizes to test the case where \r is the last
2545	n/a	# character in TextIOWrapper._pending_line.
2546	n/a	for encoding in encodings:
2547	n/a	# XXX: str.encode() should return bytes
2548	n/a	data = bytes(''.join(input_lines).encode(encoding))
2549	n/a	for do_reads in (False, True):
2550	n/a	for bufsize in range(1, 10):
2551	n/a	for newline, exp_lines in tests:
2552	n/a	bufio = self.BufferedReader(self.BytesIO(data), bufsize)
2553	n/a	textio = self.TextIOWrapper(bufio, newline=newline,
2554	n/a	encoding=encoding)
2555	n/a	if do_reads:
2556	n/a	got_lines = []
2557	n/a	while True:
2558	n/a	c2 = textio.read(2)
2559	n/a	if c2 == '':
2560	n/a	break
2561	n/a	self.assertEqual(len(c2), 2)
2562	n/a	got_lines.append(c2 + textio.readline())
2563	n/a	else:
2564	n/a	got_lines = list(textio)
2565	n/a
2566	n/a	for got_line, exp_line in zip(got_lines, exp_lines):
2567	n/a	self.assertEqual(got_line, exp_line)
2568	n/a	self.assertEqual(len(got_lines), len(exp_lines))
2569	n/a
2570	n/a	def test_newlines_input(self):
2571	n/a	testdata = b"AAA\nBB\x00B\nCCC\rDDD\rEEE\r\nFFF\r\nGGG"
2572	n/a	normalized = testdata.replace(b"\r\n", b"\n").replace(b"\r", b"\n")
2573	n/a	for newline, expected in [
2574	n/a	(None, normalized.decode("ascii").splitlines(keepends=True)),
2575	n/a	("", testdata.decode("ascii").splitlines(keepends=True)),
2576	n/a	("\n", ["AAA\n", "BB\x00B\n", "CCC\rDDD\rEEE\r\n", "FFF\r\n", "GGG"]),
2577	n/a	("\r\n", ["AAA\nBB\x00B\nCCC\rDDD\rEEE\r\n", "FFF\r\n", "GGG"]),
2578	n/a	("\r", ["AAA\nBB\x00B\nCCC\r", "DDD\r", "EEE\r", "\nFFF\r", "\nGGG"]),
2579	n/a	]:
2580	n/a	buf = self.BytesIO(testdata)
2581	n/a	txt = self.TextIOWrapper(buf, encoding="ascii", newline=newline)
2582	n/a	self.assertEqual(txt.readlines(), expected)
2583	n/a	txt.seek(0)
2584	n/a	self.assertEqual(txt.read(), "".join(expected))
2585	n/a
2586	n/a	def test_newlines_output(self):
2587	n/a	testdict = {
2588	n/a	"": b"AAA\nBBB\nCCC\nX\rY\r\nZ",
2589	n/a	"\n": b"AAA\nBBB\nCCC\nX\rY\r\nZ",
2590	n/a	"\r": b"AAA\rBBB\rCCC\rX\rY\r\rZ",
2591	n/a	"\r\n": b"AAA\r\nBBB\r\nCCC\r\nX\rY\r\r\nZ",
2592	n/a	}
2593	n/a	tests = [(None, testdict[os.linesep])] + sorted(testdict.items())
2594	n/a	for newline, expected in tests:
2595	n/a	buf = self.BytesIO()
2596	n/a	txt = self.TextIOWrapper(buf, encoding="ascii", newline=newline)
2597	n/a	txt.write("AAA\nB")
2598	n/a	txt.write("BB\nCCC\n")
2599	n/a	txt.write("X\rY\r\nZ")
2600	n/a	txt.flush()
2601	n/a	self.assertEqual(buf.closed, False)
2602	n/a	self.assertEqual(buf.getvalue(), expected)
2603	n/a
2604	n/a	def test_destructor(self):
2605	n/a	l = []
2606	n/a	base = self.BytesIO
2607	n/a	class MyBytesIO(base):
2608	n/a	def close(self):
2609	n/a	l.append(self.getvalue())
2610	n/a	base.close(self)
2611	n/a	b = MyBytesIO()
2612	n/a	t = self.TextIOWrapper(b, encoding="ascii")
2613	n/a	t.write("abc")
2614	n/a	del t
2615	n/a	support.gc_collect()
2616	n/a	self.assertEqual([b"abc"], l)
2617	n/a
2618	n/a	def test_override_destructor(self):
2619	n/a	record = []
2620	n/a	class MyTextIO(self.TextIOWrapper):
2621	n/a	def __del__(self):
2622	n/a	record.append(1)
2623	n/a	try:
2624	n/a	f = super().__del__
2625	n/a	except AttributeError:
2626	n/a	pass
2627	n/a	else:
2628	n/a	f()
2629	n/a	def close(self):
2630	n/a	record.append(2)
2631	n/a	super().close()
2632	n/a	def flush(self):
2633	n/a	record.append(3)
2634	n/a	super().flush()
2635	n/a	b = self.BytesIO()
2636	n/a	t = MyTextIO(b, encoding="ascii")
2637	n/a	del t
2638	n/a	support.gc_collect()
2639	n/a	self.assertEqual(record, [1, 2, 3])
2640	n/a
2641	n/a	def test_error_through_destructor(self):
2642	n/a	# Test that the exception state is not modified by a destructor,
2643	n/a	# even if close() fails.
2644	n/a	rawio = self.CloseFailureIO()
2645	n/a	def f():
2646	n/a	self.TextIOWrapper(rawio).xyzzy
2647	n/a	with support.captured_output("stderr") as s:
2648	n/a	self.assertRaises(AttributeError, f)
2649	n/a	s = s.getvalue().strip()
2650	n/a	if s:
2651	n/a	# The destructor may have printed an unraisable error, check it
2652	n/a	self.assertEqual(len(s.splitlines()), 1)
2653	n/a	self.assertTrue(s.startswith("Exception OSError: "), s)
2654	n/a	self.assertTrue(s.endswith(" ignored"), s)
2655	n/a
2656	n/a	# Systematic tests of the text I/O API
2657	n/a
2658	n/a	def test_basic_io(self):
2659	n/a	for chunksize in (1, 2, 3, 4, 5, 15, 16, 17, 31, 32, 33, 63, 64, 65):
2660	n/a	for enc in "ascii", "latin-1", "utf-8" :# , "utf-16-be", "utf-16-le":
2661	n/a	f = self.open(support.TESTFN, "w+", encoding=enc)
2662	n/a	f._CHUNK_SIZE = chunksize
2663	n/a	self.assertEqual(f.write("abc"), 3)
2664	n/a	f.close()
2665	n/a	f = self.open(support.TESTFN, "r+", encoding=enc)
2666	n/a	f._CHUNK_SIZE = chunksize
2667	n/a	self.assertEqual(f.tell(), 0)
2668	n/a	self.assertEqual(f.read(), "abc")
2669	n/a	cookie = f.tell()
2670	n/a	self.assertEqual(f.seek(0), 0)
2671	n/a	self.assertEqual(f.read(None), "abc")
2672	n/a	f.seek(0)
2673	n/a	self.assertEqual(f.read(2), "ab")
2674	n/a	self.assertEqual(f.read(1), "c")
2675	n/a	self.assertEqual(f.read(1), "")
2676	n/a	self.assertEqual(f.read(), "")
2677	n/a	self.assertEqual(f.tell(), cookie)
2678	n/a	self.assertEqual(f.seek(0), 0)
2679	n/a	self.assertEqual(f.seek(0, 2), cookie)
2680	n/a	self.assertEqual(f.write("def"), 3)
2681	n/a	self.assertEqual(f.seek(cookie), cookie)
2682	n/a	self.assertEqual(f.read(), "def")
2683	n/a	if enc.startswith("utf"):
2684	n/a	self.multi_line_test(f, enc)
2685	n/a	f.close()
2686	n/a
2687	n/a	def multi_line_test(self, f, enc):
2688	n/a	f.seek(0)
2689	n/a	f.truncate()
2690	n/a	sample = "s\xff\u0fff\uffff"
2691	n/a	wlines = []
2692	n/a	for size in (0, 1, 2, 3, 4, 5, 30, 31, 32, 33, 62, 63, 64, 65, 1000):
2693	n/a	chars = []
2694	n/a	for i in range(size):
2695	n/a	chars.append(sample[i % len(sample)])
2696	n/a	line = "".join(chars) + "\n"
2697	n/a	wlines.append((f.tell(), line))
2698	n/a	f.write(line)
2699	n/a	f.seek(0)
2700	n/a	rlines = []
2701	n/a	while True:
2702	n/a	pos = f.tell()
2703	n/a	line = f.readline()
2704	n/a	if not line:
2705	n/a	break
2706	n/a	rlines.append((pos, line))
2707	n/a	self.assertEqual(rlines, wlines)
2708	n/a
2709	n/a	def test_telling(self):
2710	n/a	f = self.open(support.TESTFN, "w+", encoding="utf-8")
2711	n/a	p0 = f.tell()
2712	n/a	f.write("\xff\n")
2713	n/a	p1 = f.tell()
2714	n/a	f.write("\xff\n")
2715	n/a	p2 = f.tell()
2716	n/a	f.seek(0)
2717	n/a	self.assertEqual(f.tell(), p0)
2718	n/a	self.assertEqual(f.readline(), "\xff\n")
2719	n/a	self.assertEqual(f.tell(), p1)
2720	n/a	self.assertEqual(f.readline(), "\xff\n")
2721	n/a	self.assertEqual(f.tell(), p2)
2722	n/a	f.seek(0)
2723	n/a	for line in f:
2724	n/a	self.assertEqual(line, "\xff\n")
2725	n/a	self.assertRaises(OSError, f.tell)
2726	n/a	self.assertEqual(f.tell(), p2)
2727	n/a	f.close()
2728	n/a
2729	n/a	def test_seeking(self):
2730	n/a	chunk_size = _default_chunk_size()
2731	n/a	prefix_size = chunk_size - 2
2732	n/a	u_prefix = "a" * prefix_size
2733	n/a	prefix = bytes(u_prefix.encode("utf-8"))
2734	n/a	self.assertEqual(len(u_prefix), len(prefix))
2735	n/a	u_suffix = "\u8888\n"
2736	n/a	suffix = bytes(u_suffix.encode("utf-8"))
2737	n/a	line = prefix + suffix
2738	n/a	with self.open(support.TESTFN, "wb") as f:
2739	n/a	f.write(line*2)
2740	n/a	with self.open(support.TESTFN, "r", encoding="utf-8") as f:
2741	n/a	s = f.read(prefix_size)
2742	n/a	self.assertEqual(s, str(prefix, "ascii"))
2743	n/a	self.assertEqual(f.tell(), prefix_size)
2744	n/a	self.assertEqual(f.readline(), u_suffix)
2745	n/a
2746	n/a	def test_seeking_too(self):
2747	n/a	# Regression test for a specific bug
2748	n/a	data = b'\xe0\xbf\xbf\n'
2749	n/a	with self.open(support.TESTFN, "wb") as f:
2750	n/a	f.write(data)
2751	n/a	with self.open(support.TESTFN, "r", encoding="utf-8") as f:
2752	n/a	f._CHUNK_SIZE # Just test that it exists
2753	n/a	f._CHUNK_SIZE = 2
2754	n/a	f.readline()
2755	n/a	f.tell()
2756	n/a
2757	n/a	def test_seek_and_tell(self):
2758	n/a	#Test seek/tell using the StatefulIncrementalDecoder.
2759	n/a	# Make test faster by doing smaller seeks
2760	n/a	CHUNK_SIZE = 128
2761	n/a
2762	n/a	def test_seek_and_tell_with_data(data, min_pos=0):
2763	n/a	"""Tell/seek to various points within a data stream and ensure
2764	n/a	that the decoded data returned by read() is consistent."""
2765	n/a	f = self.open(support.TESTFN, 'wb')
2766	n/a	f.write(data)
2767	n/a	f.close()
2768	n/a	f = self.open(support.TESTFN, encoding='test_decoder')
2769	n/a	f._CHUNK_SIZE = CHUNK_SIZE
2770	n/a	decoded = f.read()
2771	n/a	f.close()
2772	n/a
2773	n/a	for i in range(min_pos, len(decoded) + 1): # seek positions
2774	n/a	for j in [1, 5, len(decoded) - i]: # read lengths
2775	n/a	f = self.open(support.TESTFN, encoding='test_decoder')
2776	n/a	self.assertEqual(f.read(i), decoded[:i])
2777	n/a	cookie = f.tell()
2778	n/a	self.assertEqual(f.read(j), decoded[i:i + j])
2779	n/a	f.seek(cookie)
2780	n/a	self.assertEqual(f.read(), decoded[i:])
2781	n/a	f.close()
2782	n/a
2783	n/a	# Enable the test decoder.
2784	n/a	StatefulIncrementalDecoder.codecEnabled = 1
2785	n/a
2786	n/a	# Run the tests.
2787	n/a	try:
2788	n/a	# Try each test case.
2789	n/a	for input, _, _ in StatefulIncrementalDecoderTest.test_cases:
2790	n/a	test_seek_and_tell_with_data(input)
2791	n/a
2792	n/a	# Position each test case so that it crosses a chunk boundary.
2793	n/a	for input, _, _ in StatefulIncrementalDecoderTest.test_cases:
2794	n/a	offset = CHUNK_SIZE - len(input)//2
2795	n/a	prefix = b'.'*offset
2796	n/a	# Don't bother seeking into the prefix (takes too long).
2797	n/a	min_pos = offset*2
2798	n/a	test_seek_and_tell_with_data(prefix + input, min_pos)
2799	n/a
2800	n/a	# Ensure our test decoder won't interfere with subsequent tests.
2801	n/a	finally:
2802	n/a	StatefulIncrementalDecoder.codecEnabled = 0
2803	n/a
2804	n/a	def test_encoded_writes(self):
2805	n/a	data = "1234567890"
2806	n/a	tests = ("utf-16",
2807	n/a	"utf-16-le",
2808	n/a	"utf-16-be",
2809	n/a	"utf-32",
2810	n/a	"utf-32-le",
2811	n/a	"utf-32-be")
2812	n/a	for encoding in tests:
2813	n/a	buf = self.BytesIO()
2814	n/a	f = self.TextIOWrapper(buf, encoding=encoding)
2815	n/a	# Check if the BOM is written only once (see issue1753).
2816	n/a	f.write(data)
2817	n/a	f.write(data)
2818	n/a	f.seek(0)
2819	n/a	self.assertEqual(f.read(), data * 2)
2820	n/a	f.seek(0)
2821	n/a	self.assertEqual(f.read(), data * 2)
2822	n/a	self.assertEqual(buf.getvalue(), (data * 2).encode(encoding))
2823	n/a
2824	n/a	def test_unreadable(self):
2825	n/a	class UnReadable(self.BytesIO):
2826	n/a	def readable(self):
2827	n/a	return False
2828	n/a	txt = self.TextIOWrapper(UnReadable())
2829	n/a	self.assertRaises(OSError, txt.read)
2830	n/a
2831	n/a	def test_read_one_by_one(self):
2832	n/a	txt = self.TextIOWrapper(self.BytesIO(b"AA\r\nBB"))
2833	n/a	reads = ""
2834	n/a	while True:
2835	n/a	c = txt.read(1)
2836	n/a	if not c:
2837	n/a	break
2838	n/a	reads += c
2839	n/a	self.assertEqual(reads, "AA\nBB")
2840	n/a
2841	n/a	def test_readlines(self):
2842	n/a	txt = self.TextIOWrapper(self.BytesIO(b"AA\nBB\nCC"))
2843	n/a	self.assertEqual(txt.readlines(), ["AA\n", "BB\n", "CC"])
2844	n/a	txt.seek(0)
2845	n/a	self.assertEqual(txt.readlines(None), ["AA\n", "BB\n", "CC"])
2846	n/a	txt.seek(0)
2847	n/a	self.assertEqual(txt.readlines(5), ["AA\n", "BB\n"])
2848	n/a
2849	n/a	# read in amounts equal to TextIOWrapper._CHUNK_SIZE which is 128.
2850	n/a	def test_read_by_chunk(self):
2851	n/a	# make sure "\r\n" straddles 128 char boundary.
2852	n/a	txt = self.TextIOWrapper(self.BytesIO(b"A" * 127 + b"\r\nB"))
2853	n/a	reads = ""
2854	n/a	while True:
2855	n/a	c = txt.read(128)
2856	n/a	if not c:
2857	n/a	break
2858	n/a	reads += c
2859	n/a	self.assertEqual(reads, "A"*127+"\nB")
2860	n/a
2861	n/a	def test_writelines(self):
2862	n/a	l = ['ab', 'cd', 'ef']
2863	n/a	buf = self.BytesIO()
2864	n/a	txt = self.TextIOWrapper(buf)
2865	n/a	txt.writelines(l)
2866	n/a	txt.flush()
2867	n/a	self.assertEqual(buf.getvalue(), b'abcdef')
2868	n/a
2869	n/a	def test_writelines_userlist(self):
2870	n/a	l = UserList(['ab', 'cd', 'ef'])
2871	n/a	buf = self.BytesIO()
2872	n/a	txt = self.TextIOWrapper(buf)
2873	n/a	txt.writelines(l)
2874	n/a	txt.flush()
2875	n/a	self.assertEqual(buf.getvalue(), b'abcdef')
2876	n/a
2877	n/a	def test_writelines_error(self):
2878	n/a	txt = self.TextIOWrapper(self.BytesIO())
2879	n/a	self.assertRaises(TypeError, txt.writelines, [1, 2, 3])
2880	n/a	self.assertRaises(TypeError, txt.writelines, None)
2881	n/a	self.assertRaises(TypeError, txt.writelines, b'abc')
2882	n/a
2883	n/a	def test_issue1395_1(self):
2884	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
2885	n/a
2886	n/a	# read one char at a time
2887	n/a	reads = ""
2888	n/a	while True:
2889	n/a	c = txt.read(1)
2890	n/a	if not c:
2891	n/a	break
2892	n/a	reads += c
2893	n/a	self.assertEqual(reads, self.normalized)
2894	n/a
2895	n/a	def test_issue1395_2(self):
2896	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
2897	n/a	txt._CHUNK_SIZE = 4
2898	n/a
2899	n/a	reads = ""
2900	n/a	while True:
2901	n/a	c = txt.read(4)
2902	n/a	if not c:
2903	n/a	break
2904	n/a	reads += c
2905	n/a	self.assertEqual(reads, self.normalized)
2906	n/a
2907	n/a	def test_issue1395_3(self):
2908	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
2909	n/a	txt._CHUNK_SIZE = 4
2910	n/a
2911	n/a	reads = txt.read(4)
2912	n/a	reads += txt.read(4)
2913	n/a	reads += txt.readline()
2914	n/a	reads += txt.readline()
2915	n/a	reads += txt.readline()
2916	n/a	self.assertEqual(reads, self.normalized)
2917	n/a
2918	n/a	def test_issue1395_4(self):
2919	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
2920	n/a	txt._CHUNK_SIZE = 4
2921	n/a
2922	n/a	reads = txt.read(4)
2923	n/a	reads += txt.read()
2924	n/a	self.assertEqual(reads, self.normalized)
2925	n/a
2926	n/a	def test_issue1395_5(self):
2927	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
2928	n/a	txt._CHUNK_SIZE = 4
2929	n/a
2930	n/a	reads = txt.read(4)
2931	n/a	pos = txt.tell()
2932	n/a	txt.seek(0)
2933	n/a	txt.seek(pos)
2934	n/a	self.assertEqual(txt.read(4), "BBB\n")
2935	n/a
2936	n/a	def test_issue2282(self):
2937	n/a	buffer = self.BytesIO(self.testdata)
2938	n/a	txt = self.TextIOWrapper(buffer, encoding="ascii")
2939	n/a
2940	n/a	self.assertEqual(buffer.seekable(), txt.seekable())
2941	n/a
2942	n/a	def test_append_bom(self):
2943	n/a	# The BOM is not written again when appending to a non-empty file
2944	n/a	filename = support.TESTFN
2945	n/a	for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
2946	n/a	with self.open(filename, 'w', encoding=charset) as f:
2947	n/a	f.write('aaa')
2948	n/a	pos = f.tell()
2949	n/a	with self.open(filename, 'rb') as f:
2950	n/a	self.assertEqual(f.read(), 'aaa'.encode(charset))
2951	n/a
2952	n/a	with self.open(filename, 'a', encoding=charset) as f:
2953	n/a	f.write('xxx')
2954	n/a	with self.open(filename, 'rb') as f:
2955	n/a	self.assertEqual(f.read(), 'aaaxxx'.encode(charset))
2956	n/a
2957	n/a	def test_seek_bom(self):
2958	n/a	# Same test, but when seeking manually
2959	n/a	filename = support.TESTFN
2960	n/a	for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
2961	n/a	with self.open(filename, 'w', encoding=charset) as f:
2962	n/a	f.write('aaa')
2963	n/a	pos = f.tell()
2964	n/a	with self.open(filename, 'r+', encoding=charset) as f:
2965	n/a	f.seek(pos)
2966	n/a	f.write('zzz')
2967	n/a	f.seek(0)
2968	n/a	f.write('bbb')
2969	n/a	with self.open(filename, 'rb') as f:
2970	n/a	self.assertEqual(f.read(), 'bbbzzz'.encode(charset))
2971	n/a
2972	n/a	def test_seek_append_bom(self):
2973	n/a	# Same test, but first seek to the start and then to the end
2974	n/a	filename = support.TESTFN
2975	n/a	for charset in ('utf-8-sig', 'utf-16', 'utf-32'):
2976	n/a	with self.open(filename, 'w', encoding=charset) as f:
2977	n/a	f.write('aaa')
2978	n/a	with self.open(filename, 'a', encoding=charset) as f:
2979	n/a	f.seek(0)
2980	n/a	f.seek(0, self.SEEK_END)
2981	n/a	f.write('xxx')
2982	n/a	with self.open(filename, 'rb') as f:
2983	n/a	self.assertEqual(f.read(), 'aaaxxx'.encode(charset))
2984	n/a
2985	n/a	def test_errors_property(self):
2986	n/a	with self.open(support.TESTFN, "w") as f:
2987	n/a	self.assertEqual(f.errors, "strict")
2988	n/a	with self.open(support.TESTFN, "w", errors="replace") as f:
2989	n/a	self.assertEqual(f.errors, "replace")
2990	n/a
2991	n/a	@support.no_tracing
2992	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
2993	n/a	def test_threads_write(self):
2994	n/a	# Issue6750: concurrent writes could duplicate data
2995	n/a	event = threading.Event()
2996	n/a	with self.open(support.TESTFN, "w", buffering=1) as f:
2997	n/a	def run(n):
2998	n/a	text = "Thread%03d\n" % n
2999	n/a	event.wait()
3000	n/a	f.write(text)
3001	n/a	threads = [threading.Thread(target=run, args=(x,))
3002	n/a	for x in range(20)]
3003	n/a	with support.start_threads(threads, event.set):
3004	n/a	time.sleep(0.02)
3005	n/a	with self.open(support.TESTFN) as f:
3006	n/a	content = f.read()
3007	n/a	for n in range(20):
3008	n/a	self.assertEqual(content.count("Thread%03d\n" % n), 1)
3009	n/a
3010	n/a	def test_flush_error_on_close(self):
3011	n/a	# Test that text file is closed despite failed flush
3012	n/a	# and that flush() is called before file closed.
3013	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
3014	n/a	closed = []
3015	n/a	def bad_flush():
3016	n/a	closed[:] = [txt.closed, txt.buffer.closed]
3017	n/a	raise OSError()
3018	n/a	txt.flush = bad_flush
3019	n/a	self.assertRaises(OSError, txt.close) # exception not swallowed
3020	n/a	self.assertTrue(txt.closed)
3021	n/a	self.assertTrue(txt.buffer.closed)
3022	n/a	self.assertTrue(closed) # flush() called
3023	n/a	self.assertFalse(closed[0]) # flush() called before file closed
3024	n/a	self.assertFalse(closed[1])
3025	n/a	txt.flush = lambda: None # break reference loop
3026	n/a
3027	n/a	def test_close_error_on_close(self):
3028	n/a	buffer = self.BytesIO(self.testdata)
3029	n/a	def bad_flush():
3030	n/a	raise OSError('flush')
3031	n/a	def bad_close():
3032	n/a	raise OSError('close')
3033	n/a	buffer.close = bad_close
3034	n/a	txt = self.TextIOWrapper(buffer, encoding="ascii")
3035	n/a	txt.flush = bad_flush
3036	n/a	with self.assertRaises(OSError) as err: # exception not swallowed
3037	n/a	txt.close()
3038	n/a	self.assertEqual(err.exception.args, ('close',))
3039	n/a	self.assertIsInstance(err.exception.__context__, OSError)
3040	n/a	self.assertEqual(err.exception.__context__.args, ('flush',))
3041	n/a	self.assertFalse(txt.closed)
3042	n/a
3043	n/a	def test_nonnormalized_close_error_on_close(self):
3044	n/a	# Issue #21677
3045	n/a	buffer = self.BytesIO(self.testdata)
3046	n/a	def bad_flush():
3047	n/a	raise non_existing_flush
3048	n/a	def bad_close():
3049	n/a	raise non_existing_close
3050	n/a	buffer.close = bad_close
3051	n/a	txt = self.TextIOWrapper(buffer, encoding="ascii")
3052	n/a	txt.flush = bad_flush
3053	n/a	with self.assertRaises(NameError) as err: # exception not swallowed
3054	n/a	txt.close()
3055	n/a	self.assertIn('non_existing_close', str(err.exception))
3056	n/a	self.assertIsInstance(err.exception.__context__, NameError)
3057	n/a	self.assertIn('non_existing_flush', str(err.exception.__context__))
3058	n/a	self.assertFalse(txt.closed)
3059	n/a
3060	n/a	def test_multi_close(self):
3061	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
3062	n/a	txt.close()
3063	n/a	txt.close()
3064	n/a	txt.close()
3065	n/a	self.assertRaises(ValueError, txt.flush)
3066	n/a
3067	n/a	def test_unseekable(self):
3068	n/a	txt = self.TextIOWrapper(self.MockUnseekableIO(self.testdata))
3069	n/a	self.assertRaises(self.UnsupportedOperation, txt.tell)
3070	n/a	self.assertRaises(self.UnsupportedOperation, txt.seek, 0)
3071	n/a
3072	n/a	def test_readonly_attributes(self):
3073	n/a	txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii")
3074	n/a	buf = self.BytesIO(self.testdata)
3075	n/a	with self.assertRaises(AttributeError):
3076	n/a	txt.buffer = buf
3077	n/a
3078	n/a	def test_rawio(self):
3079	n/a	# Issue #12591: TextIOWrapper must work with raw I/O objects, so
3080	n/a	# that subprocess.Popen() can have the required unbuffered
3081	n/a	# semantics with universal_newlines=True.
3082	n/a	raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
3083	n/a	txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
3084	n/a	# Reads
3085	n/a	self.assertEqual(txt.read(4), 'abcd')
3086	n/a	self.assertEqual(txt.readline(), 'efghi\n')
3087	n/a	self.assertEqual(list(txt), ['jkl\n', 'opq\n'])
3088	n/a
3089	n/a	def test_rawio_write_through(self):
3090	n/a	# Issue #12591: with write_through=True, writes don't need a flush
3091	n/a	raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
3092	n/a	txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n',
3093	n/a	write_through=True)
3094	n/a	txt.write('1')
3095	n/a	txt.write('23\n4')
3096	n/a	txt.write('5')
3097	n/a	self.assertEqual(b''.join(raw._write_stack), b'123\n45')
3098	n/a
3099	n/a	def test_bufio_write_through(self):
3100	n/a	# Issue #21396: write_through=True doesn't force a flush()
3101	n/a	# on the underlying binary buffered object.
3102	n/a	flush_called, write_called = [], []
3103	n/a	class BufferedWriter(self.BufferedWriter):
3104	n/a	def flush(self, args, *kwargs):
3105	n/a	flush_called.append(True)
3106	n/a	return super().flush(args, *kwargs)
3107	n/a	def write(self, args, *kwargs):
3108	n/a	write_called.append(True)
3109	n/a	return super().write(args, *kwargs)
3110	n/a
3111	n/a	rawio = self.BytesIO()
3112	n/a	data = b"a"
3113	n/a	bufio = BufferedWriter(rawio, len(data)*2)
3114	n/a	textio = self.TextIOWrapper(bufio, encoding='ascii',
3115	n/a	write_through=True)
3116	n/a	# write to the buffered io but don't overflow the buffer
3117	n/a	text = data.decode('ascii')
3118	n/a	textio.write(text)
3119	n/a
3120	n/a	# buffer.flush is not called with write_through=True
3121	n/a	self.assertFalse(flush_called)
3122	n/a	# buffer.write is called with write_through=True
3123	n/a	self.assertTrue(write_called)
3124	n/a	self.assertEqual(rawio.getvalue(), b"") # no flush
3125	n/a
3126	n/a	write_called = [] # reset
3127	n/a	textio.write(text * 10) # total content is larger than bufio buffer
3128	n/a	self.assertTrue(write_called)
3129	n/a	self.assertEqual(rawio.getvalue(), data * 11) # all flushed
3130	n/a
3131	n/a	def test_read_nonbytes(self):
3132	n/a	# Issue #17106
3133	n/a	# Crash when underlying read() returns non-bytes
3134	n/a	t = self.TextIOWrapper(self.StringIO('a'))
3135	n/a	self.assertRaises(TypeError, t.read, 1)
3136	n/a	t = self.TextIOWrapper(self.StringIO('a'))
3137	n/a	self.assertRaises(TypeError, t.readline)
3138	n/a	t = self.TextIOWrapper(self.StringIO('a'))
3139	n/a	self.assertRaises(TypeError, t.read)
3140	n/a
3141	n/a	def test_illegal_decoder(self):
3142	n/a	# Issue #17106
3143	n/a	# Bypass the early encoding check added in issue 20404
3144	n/a	def _make_illegal_wrapper():
3145	n/a	quopri = codecs.lookup("quopri")
3146	n/a	quopri._is_text_encoding = True
3147	n/a	try:
3148	n/a	t = self.TextIOWrapper(self.BytesIO(b'aaaaaa'),
3149	n/a	newline='\n', encoding="quopri")
3150	n/a	finally:
3151	n/a	quopri._is_text_encoding = False
3152	n/a	return t
3153	n/a	# Crash when decoder returns non-string
3154	n/a	t = _make_illegal_wrapper()
3155	n/a	self.assertRaises(TypeError, t.read, 1)
3156	n/a	t = _make_illegal_wrapper()
3157	n/a	self.assertRaises(TypeError, t.readline)
3158	n/a	t = _make_illegal_wrapper()
3159	n/a	self.assertRaises(TypeError, t.read)
3160	n/a
3161	n/a	def _check_create_at_shutdown(self, **kwargs):
3162	n/a	# Issue #20037: creating a TextIOWrapper at shutdown
3163	n/a	# shouldn't crash the interpreter.
3164	n/a	iomod = self.io.__name__
3165	n/a	code = """if 1:
3166	n/a	import codecs
3167	n/a	import {iomod} as io
3168	n/a
3169	n/a	# Avoid looking up codecs at shutdown
3170	n/a	codecs.lookup('utf-8')
3171	n/a
3172	n/a	class C:
3173	n/a	def __init__(self):
3174	n/a	self.buf = io.BytesIO()
3175	n/a	def __del__(self):
3176	n/a	io.TextIOWrapper(self.buf, **{kwargs})
3177	n/a	print("ok")
3178	n/a	c = C()
3179	n/a	""".format(iomod=iomod, kwargs=kwargs)
3180	n/a	return assert_python_ok("-c", code)
3181	n/a
3182	n/a	@support.requires_type_collecting
3183	n/a	def test_create_at_shutdown_without_encoding(self):
3184	n/a	rc, out, err = self._check_create_at_shutdown()
3185	n/a	if err:
3186	n/a	# Can error out with a RuntimeError if the module state
3187	n/a	# isn't found.
3188	n/a	self.assertIn(self.shutdown_error, err.decode())
3189	n/a	else:
3190	n/a	self.assertEqual("ok", out.decode().strip())
3191	n/a
3192	n/a	@support.requires_type_collecting
3193	n/a	def test_create_at_shutdown_with_encoding(self):
3194	n/a	rc, out, err = self._check_create_at_shutdown(encoding='utf-8',
3195	n/a	errors='strict')
3196	n/a	self.assertFalse(err)
3197	n/a	self.assertEqual("ok", out.decode().strip())
3198	n/a
3199	n/a	def test_read_byteslike(self):
3200	n/a	r = MemviewBytesIO(b'Just some random string\n')
3201	n/a	t = self.TextIOWrapper(r, 'utf-8')
3202	n/a
3203	n/a	# TextIOwrapper will not read the full string, because
3204	n/a	# we truncate it to a multiple of the native int size
3205	n/a	# so that we can construct a more complex memoryview.
3206	n/a	bytes_val = _to_memoryview(r.getvalue()).tobytes()
3207	n/a
3208	n/a	self.assertEqual(t.read(200), bytes_val.decode('utf-8'))
3209	n/a
3210	n/a	def test_issue22849(self):
3211	n/a	class F(object):
3212	n/a	def readable(self): return True
3213	n/a	def writable(self): return True
3214	n/a	def seekable(self): return True
3215	n/a
3216	n/a	for i in range(10):
3217	n/a	try:
3218	n/a	self.TextIOWrapper(F(), encoding='utf-8')
3219	n/a	except Exception:
3220	n/a	pass
3221	n/a
3222	n/a	F.tell = lambda x: 0
3223	n/a	t = self.TextIOWrapper(F(), encoding='utf-8')
3224	n/a
3225	n/a
3226	n/a	class MemviewBytesIO(io.BytesIO):
3227	n/a	'''A BytesIO object whose read method returns memoryviews
3228	n/a	rather than bytes'''
3229	n/a
3230	n/a	def read1(self, len_):
3231	n/a	return _to_memoryview(super().read1(len_))
3232	n/a
3233	n/a	def read(self, len_):
3234	n/a	return _to_memoryview(super().read(len_))
3235	n/a
3236	n/a	def _to_memoryview(buf):
3237	n/a	'''Convert bytes-object buf to a non-trivial memoryview'''
3238	n/a
3239	n/a	arr = array.array('i')
3240	n/a	idx = len(buf) - len(buf) % arr.itemsize
3241	n/a	arr.frombytes(buf[:idx])
3242	n/a	return memoryview(arr)
3243	n/a
3244	n/a
3245	n/a	class CTextIOWrapperTest(TextIOWrapperTest):
3246	n/a	io = io
3247	n/a	shutdown_error = "RuntimeError: could not find io module state"
3248	n/a
3249	n/a	def test_initialization(self):
3250	n/a	r = self.BytesIO(b"\xc3\xa9\n\n")
3251	n/a	b = self.BufferedReader(r, 1000)
3252	n/a	t = self.TextIOWrapper(b)
3253	n/a	self.assertRaises(ValueError, t.__init__, b, newline='xyzzy')
3254	n/a	self.assertRaises(ValueError, t.read)
3255	n/a
3256	n/a	t = self.TextIOWrapper.__new__(self.TextIOWrapper)
3257	n/a	self.assertRaises(Exception, repr, t)
3258	n/a
3259	n/a	def test_garbage_collection(self):
3260	n/a	# C TextIOWrapper objects are collected, and collecting them flushes
3261	n/a	# all data to disk.
3262	n/a	# The Python version has __del__, so it ends in gc.garbage instead.
3263	n/a	with support.check_warnings(('', ResourceWarning)):
3264	n/a	rawio = io.FileIO(support.TESTFN, "wb")
3265	n/a	b = self.BufferedWriter(rawio)
3266	n/a	t = self.TextIOWrapper(b, encoding="ascii")
3267	n/a	t.write("456def")
3268	n/a	t.x = t
3269	n/a	wr = weakref.ref(t)
3270	n/a	del t
3271	n/a	support.gc_collect()
3272	n/a	self.assertIsNone(wr(), wr)
3273	n/a	with self.open(support.TESTFN, "rb") as f:
3274	n/a	self.assertEqual(f.read(), b"456def")
3275	n/a
3276	n/a	def test_rwpair_cleared_before_textio(self):
3277	n/a	# Issue 13070: TextIOWrapper's finalization would crash when called
3278	n/a	# after the reference to the underlying BufferedRWPair's writer got
3279	n/a	# cleared by the GC.
3280	n/a	for i in range(1000):
3281	n/a	b1 = self.BufferedRWPair(self.MockRawIO(), self.MockRawIO())
3282	n/a	t1 = self.TextIOWrapper(b1, encoding="ascii")
3283	n/a	b2 = self.BufferedRWPair(self.MockRawIO(), self.MockRawIO())
3284	n/a	t2 = self.TextIOWrapper(b2, encoding="ascii")
3285	n/a	# circular references
3286	n/a	t1.buddy = t2
3287	n/a	t2.buddy = t1
3288	n/a	support.gc_collect()
3289	n/a
3290	n/a
3291	n/a	class PyTextIOWrapperTest(TextIOWrapperTest):
3292	n/a	io = pyio
3293	n/a	shutdown_error = "LookupError: unknown encoding: ascii"
3294	n/a
3295	n/a
3296	n/a	class IncrementalNewlineDecoderTest(unittest.TestCase):
3297	n/a
3298	n/a	def check_newline_decoding_utf8(self, decoder):
3299	n/a	# UTF-8 specific tests for a newline decoder
3300	n/a	def _check_decode(b, s, **kwargs):
3301	n/a	# We exercise getstate() / setstate() as well as decode()
3302	n/a	state = decoder.getstate()
3303	n/a	self.assertEqual(decoder.decode(b, **kwargs), s)
3304	n/a	decoder.setstate(state)
3305	n/a	self.assertEqual(decoder.decode(b, **kwargs), s)
3306	n/a
3307	n/a	_check_decode(b'\xe8\xa2\x88', "\u8888")
3308	n/a
3309	n/a	_check_decode(b'\xe8', "")
3310	n/a	_check_decode(b'\xa2', "")
3311	n/a	_check_decode(b'\x88', "\u8888")
3312	n/a
3313	n/a	_check_decode(b'\xe8', "")
3314	n/a	_check_decode(b'\xa2', "")
3315	n/a	_check_decode(b'\x88', "\u8888")
3316	n/a
3317	n/a	_check_decode(b'\xe8', "")
3318	n/a	self.assertRaises(UnicodeDecodeError, decoder.decode, b'', final=True)
3319	n/a
3320	n/a	decoder.reset()
3321	n/a	_check_decode(b'\n', "\n")
3322	n/a	_check_decode(b'\r', "")
3323	n/a	_check_decode(b'', "\n", final=True)
3324	n/a	_check_decode(b'\r', "\n", final=True)
3325	n/a
3326	n/a	_check_decode(b'\r', "")
3327	n/a	_check_decode(b'a', "\na")
3328	n/a
3329	n/a	_check_decode(b'\r\r\n', "\n\n")
3330	n/a	_check_decode(b'\r', "")
3331	n/a	_check_decode(b'\r', "\n")
3332	n/a	_check_decode(b'\na', "\na")
3333	n/a
3334	n/a	_check_decode(b'\xe8\xa2\x88\r\n', "\u8888\n")
3335	n/a	_check_decode(b'\xe8\xa2\x88', "\u8888")
3336	n/a	_check_decode(b'\n', "\n")
3337	n/a	_check_decode(b'\xe8\xa2\x88\r', "\u8888")
3338	n/a	_check_decode(b'\n', "\n")
3339	n/a
3340	n/a	def check_newline_decoding(self, decoder, encoding):
3341	n/a	result = []
3342	n/a	if encoding is not None:
3343	n/a	encoder = codecs.getincrementalencoder(encoding)()
3344	n/a	def _decode_bytewise(s):
3345	n/a	# Decode one byte at a time
3346	n/a	for b in encoder.encode(s):
3347	n/a	result.append(decoder.decode(bytes([b])))
3348	n/a	else:
3349	n/a	encoder = None
3350	n/a	def _decode_bytewise(s):
3351	n/a	# Decode one char at a time
3352	n/a	for c in s:
3353	n/a	result.append(decoder.decode(c))
3354	n/a	self.assertEqual(decoder.newlines, None)
3355	n/a	_decode_bytewise("abc\n\r")
3356	n/a	self.assertEqual(decoder.newlines, '\n')
3357	n/a	_decode_bytewise("\nabc")
3358	n/a	self.assertEqual(decoder.newlines, ('\n', '\r\n'))
3359	n/a	_decode_bytewise("abc\r")
3360	n/a	self.assertEqual(decoder.newlines, ('\n', '\r\n'))
3361	n/a	_decode_bytewise("abc")
3362	n/a	self.assertEqual(decoder.newlines, ('\r', '\n', '\r\n'))
3363	n/a	_decode_bytewise("abc\r")
3364	n/a	self.assertEqual("".join(result), "abc\n\nabcabc\nabcabc")
3365	n/a	decoder.reset()
3366	n/a	input = "abc"
3367	n/a	if encoder is not None:
3368	n/a	encoder.reset()
3369	n/a	input = encoder.encode(input)
3370	n/a	self.assertEqual(decoder.decode(input), "abc")
3371	n/a	self.assertEqual(decoder.newlines, None)
3372	n/a
3373	n/a	def test_newline_decoder(self):
3374	n/a	encodings = (
3375	n/a	# None meaning the IncrementalNewlineDecoder takes unicode input
3376	n/a	# rather than bytes input
3377	n/a	None, 'utf-8', 'latin-1',
3378	n/a	'utf-16', 'utf-16-le', 'utf-16-be',
3379	n/a	'utf-32', 'utf-32-le', 'utf-32-be',
3380	n/a	)
3381	n/a	for enc in encodings:
3382	n/a	decoder = enc and codecs.getincrementaldecoder(enc)()
3383	n/a	decoder = self.IncrementalNewlineDecoder(decoder, translate=True)
3384	n/a	self.check_newline_decoding(decoder, enc)
3385	n/a	decoder = codecs.getincrementaldecoder("utf-8")()
3386	n/a	decoder = self.IncrementalNewlineDecoder(decoder, translate=True)
3387	n/a	self.check_newline_decoding_utf8(decoder)
3388	n/a
3389	n/a	def test_newline_bytes(self):
3390	n/a	# Issue 5433: Excessive optimization in IncrementalNewlineDecoder
3391	n/a	def _check(dec):
3392	n/a	self.assertEqual(dec.newlines, None)
3393	n/a	self.assertEqual(dec.decode("\u0D00"), "\u0D00")
3394	n/a	self.assertEqual(dec.newlines, None)
3395	n/a	self.assertEqual(dec.decode("\u0A00"), "\u0A00")
3396	n/a	self.assertEqual(dec.newlines, None)
3397	n/a	dec = self.IncrementalNewlineDecoder(None, translate=False)
3398	n/a	_check(dec)
3399	n/a	dec = self.IncrementalNewlineDecoder(None, translate=True)
3400	n/a	_check(dec)
3401	n/a
3402	n/a	class CIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
3403	n/a	pass
3404	n/a
3405	n/a	class PyIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
3406	n/a	pass
3407	n/a
3408	n/a
3409	n/a	# XXX Tests for open()
3410	n/a
3411	n/a	class MiscIOTest(unittest.TestCase):
3412	n/a
3413	n/a	def tearDown(self):
3414	n/a	support.unlink(support.TESTFN)
3415	n/a
3416	n/a	def test___all__(self):
3417	n/a	for name in self.io.__all__:
3418	n/a	obj = getattr(self.io, name, None)
3419	n/a	self.assertIsNotNone(obj, name)
3420	n/a	if name == "open":
3421	n/a	continue
3422	n/a	elif "error" in name.lower() or name == "UnsupportedOperation":
3423	n/a	self.assertTrue(issubclass(obj, Exception), name)
3424	n/a	elif not name.startswith("SEEK_"):
3425	n/a	self.assertTrue(issubclass(obj, self.IOBase))
3426	n/a
3427	n/a	def test_attributes(self):
3428	n/a	f = self.open(support.TESTFN, "wb", buffering=0)
3429	n/a	self.assertEqual(f.mode, "wb")
3430	n/a	f.close()
3431	n/a
3432	n/a	with support.check_warnings(('', DeprecationWarning)):
3433	n/a	f = self.open(support.TESTFN, "U")
3434	n/a	self.assertEqual(f.name, support.TESTFN)
3435	n/a	self.assertEqual(f.buffer.name, support.TESTFN)
3436	n/a	self.assertEqual(f.buffer.raw.name, support.TESTFN)
3437	n/a	self.assertEqual(f.mode, "U")
3438	n/a	self.assertEqual(f.buffer.mode, "rb")
3439	n/a	self.assertEqual(f.buffer.raw.mode, "rb")
3440	n/a	f.close()
3441	n/a
3442	n/a	f = self.open(support.TESTFN, "w+")
3443	n/a	self.assertEqual(f.mode, "w+")
3444	n/a	self.assertEqual(f.buffer.mode, "rb+") # Does it really matter?
3445	n/a	self.assertEqual(f.buffer.raw.mode, "rb+")
3446	n/a
3447	n/a	g = self.open(f.fileno(), "wb", closefd=False)
3448	n/a	self.assertEqual(g.mode, "wb")
3449	n/a	self.assertEqual(g.raw.mode, "wb")
3450	n/a	self.assertEqual(g.name, f.fileno())
3451	n/a	self.assertEqual(g.raw.name, f.fileno())
3452	n/a	f.close()
3453	n/a	g.close()
3454	n/a
3455	n/a	def test_io_after_close(self):
3456	n/a	for kwargs in [
3457	n/a	{"mode": "w"},
3458	n/a	{"mode": "wb"},
3459	n/a	{"mode": "w", "buffering": 1},
3460	n/a	{"mode": "w", "buffering": 2},
3461	n/a	{"mode": "wb", "buffering": 0},
3462	n/a	{"mode": "r"},
3463	n/a	{"mode": "rb"},
3464	n/a	{"mode": "r", "buffering": 1},
3465	n/a	{"mode": "r", "buffering": 2},
3466	n/a	{"mode": "rb", "buffering": 0},
3467	n/a	{"mode": "w+"},
3468	n/a	{"mode": "w+b"},
3469	n/a	{"mode": "w+", "buffering": 1},
3470	n/a	{"mode": "w+", "buffering": 2},
3471	n/a	{"mode": "w+b", "buffering": 0},
3472	n/a	]:
3473	n/a	f = self.open(support.TESTFN, **kwargs)
3474	n/a	f.close()
3475	n/a	self.assertRaises(ValueError, f.flush)
3476	n/a	self.assertRaises(ValueError, f.fileno)
3477	n/a	self.assertRaises(ValueError, f.isatty)
3478	n/a	self.assertRaises(ValueError, f.__iter__)
3479	n/a	if hasattr(f, "peek"):
3480	n/a	self.assertRaises(ValueError, f.peek, 1)
3481	n/a	self.assertRaises(ValueError, f.read)
3482	n/a	if hasattr(f, "read1"):
3483	n/a	self.assertRaises(ValueError, f.read1, 1024)
3484	n/a	self.assertRaises(ValueError, f.read1)
3485	n/a	if hasattr(f, "readall"):
3486	n/a	self.assertRaises(ValueError, f.readall)
3487	n/a	if hasattr(f, "readinto"):
3488	n/a	self.assertRaises(ValueError, f.readinto, bytearray(1024))
3489	n/a	if hasattr(f, "readinto1"):
3490	n/a	self.assertRaises(ValueError, f.readinto1, bytearray(1024))
3491	n/a	self.assertRaises(ValueError, f.readline)
3492	n/a	self.assertRaises(ValueError, f.readlines)
3493	n/a	self.assertRaises(ValueError, f.seek, 0)
3494	n/a	self.assertRaises(ValueError, f.tell)
3495	n/a	self.assertRaises(ValueError, f.truncate)
3496	n/a	self.assertRaises(ValueError, f.write,
3497	n/a	b"" if "b" in kwargs['mode'] else "")
3498	n/a	self.assertRaises(ValueError, f.writelines, [])
3499	n/a	self.assertRaises(ValueError, next, f)
3500	n/a
3501	n/a	def test_blockingioerror(self):
3502	n/a	# Various BlockingIOError issues
3503	n/a	class C(str):
3504	n/a	pass
3505	n/a	c = C("")
3506	n/a	b = self.BlockingIOError(1, c)
3507	n/a	c.b = b
3508	n/a	b.c = c
3509	n/a	wr = weakref.ref(c)
3510	n/a	del c, b
3511	n/a	support.gc_collect()
3512	n/a	self.assertIsNone(wr(), wr)
3513	n/a
3514	n/a	def test_abcs(self):
3515	n/a	# Test the visible base classes are ABCs.
3516	n/a	self.assertIsInstance(self.IOBase, abc.ABCMeta)
3517	n/a	self.assertIsInstance(self.RawIOBase, abc.ABCMeta)
3518	n/a	self.assertIsInstance(self.BufferedIOBase, abc.ABCMeta)
3519	n/a	self.assertIsInstance(self.TextIOBase, abc.ABCMeta)
3520	n/a
3521	n/a	def _check_abc_inheritance(self, abcmodule):
3522	n/a	with self.open(support.TESTFN, "wb", buffering=0) as f:
3523	n/a	self.assertIsInstance(f, abcmodule.IOBase)
3524	n/a	self.assertIsInstance(f, abcmodule.RawIOBase)
3525	n/a	self.assertNotIsInstance(f, abcmodule.BufferedIOBase)
3526	n/a	self.assertNotIsInstance(f, abcmodule.TextIOBase)
3527	n/a	with self.open(support.TESTFN, "wb") as f:
3528	n/a	self.assertIsInstance(f, abcmodule.IOBase)
3529	n/a	self.assertNotIsInstance(f, abcmodule.RawIOBase)
3530	n/a	self.assertIsInstance(f, abcmodule.BufferedIOBase)
3531	n/a	self.assertNotIsInstance(f, abcmodule.TextIOBase)
3532	n/a	with self.open(support.TESTFN, "w") as f:
3533	n/a	self.assertIsInstance(f, abcmodule.IOBase)
3534	n/a	self.assertNotIsInstance(f, abcmodule.RawIOBase)
3535	n/a	self.assertNotIsInstance(f, abcmodule.BufferedIOBase)
3536	n/a	self.assertIsInstance(f, abcmodule.TextIOBase)
3537	n/a
3538	n/a	def test_abc_inheritance(self):
3539	n/a	# Test implementations inherit from their respective ABCs
3540	n/a	self._check_abc_inheritance(self)
3541	n/a
3542	n/a	def test_abc_inheritance_official(self):
3543	n/a	# Test implementations inherit from the official ABCs of the
3544	n/a	# baseline "io" module.
3545	n/a	self._check_abc_inheritance(io)
3546	n/a
3547	n/a	def _check_warn_on_dealloc(self, args, *kwargs):
3548	n/a	f = open(args, *kwargs)
3549	n/a	r = repr(f)
3550	n/a	with self.assertWarns(ResourceWarning) as cm:
3551	n/a	f = None
3552	n/a	support.gc_collect()
3553	n/a	self.assertIn(r, str(cm.warning.args[0]))
3554	n/a
3555	n/a	def test_warn_on_dealloc(self):
3556	n/a	self._check_warn_on_dealloc(support.TESTFN, "wb", buffering=0)
3557	n/a	self._check_warn_on_dealloc(support.TESTFN, "wb")
3558	n/a	self._check_warn_on_dealloc(support.TESTFN, "w")
3559	n/a
3560	n/a	def _check_warn_on_dealloc_fd(self, args, *kwargs):
3561	n/a	fds = []
3562	n/a	def cleanup_fds():
3563	n/a	for fd in fds:
3564	n/a	try:
3565	n/a	os.close(fd)
3566	n/a	except OSError as e:
3567	n/a	if e.errno != errno.EBADF:
3568	n/a	raise
3569	n/a	self.addCleanup(cleanup_fds)
3570	n/a	r, w = os.pipe()
3571	n/a	fds += r, w
3572	n/a	self._check_warn_on_dealloc(r, args, *kwargs)
3573	n/a	# When using closefd=False, there's no warning
3574	n/a	r, w = os.pipe()
3575	n/a	fds += r, w
3576	n/a	with support.check_no_resource_warning(self):
3577	n/a	open(r, args, closefd=False, *kwargs)
3578	n/a
3579	n/a	def test_warn_on_dealloc_fd(self):
3580	n/a	self._check_warn_on_dealloc_fd("rb", buffering=0)
3581	n/a	self._check_warn_on_dealloc_fd("rb")
3582	n/a	self._check_warn_on_dealloc_fd("r")
3583	n/a
3584	n/a
3585	n/a	def test_pickling(self):
3586	n/a	# Pickling file objects is forbidden
3587	n/a	for kwargs in [
3588	n/a	{"mode": "w"},
3589	n/a	{"mode": "wb"},
3590	n/a	{"mode": "wb", "buffering": 0},
3591	n/a	{"mode": "r"},
3592	n/a	{"mode": "rb"},
3593	n/a	{"mode": "rb", "buffering": 0},
3594	n/a	{"mode": "w+"},
3595	n/a	{"mode": "w+b"},
3596	n/a	{"mode": "w+b", "buffering": 0},
3597	n/a	]:
3598	n/a	for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
3599	n/a	with self.open(support.TESTFN, **kwargs) as f:
3600	n/a	self.assertRaises(TypeError, pickle.dumps, f, protocol)
3601	n/a
3602	n/a	def test_nonblock_pipe_write_bigbuf(self):
3603	n/a	self._test_nonblock_pipe_write(16*1024)
3604	n/a
3605	n/a	def test_nonblock_pipe_write_smallbuf(self):
3606	n/a	self._test_nonblock_pipe_write(1024)
3607	n/a
3608	n/a	@unittest.skipUnless(hasattr(os, 'set_blocking'),
3609	n/a	'os.set_blocking() required for this test')
3610	n/a	def _test_nonblock_pipe_write(self, bufsize):
3611	n/a	sent = []
3612	n/a	received = []
3613	n/a	r, w = os.pipe()
3614	n/a	os.set_blocking(r, False)
3615	n/a	os.set_blocking(w, False)
3616	n/a
3617	n/a	# To exercise all code paths in the C implementation we need
3618	n/a	# to play with buffer sizes. For instance, if we choose a
3619	n/a	# buffer size less than or equal to _PIPE_BUF (4096 on Linux)
3620	n/a	# then we will never get a partial write of the buffer.
3621	n/a	rf = self.open(r, mode='rb', closefd=True, buffering=bufsize)
3622	n/a	wf = self.open(w, mode='wb', closefd=True, buffering=bufsize)
3623	n/a
3624	n/a	with rf, wf:
3625	n/a	for N in 9999, 73, 7574:
3626	n/a	try:
3627	n/a	i = 0
3628	n/a	while True:
3629	n/a	msg = bytes([i % 26 + 97]) * N
3630	n/a	sent.append(msg)
3631	n/a	wf.write(msg)
3632	n/a	i += 1
3633	n/a
3634	n/a	except self.BlockingIOError as e:
3635	n/a	self.assertEqual(e.args[0], errno.EAGAIN)
3636	n/a	self.assertEqual(e.args[2], e.characters_written)
3637	n/a	sent[-1] = sent[-1][:e.characters_written]
3638	n/a	received.append(rf.read())
3639	n/a	msg = b'BLOCKED'
3640	n/a	wf.write(msg)
3641	n/a	sent.append(msg)
3642	n/a
3643	n/a	while True:
3644	n/a	try:
3645	n/a	wf.flush()
3646	n/a	break
3647	n/a	except self.BlockingIOError as e:
3648	n/a	self.assertEqual(e.args[0], errno.EAGAIN)
3649	n/a	self.assertEqual(e.args[2], e.characters_written)
3650	n/a	self.assertEqual(e.characters_written, 0)
3651	n/a	received.append(rf.read())
3652	n/a
3653	n/a	received += iter(rf.read, None)
3654	n/a
3655	n/a	sent, received = b''.join(sent), b''.join(received)
3656	n/a	self.assertEqual(sent, received)
3657	n/a	self.assertTrue(wf.closed)
3658	n/a	self.assertTrue(rf.closed)
3659	n/a
3660	n/a	def test_create_fail(self):
3661	n/a	# 'x' mode fails if file is existing
3662	n/a	with self.open(support.TESTFN, 'w'):
3663	n/a	pass
3664	n/a	self.assertRaises(FileExistsError, self.open, support.TESTFN, 'x')
3665	n/a
3666	n/a	def test_create_writes(self):
3667	n/a	# 'x' mode opens for writing
3668	n/a	with self.open(support.TESTFN, 'xb') as f:
3669	n/a	f.write(b"spam")
3670	n/a	with self.open(support.TESTFN, 'rb') as f:
3671	n/a	self.assertEqual(b"spam", f.read())
3672	n/a
3673	n/a	def test_open_allargs(self):
3674	n/a	# there used to be a buffer overflow in the parser for rawmode
3675	n/a	self.assertRaises(ValueError, self.open, support.TESTFN, 'rwax+')
3676	n/a
3677	n/a
3678	n/a	class CMiscIOTest(MiscIOTest):
3679	n/a	io = io
3680	n/a
3681	n/a	def test_readinto_buffer_overflow(self):
3682	n/a	# Issue #18025
3683	n/a	class BadReader(self.io.BufferedIOBase):
3684	n/a	def read(self, n=-1):
3685	n/a	return b'x' * 10**6
3686	n/a	bufio = BadReader()
3687	n/a	b = bytearray(2)
3688	n/a	self.assertRaises(ValueError, bufio.readinto, b)
3689	n/a
3690	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
3691	n/a	def check_daemon_threads_shutdown_deadlock(self, stream_name):
3692	n/a	# Issue #23309: deadlocks at shutdown should be avoided when a
3693	n/a	# daemon thread and the main thread both write to a file.
3694	n/a	code = """if 1:
3695	n/a	import sys
3696	n/a	import time
3697	n/a	import threading
3698	n/a
3699	n/a	file = sys.{stream_name}
3700	n/a
3701	n/a	def run():
3702	n/a	while True:
3703	n/a	file.write('.')
3704	n/a	file.flush()
3705	n/a
3706	n/a	thread = threading.Thread(target=run)
3707	n/a	thread.daemon = True
3708	n/a	thread.start()
3709	n/a
3710	n/a	time.sleep(0.5)
3711	n/a	file.write('!')
3712	n/a	file.flush()
3713	n/a	""".format_map(locals())
3714	n/a	res, _ = run_python_until_end("-c", code)
3715	n/a	err = res.err.decode()
3716	n/a	if res.rc != 0:
3717	n/a	# Failure: should be a fatal error
3718	n/a	self.assertIn("Fatal Python error: could not acquire lock "
3719	n/a	"for <_io.BufferedWriter name='<{stream_name}>'> "
3720	n/a	"at interpreter shutdown, possibly due to "
3721	n/a	"daemon threads".format_map(locals()),
3722	n/a	err)
3723	n/a	else:
3724	n/a	self.assertFalse(err.strip('.!'))
3725	n/a
3726	n/a	def test_daemon_threads_shutdown_stdout_deadlock(self):
3727	n/a	self.check_daemon_threads_shutdown_deadlock('stdout')
3728	n/a
3729	n/a	def test_daemon_threads_shutdown_stderr_deadlock(self):
3730	n/a	self.check_daemon_threads_shutdown_deadlock('stderr')
3731	n/a
3732	n/a
3733	n/a	class PyMiscIOTest(MiscIOTest):
3734	n/a	io = pyio
3735	n/a
3736	n/a
3737	n/a	@unittest.skipIf(os.name == 'nt', 'POSIX signals required for this test.')
3738	n/a	class SignalsTest(unittest.TestCase):
3739	n/a
3740	n/a	def setUp(self):
3741	n/a	self.oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
3742	n/a
3743	n/a	def tearDown(self):
3744	n/a	signal.signal(signal.SIGALRM, self.oldalrm)
3745	n/a
3746	n/a	def alarm_interrupt(self, sig, frame):
3747	n/a	1/0
3748	n/a
3749	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
3750	n/a	def check_interrupted_write(self, item, bytes, **fdopen_kwargs):
3751	n/a	"""Check that a partial write, when it gets interrupted, properly
3752	n/a	invokes the signal handler, and bubbles up the exception raised
3753	n/a	in the latter."""
3754	n/a	read_results = []
3755	n/a	def _read():
3756	n/a	if hasattr(signal, 'pthread_sigmask'):
3757	n/a	signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGALRM])
3758	n/a	s = os.read(r, 1)
3759	n/a	read_results.append(s)
3760	n/a	t = threading.Thread(target=_read)
3761	n/a	t.daemon = True
3762	n/a	r, w = os.pipe()
3763	n/a	fdopen_kwargs["closefd"] = False
3764	n/a	large_data = item * (support.PIPE_MAX_SIZE // len(item) + 1)
3765	n/a	try:
3766	n/a	wio = self.io.open(w, **fdopen_kwargs)
3767	n/a	t.start()
3768	n/a	# Fill the pipe enough that the write will be blocking.
3769	n/a	# It will be interrupted by the timer armed above. Since the
3770	n/a	# other thread has read one byte, the low-level write will
3771	n/a	# return with a successful (partial) result rather than an EINTR.
3772	n/a	# The buffered IO layer must check for pending signal
3773	n/a	# handlers, which in this case will invoke alarm_interrupt().
3774	n/a	signal.alarm(1)
3775	n/a	try:
3776	n/a	self.assertRaises(ZeroDivisionError, wio.write, large_data)
3777	n/a	finally:
3778	n/a	signal.alarm(0)
3779	n/a	t.join()
3780	n/a	# We got one byte, get another one and check that it isn't a
3781	n/a	# repeat of the first one.
3782	n/a	read_results.append(os.read(r, 1))
3783	n/a	self.assertEqual(read_results, [bytes[0:1], bytes[1:2]])
3784	n/a	finally:
3785	n/a	os.close(w)
3786	n/a	os.close(r)
3787	n/a	# This is deliberate. If we didn't close the file descriptor
3788	n/a	# before closing wio, wio would try to flush its internal
3789	n/a	# buffer, and block again.
3790	n/a	try:
3791	n/a	wio.close()
3792	n/a	except OSError as e:
3793	n/a	if e.errno != errno.EBADF:
3794	n/a	raise
3795	n/a
3796	n/a	def test_interrupted_write_unbuffered(self):
3797	n/a	self.check_interrupted_write(b"xy", b"xy", mode="wb", buffering=0)
3798	n/a
3799	n/a	def test_interrupted_write_buffered(self):
3800	n/a	self.check_interrupted_write(b"xy", b"xy", mode="wb")
3801	n/a
3802	n/a	# Issue #22331: The test hangs on FreeBSD 7.2
3803	n/a	@support.requires_freebsd_version(8)
3804	n/a	def test_interrupted_write_text(self):
3805	n/a	self.check_interrupted_write("xy", b"xy", mode="w", encoding="ascii")
3806	n/a
3807	n/a	@support.no_tracing
3808	n/a	def check_reentrant_write(self, data, **fdopen_kwargs):
3809	n/a	def on_alarm(*args):
3810	n/a	# Will be called reentrantly from the same thread
3811	n/a	wio.write(data)
3812	n/a	1/0
3813	n/a	signal.signal(signal.SIGALRM, on_alarm)
3814	n/a	r, w = os.pipe()
3815	n/a	wio = self.io.open(w, **fdopen_kwargs)
3816	n/a	try:
3817	n/a	signal.alarm(1)
3818	n/a	# Either the reentrant call to wio.write() fails with RuntimeError,
3819	n/a	# or the signal handler raises ZeroDivisionError.
3820	n/a	with self.assertRaises((ZeroDivisionError, RuntimeError)) as cm:
3821	n/a	while 1:
3822	n/a	for i in range(100):
3823	n/a	wio.write(data)
3824	n/a	wio.flush()
3825	n/a	# Make sure the buffer doesn't fill up and block further writes
3826	n/a	os.read(r, len(data) * 100)
3827	n/a	exc = cm.exception
3828	n/a	if isinstance(exc, RuntimeError):
3829	n/a	self.assertTrue(str(exc).startswith("reentrant call"), str(exc))
3830	n/a	finally:
3831	n/a	wio.close()
3832	n/a	os.close(r)
3833	n/a
3834	n/a	def test_reentrant_write_buffered(self):
3835	n/a	self.check_reentrant_write(b"xy", mode="wb")
3836	n/a
3837	n/a	def test_reentrant_write_text(self):
3838	n/a	self.check_reentrant_write("xy", mode="w", encoding="ascii")
3839	n/a
3840	n/a	def check_interrupted_read_retry(self, decode, **fdopen_kwargs):
3841	n/a	"""Check that a buffered read, when it gets interrupted (either
3842	n/a	returning a partial result or EINTR), properly invokes the signal
3843	n/a	handler and retries if the latter returned successfully."""
3844	n/a	r, w = os.pipe()
3845	n/a	fdopen_kwargs["closefd"] = False
3846	n/a	def alarm_handler(sig, frame):
3847	n/a	os.write(w, b"bar")
3848	n/a	signal.signal(signal.SIGALRM, alarm_handler)
3849	n/a	try:
3850	n/a	rio = self.io.open(r, **fdopen_kwargs)
3851	n/a	os.write(w, b"foo")
3852	n/a	signal.alarm(1)
3853	n/a	# Expected behaviour:
3854	n/a	# - first raw read() returns partial b"foo"
3855	n/a	# - second raw read() returns EINTR
3856	n/a	# - third raw read() returns b"bar"
3857	n/a	self.assertEqual(decode(rio.read(6)), "foobar")
3858	n/a	finally:
3859	n/a	rio.close()
3860	n/a	os.close(w)
3861	n/a	os.close(r)
3862	n/a
3863	n/a	def test_interrupted_read_retry_buffered(self):
3864	n/a	self.check_interrupted_read_retry(lambda x: x.decode('latin1'),
3865	n/a	mode="rb")
3866	n/a
3867	n/a	def test_interrupted_read_retry_text(self):
3868	n/a	self.check_interrupted_read_retry(lambda x: x,
3869	n/a	mode="r")
3870	n/a
3871	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
3872	n/a	def check_interrupted_write_retry(self, item, **fdopen_kwargs):
3873	n/a	"""Check that a buffered write, when it gets interrupted (either
3874	n/a	returning a partial result or EINTR), properly invokes the signal
3875	n/a	handler and retries if the latter returned successfully."""
3876	n/a	select = support.import_module("select")
3877	n/a
3878	n/a	# A quantity that exceeds the buffer size of an anonymous pipe's
3879	n/a	# write end.
3880	n/a	N = support.PIPE_MAX_SIZE
3881	n/a	r, w = os.pipe()
3882	n/a	fdopen_kwargs["closefd"] = False
3883	n/a
3884	n/a	# We need a separate thread to read from the pipe and allow the
3885	n/a	# write() to finish. This thread is started after the SIGALRM is
3886	n/a	# received (forcing a first EINTR in write()).
3887	n/a	read_results = []
3888	n/a	write_finished = False
3889	n/a	error = None
3890	n/a	def _read():
3891	n/a	try:
3892	n/a	while not write_finished:
3893	n/a	while r in select.select([r], [], [], 1.0)[0]:
3894	n/a	s = os.read(r, 1024)
3895	n/a	read_results.append(s)
3896	n/a	except BaseException as exc:
3897	n/a	nonlocal error
3898	n/a	error = exc
3899	n/a	t = threading.Thread(target=_read)
3900	n/a	t.daemon = True
3901	n/a	def alarm1(sig, frame):
3902	n/a	signal.signal(signal.SIGALRM, alarm2)
3903	n/a	signal.alarm(1)
3904	n/a	def alarm2(sig, frame):
3905	n/a	t.start()
3906	n/a
3907	n/a	large_data = item * N
3908	n/a	signal.signal(signal.SIGALRM, alarm1)
3909	n/a	try:
3910	n/a	wio = self.io.open(w, **fdopen_kwargs)
3911	n/a	signal.alarm(1)
3912	n/a	# Expected behaviour:
3913	n/a	# - first raw write() is partial (because of the limited pipe buffer
3914	n/a	# and the first alarm)
3915	n/a	# - second raw write() returns EINTR (because of the second alarm)
3916	n/a	# - subsequent write()s are successful (either partial or complete)
3917	n/a	written = wio.write(large_data)
3918	n/a	self.assertEqual(N, written)
3919	n/a
3920	n/a	wio.flush()
3921	n/a	write_finished = True
3922	n/a	t.join()
3923	n/a
3924	n/a	self.assertIsNone(error)
3925	n/a	self.assertEqual(N, sum(len(x) for x in read_results))
3926	n/a	finally:
3927	n/a	write_finished = True
3928	n/a	os.close(w)
3929	n/a	os.close(r)
3930	n/a	# This is deliberate. If we didn't close the file descriptor
3931	n/a	# before closing wio, wio would try to flush its internal
3932	n/a	# buffer, and could block (in case of failure).
3933	n/a	try:
3934	n/a	wio.close()
3935	n/a	except OSError as e:
3936	n/a	if e.errno != errno.EBADF:
3937	n/a	raise
3938	n/a
3939	n/a	def test_interrupted_write_retry_buffered(self):
3940	n/a	self.check_interrupted_write_retry(b"x", mode="wb")
3941	n/a
3942	n/a	def test_interrupted_write_retry_text(self):
3943	n/a	self.check_interrupted_write_retry("x", mode="w", encoding="latin1")
3944	n/a
3945	n/a
3946	n/a	class CSignalsTest(SignalsTest):
3947	n/a	io = io
3948	n/a
3949	n/a	class PySignalsTest(SignalsTest):
3950	n/a	io = pyio
3951	n/a
3952	n/a	# Handling reentrancy issues would slow down _pyio even more, so the
3953	n/a	# tests are disabled.
3954	n/a	test_reentrant_write_buffered = None
3955	n/a	test_reentrant_write_text = None
3956	n/a
3957	n/a
3958	n/a	def load_tests(*args):
3959	n/a	tests = (CIOTest, PyIOTest, APIMismatchTest,
3960	n/a	CBufferedReaderTest, PyBufferedReaderTest,
3961	n/a	CBufferedWriterTest, PyBufferedWriterTest,
3962	n/a	CBufferedRWPairTest, PyBufferedRWPairTest,
3963	n/a	CBufferedRandomTest, PyBufferedRandomTest,
3964	n/a	StatefulIncrementalDecoderTest,
3965	n/a	CIncrementalNewlineDecoderTest, PyIncrementalNewlineDecoderTest,
3966	n/a	CTextIOWrapperTest, PyTextIOWrapperTest,
3967	n/a	CMiscIOTest, PyMiscIOTest,
3968	n/a	CSignalsTest, PySignalsTest,
3969	n/a	)
3970	n/a
3971	n/a	# Put the namespaces of the IO module we are testing and some useful mock
3972	n/a	# classes in the __dict__ of each test.
3973	n/a	mocks = (MockRawIO, MisbehavedRawIO, MockFileIO, CloseFailureIO,
3974	n/a	MockNonBlockWriterIO, MockUnseekableIO, MockRawIOWithoutRead)
3975	n/a	all_members = io.__all__ + ["IncrementalNewlineDecoder"]
3976	n/a	c_io_ns = {name : getattr(io, name) for name in all_members}
3977	n/a	py_io_ns = {name : getattr(pyio, name) for name in all_members}
3978	n/a	globs = globals()
3979	n/a	c_io_ns.update((x.__name__, globs["C" + x.__name__]) for x in mocks)
3980	n/a	py_io_ns.update((x.__name__, globs["Py" + x.__name__]) for x in mocks)
3981	n/a	# Avoid turning open into a bound method.
3982	n/a	py_io_ns["open"] = pyio.OpenWrapper
3983	n/a	for test in tests:
3984	n/a	if test.__name__.startswith("C"):
3985	n/a	for name, obj in c_io_ns.items():
3986	n/a	setattr(test, name, obj)
3987	n/a	elif test.__name__.startswith("Py"):
3988	n/a	for name, obj in py_io_ns.items():
3989	n/a	setattr(test, name, obj)
3990	n/a
3991	n/a	suite = unittest.TestSuite([unittest.makeSuite(test) for test in tests])
3992	n/a	return suite
3993	n/a
3994	n/a	if __name__ == "__main__":
3995	n/a	unittest.main()