1 | '''
|
---|
2 | Tests for fileinput module.
|
---|
3 | Nick Mathewson
|
---|
4 | '''
|
---|
5 |
|
---|
6 | import unittest
|
---|
7 | from test.test_support import verbose, TESTFN, run_unittest
|
---|
8 | from test.test_support import unlink as safe_unlink
|
---|
9 | import sys, re
|
---|
10 | from StringIO import StringIO
|
---|
11 | from fileinput import FileInput, hook_encoded
|
---|
12 |
|
---|
13 | # The fileinput module has 2 interfaces: the FileInput class which does
|
---|
14 | # all the work, and a few functions (input, etc.) that use a global _state
|
---|
15 | # variable. We only test the FileInput class, since the other functions
|
---|
16 | # only provide a thin facade over FileInput.
|
---|
17 |
|
---|
18 | # Write lines (a list of lines) to temp file number i, and return the
|
---|
19 | # temp file's name.
|
---|
20 | def writeTmp(i, lines, mode='w'): # opening in text mode is the default
|
---|
21 | name = TESTFN + str(i)
|
---|
22 | f = open(name, mode)
|
---|
23 | f.writelines(lines)
|
---|
24 | f.close()
|
---|
25 | return name
|
---|
26 |
|
---|
27 | def remove_tempfiles(*names):
|
---|
28 | for name in names:
|
---|
29 | safe_unlink(name)
|
---|
30 |
|
---|
31 | class BufferSizesTests(unittest.TestCase):
|
---|
32 | def test_buffer_sizes(self):
|
---|
33 | # First, run the tests with default and teeny buffer size.
|
---|
34 | for round, bs in (0, 0), (1, 30):
|
---|
35 | try:
|
---|
36 | t1 = writeTmp(1, ["Line %s of file 1\n" % (i+1) for i in range(15)])
|
---|
37 | t2 = writeTmp(2, ["Line %s of file 2\n" % (i+1) for i in range(10)])
|
---|
38 | t3 = writeTmp(3, ["Line %s of file 3\n" % (i+1) for i in range(5)])
|
---|
39 | t4 = writeTmp(4, ["Line %s of file 4\n" % (i+1) for i in range(1)])
|
---|
40 | self.buffer_size_test(t1, t2, t3, t4, bs, round)
|
---|
41 | finally:
|
---|
42 | remove_tempfiles(t1, t2, t3, t4)
|
---|
43 |
|
---|
44 | def buffer_size_test(self, t1, t2, t3, t4, bs=0, round=0):
|
---|
45 | pat = re.compile(r'LINE (\d+) OF FILE (\d+)')
|
---|
46 |
|
---|
47 | start = 1 + round*6
|
---|
48 | if verbose:
|
---|
49 | print '%s. Simple iteration (bs=%s)' % (start+0, bs)
|
---|
50 | fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
|
---|
51 | lines = list(fi)
|
---|
52 | fi.close()
|
---|
53 | self.assertEqual(len(lines), 31)
|
---|
54 | self.assertEqual(lines[4], 'Line 5 of file 1\n')
|
---|
55 | self.assertEqual(lines[30], 'Line 1 of file 4\n')
|
---|
56 | self.assertEqual(fi.lineno(), 31)
|
---|
57 | self.assertEqual(fi.filename(), t4)
|
---|
58 |
|
---|
59 | if verbose:
|
---|
60 | print '%s. Status variables (bs=%s)' % (start+1, bs)
|
---|
61 | fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
|
---|
62 | s = "x"
|
---|
63 | while s and s != 'Line 6 of file 2\n':
|
---|
64 | s = fi.readline()
|
---|
65 | self.assertEqual(fi.filename(), t2)
|
---|
66 | self.assertEqual(fi.lineno(), 21)
|
---|
67 | self.assertEqual(fi.filelineno(), 6)
|
---|
68 | self.assertFalse(fi.isfirstline())
|
---|
69 | self.assertFalse(fi.isstdin())
|
---|
70 |
|
---|
71 | if verbose:
|
---|
72 | print '%s. Nextfile (bs=%s)' % (start+2, bs)
|
---|
73 | fi.nextfile()
|
---|
74 | self.assertEqual(fi.readline(), 'Line 1 of file 3\n')
|
---|
75 | self.assertEqual(fi.lineno(), 22)
|
---|
76 | fi.close()
|
---|
77 |
|
---|
78 | if verbose:
|
---|
79 | print '%s. Stdin (bs=%s)' % (start+3, bs)
|
---|
80 | fi = FileInput(files=(t1, t2, t3, t4, '-'), bufsize=bs)
|
---|
81 | savestdin = sys.stdin
|
---|
82 | try:
|
---|
83 | sys.stdin = StringIO("Line 1 of stdin\nLine 2 of stdin\n")
|
---|
84 | lines = list(fi)
|
---|
85 | self.assertEqual(len(lines), 33)
|
---|
86 | self.assertEqual(lines[32], 'Line 2 of stdin\n')
|
---|
87 | self.assertEqual(fi.filename(), '<stdin>')
|
---|
88 | fi.nextfile()
|
---|
89 | finally:
|
---|
90 | sys.stdin = savestdin
|
---|
91 |
|
---|
92 | if verbose:
|
---|
93 | print '%s. Boundary conditions (bs=%s)' % (start+4, bs)
|
---|
94 | fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
|
---|
95 | self.assertEqual(fi.lineno(), 0)
|
---|
96 | self.assertEqual(fi.filename(), None)
|
---|
97 | fi.nextfile()
|
---|
98 | self.assertEqual(fi.lineno(), 0)
|
---|
99 | self.assertEqual(fi.filename(), None)
|
---|
100 |
|
---|
101 | if verbose:
|
---|
102 | print '%s. Inplace (bs=%s)' % (start+5, bs)
|
---|
103 | savestdout = sys.stdout
|
---|
104 | try:
|
---|
105 | fi = FileInput(files=(t1, t2, t3, t4), inplace=1, bufsize=bs)
|
---|
106 | for line in fi:
|
---|
107 | line = line[:-1].upper()
|
---|
108 | print line
|
---|
109 | fi.close()
|
---|
110 | finally:
|
---|
111 | sys.stdout = savestdout
|
---|
112 |
|
---|
113 | fi = FileInput(files=(t1, t2, t3, t4), bufsize=bs)
|
---|
114 | for line in fi:
|
---|
115 | self.assertEqual(line[-1], '\n')
|
---|
116 | m = pat.match(line[:-1])
|
---|
117 | self.assertNotEqual(m, None)
|
---|
118 | self.assertEqual(int(m.group(1)), fi.filelineno())
|
---|
119 | fi.close()
|
---|
120 |
|
---|
121 | class FileInputTests(unittest.TestCase):
|
---|
122 | def test_zero_byte_files(self):
|
---|
123 | try:
|
---|
124 | t1 = writeTmp(1, [""])
|
---|
125 | t2 = writeTmp(2, [""])
|
---|
126 | t3 = writeTmp(3, ["The only line there is.\n"])
|
---|
127 | t4 = writeTmp(4, [""])
|
---|
128 | fi = FileInput(files=(t1, t2, t3, t4))
|
---|
129 |
|
---|
130 | line = fi.readline()
|
---|
131 | self.assertEqual(line, 'The only line there is.\n')
|
---|
132 | self.assertEqual(fi.lineno(), 1)
|
---|
133 | self.assertEqual(fi.filelineno(), 1)
|
---|
134 | self.assertEqual(fi.filename(), t3)
|
---|
135 |
|
---|
136 | line = fi.readline()
|
---|
137 | self.assertFalse(line)
|
---|
138 | self.assertEqual(fi.lineno(), 1)
|
---|
139 | self.assertEqual(fi.filelineno(), 0)
|
---|
140 | self.assertEqual(fi.filename(), t4)
|
---|
141 | fi.close()
|
---|
142 | finally:
|
---|
143 | remove_tempfiles(t1, t2, t3, t4)
|
---|
144 |
|
---|
145 | def test_files_that_dont_end_with_newline(self):
|
---|
146 | try:
|
---|
147 | t1 = writeTmp(1, ["A\nB\nC"])
|
---|
148 | t2 = writeTmp(2, ["D\nE\nF"])
|
---|
149 | fi = FileInput(files=(t1, t2))
|
---|
150 | lines = list(fi)
|
---|
151 | self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"])
|
---|
152 | self.assertEqual(fi.filelineno(), 3)
|
---|
153 | self.assertEqual(fi.lineno(), 6)
|
---|
154 | finally:
|
---|
155 | remove_tempfiles(t1, t2)
|
---|
156 |
|
---|
157 | def test_unicode_filenames(self):
|
---|
158 | try:
|
---|
159 | t1 = writeTmp(1, ["A\nB"])
|
---|
160 | encoding = sys.getfilesystemencoding()
|
---|
161 | if encoding is None:
|
---|
162 | encoding = 'ascii'
|
---|
163 | fi = FileInput(files=unicode(t1, encoding))
|
---|
164 | lines = list(fi)
|
---|
165 | self.assertEqual(lines, ["A\n", "B"])
|
---|
166 | finally:
|
---|
167 | remove_tempfiles(t1)
|
---|
168 |
|
---|
169 | def test_fileno(self):
|
---|
170 | try:
|
---|
171 | t1 = writeTmp(1, ["A\nB"])
|
---|
172 | t2 = writeTmp(2, ["C\nD"])
|
---|
173 | fi = FileInput(files=(t1, t2))
|
---|
174 | self.assertEqual(fi.fileno(), -1)
|
---|
175 | line = fi.next()
|
---|
176 | self.assertNotEqual(fi.fileno(), -1)
|
---|
177 | fi.nextfile()
|
---|
178 | self.assertEqual(fi.fileno(), -1)
|
---|
179 | line = list(fi)
|
---|
180 | self.assertEqual(fi.fileno(), -1)
|
---|
181 | finally:
|
---|
182 | remove_tempfiles(t1, t2)
|
---|
183 |
|
---|
184 | def test_opening_mode(self):
|
---|
185 | try:
|
---|
186 | # invalid mode, should raise ValueError
|
---|
187 | fi = FileInput(mode="w")
|
---|
188 | self.fail("FileInput should reject invalid mode argument")
|
---|
189 | except ValueError:
|
---|
190 | pass
|
---|
191 | try:
|
---|
192 | # try opening in universal newline mode
|
---|
193 | t1 = writeTmp(1, ["A\nB\r\nC\rD"], mode="wb")
|
---|
194 | fi = FileInput(files=t1, mode="U")
|
---|
195 | lines = list(fi)
|
---|
196 | self.assertEqual(lines, ["A\n", "B\n", "C\n", "D"])
|
---|
197 | finally:
|
---|
198 | remove_tempfiles(t1)
|
---|
199 |
|
---|
200 | def test_file_opening_hook(self):
|
---|
201 | try:
|
---|
202 | # cannot use openhook and inplace mode
|
---|
203 | fi = FileInput(inplace=1, openhook=lambda f,m: None)
|
---|
204 | self.fail("FileInput should raise if both inplace "
|
---|
205 | "and openhook arguments are given")
|
---|
206 | except ValueError:
|
---|
207 | pass
|
---|
208 | try:
|
---|
209 | fi = FileInput(openhook=1)
|
---|
210 | self.fail("FileInput should check openhook for being callable")
|
---|
211 | except ValueError:
|
---|
212 | pass
|
---|
213 | try:
|
---|
214 | t1 = writeTmp(1, ["A\nB"], mode="wb")
|
---|
215 | fi = FileInput(files=t1, openhook=hook_encoded("rot13"))
|
---|
216 | lines = list(fi)
|
---|
217 | self.assertEqual(lines, ["N\n", "O"])
|
---|
218 | finally:
|
---|
219 | remove_tempfiles(t1)
|
---|
220 |
|
---|
221 | def test_main():
|
---|
222 | run_unittest(BufferSizesTests, FileInputTests)
|
---|
223 |
|
---|
224 | if __name__ == "__main__":
|
---|
225 | test_main()
|
---|