test.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919
  1. # -*- coding: utf-8 -*-
  2. # Copyright (C) 2001,2002 Python Software Foundation
  3. # csv package unit tests
  4. import array
  5. import decimal
  6. import os
  7. import string
  8. import sys
  9. import tempfile
  10. import unittest2 as unittest
  11. from codecs import EncodedFile
  12. from io import BytesIO
  13. import unicodecsv as csv
  14. try:
  15. # Python 2
  16. chr = unichr
  17. except:
  18. pass
  19. # pypy and cpython differ under which exception is raised under some
  20. # circumstances e.g. whether a module is written in C or not.
  21. py_compat_exc = (TypeError, AttributeError)
  22. class Test_Csv(unittest.TestCase):
  23. """
  24. Test the underlying C csv parser in ways that are not appropriate
  25. from the high level interface. Further tests of this nature are done
  26. in TestDialectRegistry.
  27. """
  28. def _test_arg_valid(self, ctor, arg):
  29. self.assertRaises(py_compat_exc, ctor)
  30. self.assertRaises(py_compat_exc, ctor, None)
  31. self.assertRaises(py_compat_exc, ctor, arg, bad_attr=0)
  32. self.assertRaises(py_compat_exc, ctor, arg, delimiter=0)
  33. self.assertRaises(py_compat_exc, ctor, arg, delimiter='XX')
  34. self.assertRaises(csv.Error, ctor, arg, 'foo')
  35. self.assertRaises(py_compat_exc, ctor, arg, delimiter=None)
  36. self.assertRaises(py_compat_exc, ctor, arg, delimiter=1)
  37. self.assertRaises(py_compat_exc, ctor, arg, quotechar=1)
  38. self.assertRaises(py_compat_exc, ctor, arg, lineterminator=None)
  39. self.assertRaises(py_compat_exc, ctor, arg, lineterminator=1)
  40. self.assertRaises(py_compat_exc, ctor, arg, quoting=None)
  41. self.assertRaises(py_compat_exc, ctor, arg,
  42. quoting=csv.QUOTE_ALL, quotechar='')
  43. self.assertRaises(py_compat_exc, ctor, arg,
  44. quoting=csv.QUOTE_ALL, quotechar=None)
  45. def test_reader_arg_valid(self):
  46. self._test_arg_valid(csv.reader, [])
  47. def test_writer_arg_valid(self):
  48. self._test_arg_valid(csv.writer, BytesIO())
  49. def _test_default_attrs(self, ctor, *args):
  50. obj = ctor(*args)
  51. # Check defaults
  52. self.assertEqual(obj.dialect.delimiter, ',')
  53. self.assertEqual(obj.dialect.doublequote, True)
  54. self.assertEqual(obj.dialect.escapechar, None)
  55. self.assertEqual(obj.dialect.lineterminator, "\r\n")
  56. self.assertEqual(obj.dialect.quotechar, '"')
  57. self.assertEqual(obj.dialect.quoting, csv.QUOTE_MINIMAL)
  58. self.assertEqual(obj.dialect.skipinitialspace, False)
  59. self.assertEqual(obj.dialect.strict, False)
  60. # Try deleting or changing attributes (they are read-only)
  61. self.assertRaises(py_compat_exc, delattr,
  62. obj.dialect, 'delimiter')
  63. self.assertRaises(py_compat_exc, setattr,
  64. obj.dialect, 'delimiter', ':')
  65. self.assertRaises(py_compat_exc, delattr,
  66. obj.dialect, 'quoting')
  67. self.assertRaises(py_compat_exc, setattr,
  68. obj.dialect, 'quoting', None)
  69. def test_reader_attrs(self):
  70. self._test_default_attrs(csv.reader, [])
  71. def test_writer_attrs(self):
  72. self._test_default_attrs(csv.writer, BytesIO())
  73. def _test_kw_attrs(self, ctor, *args):
  74. # Now try with alternate options
  75. kwargs = dict(delimiter=':', doublequote=False, escapechar='\\',
  76. lineterminator='\r', quotechar='*',
  77. quoting=csv.QUOTE_NONE, skipinitialspace=True,
  78. strict=True)
  79. obj = ctor(*args, **kwargs)
  80. self.assertEqual(obj.dialect.delimiter, ':')
  81. self.assertEqual(obj.dialect.doublequote, False)
  82. self.assertEqual(obj.dialect.escapechar, '\\')
  83. self.assertEqual(obj.dialect.lineterminator, "\r")
  84. self.assertEqual(obj.dialect.quotechar, '*')
  85. self.assertEqual(obj.dialect.quoting, csv.QUOTE_NONE)
  86. self.assertEqual(obj.dialect.skipinitialspace, True)
  87. self.assertEqual(obj.dialect.strict, True)
  88. def test_reader_kw_attrs(self):
  89. self._test_kw_attrs(csv.reader, [])
  90. def test_writer_kw_attrs(self):
  91. self._test_kw_attrs(csv.writer, BytesIO())
  92. def _test_dialect_attrs(self, ctor, *args):
  93. # Now try with dialect-derived options
  94. class dialect:
  95. delimiter = '-'
  96. doublequote = False
  97. escapechar = '^'
  98. lineterminator = '$'
  99. quotechar = '#'
  100. quoting = csv.QUOTE_ALL
  101. skipinitialspace = True
  102. strict = False
  103. args = args + (dialect,)
  104. obj = ctor(*args)
  105. self.assertEqual(obj.dialect.delimiter, '-')
  106. self.assertEqual(obj.dialect.doublequote, False)
  107. self.assertEqual(obj.dialect.escapechar, '^')
  108. self.assertEqual(obj.dialect.lineterminator, "$")
  109. self.assertEqual(obj.dialect.quotechar, '#')
  110. self.assertEqual(obj.dialect.quoting, csv.QUOTE_ALL)
  111. self.assertEqual(obj.dialect.skipinitialspace, True)
  112. self.assertEqual(obj.dialect.strict, False)
  113. def test_reader_dialect_attrs(self):
  114. self._test_dialect_attrs(csv.reader, [])
  115. def test_writer_dialect_attrs(self):
  116. self._test_dialect_attrs(csv.writer, BytesIO())
  117. def _write_test(self, fields, expect, **kwargs):
  118. fd, name = tempfile.mkstemp()
  119. fileobj = os.fdopen(fd, "w+b")
  120. try:
  121. writer = csv.writer(fileobj, **kwargs)
  122. writer.writerow(fields)
  123. fileobj.seek(0)
  124. self.assertEqual(fileobj.read(),
  125. expect + writer.dialect.lineterminator.encode('utf-8'))
  126. finally:
  127. fileobj.close()
  128. os.unlink(name)
  129. def test_write_arg_valid(self):
  130. import sys
  131. pypy3 = hasattr(sys, 'pypy_version_info') and sys.version_info.major == 3
  132. self.assertRaises(TypeError if pypy3 else csv.Error, self._write_test, None, '')
  133. self._write_test((), b'')
  134. self._write_test([None], b'""')
  135. self.assertRaises(csv.Error, self._write_test,
  136. [None], None, quoting=csv.QUOTE_NONE)
  137. # Check that exceptions are passed up the chain
  138. class BadList:
  139. def __len__(self):
  140. return 10
  141. def __getitem__(self, i):
  142. if i > 2:
  143. raise IOError
  144. self.assertRaises(IOError, self._write_test, BadList(), '')
  145. class BadItem:
  146. def __str__(self):
  147. raise IOError
  148. self.assertRaises(IOError, self._write_test, [BadItem()], '')
  149. def test_write_bigfield(self):
  150. # This exercises the buffer realloc functionality
  151. bigstring = 'X' * 50000
  152. self._write_test([bigstring, bigstring],
  153. b','.join([bigstring.encode('utf-8')] * 2))
  154. def test_write_quoting(self):
  155. self._write_test(['a', 1, 'p,q'], b'a,1,"p,q"')
  156. self.assertRaises(csv.Error,
  157. self._write_test,
  158. ['a', 1, 'p,q'], b'a,1,p,q',
  159. quoting=csv.QUOTE_NONE)
  160. self._write_test(['a', 1, 'p,q'], b'a,1,"p,q"',
  161. quoting=csv.QUOTE_MINIMAL)
  162. self._write_test(['a', 1, 'p,q'], b'"a",1,"p,q"',
  163. quoting=csv.QUOTE_NONNUMERIC)
  164. self._write_test(['a', 1, 'p,q'], b'"a","1","p,q"',
  165. quoting=csv.QUOTE_ALL)
  166. self._write_test(['a\nb', 1], b'"a\nb","1"',
  167. quoting=csv.QUOTE_ALL)
  168. def test_write_decimal(self):
  169. self._write_test(['a', decimal.Decimal("1.1"), 'p,q'], b'"a",1.1,"p,q"',
  170. quoting=csv.QUOTE_NONNUMERIC)
  171. def test_write_escape(self):
  172. self._write_test(['a', 1, 'p,q'], b'a,1,"p,q"',
  173. escapechar='\\')
  174. self.assertRaises(csv.Error,
  175. self._write_test,
  176. ['a', 1, 'p,"q"'], b'a,1,"p,\\"q\\""',
  177. escapechar=None, doublequote=False)
  178. self._write_test(['a', 1, 'p,"q"'], b'a,1,"p,\\"q\\""',
  179. escapechar='\\', doublequote=False)
  180. self._write_test(['"'], b'""""',
  181. escapechar='\\', quoting=csv.QUOTE_MINIMAL)
  182. self._write_test(['"'], b'\\"',
  183. escapechar='\\', quoting=csv.QUOTE_MINIMAL,
  184. doublequote=False)
  185. self._write_test(['"'], b'\\"',
  186. escapechar='\\', quoting=csv.QUOTE_NONE)
  187. self._write_test(['a', 1, 'p,q'], b'a,1,p\\,q',
  188. escapechar='\\', quoting=csv.QUOTE_NONE)
  189. def test_writerows(self):
  190. class BrokenFile:
  191. def write(self, buf):
  192. raise IOError
  193. writer = csv.writer(BrokenFile())
  194. self.assertRaises(IOError, writer.writerows, [['a']])
  195. fd, name = tempfile.mkstemp()
  196. fileobj = os.fdopen(fd, "w+b")
  197. try:
  198. writer = csv.writer(fileobj)
  199. self.assertRaises(TypeError, writer.writerows, None)
  200. writer.writerows([['a', 'b'], ['c', 'd']])
  201. fileobj.seek(0)
  202. self.assertEqual(fileobj.read(), b"a,b\r\nc,d\r\n")
  203. finally:
  204. fileobj.close()
  205. os.unlink(name)
  206. def _read_test(self, input, expect, **kwargs):
  207. reader = csv.reader(input, **kwargs)
  208. result = list(reader)
  209. self.assertEqual(result, expect)
  210. def test_read_oddinputs(self):
  211. self._read_test([], [])
  212. self._read_test([b''], [[]])
  213. self.assertRaises(csv.Error, self._read_test,
  214. [b'"ab"c'], None, strict=1)
  215. # cannot handle null bytes for the moment
  216. self.assertRaises(csv.Error, self._read_test,
  217. [b'ab\0c'], None, strict=1)
  218. self._read_test([b'"ab"c'], [['abc']], doublequote=0)
  219. def test_read_eol(self):
  220. self._read_test([b'a,b'], [['a', 'b']])
  221. self._read_test([b'a,b\n'], [['a', 'b']])
  222. self._read_test([b'a,b\r\n'], [['a', 'b']])
  223. self._read_test([b'a,b\r'], [['a', 'b']])
  224. self.assertRaises(csv.Error, self._read_test, [b'a,b\rc,d'], [])
  225. self.assertRaises(csv.Error, self._read_test, [b'a,b\nc,d'], [])
  226. self.assertRaises(csv.Error, self._read_test, [b'a,b\r\nc,d'], [])
  227. def test_read_escape(self):
  228. self._read_test([b'a,\\b,c'], [['a', 'b', 'c']], escapechar='\\')
  229. self._read_test([b'a,b\\,c'], [['a', 'b,c']], escapechar='\\')
  230. self._read_test([b'a,"b\\,c"'], [['a', 'b,c']], escapechar='\\')
  231. self._read_test([b'a,"b,\\c"'], [['a', 'b,c']], escapechar='\\')
  232. self._read_test([b'a,"b,c\\""'], [['a', 'b,c"']], escapechar='\\')
  233. self._read_test([b'a,"b,c"\\'], [['a', 'b,c\\']], escapechar='\\')
  234. def test_read_quoting(self):
  235. self._read_test([b'1,",3,",5'], [['1', ',3,', '5']])
  236. self._read_test([b'1,",3,",5'], [['1', '"', '3', '"', '5']],
  237. quotechar=None, escapechar='\\')
  238. self._read_test([b'1,",3,",5'], [['1', '"', '3', '"', '5']],
  239. quoting=csv.QUOTE_NONE, escapechar='\\')
  240. # will this fail where locale uses comma for decimals?
  241. self._read_test([b',3,"5",7.3, 9'], [['', 3, '5', 7.3, 9]],
  242. quoting=csv.QUOTE_NONNUMERIC)
  243. self._read_test([b'"a\nb", 7'], [['a\nb', ' 7']])
  244. self.assertRaises(ValueError, self._read_test,
  245. [b'abc,3'], [[]],
  246. quoting=csv.QUOTE_NONNUMERIC)
  247. def test_read_linenum(self):
  248. for r in (csv.reader([b'line,1', b'line,2', b'line,3']),
  249. csv.DictReader([b'line,1', b'line,2', b'line,3'],
  250. fieldnames=['a', 'b', 'c'])):
  251. self.assertEqual(r.line_num, 0)
  252. next(r)
  253. self.assertEqual(r.line_num, 1)
  254. next(r)
  255. self.assertEqual(r.line_num, 2)
  256. next(r)
  257. self.assertEqual(r.line_num, 3)
  258. self.assertRaises(StopIteration, next, r)
  259. self.assertEqual(r.line_num, 3)
  260. def test_roundtrip_quoteed_newlines(self):
  261. fd, name = tempfile.mkstemp()
  262. fileobj = os.fdopen(fd, "w+b")
  263. try:
  264. writer = csv.writer(fileobj)
  265. self.assertRaises(TypeError, writer.writerows, None)
  266. rows = [['a\nb', 'b'], ['c', 'x\r\nd']]
  267. writer.writerows(rows)
  268. fileobj.seek(0)
  269. for i, row in enumerate(csv.reader(fileobj)):
  270. self.assertEqual(row, rows[i])
  271. finally:
  272. fileobj.close()
  273. os.unlink(name)
  274. class TestDialectRegistry(unittest.TestCase):
  275. def test_registry_badargs(self):
  276. self.assertRaises(TypeError, csv.list_dialects, None)
  277. self.assertRaises(TypeError, csv.get_dialect)
  278. self.assertRaises(csv.Error, csv.get_dialect, None)
  279. self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
  280. self.assertRaises(TypeError, csv.unregister_dialect)
  281. self.assertRaises(csv.Error, csv.unregister_dialect, None)
  282. self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
  283. self.assertRaises(TypeError, csv.register_dialect, None)
  284. self.assertRaises(TypeError, csv.register_dialect, None, None)
  285. self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
  286. self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
  287. badargument=None)
  288. self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
  289. quoting=None)
  290. self.assertRaises(TypeError, csv.register_dialect, [])
  291. def test_registry(self):
  292. class myexceltsv(csv.excel):
  293. delimiter = "\t"
  294. name = "myexceltsv"
  295. expected_dialects = csv.list_dialects() + [name]
  296. expected_dialects.sort()
  297. csv.register_dialect(name, myexceltsv)
  298. try:
  299. self.assertEqual(csv.get_dialect(name).delimiter, '\t')
  300. got_dialects = csv.list_dialects()
  301. got_dialects.sort()
  302. self.assertEqual(expected_dialects, got_dialects)
  303. finally:
  304. csv.unregister_dialect(name)
  305. def test_register_kwargs(self):
  306. name = 'fedcba'
  307. csv.register_dialect(name, delimiter=';')
  308. try:
  309. self.assertNotEqual(csv.get_dialect(name).delimiter, '\t')
  310. self.assertEqual(list(csv.reader([b'X;Y;Z'], name)), [[u'X', u'Y', u'Z']])
  311. finally:
  312. csv.unregister_dialect(name)
  313. def test_incomplete_dialect(self):
  314. class myexceltsv(csv.Dialect):
  315. delimiter = "\t"
  316. self.assertRaises(csv.Error, myexceltsv)
  317. def test_space_dialect(self):
  318. class space(csv.excel):
  319. delimiter = " "
  320. quoting = csv.QUOTE_NONE
  321. escapechar = "\\"
  322. fd, name = tempfile.mkstemp()
  323. fileobj = os.fdopen(fd, "w+b")
  324. try:
  325. fileobj.write(b"abc def\nc1ccccc1 benzene\n")
  326. fileobj.seek(0)
  327. rdr = csv.reader(fileobj, dialect=space())
  328. self.assertEqual(next(rdr), ["abc", "def"])
  329. self.assertEqual(next(rdr), ["c1ccccc1", "benzene"])
  330. finally:
  331. fileobj.close()
  332. os.unlink(name)
  333. def test_dialect_apply(self):
  334. class testA(csv.excel):
  335. delimiter = "\t"
  336. class testB(csv.excel):
  337. delimiter = ":"
  338. class testC(csv.excel):
  339. delimiter = "|"
  340. csv.register_dialect('testC', testC)
  341. try:
  342. fd, name = tempfile.mkstemp()
  343. fileobj = os.fdopen(fd, "w+b")
  344. try:
  345. writer = csv.writer(fileobj)
  346. writer.writerow([1, 2, 3])
  347. fileobj.seek(0)
  348. self.assertEqual(fileobj.read(), b"1,2,3\r\n")
  349. finally:
  350. fileobj.close()
  351. os.unlink(name)
  352. fd, name = tempfile.mkstemp()
  353. fileobj = os.fdopen(fd, "w+b")
  354. try:
  355. writer = csv.writer(fileobj, testA)
  356. writer.writerow([1, 2, 3])
  357. fileobj.seek(0)
  358. self.assertEqual(fileobj.read(), b"1\t2\t3\r\n")
  359. finally:
  360. fileobj.close()
  361. os.unlink(name)
  362. fd, name = tempfile.mkstemp()
  363. fileobj = os.fdopen(fd, "w+b")
  364. try:
  365. writer = csv.writer(fileobj, dialect=testB())
  366. writer.writerow([1, 2, 3])
  367. fileobj.seek(0)
  368. self.assertEqual(fileobj.read(), b"1:2:3\r\n")
  369. finally:
  370. fileobj.close()
  371. os.unlink(name)
  372. fd, name = tempfile.mkstemp()
  373. fileobj = os.fdopen(fd, "w+b")
  374. try:
  375. writer = csv.writer(fileobj, dialect='testC')
  376. writer.writerow([1, 2, 3])
  377. fileobj.seek(0)
  378. self.assertEqual(fileobj.read(), b"1|2|3\r\n")
  379. finally:
  380. fileobj.close()
  381. os.unlink(name)
  382. fd, name = tempfile.mkstemp()
  383. fileobj = os.fdopen(fd, "w+b")
  384. try:
  385. writer = csv.writer(fileobj, dialect=testA, delimiter=';')
  386. writer.writerow([1, 2, 3])
  387. fileobj.seek(0)
  388. self.assertEqual(fileobj.read(), b"1;2;3\r\n")
  389. finally:
  390. fileobj.close()
  391. os.unlink(name)
  392. finally:
  393. csv.unregister_dialect('testC')
  394. def test_bad_dialect(self):
  395. # Unknown parameter
  396. self.assertRaises(TypeError, csv.reader, [], bad_attr=0)
  397. # Bad values
  398. self.assertRaises(TypeError, csv.reader, [], delimiter=None)
  399. self.assertRaises(TypeError, csv.reader, [], quoting=-1)
  400. self.assertRaises(TypeError, csv.reader, [], quoting=100)
  401. class TestCsvBase(unittest.TestCase):
  402. def readerAssertEqual(self, input, expected_result):
  403. fd, name = tempfile.mkstemp()
  404. fileobj = os.fdopen(fd, "w+b")
  405. try:
  406. fileobj.write(input)
  407. fileobj.seek(0)
  408. reader = csv.reader(fileobj, dialect=self.dialect)
  409. fields = list(reader)
  410. self.assertEqual(fields, expected_result)
  411. finally:
  412. fileobj.close()
  413. os.unlink(name)
  414. def writerAssertEqual(self, input, expected_result):
  415. fd, name = tempfile.mkstemp()
  416. fileobj = os.fdopen(fd, "w+b")
  417. try:
  418. writer = csv.writer(fileobj, dialect=self.dialect)
  419. writer.writerows(input)
  420. fileobj.seek(0)
  421. self.assertEqual(fileobj.read(), expected_result)
  422. finally:
  423. fileobj.close()
  424. os.unlink(name)
  425. class TestDialectExcel(TestCsvBase):
  426. dialect = 'excel'
  427. def test_single(self):
  428. self.readerAssertEqual(b'abc', [['abc']])
  429. def test_simple(self):
  430. self.readerAssertEqual(b'1,2,3,4,5', [['1', '2', '3', '4', '5']])
  431. def test_blankline(self):
  432. self.readerAssertEqual(b'', [])
  433. def test_empty_fields(self):
  434. self.readerAssertEqual(b',', [['', '']])
  435. def test_singlequoted(self):
  436. self.readerAssertEqual(b'""', [['']])
  437. def test_singlequoted_left_empty(self):
  438. self.readerAssertEqual(b'"",', [['', '']])
  439. def test_singlequoted_right_empty(self):
  440. self.readerAssertEqual(b',""', [['', '']])
  441. def test_single_quoted_quote(self):
  442. self.readerAssertEqual(b'""""', [['"']])
  443. def test_quoted_quotes(self):
  444. self.readerAssertEqual(b'""""""', [['""']])
  445. def test_inline_quote(self):
  446. self.readerAssertEqual(b'a""b', [['a""b']])
  447. def test_inline_quotes(self):
  448. self.readerAssertEqual(b'a"b"c', [['a"b"c']])
  449. def test_quotes_and_more(self):
  450. # Excel would never write a field containing '"a"b', but when
  451. # reading one, it will return 'ab'.
  452. self.readerAssertEqual(b'"a"b', [['ab']])
  453. def test_lone_quote(self):
  454. self.readerAssertEqual(b'a"b', [['a"b']])
  455. def test_quote_and_quote(self):
  456. # Excel would never write a field containing '"a" "b"', but when
  457. # reading one, it will return 'a "b"'.
  458. self.readerAssertEqual(b'"a" "b"', [['a "b"']])
  459. def test_space_and_quote(self):
  460. self.readerAssertEqual(b' "a"', [[' "a"']])
  461. def test_quoted(self):
  462. self.readerAssertEqual(b'1,2,3,"I think, therefore I am",5,6',
  463. [['1', '2', '3',
  464. 'I think, therefore I am',
  465. '5', '6']])
  466. def test_quoted_quote(self):
  467. value = b'1,2,3,"""I see,"" said the blind man","as he picked up his hammer and saw"'
  468. self.readerAssertEqual(value,
  469. [['1', '2', '3',
  470. '"I see," said the blind man',
  471. 'as he picked up his hammer and saw']])
  472. def test_quoted_nl(self):
  473. input = b'''\
  474. 1,2,3,"""I see,""
  475. said the blind man","as he picked up his
  476. hammer and saw"
  477. 9,8,7,6'''
  478. self.readerAssertEqual(input,
  479. [['1', '2', '3',
  480. '"I see,"\nsaid the blind man',
  481. 'as he picked up his\nhammer and saw'],
  482. ['9', '8', '7', '6']])
  483. def test_dubious_quote(self):
  484. self.readerAssertEqual(b'12,12,1",', [['12', '12', '1"', '']])
  485. def test_null(self):
  486. self.writerAssertEqual([], b'')
  487. def test_single_writer(self):
  488. self.writerAssertEqual([['abc']], b'abc\r\n')
  489. def test_simple_writer(self):
  490. self.writerAssertEqual([[1, 2, 'abc', 3, 4]],
  491. b'1,2,abc,3,4\r\n')
  492. def test_quotes(self):
  493. self.writerAssertEqual([[1, 2, 'a"bc"', 3, 4]],
  494. b'1,2,"a""bc""",3,4\r\n')
  495. def test_quote_fieldsep(self):
  496. self.writerAssertEqual([['abc,def']],
  497. b'"abc,def"\r\n')
  498. def test_newlines(self):
  499. self.writerAssertEqual([[1, 2, 'a\nbc', 3, 4]],
  500. b'1,2,"a\nbc",3,4\r\n')
  501. class EscapedExcel(csv.excel):
  502. quoting = csv.QUOTE_NONE
  503. escapechar = '\\'
  504. class TestEscapedExcel(TestCsvBase):
  505. dialect = EscapedExcel()
  506. def test_escape_fieldsep(self):
  507. self.writerAssertEqual([['abc,def']], b'abc\\,def\r\n')
  508. def test_read_escape_fieldsep(self):
  509. self.readerAssertEqual(b'abc\\,def\r\n', [['abc,def']])
  510. class QuotedEscapedExcel(csv.excel):
  511. quoting = csv.QUOTE_NONNUMERIC
  512. escapechar = '\\'
  513. class TestQuotedEscapedExcel(TestCsvBase):
  514. dialect = QuotedEscapedExcel()
  515. def test_write_escape_fieldsep(self):
  516. self.writerAssertEqual([['abc,def']], b'"abc,def"\r\n')
  517. def test_read_escape_fieldsep(self):
  518. self.readerAssertEqual(b'"abc\\,def"\r\n', [['abc,def']])
  519. class TestDictFields(unittest.TestCase):
  520. # "long" means the row is longer than the number of fieldnames
  521. # "short" means there are fewer elements in the row than fieldnames
  522. def test_write_simple_dict(self):
  523. fd, name = tempfile.mkstemp()
  524. fileobj = open(name, 'w+b')
  525. try:
  526. writer = csv.DictWriter(fileobj, fieldnames=["f1", "f2", "f3"])
  527. writer.writeheader()
  528. fileobj.seek(0)
  529. self.assertEqual(fileobj.readline(), b"f1,f2,f3\r\n")
  530. writer.writerow({"f1": 10, "f3": "abc"})
  531. fileobj.seek(0)
  532. fileobj.readline() # header
  533. self.assertEqual(fileobj.read(), b"10,,abc\r\n")
  534. finally:
  535. fileobj.close()
  536. os.unlink(name)
  537. def test_write_unicode_header_dict(self):
  538. fd, name = tempfile.mkstemp()
  539. fileobj = open(name, 'w+b')
  540. try:
  541. writer = csv.DictWriter(fileobj, fieldnames=[u"ñ", u"ö"])
  542. writer.writeheader()
  543. fileobj.seek(0)
  544. self.assertEqual(fileobj.readline().decode('utf-8'), u"ñ,ö\r\n")
  545. finally:
  546. fileobj.close()
  547. os.unlink(name)
  548. def test_write_no_fields(self):
  549. fileobj = BytesIO()
  550. self.assertRaises(TypeError, csv.DictWriter, fileobj)
  551. def test_read_dict_fields(self):
  552. fd, name = tempfile.mkstemp()
  553. fileobj = os.fdopen(fd, "w+b")
  554. try:
  555. fileobj.write(b"1,2,abc\r\n")
  556. fileobj.seek(0)
  557. reader = csv.DictReader(fileobj,
  558. fieldnames=["f1", "f2", "f3"])
  559. self.assertEqual(next(reader),
  560. {"f1": '1', "f2": '2', "f3": 'abc'})
  561. finally:
  562. fileobj.close()
  563. os.unlink(name)
  564. def test_read_dict_no_fieldnames(self):
  565. fd, name = tempfile.mkstemp()
  566. fileobj = os.fdopen(fd, "w+b")
  567. try:
  568. fileobj.write(b"f1,f2,f3\r\n1,2,abc\r\n")
  569. fileobj.seek(0)
  570. reader = csv.DictReader(fileobj)
  571. self.assertEqual(reader.fieldnames,
  572. ["f1", "f2", "f3"])
  573. self.assertEqual(next(reader),
  574. {"f1": '1', "f2": '2', "f3": 'abc'})
  575. finally:
  576. fileobj.close()
  577. os.unlink(name)
  578. # Two test cases to make sure existing ways of implicitly setting
  579. # fieldnames continue to work. Both arise from discussion in issue3436.
  580. def test_read_dict_fieldnames_from_file(self):
  581. fd, name = tempfile.mkstemp()
  582. f = os.fdopen(fd, "w+b")
  583. try:
  584. f.write(b"f1,f2,f3\r\n1,2,abc\r\n")
  585. f.seek(0)
  586. reader = csv.DictReader(f, fieldnames=next(csv.reader(f)))
  587. self.assertEqual(reader.fieldnames,
  588. ["f1", "f2", "f3"])
  589. self.assertEqual(next(reader),
  590. {"f1": '1', "f2": '2', "f3": 'abc'})
  591. finally:
  592. f.close()
  593. os.unlink(name)
  594. def test_read_dict_fieldnames_chain(self):
  595. import itertools
  596. fd, name = tempfile.mkstemp()
  597. f = os.fdopen(fd, "w+b")
  598. try:
  599. f.write(b"f1,f2,f3\r\n1,2,abc\r\n")
  600. f.seek(0)
  601. reader = csv.DictReader(f)
  602. first = next(reader)
  603. for row in itertools.chain([first], reader):
  604. self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
  605. self.assertEqual(row, {"f1": '1', "f2": '2', "f3": 'abc'})
  606. finally:
  607. f.close()
  608. os.unlink(name)
  609. def test_read_long(self):
  610. fd, name = tempfile.mkstemp()
  611. fileobj = os.fdopen(fd, "w+b")
  612. try:
  613. fileobj.write(b"1,2,abc,4,5,6\r\n")
  614. fileobj.seek(0)
  615. reader = csv.DictReader(fileobj,
  616. fieldnames=["f1", "f2"])
  617. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  618. None: ["abc", "4", "5", "6"]})
  619. finally:
  620. fileobj.close()
  621. os.unlink(name)
  622. def test_read_long_with_rest(self):
  623. fd, name = tempfile.mkstemp()
  624. fileobj = os.fdopen(fd, "w+b")
  625. try:
  626. fileobj.write(b"1,2,abc,4,5,6\r\n")
  627. fileobj.seek(0)
  628. reader = csv.DictReader(fileobj,
  629. fieldnames=["f1", "f2"], restkey="_rest")
  630. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  631. "_rest": ["abc", "4", "5", "6"]})
  632. finally:
  633. fileobj.close()
  634. os.unlink(name)
  635. def test_read_long_with_rest_no_fieldnames(self):
  636. fd, name = tempfile.mkstemp()
  637. fileobj = os.fdopen(fd, "w+b")
  638. try:
  639. fileobj.write(b"f1,f2\r\n1,2,abc,4,5,6\r\n")
  640. fileobj.seek(0)
  641. reader = csv.DictReader(fileobj, restkey="_rest")
  642. self.assertEqual(reader.fieldnames, ["f1", "f2"])
  643. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  644. "_rest": ["abc", "4", "5", "6"]})
  645. finally:
  646. fileobj.close()
  647. os.unlink(name)
  648. def test_read_short(self):
  649. fd, name = tempfile.mkstemp()
  650. fileobj = os.fdopen(fd, "w+b")
  651. try:
  652. fileobj.write(b"1,2,abc,4,5,6\r\n1,2,abc\r\n")
  653. fileobj.seek(0)
  654. reader = csv.DictReader(fileobj,
  655. fieldnames="1 2 3 4 5 6".split(),
  656. restval="DEFAULT")
  657. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  658. "4": '4', "5": '5', "6": '6'})
  659. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  660. "4": 'DEFAULT', "5": 'DEFAULT',
  661. "6": 'DEFAULT'})
  662. finally:
  663. fileobj.close()
  664. os.unlink(name)
  665. def test_read_multi(self):
  666. sample = [
  667. b'2147483648,43.0e12,17,abc,def\r\n',
  668. b'147483648,43.0e2,17,abc,def\r\n',
  669. b'47483648,43.0,170,abc,def\r\n'
  670. ]
  671. reader = csv.DictReader(sample,
  672. fieldnames="i1 float i2 s1 s2".split())
  673. self.assertEqual(next(reader), {"i1": '2147483648',
  674. "float": '43.0e12',
  675. "i2": '17',
  676. "s1": 'abc',
  677. "s2": 'def'})
  678. def test_read_with_blanks(self):
  679. reader = csv.DictReader([b"1,2,abc,4,5,6\r\n", b"\r\n",
  680. b"1,2,abc,4,5,6\r\n"],
  681. fieldnames="1 2 3 4 5 6".split())
  682. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  683. "4": '4', "5": '5', "6": '6'})
  684. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  685. "4": '4', "5": '5', "6": '6'})
  686. def test_read_semi_sep(self):
  687. reader = csv.DictReader([b"1;2;abc;4;5;6\r\n"],
  688. fieldnames="1 2 3 4 5 6".split(),
  689. delimiter=';')
  690. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  691. "4": '4', "5": '5', "6": '6'})
  692. def test_empty_file(self):
  693. csv.DictReader(BytesIO())
  694. class TestArrayWrites(unittest.TestCase):
  695. def test_int_write(self):
  696. contents = [(20-i) for i in range(20)]
  697. a = array.array('i', contents)
  698. fd, name = tempfile.mkstemp()
  699. fileobj = os.fdopen(fd, "w+b")
  700. try:
  701. writer = csv.writer(fileobj, dialect="excel")
  702. writer.writerow(a)
  703. expected = b",".join([str(i).encode('utf-8') for i in a])+b"\r\n"
  704. fileobj.seek(0)
  705. self.assertEqual(fileobj.read(), expected)
  706. finally:
  707. fileobj.close()
  708. os.unlink(name)
  709. def test_double_write(self):
  710. contents = [(20-i)*0.1 for i in range(20)]
  711. a = array.array('d', contents)
  712. fd, name = tempfile.mkstemp()
  713. fileobj = os.fdopen(fd, "w+b")
  714. try:
  715. writer = csv.writer(fileobj, dialect="excel")
  716. writer.writerow(a)
  717. float_repr = str
  718. if sys.version_info >= (2, 7, 3):
  719. float_repr = repr
  720. expected = b",".join([float_repr(i).encode('utf-8') for i in a])+b"\r\n"
  721. fileobj.seek(0)
  722. self.assertEqual(fileobj.read(), expected)
  723. finally:
  724. fileobj.close()
  725. os.unlink(name)
  726. def test_float_write(self):
  727. contents = [(20-i)*0.1 for i in range(20)]
  728. a = array.array('f', contents)
  729. fd, name = tempfile.mkstemp()
  730. fileobj = os.fdopen(fd, "w+b")
  731. try:
  732. writer = csv.writer(fileobj, dialect="excel")
  733. writer.writerow(a)
  734. float_repr = str
  735. if sys.version_info >= (2, 7, 3):
  736. float_repr = repr
  737. expected = b",".join([float_repr(i).encode('utf-8') for i in a])+b"\r\n"
  738. fileobj.seek(0)
  739. self.assertEqual(fileobj.read(), expected)
  740. finally:
  741. fileobj.close()
  742. os.unlink(name)
  743. def test_char_write(self):
  744. a = string.ascii_letters
  745. fd, name = tempfile.mkstemp()
  746. fileobj = os.fdopen(fd, "w+b")
  747. try:
  748. writer = csv.writer(fileobj, dialect="excel")
  749. writer.writerow(a)
  750. expected = ",".join(a).encode('utf-8')+b"\r\n"
  751. fileobj.seek(0)
  752. self.assertEqual(fileobj.read(), expected)
  753. finally:
  754. fileobj.close()
  755. os.unlink(name)
  756. class TestUnicode(unittest.TestCase):
  757. def test_unicode_read(self):
  758. f = EncodedFile(BytesIO((u"Martin von Löwis,"
  759. u"Marc André Lemburg,"
  760. u"Guido van Rossum,"
  761. u"François Pinard\r\n").encode('iso-8859-1')),
  762. data_encoding='iso-8859-1')
  763. reader = csv.reader(f, encoding='iso-8859-1')
  764. self.assertEqual(list(reader), [[u"Martin von Löwis",
  765. u"Marc André Lemburg",
  766. u"Guido van Rossum",
  767. u"François Pinard"]])
  768. class TestUnicodeErrors(unittest.TestCase):
  769. def test_encode_error(self):
  770. fd = BytesIO()
  771. writer = csv.writer(fd, encoding='cp1252', errors='xmlcharrefreplace')
  772. writer.writerow(['hello', chr(2603)])
  773. self.assertEqual(fd.getvalue(), b'hello,ਫ\r\n')
  774. def test_encode_error_dictwriter(self):
  775. fd = BytesIO()
  776. dw = csv.DictWriter(fd, ['col1'],
  777. encoding='cp1252', errors='xmlcharrefreplace')
  778. dw.writerow({'col1': chr(2604)})
  779. self.assertEqual(fd.getvalue(), b'ਬ\r\n')
  780. def test_decode_error(self):
  781. """Make sure the specified error-handling mode is obeyed on readers."""
  782. file = EncodedFile(BytesIO(u'Löwis,2,3'.encode('iso-8859-1')),
  783. data_encoding='iso-8859-1')
  784. reader = csv.reader(file, encoding='ascii', errors='ignore')
  785. self.assertEqual(list(reader)[0][0], 'Lwis')
  786. def test_decode_error_dictreader(self):
  787. """Make sure the error-handling mode is obeyed on DictReaders."""
  788. file = EncodedFile(BytesIO(u'name,height,weight\nLöwis,2,3'.encode('iso-8859-1')),
  789. data_encoding='iso-8859-1')
  790. reader = csv.DictReader(file, encoding='ascii', errors='ignore')
  791. self.assertEqual(list(reader)[0]['name'], 'Lwis')