| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  | '''
 | 
					
						
							|  |  |  | Tests for fileinput module. | 
					
						
							|  |  |  | Nick Mathewson | 
					
						
							|  |  |  | '''
 | 
					
						
							|  |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  | import unittest | 
					
						
							|  |  |  | from test.test_support import verbose, TESTFN, run_unittest | 
					
						
							|  |  |  | from test.test_support import unlink as safe_unlink | 
					
						
							| 
									
										
										
										
											2008-02-23 17:40:11 +00:00
										 |  |  | import sys, re | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  | from StringIO import StringIO | 
					
						
							| 
									
										
										
										
											2006-02-19 14:57:47 +00:00
										 |  |  | from fileinput import FileInput, hook_encoded | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  | 
 | 
					
						
							|  |  |  | # The fileinput module has 2 interfaces: the FileInput class which does | 
					
						
							|  |  |  | # all the work, and a few functions (input, etc.) that use a global _state | 
					
						
							|  |  |  | # variable.  We only test the FileInput class, since the other functions | 
					
						
							|  |  |  | # only provide a thin facade over FileInput. | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  | # Write lines (a list of lines) to temp file number i, and return the | 
					
						
							|  |  |  | # temp file's name. | 
					
						
							| 
									
										
										
										
											2006-02-19 21:22:10 +00:00
										 |  |  | def writeTmp(i, lines, mode='w'):  # opening in text mode is the default | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |     name = TESTFN + str(i) | 
					
						
							| 
									
										
										
										
											2006-02-19 21:22:10 +00:00
										 |  |  |     f = open(name, mode) | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |     f.writelines(lines) | 
					
						
							|  |  |  |     f.close() | 
					
						
							|  |  |  |     return name | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  | def remove_tempfiles(*names): | 
					
						
							|  |  |  |     for name in names: | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |         safe_unlink(name) | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  | class BufferSizesTests(unittest.TestCase): | 
					
						
							|  |  |  |     def test_buffer_sizes(self): | 
					
						
							|  |  |  |         # First, run the tests with default and teeny buffer size. | 
					
						
							|  |  |  |         for round, bs in (0, 0), (1, 30): | 
					
						
							|  |  |  |             try: | 
					
						
							|  |  |  |                 t1 = writeTmp(1, ["Line %s of file 1\n" % (i+1) for i in range(15)]) | 
					
						
							|  |  |  |                 t2 = writeTmp(2, ["Line %s of file 2\n" % (i+1) for i in range(10)]) | 
					
						
							|  |  |  |                 t3 = writeTmp(3, ["Line %s of file 3\n" % (i+1) for i in range(5)]) | 
					
						
							|  |  |  |                 t4 = writeTmp(4, ["Line %s of file 4\n" % (i+1) for i in range(1)]) | 
					
						
							|  |  |  |                 self.buffer_size_test(t1, t2, t3, t4, bs, round) | 
					
						
							|  |  |  |             finally: | 
					
						
							|  |  |  |                 remove_tempfiles(t1, t2, t3, t4) | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |     def buffer_size_test(self, t1, t2, t3, t4, bs=0, round=0): | 
					
						
							|  |  |  |         pat = re.compile(r'LINE (\d+) OF FILE (\d+)') | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |         start = 1 + round*6 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Simple iteration (bs=%s)' % (start+0, bs) | 
					
						
							|  |  |  |         fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs) | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |         lines = list(fi) | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |         fi.close() | 
					
						
							|  |  |  |         self.assertEqual(len(lines), 31) | 
					
						
							|  |  |  |         self.assertEqual(lines[4], 'Line 5 of file 1\n') | 
					
						
							|  |  |  |         self.assertEqual(lines[30], 'Line 1 of file 4\n') | 
					
						
							|  |  |  |         self.assertEqual(fi.lineno(), 31) | 
					
						
							|  |  |  |         self.assertEqual(fi.filename(), t4) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Status variables (bs=%s)' % (start+1, bs) | 
					
						
							|  |  |  |         fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs) | 
					
						
							|  |  |  |         s = "x" | 
					
						
							|  |  |  |         while s and s != 'Line 6 of file 2\n': | 
					
						
							|  |  |  |             s = fi.readline() | 
					
						
							|  |  |  |         self.assertEqual(fi.filename(), t2) | 
					
						
							|  |  |  |         self.assertEqual(fi.lineno(), 21) | 
					
						
							|  |  |  |         self.assertEqual(fi.filelineno(), 6) | 
					
						
							| 
									
										
										
										
											2009-06-30 22:57:08 +00:00
										 |  |  |         self.assertFalse(fi.isfirstline()) | 
					
						
							|  |  |  |         self.assertFalse(fi.isstdin()) | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  | 
 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Nextfile (bs=%s)' % (start+2, bs) | 
					
						
							|  |  |  |         fi.nextfile() | 
					
						
							|  |  |  |         self.assertEqual(fi.readline(), 'Line 1 of file 3\n') | 
					
						
							|  |  |  |         self.assertEqual(fi.lineno(), 22) | 
					
						
							|  |  |  |         fi.close() | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Stdin (bs=%s)' % (start+3, bs) | 
					
						
							|  |  |  |         fi = FileInput(files=(t1, t2, t3, t4, '-'), bufsize=bs) | 
					
						
							|  |  |  |         savestdin = sys.stdin | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             sys.stdin = StringIO("Line 1 of stdin\nLine 2 of stdin\n") | 
					
						
							|  |  |  |             lines = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(len(lines), 33) | 
					
						
							|  |  |  |             self.assertEqual(lines[32], 'Line 2 of stdin\n') | 
					
						
							|  |  |  |             self.assertEqual(fi.filename(), '<stdin>') | 
					
						
							|  |  |  |             fi.nextfile() | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             sys.stdin = savestdin | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Boundary conditions (bs=%s)' % (start+4, bs) | 
					
						
							|  |  |  |         fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs) | 
					
						
							|  |  |  |         self.assertEqual(fi.lineno(), 0) | 
					
						
							|  |  |  |         self.assertEqual(fi.filename(), None) | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |         fi.nextfile() | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |         self.assertEqual(fi.lineno(), 0) | 
					
						
							|  |  |  |         self.assertEqual(fi.filename(), None) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |         if verbose: | 
					
						
							|  |  |  |             print '%s. Inplace (bs=%s)' % (start+5, bs) | 
					
						
							|  |  |  |         savestdout = sys.stdout | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             fi = FileInput(files=(t1, t2, t3, t4), inplace=1, bufsize=bs) | 
					
						
							|  |  |  |             for line in fi: | 
					
						
							|  |  |  |                 line = line[:-1].upper() | 
					
						
							|  |  |  |                 print line | 
					
						
							|  |  |  |             fi.close() | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             sys.stdout = savestdout | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |         fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs) | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |         for line in fi: | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |             self.assertEqual(line[-1], '\n') | 
					
						
							|  |  |  |             m = pat.match(line[:-1]) | 
					
						
							|  |  |  |             self.assertNotEqual(m, None) | 
					
						
							|  |  |  |             self.assertEqual(int(m.group(1)), fi.filelineno()) | 
					
						
							| 
									
										
										
										
											2001-07-11 22:21:17 +00:00
										 |  |  |         fi.close() | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  | class FileInputTests(unittest.TestCase): | 
					
						
							|  |  |  |     def test_zero_byte_files(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             t1 = writeTmp(1, [""]) | 
					
						
							|  |  |  |             t2 = writeTmp(2, [""]) | 
					
						
							|  |  |  |             t3 = writeTmp(3, ["The only line there is.\n"]) | 
					
						
							|  |  |  |             t4 = writeTmp(4, [""]) | 
					
						
							|  |  |  |             fi = FileInput(files=(t1, t2, t3, t4)) | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |             line = fi.readline() | 
					
						
							|  |  |  |             self.assertEqual(line, 'The only line there is.\n') | 
					
						
							|  |  |  |             self.assertEqual(fi.lineno(), 1) | 
					
						
							|  |  |  |             self.assertEqual(fi.filelineno(), 1) | 
					
						
							|  |  |  |             self.assertEqual(fi.filename(), t3) | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |             line = fi.readline() | 
					
						
							| 
									
										
										
										
											2009-06-30 22:57:08 +00:00
										 |  |  |             self.assertFalse(line) | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  |             self.assertEqual(fi.lineno(), 1) | 
					
						
							|  |  |  |             self.assertEqual(fi.filelineno(), 0) | 
					
						
							|  |  |  |             self.assertEqual(fi.filename(), t4) | 
					
						
							|  |  |  |             fi.close() | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1, t2, t3, t4) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     def test_files_that_dont_end_with_newline(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             t1 = writeTmp(1, ["A\nB\nC"]) | 
					
						
							|  |  |  |             t2 = writeTmp(2, ["D\nE\nF"]) | 
					
						
							|  |  |  |             fi = FileInput(files=(t1, t2)) | 
					
						
							|  |  |  |             lines = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"]) | 
					
						
							|  |  |  |             self.assertEqual(fi.filelineno(), 3) | 
					
						
							|  |  |  |             self.assertEqual(fi.lineno(), 6) | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1, t2) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     def test_unicode_filenames(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             t1 = writeTmp(1, ["A\nB"]) | 
					
						
							|  |  |  |             encoding = sys.getfilesystemencoding() | 
					
						
							|  |  |  |             if encoding is None: | 
					
						
							|  |  |  |                 encoding = 'ascii' | 
					
						
							|  |  |  |             fi = FileInput(files=unicode(t1, encoding)) | 
					
						
							|  |  |  |             lines = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(lines, ["A\n", "B"]) | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     def test_fileno(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             t1 = writeTmp(1, ["A\nB"]) | 
					
						
							|  |  |  |             t2 = writeTmp(2, ["C\nD"]) | 
					
						
							|  |  |  |             fi = FileInput(files=(t1, t2)) | 
					
						
							|  |  |  |             self.assertEqual(fi.fileno(), -1) | 
					
						
							|  |  |  |             line = fi.next() | 
					
						
							|  |  |  |             self.assertNotEqual(fi.fileno(), -1) | 
					
						
							|  |  |  |             fi.nextfile() | 
					
						
							|  |  |  |             self.assertEqual(fi.fileno(), -1) | 
					
						
							|  |  |  |             line = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(fi.fileno(), -1) | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1, t2) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     def test_opening_mode(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             # invalid mode, should raise ValueError | 
					
						
							|  |  |  |             fi = FileInput(mode="w") | 
					
						
							|  |  |  |             self.fail("FileInput should reject invalid mode argument") | 
					
						
							|  |  |  |         except ValueError: | 
					
						
							|  |  |  |             pass | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             # try opening in universal newline mode | 
					
						
							|  |  |  |             t1 = writeTmp(1, ["A\nB\r\nC\rD"], mode="wb") | 
					
						
							|  |  |  |             fi = FileInput(files=t1, mode="U") | 
					
						
							|  |  |  |             lines = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(lines, ["A\n", "B\n", "C\n", "D"]) | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     def test_file_opening_hook(self): | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             # cannot use openhook and inplace mode | 
					
						
							|  |  |  |             fi = FileInput(inplace=1, openhook=lambda f,m: None) | 
					
						
							|  |  |  |             self.fail("FileInput should raise if both inplace " | 
					
						
							|  |  |  |                              "and openhook arguments are given") | 
					
						
							|  |  |  |         except ValueError: | 
					
						
							|  |  |  |             pass | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             fi = FileInput(openhook=1) | 
					
						
							|  |  |  |             self.fail("FileInput should check openhook for being callable") | 
					
						
							|  |  |  |         except ValueError: | 
					
						
							|  |  |  |             pass | 
					
						
							|  |  |  |         try: | 
					
						
							|  |  |  |             t1 = writeTmp(1, ["A\nB"], mode="wb") | 
					
						
							|  |  |  |             fi = FileInput(files=t1, openhook=hook_encoded("rot13")) | 
					
						
							|  |  |  |             lines = list(fi) | 
					
						
							|  |  |  |             self.assertEqual(lines, ["N\n", "O"]) | 
					
						
							|  |  |  |         finally: | 
					
						
							|  |  |  |             remove_tempfiles(t1) | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  | def test_main(): | 
					
						
							|  |  |  |     run_unittest(BufferSizesTests, FileInputTests) | 
					
						
							| 
									
										
										
										
											2007-04-25 06:30:05 +00:00
										 |  |  | 
 | 
					
						
							| 
									
										
										
										
											2007-04-07 04:40:43 +00:00
										 |  |  | if __name__ == "__main__": | 
					
						
							|  |  |  |     test_main() |