test_fileinput.py 37 KB

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