Python code coverage for Lib/test/test_fileinput.py

#	count	content
1	n/a	'''
2	n/a	Tests for fileinput module.
3	n/a	Nick Mathewson
4	n/a	'''
5	n/a	import os
6	n/a	import sys
7	n/a	import re
8	n/a	import fileinput
9	n/a	import collections
10	n/a	import builtins
11	n/a	import unittest
12	n/a
13	n/a	try:
14	n/a	import bz2
15	n/a	except ImportError:
16	n/a	bz2 = None
17	n/a	try:
18	n/a	import gzip
19	n/a	except ImportError:
20	n/a	gzip = None
21	n/a
22	n/a	from io import BytesIO, StringIO
23	n/a	from fileinput import FileInput, hook_encoded
24	n/a
25	n/a	from test.support import verbose, TESTFN, check_warnings
26	n/a	from test.support import unlink as safe_unlink
27	n/a	from test import support
28	n/a	from unittest import mock
29	n/a
30	n/a
31	n/a	# The fileinput module has 2 interfaces: the FileInput class which does
32	n/a	# all the work, and a few functions (input, etc.) that use a global _state
33	n/a	# variable.
34	n/a
35	n/a	# Write lines (a list of lines) to temp file number i, and return the
36	n/a	# temp file's name.
37	n/a	def writeTmp(i, lines, mode='w'): # opening in text mode is the default
38	n/a	name = TESTFN + str(i)
39	n/a	f = open(name, mode)
40	n/a	for line in lines:
41	n/a	f.write(line)
42	n/a	f.close()
43	n/a	return name
44	n/a
45	n/a	def remove_tempfiles(*names):
46	n/a	for name in names:
47	n/a	if name:
48	n/a	safe_unlink(name)
49	n/a
50	n/a	class LineReader:
51	n/a
52	n/a	def __init__(self):
53	n/a	self._linesread = []
54	n/a
55	n/a	@property
56	n/a	def linesread(self):
57	n/a	try:
58	n/a	return self._linesread[:]
59	n/a	finally:
60	n/a	self._linesread = []
61	n/a
62	n/a	def openhook(self, filename, mode):
63	n/a	self.it = iter(filename.splitlines(True))
64	n/a	return self
65	n/a
66	n/a	def readline(self, size=None):
67	n/a	line = next(self.it, '')
68	n/a	self._linesread.append(line)
69	n/a	return line
70	n/a
71	n/a	def readlines(self, hint=-1):
72	n/a	lines = []
73	n/a	size = 0
74	n/a	while True:
75	n/a	line = self.readline()
76	n/a	if not line:
77	n/a	return lines
78	n/a	lines.append(line)
79	n/a	size += len(line)
80	n/a	if size >= hint:
81	n/a	return lines
82	n/a
83	n/a	def close(self):
84	n/a	pass
85	n/a
86	n/a	class BufferSizesTests(unittest.TestCase):
87	n/a	def test_buffer_sizes(self):
88	n/a	# First, run the tests with default and teeny buffer size.
89	n/a	for round, bs in (0, 0), (1, 30):
90	n/a	t1 = t2 = t3 = t4 = None
91	n/a	try:
92	n/a	t1 = writeTmp(1, ["Line %s of file 1\n" % (i+1) for i in range(15)])
93	n/a	t2 = writeTmp(2, ["Line %s of file 2\n" % (i+1) for i in range(10)])
94	n/a	t3 = writeTmp(3, ["Line %s of file 3\n" % (i+1) for i in range(5)])
95	n/a	t4 = writeTmp(4, ["Line %s of file 4\n" % (i+1) for i in range(1)])
96	n/a	if bs:
97	n/a	with self.assertWarns(DeprecationWarning):
98	n/a	self.buffer_size_test(t1, t2, t3, t4, bs, round)
99	n/a	else:
100	n/a	self.buffer_size_test(t1, t2, t3, t4, bs, round)
101	n/a	finally:
102	n/a	remove_tempfiles(t1, t2, t3, t4)
103	n/a
104	n/a	def buffer_size_test(self, t1, t2, t3, t4, bs=0, round=0):
105	n/a	pat = re.compile(r'LINE (\d+) OF FILE (\d+)')
106	n/a
107	n/a	start = 1 + round*6
108	n/a	if verbose:
109	n/a	print('%s. Simple iteration (bs=%s)' % (start+0, bs))
110	n/a	fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
111	n/a	lines = list(fi)
112	n/a	fi.close()
113	n/a	self.assertEqual(len(lines), 31)
114	n/a	self.assertEqual(lines[4], 'Line 5 of file 1\n')
115	n/a	self.assertEqual(lines[30], 'Line 1 of file 4\n')
116	n/a	self.assertEqual(fi.lineno(), 31)
117	n/a	self.assertEqual(fi.filename(), t4)
118	n/a
119	n/a	if verbose:
120	n/a	print('%s. Status variables (bs=%s)' % (start+1, bs))
121	n/a	fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
122	n/a	s = "x"
123	n/a	while s and s != 'Line 6 of file 2\n':
124	n/a	s = fi.readline()
125	n/a	self.assertEqual(fi.filename(), t2)
126	n/a	self.assertEqual(fi.lineno(), 21)
127	n/a	self.assertEqual(fi.filelineno(), 6)
128	n/a	self.assertFalse(fi.isfirstline())
129	n/a	self.assertFalse(fi.isstdin())
130	n/a
131	n/a	if verbose:
132	n/a	print('%s. Nextfile (bs=%s)' % (start+2, bs))
133	n/a	fi.nextfile()
134	n/a	self.assertEqual(fi.readline(), 'Line 1 of file 3\n')
135	n/a	self.assertEqual(fi.lineno(), 22)
136	n/a	fi.close()
137	n/a
138	n/a	if verbose:
139	n/a	print('%s. Stdin (bs=%s)' % (start+3, bs))
140	n/a	fi = FileInput(files=(t1, t2, t3, t4, '-'), bufsize=bs)
141	n/a	savestdin = sys.stdin
142	n/a	try:
143	n/a	sys.stdin = StringIO("Line 1 of stdin\nLine 2 of stdin\n")
144	n/a	lines = list(fi)
145	n/a	self.assertEqual(len(lines), 33)
146	n/a	self.assertEqual(lines[32], 'Line 2 of stdin\n')
147	n/a	self.assertEqual(fi.filename(), '<stdin>')
148	n/a	fi.nextfile()
149	n/a	finally:
150	n/a	sys.stdin = savestdin
151	n/a
152	n/a	if verbose:
153	n/a	print('%s. Boundary conditions (bs=%s)' % (start+4, bs))
154	n/a	fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
155	n/a	self.assertEqual(fi.lineno(), 0)
156	n/a	self.assertEqual(fi.filename(), None)
157	n/a	fi.nextfile()
158	n/a	self.assertEqual(fi.lineno(), 0)
159	n/a	self.assertEqual(fi.filename(), None)
160	n/a
161	n/a	if verbose:
162	n/a	print('%s. Inplace (bs=%s)' % (start+5, bs))
163	n/a	savestdout = sys.stdout
164	n/a	try:
165	n/a	fi = FileInput(files=(t1, t2, t3, t4), inplace=1, bufsize=bs)
166	n/a	for line in fi:
167	n/a	line = line[:-1].upper()
168	n/a	print(line)
169	n/a	fi.close()
170	n/a	finally:
171	n/a	sys.stdout = savestdout
172	n/a
173	n/a	fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
174	n/a	for line in fi:
175	n/a	self.assertEqual(line[-1], '\n')
176	n/a	m = pat.match(line[:-1])
177	n/a	self.assertNotEqual(m, None)
178	n/a	self.assertEqual(int(m.group(1)), fi.filelineno())
179	n/a	fi.close()
180	n/a
181	n/a	class UnconditionallyRaise:
182	n/a	def __init__(self, exception_type):
183	n/a	self.exception_type = exception_type
184	n/a	self.invoked = False
185	n/a	def __call__(self, args, *kwargs):
186	n/a	self.invoked = True
187	n/a	raise self.exception_type()
188	n/a
189	n/a	class FileInputTests(unittest.TestCase):
190	n/a
191	n/a	def test_zero_byte_files(self):
192	n/a	t1 = t2 = t3 = t4 = None
193	n/a	try:
194	n/a	t1 = writeTmp(1, [""])
195	n/a	t2 = writeTmp(2, [""])
196	n/a	t3 = writeTmp(3, ["The only line there is.\n"])
197	n/a	t4 = writeTmp(4, [""])
198	n/a	fi = FileInput(files=(t1, t2, t3, t4))
199	n/a
200	n/a	line = fi.readline()
201	n/a	self.assertEqual(line, 'The only line there is.\n')
202	n/a	self.assertEqual(fi.lineno(), 1)
203	n/a	self.assertEqual(fi.filelineno(), 1)
204	n/a	self.assertEqual(fi.filename(), t3)
205	n/a
206	n/a	line = fi.readline()
207	n/a	self.assertFalse(line)
208	n/a	self.assertEqual(fi.lineno(), 1)
209	n/a	self.assertEqual(fi.filelineno(), 0)
210	n/a	self.assertEqual(fi.filename(), t4)
211	n/a	fi.close()
212	n/a	finally:
213	n/a	remove_tempfiles(t1, t2, t3, t4)
214	n/a
215	n/a	def test_files_that_dont_end_with_newline(self):
216	n/a	t1 = t2 = None
217	n/a	try:
218	n/a	t1 = writeTmp(1, ["A\nB\nC"])
219	n/a	t2 = writeTmp(2, ["D\nE\nF"])
220	n/a	fi = FileInput(files=(t1, t2))
221	n/a	lines = list(fi)
222	n/a	self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"])
223	n/a	self.assertEqual(fi.filelineno(), 3)
224	n/a	self.assertEqual(fi.lineno(), 6)
225	n/a	finally:
226	n/a	remove_tempfiles(t1, t2)
227	n/a
228	n/a	## def test_unicode_filenames(self):
229	n/a	## # XXX A unicode string is always returned by writeTmp.
230	n/a	## # So is this needed?
231	n/a	## try:
232	n/a	## t1 = writeTmp(1, ["A\nB"])
233	n/a	## encoding = sys.getfilesystemencoding()
234	n/a	## if encoding is None:
235	n/a	## encoding = 'ascii'
236	n/a	## fi = FileInput(files=str(t1, encoding))
237	n/a	## lines = list(fi)
238	n/a	## self.assertEqual(lines, ["A\n", "B"])
239	n/a	## finally:
240	n/a	## remove_tempfiles(t1)
241	n/a
242	n/a	def test_fileno(self):
243	n/a	t1 = t2 = None
244	n/a	try:
245	n/a	t1 = writeTmp(1, ["A\nB"])
246	n/a	t2 = writeTmp(2, ["C\nD"])
247	n/a	fi = FileInput(files=(t1, t2))
248	n/a	self.assertEqual(fi.fileno(), -1)
249	n/a	line =next( fi)
250	n/a	self.assertNotEqual(fi.fileno(), -1)
251	n/a	fi.nextfile()
252	n/a	self.assertEqual(fi.fileno(), -1)
253	n/a	line = list(fi)
254	n/a	self.assertEqual(fi.fileno(), -1)
255	n/a	finally:
256	n/a	remove_tempfiles(t1, t2)
257	n/a
258	n/a	def test_opening_mode(self):
259	n/a	try:
260	n/a	# invalid mode, should raise ValueError
261	n/a	fi = FileInput(mode="w")
262	n/a	self.fail("FileInput should reject invalid mode argument")
263	n/a	except ValueError:
264	n/a	pass
265	n/a	t1 = None
266	n/a	try:
267	n/a	# try opening in universal newline mode
268	n/a	t1 = writeTmp(1, [b"A\nB\r\nC\rD"], mode="wb")
269	n/a	with check_warnings(('', DeprecationWarning)):
270	n/a	fi = FileInput(files=t1, mode="U")
271	n/a	with check_warnings(('', DeprecationWarning)):
272	n/a	lines = list(fi)
273	n/a	self.assertEqual(lines, ["A\n", "B\n", "C\n", "D"])
274	n/a	finally:
275	n/a	remove_tempfiles(t1)
276	n/a
277	n/a	def test_stdin_binary_mode(self):
278	n/a	with mock.patch('sys.stdin') as m_stdin:
279	n/a	m_stdin.buffer = BytesIO(b'spam, bacon, sausage, and spam')
280	n/a	fi = FileInput(files=['-'], mode='rb')
281	n/a	lines = list(fi)
282	n/a	self.assertEqual(lines, [b'spam, bacon, sausage, and spam'])
283	n/a
284	n/a	def test_detached_stdin_binary_mode(self):
285	n/a	orig_stdin = sys.stdin
286	n/a	try:
287	n/a	sys.stdin = BytesIO(b'spam, bacon, sausage, and spam')
288	n/a	self.assertFalse(hasattr(sys.stdin, 'buffer'))
289	n/a	fi = FileInput(files=['-'], mode='rb')
290	n/a	lines = list(fi)
291	n/a	self.assertEqual(lines, [b'spam, bacon, sausage, and spam'])
292	n/a	finally:
293	n/a	sys.stdin = orig_stdin
294	n/a
295	n/a	def test_file_opening_hook(self):
296	n/a	try:
297	n/a	# cannot use openhook and inplace mode
298	n/a	fi = FileInput(inplace=1, openhook=lambda f, m: None)
299	n/a	self.fail("FileInput should raise if both inplace "
300	n/a	"and openhook arguments are given")
301	n/a	except ValueError:
302	n/a	pass
303	n/a	try:
304	n/a	fi = FileInput(openhook=1)
305	n/a	self.fail("FileInput should check openhook for being callable")
306	n/a	except ValueError:
307	n/a	pass
308	n/a
309	n/a	class CustomOpenHook:
310	n/a	def __init__(self):
311	n/a	self.invoked = False
312	n/a	def __call__(self, *args):
313	n/a	self.invoked = True
314	n/a	return open(*args)
315	n/a
316	n/a	t = writeTmp(1, ["\n"])
317	n/a	self.addCleanup(remove_tempfiles, t)
318	n/a	custom_open_hook = CustomOpenHook()
319	n/a	with FileInput([t], openhook=custom_open_hook) as fi:
320	n/a	fi.readline()
321	n/a	self.assertTrue(custom_open_hook.invoked, "openhook not invoked")
322	n/a
323	n/a	def test_readline(self):
324	n/a	with open(TESTFN, 'wb') as f:
325	n/a	f.write(b'A\nB\r\nC\r')
326	n/a	# Fill TextIOWrapper buffer.
327	n/a	f.write(b'123456789\n' * 1000)
328	n/a	# Issue #20501: readline() shouldn't read whole file.
329	n/a	f.write(b'\x80')
330	n/a	self.addCleanup(safe_unlink, TESTFN)
331	n/a
332	n/a	with FileInput(files=TESTFN,
333	n/a	openhook=hook_encoded('ascii')) as fi:
334	n/a	try:
335	n/a	self.assertEqual(fi.readline(), 'A\n')
336	n/a	self.assertEqual(fi.readline(), 'B\n')
337	n/a	self.assertEqual(fi.readline(), 'C\n')
338	n/a	except UnicodeDecodeError:
339	n/a	self.fail('Read to end of file')
340	n/a	with self.assertRaises(UnicodeDecodeError):
341	n/a	# Read to the end of file.
342	n/a	list(fi)
343	n/a	self.assertEqual(fi.readline(), '')
344	n/a	self.assertEqual(fi.readline(), '')
345	n/a
346	n/a	def test_readline_binary_mode(self):
347	n/a	with open(TESTFN, 'wb') as f:
348	n/a	f.write(b'A\nB\r\nC\rD')
349	n/a	self.addCleanup(safe_unlink, TESTFN)
350	n/a
351	n/a	with FileInput(files=TESTFN, mode='rb') as fi:
352	n/a	self.assertEqual(fi.readline(), b'A\n')
353	n/a	self.assertEqual(fi.readline(), b'B\r\n')
354	n/a	self.assertEqual(fi.readline(), b'C\rD')
355	n/a	# Read to the end of file.
356	n/a	self.assertEqual(fi.readline(), b'')
357	n/a	self.assertEqual(fi.readline(), b'')
358	n/a
359	n/a	def test_context_manager(self):
360	n/a	try:
361	n/a	t1 = writeTmp(1, ["A\nB\nC"])
362	n/a	t2 = writeTmp(2, ["D\nE\nF"])
363	n/a	with FileInput(files=(t1, t2)) as fi:
364	n/a	lines = list(fi)
365	n/a	self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"])
366	n/a	self.assertEqual(fi.filelineno(), 3)
367	n/a	self.assertEqual(fi.lineno(), 6)
368	n/a	self.assertEqual(fi._files, ())
369	n/a	finally:
370	n/a	remove_tempfiles(t1, t2)
371	n/a
372	n/a	def test_close_on_exception(self):
373	n/a	try:
374	n/a	t1 = writeTmp(1, [""])
375	n/a	with FileInput(files=t1) as fi:
376	n/a	raise OSError
377	n/a	except OSError:
378	n/a	self.assertEqual(fi._files, ())
379	n/a	finally:
380	n/a	remove_tempfiles(t1)
381	n/a
382	n/a	def test_empty_files_list_specified_to_constructor(self):
383	n/a	with FileInput(files=[]) as fi:
384	n/a	self.assertEqual(fi._files, ('-',))
385	n/a
386	n/a	def test__getitem__(self):
387	n/a	"""Tests invoking FileInput.__getitem__() with the current
388	n/a	line number"""
389	n/a	t = writeTmp(1, ["line1\n", "line2\n"])
390	n/a	self.addCleanup(remove_tempfiles, t)
391	n/a	with FileInput(files=[t]) as fi:
392	n/a	retval1 = fi[0]
393	n/a	self.assertEqual(retval1, "line1\n")
394	n/a	retval2 = fi[1]
395	n/a	self.assertEqual(retval2, "line2\n")
396	n/a
397	n/a	def test__getitem__invalid_key(self):
398	n/a	"""Tests invoking FileInput.__getitem__() with an index unequal to
399	n/a	the line number"""
400	n/a	t = writeTmp(1, ["line1\n", "line2\n"])
401	n/a	self.addCleanup(remove_tempfiles, t)
402	n/a	with FileInput(files=[t]) as fi:
403	n/a	with self.assertRaises(RuntimeError) as cm:
404	n/a	fi[1]
405	n/a	self.assertEqual(cm.exception.args, ("accessing lines out of order",))
406	n/a
407	n/a	def test__getitem__eof(self):
408	n/a	"""Tests invoking FileInput.__getitem__() with the line number but at
409	n/a	end-of-input"""
410	n/a	t = writeTmp(1, [])
411	n/a	self.addCleanup(remove_tempfiles, t)
412	n/a	with FileInput(files=[t]) as fi:
413	n/a	with self.assertRaises(IndexError) as cm:
414	n/a	fi[0]
415	n/a	self.assertEqual(cm.exception.args, ("end of input reached",))
416	n/a
417	n/a	def test_nextfile_oserror_deleting_backup(self):
418	n/a	"""Tests invoking FileInput.nextfile() when the attempt to delete
419	n/a	the backup file would raise OSError. This error is expected to be
420	n/a	silently ignored"""
421	n/a
422	n/a	os_unlink_orig = os.unlink
423	n/a	os_unlink_replacement = UnconditionallyRaise(OSError)
424	n/a	try:
425	n/a	t = writeTmp(1, ["\n"])
426	n/a	self.addCleanup(remove_tempfiles, t)
427	n/a	with FileInput(files=[t], inplace=True) as fi:
428	n/a	next(fi) # make sure the file is opened
429	n/a	os.unlink = os_unlink_replacement
430	n/a	fi.nextfile()
431	n/a	finally:
432	n/a	os.unlink = os_unlink_orig
433	n/a
434	n/a	# sanity check to make sure that our test scenario was actually hit
435	n/a	self.assertTrue(os_unlink_replacement.invoked,
436	n/a	"os.unlink() was not invoked")
437	n/a
438	n/a	def test_readline_os_fstat_raises_OSError(self):
439	n/a	"""Tests invoking FileInput.readline() when os.fstat() raises OSError.
440	n/a	This exception should be silently discarded."""
441	n/a
442	n/a	os_fstat_orig = os.fstat
443	n/a	os_fstat_replacement = UnconditionallyRaise(OSError)
444	n/a	try:
445	n/a	t = writeTmp(1, ["\n"])
446	n/a	self.addCleanup(remove_tempfiles, t)
447	n/a	with FileInput(files=[t], inplace=True) as fi:
448	n/a	os.fstat = os_fstat_replacement
449	n/a	fi.readline()
450	n/a	finally:
451	n/a	os.fstat = os_fstat_orig
452	n/a
453	n/a	# sanity check to make sure that our test scenario was actually hit
454	n/a	self.assertTrue(os_fstat_replacement.invoked,
455	n/a	"os.fstat() was not invoked")
456	n/a
457	n/a	@unittest.skipIf(not hasattr(os, "chmod"), "os.chmod does not exist")
458	n/a	def test_readline_os_chmod_raises_OSError(self):
459	n/a	"""Tests invoking FileInput.readline() when os.chmod() raises OSError.
460	n/a	This exception should be silently discarded."""
461	n/a
462	n/a	os_chmod_orig = os.chmod
463	n/a	os_chmod_replacement = UnconditionallyRaise(OSError)
464	n/a	try:
465	n/a	t = writeTmp(1, ["\n"])
466	n/a	self.addCleanup(remove_tempfiles, t)
467	n/a	with FileInput(files=[t], inplace=True) as fi:
468	n/a	os.chmod = os_chmod_replacement
469	n/a	fi.readline()
470	n/a	finally:
471	n/a	os.chmod = os_chmod_orig
472	n/a
473	n/a	# sanity check to make sure that our test scenario was actually hit
474	n/a	self.assertTrue(os_chmod_replacement.invoked,
475	n/a	"os.fstat() was not invoked")
476	n/a
477	n/a	def test_fileno_when_ValueError_raised(self):
478	n/a	class FilenoRaisesValueError(UnconditionallyRaise):
479	n/a	def __init__(self):
480	n/a	UnconditionallyRaise.__init__(self, ValueError)
481	n/a	def fileno(self):
482	n/a	self.__call__()
483	n/a
484	n/a	unconditionally_raise_ValueError = FilenoRaisesValueError()
485	n/a	t = writeTmp(1, ["\n"])
486	n/a	self.addCleanup(remove_tempfiles, t)
487	n/a	with FileInput(files=[t]) as fi:
488	n/a	file_backup = fi._file
489	n/a	try:
490	n/a	fi._file = unconditionally_raise_ValueError
491	n/a	result = fi.fileno()
492	n/a	finally:
493	n/a	fi._file = file_backup # make sure the file gets cleaned up
494	n/a
495	n/a	# sanity check to make sure that our test scenario was actually hit
496	n/a	self.assertTrue(unconditionally_raise_ValueError.invoked,
497	n/a	"_file.fileno() was not invoked")
498	n/a
499	n/a	self.assertEqual(result, -1, "fileno() should return -1")
500	n/a
501	n/a	def test_readline_buffering(self):
502	n/a	src = LineReader()
503	n/a	with FileInput(files=['line1\nline2', 'line3\n'],
504	n/a	openhook=src.openhook) as fi:
505	n/a	self.assertEqual(src.linesread, [])
506	n/a	self.assertEqual(fi.readline(), 'line1\n')
507	n/a	self.assertEqual(src.linesread, ['line1\n'])
508	n/a	self.assertEqual(fi.readline(), 'line2')
509	n/a	self.assertEqual(src.linesread, ['line2'])
510	n/a	self.assertEqual(fi.readline(), 'line3\n')
511	n/a	self.assertEqual(src.linesread, ['', 'line3\n'])
512	n/a	self.assertEqual(fi.readline(), '')
513	n/a	self.assertEqual(src.linesread, [''])
514	n/a	self.assertEqual(fi.readline(), '')
515	n/a	self.assertEqual(src.linesread, [])
516	n/a
517	n/a	def test_iteration_buffering(self):
518	n/a	src = LineReader()
519	n/a	with FileInput(files=['line1\nline2', 'line3\n'],
520	n/a	openhook=src.openhook) as fi:
521	n/a	self.assertEqual(src.linesread, [])
522	n/a	self.assertEqual(next(fi), 'line1\n')
523	n/a	self.assertEqual(src.linesread, ['line1\n'])
524	n/a	self.assertEqual(next(fi), 'line2')
525	n/a	self.assertEqual(src.linesread, ['line2'])
526	n/a	self.assertEqual(next(fi), 'line3\n')
527	n/a	self.assertEqual(src.linesread, ['', 'line3\n'])
528	n/a	self.assertRaises(StopIteration, next, fi)
529	n/a	self.assertEqual(src.linesread, [''])
530	n/a	self.assertRaises(StopIteration, next, fi)
531	n/a	self.assertEqual(src.linesread, [])
532	n/a
533	n/a	class MockFileInput:
534	n/a	"""A class that mocks out fileinput.FileInput for use during unit tests"""
535	n/a
536	n/a	def __init__(self, files=None, inplace=False, backup="", bufsize=0,
537	n/a	mode="r", openhook=None):
538	n/a	self.files = files
539	n/a	self.inplace = inplace
540	n/a	self.backup = backup
541	n/a	self.bufsize = bufsize
542	n/a	self.mode = mode
543	n/a	self.openhook = openhook
544	n/a	self._file = None
545	n/a	self.invocation_counts = collections.defaultdict(lambda: 0)
546	n/a	self.return_values = {}
547	n/a
548	n/a	def close(self):
549	n/a	self.invocation_counts["close"] += 1
550	n/a
551	n/a	def nextfile(self):
552	n/a	self.invocation_counts["nextfile"] += 1
553	n/a	return self.return_values["nextfile"]
554	n/a
555	n/a	def filename(self):
556	n/a	self.invocation_counts["filename"] += 1
557	n/a	return self.return_values["filename"]
558	n/a
559	n/a	def lineno(self):
560	n/a	self.invocation_counts["lineno"] += 1
561	n/a	return self.return_values["lineno"]
562	n/a
563	n/a	def filelineno(self):
564	n/a	self.invocation_counts["filelineno"] += 1
565	n/a	return self.return_values["filelineno"]
566	n/a
567	n/a	def fileno(self):
568	n/a	self.invocation_counts["fileno"] += 1
569	n/a	return self.return_values["fileno"]
570	n/a
571	n/a	def isfirstline(self):
572	n/a	self.invocation_counts["isfirstline"] += 1
573	n/a	return self.return_values["isfirstline"]
574	n/a
575	n/a	def isstdin(self):
576	n/a	self.invocation_counts["isstdin"] += 1
577	n/a	return self.return_values["isstdin"]
578	n/a
579	n/a	class BaseFileInputGlobalMethodsTest(unittest.TestCase):
580	n/a	"""Base class for unit tests for the global function of
581	n/a	the fileinput module."""
582	n/a
583	n/a	def setUp(self):
584	n/a	self._orig_state = fileinput._state
585	n/a	self._orig_FileInput = fileinput.FileInput
586	n/a	fileinput.FileInput = MockFileInput
587	n/a
588	n/a	def tearDown(self):
589	n/a	fileinput.FileInput = self._orig_FileInput
590	n/a	fileinput._state = self._orig_state
591	n/a
592	n/a	def assertExactlyOneInvocation(self, mock_file_input, method_name):
593	n/a	# assert that the method with the given name was invoked once
594	n/a	actual_count = mock_file_input.invocation_counts[method_name]
595	n/a	self.assertEqual(actual_count, 1, method_name)
596	n/a	# assert that no other unexpected methods were invoked
597	n/a	actual_total_count = len(mock_file_input.invocation_counts)
598	n/a	self.assertEqual(actual_total_count, 1)
599	n/a
600	n/a	class Test_fileinput_input(BaseFileInputGlobalMethodsTest):
601	n/a	"""Unit tests for fileinput.input()"""
602	n/a
603	n/a	def test_state_is_not_None_and_state_file_is_not_None(self):
604	n/a	"""Tests invoking fileinput.input() when fileinput._state is not None
605	n/a	and its _file attribute is also not None. Expect RuntimeError to
606	n/a	be raised with a meaningful error message and for fileinput._state
607	n/a	to not be modified."""
608	n/a	instance = MockFileInput()
609	n/a	instance._file = object()
610	n/a	fileinput._state = instance
611	n/a	with self.assertRaises(RuntimeError) as cm:
612	n/a	fileinput.input()
613	n/a	self.assertEqual(("input() already active",), cm.exception.args)
614	n/a	self.assertIs(instance, fileinput._state, "fileinput._state")
615	n/a
616	n/a	def test_state_is_not_None_and_state_file_is_None(self):
617	n/a	"""Tests invoking fileinput.input() when fileinput._state is not None
618	n/a	but its _file attribute is None. Expect it to create and return
619	n/a	a new fileinput.FileInput object with all method parameters passed
620	n/a	explicitly to the __init__() method; also ensure that
621	n/a	fileinput._state is set to the returned instance."""
622	n/a	instance = MockFileInput()
623	n/a	instance._file = None
624	n/a	fileinput._state = instance
625	n/a	self.do_test_call_input()
626	n/a
627	n/a	def test_state_is_None(self):
628	n/a	"""Tests invoking fileinput.input() when fileinput._state is None
629	n/a	Expect it to create and return a new fileinput.FileInput object
630	n/a	with all method parameters passed explicitly to the __init__()
631	n/a	method; also ensure that fileinput._state is set to the returned
632	n/a	instance."""
633	n/a	fileinput._state = None
634	n/a	self.do_test_call_input()
635	n/a
636	n/a	def do_test_call_input(self):
637	n/a	"""Tests that fileinput.input() creates a new fileinput.FileInput
638	n/a	object, passing the given parameters unmodified to
639	n/a	fileinput.FileInput.__init__(). Note that this test depends on the
640	n/a	monkey patching of fileinput.FileInput done by setUp()."""
641	n/a	files = object()
642	n/a	inplace = object()
643	n/a	backup = object()
644	n/a	bufsize = object()
645	n/a	mode = object()
646	n/a	openhook = object()
647	n/a
648	n/a	# call fileinput.input() with different values for each argument
649	n/a	result = fileinput.input(files=files, inplace=inplace, backup=backup,
650	n/a	bufsize=bufsize,
651	n/a	mode=mode, openhook=openhook)
652	n/a
653	n/a	# ensure fileinput._state was set to the returned object
654	n/a	self.assertIs(result, fileinput._state, "fileinput._state")
655	n/a
656	n/a	# ensure the parameters to fileinput.input() were passed directly
657	n/a	# to FileInput.__init__()
658	n/a	self.assertIs(files, result.files, "files")
659	n/a	self.assertIs(inplace, result.inplace, "inplace")
660	n/a	self.assertIs(backup, result.backup, "backup")
661	n/a	self.assertIs(bufsize, result.bufsize, "bufsize")
662	n/a	self.assertIs(mode, result.mode, "mode")
663	n/a	self.assertIs(openhook, result.openhook, "openhook")
664	n/a
665	n/a	class Test_fileinput_close(BaseFileInputGlobalMethodsTest):
666	n/a	"""Unit tests for fileinput.close()"""
667	n/a
668	n/a	def test_state_is_None(self):
669	n/a	"""Tests that fileinput.close() does nothing if fileinput._state
670	n/a	is None"""
671	n/a	fileinput._state = None
672	n/a	fileinput.close()
673	n/a	self.assertIsNone(fileinput._state)
674	n/a
675	n/a	def test_state_is_not_None(self):
676	n/a	"""Tests that fileinput.close() invokes close() on fileinput._state
677	n/a	and sets _state=None"""
678	n/a	instance = MockFileInput()
679	n/a	fileinput._state = instance
680	n/a	fileinput.close()
681	n/a	self.assertExactlyOneInvocation(instance, "close")
682	n/a	self.assertIsNone(fileinput._state)
683	n/a
684	n/a	class Test_fileinput_nextfile(BaseFileInputGlobalMethodsTest):
685	n/a	"""Unit tests for fileinput.nextfile()"""
686	n/a
687	n/a	def test_state_is_None(self):
688	n/a	"""Tests fileinput.nextfile() when fileinput._state is None.
689	n/a	Ensure that it raises RuntimeError with a meaningful error message
690	n/a	and does not modify fileinput._state"""
691	n/a	fileinput._state = None
692	n/a	with self.assertRaises(RuntimeError) as cm:
693	n/a	fileinput.nextfile()
694	n/a	self.assertEqual(("no active input()",), cm.exception.args)
695	n/a	self.assertIsNone(fileinput._state)
696	n/a
697	n/a	def test_state_is_not_None(self):
698	n/a	"""Tests fileinput.nextfile() when fileinput._state is not None.
699	n/a	Ensure that it invokes fileinput._state.nextfile() exactly once,
700	n/a	returns whatever it returns, and does not modify fileinput._state
701	n/a	to point to a different object."""
702	n/a	nextfile_retval = object()
703	n/a	instance = MockFileInput()
704	n/a	instance.return_values["nextfile"] = nextfile_retval
705	n/a	fileinput._state = instance
706	n/a	retval = fileinput.nextfile()
707	n/a	self.assertExactlyOneInvocation(instance, "nextfile")
708	n/a	self.assertIs(retval, nextfile_retval)
709	n/a	self.assertIs(fileinput._state, instance)
710	n/a
711	n/a	class Test_fileinput_filename(BaseFileInputGlobalMethodsTest):
712	n/a	"""Unit tests for fileinput.filename()"""
713	n/a
714	n/a	def test_state_is_None(self):
715	n/a	"""Tests fileinput.filename() when fileinput._state is None.
716	n/a	Ensure that it raises RuntimeError with a meaningful error message
717	n/a	and does not modify fileinput._state"""
718	n/a	fileinput._state = None
719	n/a	with self.assertRaises(RuntimeError) as cm:
720	n/a	fileinput.filename()
721	n/a	self.assertEqual(("no active input()",), cm.exception.args)
722	n/a	self.assertIsNone(fileinput._state)
723	n/a
724	n/a	def test_state_is_not_None(self):
725	n/a	"""Tests fileinput.filename() when fileinput._state is not None.
726	n/a	Ensure that it invokes fileinput._state.filename() exactly once,
727	n/a	returns whatever it returns, and does not modify fileinput._state
728	n/a	to point to a different object."""
729	n/a	filename_retval = object()
730	n/a	instance = MockFileInput()
731	n/a	instance.return_values["filename"] = filename_retval
732	n/a	fileinput._state = instance
733	n/a	retval = fileinput.filename()
734	n/a	self.assertExactlyOneInvocation(instance, "filename")
735	n/a	self.assertIs(retval, filename_retval)
736	n/a	self.assertIs(fileinput._state, instance)
737	n/a
738	n/a	class Test_fileinput_lineno(BaseFileInputGlobalMethodsTest):
739	n/a	"""Unit tests for fileinput.lineno()"""
740	n/a
741	n/a	def test_state_is_None(self):
742	n/a	"""Tests fileinput.lineno() when fileinput._state is None.
743	n/a	Ensure that it raises RuntimeError with a meaningful error message
744	n/a	and does not modify fileinput._state"""
745	n/a	fileinput._state = None
746	n/a	with self.assertRaises(RuntimeError) as cm:
747	n/a	fileinput.lineno()
748	n/a	self.assertEqual(("no active input()",), cm.exception.args)
749	n/a	self.assertIsNone(fileinput._state)
750	n/a
751	n/a	def test_state_is_not_None(self):
752	n/a	"""Tests fileinput.lineno() when fileinput._state is not None.
753	n/a	Ensure that it invokes fileinput._state.lineno() exactly once,
754	n/a	returns whatever it returns, and does not modify fileinput._state
755	n/a	to point to a different object."""
756	n/a	lineno_retval = object()
757	n/a	instance = MockFileInput()
758	n/a	instance.return_values["lineno"] = lineno_retval
759	n/a	fileinput._state = instance
760	n/a	retval = fileinput.lineno()
761	n/a	self.assertExactlyOneInvocation(instance, "lineno")
762	n/a	self.assertIs(retval, lineno_retval)
763	n/a	self.assertIs(fileinput._state, instance)
764	n/a
765	n/a	class Test_fileinput_filelineno(BaseFileInputGlobalMethodsTest):
766	n/a	"""Unit tests for fileinput.filelineno()"""
767	n/a
768	n/a	def test_state_is_None(self):
769	n/a	"""Tests fileinput.filelineno() when fileinput._state is None.
770	n/a	Ensure that it raises RuntimeError with a meaningful error message
771	n/a	and does not modify fileinput._state"""
772	n/a	fileinput._state = None
773	n/a	with self.assertRaises(RuntimeError) as cm:
774	n/a	fileinput.filelineno()
775	n/a	self.assertEqual(("no active input()",), cm.exception.args)
776	n/a	self.assertIsNone(fileinput._state)
777	n/a
778	n/a	def test_state_is_not_None(self):
779	n/a	"""Tests fileinput.filelineno() when fileinput._state is not None.
780	n/a	Ensure that it invokes fileinput._state.filelineno() exactly once,
781	n/a	returns whatever it returns, and does not modify fileinput._state
782	n/a	to point to a different object."""
783	n/a	filelineno_retval = object()
784	n/a	instance = MockFileInput()
785	n/a	instance.return_values["filelineno"] = filelineno_retval
786	n/a	fileinput._state = instance
787	n/a	retval = fileinput.filelineno()
788	n/a	self.assertExactlyOneInvocation(instance, "filelineno")
789	n/a	self.assertIs(retval, filelineno_retval)
790	n/a	self.assertIs(fileinput._state, instance)
791	n/a
792	n/a	class Test_fileinput_fileno(BaseFileInputGlobalMethodsTest):
793	n/a	"""Unit tests for fileinput.fileno()"""
794	n/a
795	n/a	def test_state_is_None(self):
796	n/a	"""Tests fileinput.fileno() when fileinput._state is None.
797	n/a	Ensure that it raises RuntimeError with a meaningful error message
798	n/a	and does not modify fileinput._state"""
799	n/a	fileinput._state = None
800	n/a	with self.assertRaises(RuntimeError) as cm:
801	n/a	fileinput.fileno()
802	n/a	self.assertEqual(("no active input()",), cm.exception.args)
803	n/a	self.assertIsNone(fileinput._state)
804	n/a
805	n/a	def test_state_is_not_None(self):
806	n/a	"""Tests fileinput.fileno() when fileinput._state is not None.
807	n/a	Ensure that it invokes fileinput._state.fileno() exactly once,
808	n/a	returns whatever it returns, and does not modify fileinput._state
809	n/a	to point to a different object."""
810	n/a	fileno_retval = object()
811	n/a	instance = MockFileInput()
812	n/a	instance.return_values["fileno"] = fileno_retval
813	n/a	instance.fileno_retval = fileno_retval
814	n/a	fileinput._state = instance
815	n/a	retval = fileinput.fileno()
816	n/a	self.assertExactlyOneInvocation(instance, "fileno")
817	n/a	self.assertIs(retval, fileno_retval)
818	n/a	self.assertIs(fileinput._state, instance)
819	n/a
820	n/a	class Test_fileinput_isfirstline(BaseFileInputGlobalMethodsTest):
821	n/a	"""Unit tests for fileinput.isfirstline()"""
822	n/a
823	n/a	def test_state_is_None(self):
824	n/a	"""Tests fileinput.isfirstline() when fileinput._state is None.
825	n/a	Ensure that it raises RuntimeError with a meaningful error message
826	n/a	and does not modify fileinput._state"""
827	n/a	fileinput._state = None
828	n/a	with self.assertRaises(RuntimeError) as cm:
829	n/a	fileinput.isfirstline()
830	n/a	self.assertEqual(("no active input()",), cm.exception.args)
831	n/a	self.assertIsNone(fileinput._state)
832	n/a
833	n/a	def test_state_is_not_None(self):
834	n/a	"""Tests fileinput.isfirstline() when fileinput._state is not None.
835	n/a	Ensure that it invokes fileinput._state.isfirstline() exactly once,
836	n/a	returns whatever it returns, and does not modify fileinput._state
837	n/a	to point to a different object."""
838	n/a	isfirstline_retval = object()
839	n/a	instance = MockFileInput()
840	n/a	instance.return_values["isfirstline"] = isfirstline_retval
841	n/a	fileinput._state = instance
842	n/a	retval = fileinput.isfirstline()
843	n/a	self.assertExactlyOneInvocation(instance, "isfirstline")
844	n/a	self.assertIs(retval, isfirstline_retval)
845	n/a	self.assertIs(fileinput._state, instance)
846	n/a
847	n/a	class Test_fileinput_isstdin(BaseFileInputGlobalMethodsTest):
848	n/a	"""Unit tests for fileinput.isstdin()"""
849	n/a
850	n/a	def test_state_is_None(self):
851	n/a	"""Tests fileinput.isstdin() when fileinput._state is None.
852	n/a	Ensure that it raises RuntimeError with a meaningful error message
853	n/a	and does not modify fileinput._state"""
854	n/a	fileinput._state = None
855	n/a	with self.assertRaises(RuntimeError) as cm:
856	n/a	fileinput.isstdin()
857	n/a	self.assertEqual(("no active input()",), cm.exception.args)
858	n/a	self.assertIsNone(fileinput._state)
859	n/a
860	n/a	def test_state_is_not_None(self):
861	n/a	"""Tests fileinput.isstdin() when fileinput._state is not None.
862	n/a	Ensure that it invokes fileinput._state.isstdin() exactly once,
863	n/a	returns whatever it returns, and does not modify fileinput._state
864	n/a	to point to a different object."""
865	n/a	isstdin_retval = object()
866	n/a	instance = MockFileInput()
867	n/a	instance.return_values["isstdin"] = isstdin_retval
868	n/a	fileinput._state = instance
869	n/a	retval = fileinput.isstdin()
870	n/a	self.assertExactlyOneInvocation(instance, "isstdin")
871	n/a	self.assertIs(retval, isstdin_retval)
872	n/a	self.assertIs(fileinput._state, instance)
873	n/a
874	n/a	class InvocationRecorder:
875	n/a	def __init__(self):
876	n/a	self.invocation_count = 0
877	n/a	def __call__(self, args, *kwargs):
878	n/a	self.invocation_count += 1
879	n/a	self.last_invocation = (args, kwargs)
880	n/a
881	n/a	class Test_hook_compressed(unittest.TestCase):
882	n/a	"""Unit tests for fileinput.hook_compressed()"""
883	n/a
884	n/a	def setUp(self):
885	n/a	self.fake_open = InvocationRecorder()
886	n/a
887	n/a	def test_empty_string(self):
888	n/a	self.do_test_use_builtin_open("", 1)
889	n/a
890	n/a	def test_no_ext(self):
891	n/a	self.do_test_use_builtin_open("abcd", 2)
892	n/a
893	n/a	@unittest.skipUnless(gzip, "Requires gzip and zlib")
894	n/a	def test_gz_ext_fake(self):
895	n/a	original_open = gzip.open
896	n/a	gzip.open = self.fake_open
897	n/a	try:
898	n/a	result = fileinput.hook_compressed("test.gz", 3)
899	n/a	finally:
900	n/a	gzip.open = original_open
901	n/a
902	n/a	self.assertEqual(self.fake_open.invocation_count, 1)
903	n/a	self.assertEqual(self.fake_open.last_invocation, (("test.gz", 3), {}))
904	n/a
905	n/a	@unittest.skipUnless(bz2, "Requires bz2")
906	n/a	def test_bz2_ext_fake(self):
907	n/a	original_open = bz2.BZ2File
908	n/a	bz2.BZ2File = self.fake_open
909	n/a	try:
910	n/a	result = fileinput.hook_compressed("test.bz2", 4)
911	n/a	finally:
912	n/a	bz2.BZ2File = original_open
913	n/a
914	n/a	self.assertEqual(self.fake_open.invocation_count, 1)
915	n/a	self.assertEqual(self.fake_open.last_invocation, (("test.bz2", 4), {}))
916	n/a
917	n/a	def test_blah_ext(self):
918	n/a	self.do_test_use_builtin_open("abcd.blah", 5)
919	n/a
920	n/a	def test_gz_ext_builtin(self):
921	n/a	self.do_test_use_builtin_open("abcd.Gz", 6)
922	n/a
923	n/a	def test_bz2_ext_builtin(self):
924	n/a	self.do_test_use_builtin_open("abcd.Bz2", 7)
925	n/a
926	n/a	def do_test_use_builtin_open(self, filename, mode):
927	n/a	original_open = self.replace_builtin_open(self.fake_open)
928	n/a	try:
929	n/a	result = fileinput.hook_compressed(filename, mode)
930	n/a	finally:
931	n/a	self.replace_builtin_open(original_open)
932	n/a
933	n/a	self.assertEqual(self.fake_open.invocation_count, 1)
934	n/a	self.assertEqual(self.fake_open.last_invocation,
935	n/a	((filename, mode), {}))
936	n/a
937	n/a	@staticmethod
938	n/a	def replace_builtin_open(new_open_func):
939	n/a	original_open = builtins.open
940	n/a	builtins.open = new_open_func
941	n/a	return original_open
942	n/a
943	n/a	class Test_hook_encoded(unittest.TestCase):
944	n/a	"""Unit tests for fileinput.hook_encoded()"""
945	n/a
946	n/a	def test(self):
947	n/a	encoding = object()
948	n/a	errors = object()
949	n/a	result = fileinput.hook_encoded(encoding, errors=errors)
950	n/a
951	n/a	fake_open = InvocationRecorder()
952	n/a	original_open = builtins.open
953	n/a	builtins.open = fake_open
954	n/a	try:
955	n/a	filename = object()
956	n/a	mode = object()
957	n/a	open_result = result(filename, mode)
958	n/a	finally:
959	n/a	builtins.open = original_open
960	n/a
961	n/a	self.assertEqual(fake_open.invocation_count, 1)
962	n/a
963	n/a	args, kwargs = fake_open.last_invocation
964	n/a	self.assertIs(args[0], filename)
965	n/a	self.assertIs(args[1], mode)
966	n/a	self.assertIs(kwargs.pop('encoding'), encoding)
967	n/a	self.assertIs(kwargs.pop('errors'), errors)
968	n/a	self.assertFalse(kwargs)
969	n/a
970	n/a	def test_errors(self):
971	n/a	with open(TESTFN, 'wb') as f:
972	n/a	f.write(b'\x80abc')
973	n/a	self.addCleanup(safe_unlink, TESTFN)
974	n/a
975	n/a	def check(errors, expected_lines):
976	n/a	with FileInput(files=TESTFN, mode='r',
977	n/a	openhook=hook_encoded('utf-8', errors=errors)) as fi:
978	n/a	lines = list(fi)
979	n/a	self.assertEqual(lines, expected_lines)
980	n/a
981	n/a	check('ignore', ['abc'])
982	n/a	with self.assertRaises(UnicodeDecodeError):
983	n/a	check('strict', ['abc'])
984	n/a	check('replace', ['\ufffdabc'])
985	n/a	check('backslashreplace', ['\\x80abc'])
986	n/a
987	n/a	def test_modes(self):
988	n/a	with open(TESTFN, 'wb') as f:
989	n/a	# UTF-7 is a convenient, seldom used encoding
990	n/a	f.write(b'A\nB\r\nC\rD+IKw-')
991	n/a	self.addCleanup(safe_unlink, TESTFN)
992	n/a
993	n/a	def check(mode, expected_lines):
994	n/a	with FileInput(files=TESTFN, mode=mode,
995	n/a	openhook=hook_encoded('utf-7')) as fi:
996	n/a	lines = list(fi)
997	n/a	self.assertEqual(lines, expected_lines)
998	n/a
999	n/a	check('r', ['A\n', 'B\n', 'C\n', 'D\u20ac'])
1000	n/a	with self.assertWarns(DeprecationWarning):
1001	n/a	check('rU', ['A\n', 'B\n', 'C\n', 'D\u20ac'])
1002	n/a	with self.assertWarns(DeprecationWarning):
1003	n/a	check('U', ['A\n', 'B\n', 'C\n', 'D\u20ac'])
1004	n/a	with self.assertRaises(ValueError):
1005	n/a	check('rb', ['A\n', 'B\r\n', 'C\r', 'D\u20ac'])
1006	n/a
1007	n/a
1008	n/a	class MiscTest(unittest.TestCase):
1009	n/a
1010	n/a	def test_all(self):
1011	n/a	support.check__all__(self, fileinput)
1012	n/a
1013	n/a
1014	n/a	if __name__ == "__main__":
1015	n/a	unittest.main()