test_csv.py 55 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394
  1. # Copyright (C) 2001,2002 Python Software Foundation
  2. # csv package unit tests
  3. import copy
  4. import sys
  5. import unittest
  6. from io import StringIO
  7. from tempfile import TemporaryFile
  8. import csv
  9. import gc
  10. import pickle
  11. from test import support
  12. from test.support import warnings_helper
  13. from itertools import permutations
  14. from textwrap import dedent
  15. from collections import OrderedDict
  16. class BadIterable:
  17. def __iter__(self):
  18. raise OSError
  19. class Test_Csv(unittest.TestCase):
  20. """
  21. Test the underlying C csv parser in ways that are not appropriate
  22. from the high level interface. Further tests of this nature are done
  23. in TestDialectRegistry.
  24. """
  25. def _test_arg_valid(self, ctor, arg):
  26. self.assertRaises(TypeError, ctor)
  27. self.assertRaises(TypeError, ctor, None)
  28. self.assertRaises(TypeError, ctor, arg, bad_attr = 0)
  29. self.assertRaises(TypeError, ctor, arg, delimiter = 0)
  30. self.assertRaises(TypeError, ctor, arg, delimiter = 'XX')
  31. self.assertRaises(csv.Error, ctor, arg, 'foo')
  32. self.assertRaises(TypeError, ctor, arg, delimiter=None)
  33. self.assertRaises(TypeError, ctor, arg, delimiter=1)
  34. self.assertRaises(TypeError, ctor, arg, quotechar=1)
  35. self.assertRaises(TypeError, ctor, arg, lineterminator=None)
  36. self.assertRaises(TypeError, ctor, arg, lineterminator=1)
  37. self.assertRaises(TypeError, ctor, arg, quoting=None)
  38. self.assertRaises(TypeError, ctor, arg,
  39. quoting=csv.QUOTE_ALL, quotechar='')
  40. self.assertRaises(TypeError, ctor, arg,
  41. quoting=csv.QUOTE_ALL, quotechar=None)
  42. self.assertRaises(TypeError, ctor, arg,
  43. quoting=csv.QUOTE_NONE, quotechar='')
  44. def test_reader_arg_valid(self):
  45. self._test_arg_valid(csv.reader, [])
  46. self.assertRaises(OSError, csv.reader, BadIterable())
  47. def test_writer_arg_valid(self):
  48. self._test_arg_valid(csv.writer, StringIO())
  49. class BadWriter:
  50. @property
  51. def write(self):
  52. raise OSError
  53. self.assertRaises(OSError, csv.writer, BadWriter())
  54. def _test_default_attrs(self, ctor, *args):
  55. obj = ctor(*args)
  56. # Check defaults
  57. self.assertEqual(obj.dialect.delimiter, ',')
  58. self.assertIs(obj.dialect.doublequote, True)
  59. self.assertEqual(obj.dialect.escapechar, None)
  60. self.assertEqual(obj.dialect.lineterminator, "\r\n")
  61. self.assertEqual(obj.dialect.quotechar, '"')
  62. self.assertEqual(obj.dialect.quoting, csv.QUOTE_MINIMAL)
  63. self.assertIs(obj.dialect.skipinitialspace, False)
  64. self.assertIs(obj.dialect.strict, False)
  65. # Try deleting or changing attributes (they are read-only)
  66. self.assertRaises(AttributeError, delattr, obj.dialect, 'delimiter')
  67. self.assertRaises(AttributeError, setattr, obj.dialect, 'delimiter', ':')
  68. self.assertRaises(AttributeError, delattr, obj.dialect, 'quoting')
  69. self.assertRaises(AttributeError, setattr, obj.dialect,
  70. 'quoting', None)
  71. def test_reader_attrs(self):
  72. self._test_default_attrs(csv.reader, [])
  73. def test_writer_attrs(self):
  74. self._test_default_attrs(csv.writer, StringIO())
  75. def _test_kw_attrs(self, ctor, *args):
  76. # Now try with alternate options
  77. kwargs = dict(delimiter=':', doublequote=False, escapechar='\\',
  78. lineterminator='\r', quotechar='*',
  79. quoting=csv.QUOTE_NONE, skipinitialspace=True,
  80. strict=True)
  81. obj = ctor(*args, **kwargs)
  82. self.assertEqual(obj.dialect.delimiter, ':')
  83. self.assertIs(obj.dialect.doublequote, False)
  84. self.assertEqual(obj.dialect.escapechar, '\\')
  85. self.assertEqual(obj.dialect.lineterminator, "\r")
  86. self.assertEqual(obj.dialect.quotechar, '*')
  87. self.assertEqual(obj.dialect.quoting, csv.QUOTE_NONE)
  88. self.assertIs(obj.dialect.skipinitialspace, True)
  89. self.assertIs(obj.dialect.strict, True)
  90. def test_reader_kw_attrs(self):
  91. self._test_kw_attrs(csv.reader, [])
  92. def test_writer_kw_attrs(self):
  93. self._test_kw_attrs(csv.writer, StringIO())
  94. def _test_dialect_attrs(self, ctor, *args):
  95. # Now try with dialect-derived options
  96. class dialect:
  97. delimiter='-'
  98. doublequote=False
  99. escapechar='^'
  100. lineterminator='$'
  101. quotechar='#'
  102. quoting=csv.QUOTE_ALL
  103. skipinitialspace=True
  104. strict=False
  105. args = args + (dialect,)
  106. obj = ctor(*args)
  107. self.assertEqual(obj.dialect.delimiter, '-')
  108. self.assertIs(obj.dialect.doublequote, False)
  109. self.assertEqual(obj.dialect.escapechar, '^')
  110. self.assertEqual(obj.dialect.lineterminator, "$")
  111. self.assertEqual(obj.dialect.quotechar, '#')
  112. self.assertEqual(obj.dialect.quoting, csv.QUOTE_ALL)
  113. self.assertIs(obj.dialect.skipinitialspace, True)
  114. self.assertIs(obj.dialect.strict, False)
  115. def test_reader_dialect_attrs(self):
  116. self._test_dialect_attrs(csv.reader, [])
  117. def test_writer_dialect_attrs(self):
  118. self._test_dialect_attrs(csv.writer, StringIO())
  119. def _write_test(self, fields, expect, **kwargs):
  120. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  121. writer = csv.writer(fileobj, **kwargs)
  122. writer.writerow(fields)
  123. fileobj.seek(0)
  124. self.assertEqual(fileobj.read(),
  125. expect + writer.dialect.lineterminator)
  126. def _write_error_test(self, exc, fields, **kwargs):
  127. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  128. writer = csv.writer(fileobj, **kwargs)
  129. with self.assertRaises(exc):
  130. writer.writerow(fields)
  131. fileobj.seek(0)
  132. self.assertEqual(fileobj.read(), '')
  133. def test_write_arg_valid(self):
  134. self._write_error_test(csv.Error, None)
  135. self._write_test((), '')
  136. self._write_test([None], '""')
  137. self._write_error_test(csv.Error, [None], quoting = csv.QUOTE_NONE)
  138. # Check that exceptions are passed up the chain
  139. self._write_error_test(OSError, BadIterable())
  140. class BadList:
  141. def __len__(self):
  142. return 10
  143. def __getitem__(self, i):
  144. if i > 2:
  145. raise OSError
  146. self._write_error_test(OSError, BadList())
  147. class BadItem:
  148. def __str__(self):
  149. raise OSError
  150. self._write_error_test(OSError, [BadItem()])
  151. def test_write_bigfield(self):
  152. # This exercises the buffer realloc functionality
  153. bigstring = 'X' * 50000
  154. self._write_test([bigstring,bigstring], '%s,%s' % \
  155. (bigstring, bigstring))
  156. def test_write_quoting(self):
  157. self._write_test(['a',1,'p,q'], 'a,1,"p,q"')
  158. self._write_error_test(csv.Error, ['a',1,'p,q'],
  159. quoting = csv.QUOTE_NONE)
  160. self._write_test(['a',1,'p,q'], 'a,1,"p,q"',
  161. quoting = csv.QUOTE_MINIMAL)
  162. self._write_test(['a',1,'p,q'], '"a",1,"p,q"',
  163. quoting = csv.QUOTE_NONNUMERIC)
  164. self._write_test(['a',1,'p,q'], '"a","1","p,q"',
  165. quoting = csv.QUOTE_ALL)
  166. self._write_test(['a\nb',1], '"a\nb","1"',
  167. quoting = csv.QUOTE_ALL)
  168. def test_write_escape(self):
  169. self._write_test(['a',1,'p,q'], 'a,1,"p,q"',
  170. escapechar='\\')
  171. self._write_error_test(csv.Error, ['a',1,'p,"q"'],
  172. escapechar=None, doublequote=False)
  173. self._write_test(['a',1,'p,"q"'], 'a,1,"p,\\"q\\""',
  174. escapechar='\\', doublequote = False)
  175. self._write_test(['"'], '""""',
  176. escapechar='\\', quoting = csv.QUOTE_MINIMAL)
  177. self._write_test(['"'], '\\"',
  178. escapechar='\\', quoting = csv.QUOTE_MINIMAL,
  179. doublequote = False)
  180. self._write_test(['"'], '\\"',
  181. escapechar='\\', quoting = csv.QUOTE_NONE)
  182. self._write_test(['a',1,'p,q'], 'a,1,p\\,q',
  183. escapechar='\\', quoting = csv.QUOTE_NONE)
  184. self._write_test(['\\', 'a'], '\\\\,a',
  185. escapechar='\\', quoting=csv.QUOTE_NONE)
  186. self._write_test(['\\', 'a'], '\\\\,a',
  187. escapechar='\\', quoting=csv.QUOTE_MINIMAL)
  188. self._write_test(['\\', 'a'], '"\\\\","a"',
  189. escapechar='\\', quoting=csv.QUOTE_ALL)
  190. self._write_test(['\\ ', 'a'], '\\\\ ,a',
  191. escapechar='\\', quoting=csv.QUOTE_MINIMAL)
  192. self._write_test(['\\,', 'a'], '\\\\\\,,a',
  193. escapechar='\\', quoting=csv.QUOTE_NONE)
  194. self._write_test([',\\', 'a'], '",\\\\",a',
  195. escapechar='\\', quoting=csv.QUOTE_MINIMAL)
  196. self._write_test(['C\\', '6', '7', 'X"'], 'C\\\\,6,7,"X"""',
  197. escapechar='\\', quoting=csv.QUOTE_MINIMAL)
  198. def test_write_lineterminator(self):
  199. for lineterminator in '\r\n', '\n', '\r', '!@#', '\0':
  200. with self.subTest(lineterminator=lineterminator):
  201. with StringIO() as sio:
  202. writer = csv.writer(sio, lineterminator=lineterminator)
  203. writer.writerow(['a', 'b'])
  204. writer.writerow([1, 2])
  205. self.assertEqual(sio.getvalue(),
  206. f'a,b{lineterminator}'
  207. f'1,2{lineterminator}')
  208. def test_write_iterable(self):
  209. self._write_test(iter(['a', 1, 'p,q']), 'a,1,"p,q"')
  210. self._write_test(iter(['a', 1, None]), 'a,1,')
  211. self._write_test(iter([]), '')
  212. self._write_test(iter([None]), '""')
  213. self._write_error_test(csv.Error, iter([None]), quoting=csv.QUOTE_NONE)
  214. self._write_test(iter([None, None]), ',')
  215. def test_writerows(self):
  216. class BrokenFile:
  217. def write(self, buf):
  218. raise OSError
  219. writer = csv.writer(BrokenFile())
  220. self.assertRaises(OSError, writer.writerows, [['a']])
  221. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  222. writer = csv.writer(fileobj)
  223. self.assertRaises(TypeError, writer.writerows, None)
  224. writer.writerows([['a', 'b'], ['c', 'd']])
  225. fileobj.seek(0)
  226. self.assertEqual(fileobj.read(), "a,b\r\nc,d\r\n")
  227. def test_writerows_with_none(self):
  228. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  229. writer = csv.writer(fileobj)
  230. writer.writerows([['a', None], [None, 'd']])
  231. fileobj.seek(0)
  232. self.assertEqual(fileobj.read(), "a,\r\n,d\r\n")
  233. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  234. writer = csv.writer(fileobj)
  235. writer.writerows([[None], ['a']])
  236. fileobj.seek(0)
  237. self.assertEqual(fileobj.read(), '""\r\na\r\n')
  238. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  239. writer = csv.writer(fileobj)
  240. writer.writerows([['a'], [None]])
  241. fileobj.seek(0)
  242. self.assertEqual(fileobj.read(), 'a\r\n""\r\n')
  243. def test_writerows_errors(self):
  244. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  245. writer = csv.writer(fileobj)
  246. self.assertRaises(TypeError, writer.writerows, None)
  247. self.assertRaises(OSError, writer.writerows, BadIterable())
  248. @support.cpython_only
  249. @support.requires_legacy_unicode_capi
  250. @warnings_helper.ignore_warnings(category=DeprecationWarning)
  251. def test_writerows_legacy_strings(self):
  252. import _testcapi
  253. c = _testcapi.unicode_legacy_string('a')
  254. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  255. writer = csv.writer(fileobj)
  256. writer.writerows([[c]])
  257. fileobj.seek(0)
  258. self.assertEqual(fileobj.read(), "a\r\n")
  259. def _read_test(self, input, expect, **kwargs):
  260. reader = csv.reader(input, **kwargs)
  261. result = list(reader)
  262. self.assertEqual(result, expect)
  263. def test_read_oddinputs(self):
  264. self._read_test([], [])
  265. self._read_test([''], [[]])
  266. self.assertRaises(csv.Error, self._read_test,
  267. ['"ab"c'], None, strict = 1)
  268. self._read_test(['"ab"c'], [['abc']], doublequote = 0)
  269. self.assertRaises(csv.Error, self._read_test,
  270. [b'abc'], None)
  271. def test_read_eol(self):
  272. self._read_test(['a,b'], [['a','b']])
  273. self._read_test(['a,b\n'], [['a','b']])
  274. self._read_test(['a,b\r\n'], [['a','b']])
  275. self._read_test(['a,b\r'], [['a','b']])
  276. self.assertRaises(csv.Error, self._read_test, ['a,b\rc,d'], [])
  277. self.assertRaises(csv.Error, self._read_test, ['a,b\nc,d'], [])
  278. self.assertRaises(csv.Error, self._read_test, ['a,b\r\nc,d'], [])
  279. def test_read_eof(self):
  280. self._read_test(['a,"'], [['a', '']])
  281. self._read_test(['"a'], [['a']])
  282. self._read_test(['^'], [['\n']], escapechar='^')
  283. self.assertRaises(csv.Error, self._read_test, ['a,"'], [], strict=True)
  284. self.assertRaises(csv.Error, self._read_test, ['"a'], [], strict=True)
  285. self.assertRaises(csv.Error, self._read_test,
  286. ['^'], [], escapechar='^', strict=True)
  287. def test_read_nul(self):
  288. self._read_test(['\0'], [['\0']])
  289. self._read_test(['a,\0b,c'], [['a', '\0b', 'c']])
  290. self._read_test(['a,b\0,c'], [['a', 'b\0', 'c']])
  291. self._read_test(['a,b\\\0,c'], [['a', 'b\0', 'c']], escapechar='\\')
  292. self._read_test(['a,"\0b",c'], [['a', '\0b', 'c']])
  293. def test_read_delimiter(self):
  294. self._read_test(['a,b,c'], [['a', 'b', 'c']])
  295. self._read_test(['a;b;c'], [['a', 'b', 'c']], delimiter=';')
  296. self._read_test(['a\0b\0c'], [['a', 'b', 'c']], delimiter='\0')
  297. def test_read_escape(self):
  298. self._read_test(['a,\\b,c'], [['a', 'b', 'c']], escapechar='\\')
  299. self._read_test(['a,b\\,c'], [['a', 'b,c']], escapechar='\\')
  300. self._read_test(['a,"b\\,c"'], [['a', 'b,c']], escapechar='\\')
  301. self._read_test(['a,"b,\\c"'], [['a', 'b,c']], escapechar='\\')
  302. self._read_test(['a,"b,c\\""'], [['a', 'b,c"']], escapechar='\\')
  303. self._read_test(['a,"b,c"\\'], [['a', 'b,c\\']], escapechar='\\')
  304. self._read_test(['a,^b,c'], [['a', 'b', 'c']], escapechar='^')
  305. self._read_test(['a,\0b,c'], [['a', 'b', 'c']], escapechar='\0')
  306. self._read_test(['a,\\b,c'], [['a', '\\b', 'c']], escapechar=None)
  307. self._read_test(['a,\\b,c'], [['a', '\\b', 'c']])
  308. def test_read_quoting(self):
  309. self._read_test(['1,",3,",5'], [['1', ',3,', '5']])
  310. self._read_test(['1,",3,",5'], [['1', '"', '3', '"', '5']],
  311. quotechar=None, escapechar='\\')
  312. self._read_test(['1,",3,",5'], [['1', '"', '3', '"', '5']],
  313. quoting=csv.QUOTE_NONE, escapechar='\\')
  314. # will this fail where locale uses comma for decimals?
  315. self._read_test([',3,"5",7.3, 9'], [['', 3, '5', 7.3, 9]],
  316. quoting=csv.QUOTE_NONNUMERIC)
  317. self._read_test(['"a\nb", 7'], [['a\nb', ' 7']])
  318. self.assertRaises(ValueError, self._read_test,
  319. ['abc,3'], [[]],
  320. quoting=csv.QUOTE_NONNUMERIC)
  321. self._read_test(['1,@,3,@,5'], [['1', ',3,', '5']], quotechar='@')
  322. self._read_test(['1,\0,3,\0,5'], [['1', ',3,', '5']], quotechar='\0')
  323. def test_read_skipinitialspace(self):
  324. self._read_test(['no space, space, spaces,\ttab'],
  325. [['no space', 'space', 'spaces', '\ttab']],
  326. skipinitialspace=True)
  327. def test_read_bigfield(self):
  328. # This exercises the buffer realloc functionality and field size
  329. # limits.
  330. limit = csv.field_size_limit()
  331. try:
  332. size = 50000
  333. bigstring = 'X' * size
  334. bigline = '%s,%s' % (bigstring, bigstring)
  335. self._read_test([bigline], [[bigstring, bigstring]])
  336. csv.field_size_limit(size)
  337. self._read_test([bigline], [[bigstring, bigstring]])
  338. self.assertEqual(csv.field_size_limit(), size)
  339. csv.field_size_limit(size-1)
  340. self.assertRaises(csv.Error, self._read_test, [bigline], [])
  341. self.assertRaises(TypeError, csv.field_size_limit, None)
  342. self.assertRaises(TypeError, csv.field_size_limit, 1, None)
  343. finally:
  344. csv.field_size_limit(limit)
  345. def test_read_linenum(self):
  346. r = csv.reader(['line,1', 'line,2', 'line,3'])
  347. self.assertEqual(r.line_num, 0)
  348. next(r)
  349. self.assertEqual(r.line_num, 1)
  350. next(r)
  351. self.assertEqual(r.line_num, 2)
  352. next(r)
  353. self.assertEqual(r.line_num, 3)
  354. self.assertRaises(StopIteration, next, r)
  355. self.assertEqual(r.line_num, 3)
  356. def test_roundtrip_quoteed_newlines(self):
  357. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  358. writer = csv.writer(fileobj)
  359. rows = [['a\nb','b'],['c','x\r\nd']]
  360. writer.writerows(rows)
  361. fileobj.seek(0)
  362. for i, row in enumerate(csv.reader(fileobj)):
  363. self.assertEqual(row, rows[i])
  364. def test_roundtrip_escaped_unquoted_newlines(self):
  365. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  366. writer = csv.writer(fileobj,quoting=csv.QUOTE_NONE,escapechar="\\")
  367. rows = [['a\nb','b'],['c','x\r\nd']]
  368. writer.writerows(rows)
  369. fileobj.seek(0)
  370. for i, row in enumerate(csv.reader(fileobj,quoting=csv.QUOTE_NONE,escapechar="\\")):
  371. self.assertEqual(row,rows[i])
  372. class TestDialectRegistry(unittest.TestCase):
  373. def test_registry_badargs(self):
  374. self.assertRaises(TypeError, csv.list_dialects, None)
  375. self.assertRaises(TypeError, csv.get_dialect)
  376. self.assertRaises(csv.Error, csv.get_dialect, None)
  377. self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
  378. self.assertRaises(TypeError, csv.unregister_dialect)
  379. self.assertRaises(csv.Error, csv.unregister_dialect, None)
  380. self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
  381. self.assertRaises(TypeError, csv.register_dialect, None)
  382. self.assertRaises(TypeError, csv.register_dialect, None, None)
  383. self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
  384. self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
  385. badargument=None)
  386. self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
  387. quoting=None)
  388. self.assertRaises(TypeError, csv.register_dialect, [])
  389. def test_registry(self):
  390. class myexceltsv(csv.excel):
  391. delimiter = "\t"
  392. name = "myexceltsv"
  393. expected_dialects = csv.list_dialects() + [name]
  394. expected_dialects.sort()
  395. csv.register_dialect(name, myexceltsv)
  396. self.addCleanup(csv.unregister_dialect, name)
  397. self.assertEqual(csv.get_dialect(name).delimiter, '\t')
  398. got_dialects = sorted(csv.list_dialects())
  399. self.assertEqual(expected_dialects, got_dialects)
  400. def test_register_kwargs(self):
  401. name = 'fedcba'
  402. csv.register_dialect(name, delimiter=';')
  403. self.addCleanup(csv.unregister_dialect, name)
  404. self.assertEqual(csv.get_dialect(name).delimiter, ';')
  405. self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
  406. def test_register_kwargs_override(self):
  407. class mydialect(csv.Dialect):
  408. delimiter = "\t"
  409. quotechar = '"'
  410. doublequote = True
  411. skipinitialspace = False
  412. lineterminator = '\r\n'
  413. quoting = csv.QUOTE_MINIMAL
  414. name = 'test_dialect'
  415. csv.register_dialect(name, mydialect,
  416. delimiter=';',
  417. quotechar="'",
  418. doublequote=False,
  419. skipinitialspace=True,
  420. lineterminator='\n',
  421. quoting=csv.QUOTE_ALL)
  422. self.addCleanup(csv.unregister_dialect, name)
  423. # Ensure that kwargs do override attributes of a dialect class:
  424. dialect = csv.get_dialect(name)
  425. self.assertEqual(dialect.delimiter, ';')
  426. self.assertEqual(dialect.quotechar, "'")
  427. self.assertEqual(dialect.doublequote, False)
  428. self.assertEqual(dialect.skipinitialspace, True)
  429. self.assertEqual(dialect.lineterminator, '\n')
  430. self.assertEqual(dialect.quoting, csv.QUOTE_ALL)
  431. def test_incomplete_dialect(self):
  432. class myexceltsv(csv.Dialect):
  433. delimiter = "\t"
  434. self.assertRaises(csv.Error, myexceltsv)
  435. def test_space_dialect(self):
  436. class space(csv.excel):
  437. delimiter = " "
  438. quoting = csv.QUOTE_NONE
  439. escapechar = "\\"
  440. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  441. fileobj.write("abc def\nc1ccccc1 benzene\n")
  442. fileobj.seek(0)
  443. reader = csv.reader(fileobj, dialect=space())
  444. self.assertEqual(next(reader), ["abc", "def"])
  445. self.assertEqual(next(reader), ["c1ccccc1", "benzene"])
  446. def compare_dialect_123(self, expected, *writeargs, **kwwriteargs):
  447. with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
  448. writer = csv.writer(fileobj, *writeargs, **kwwriteargs)
  449. writer.writerow([1,2,3])
  450. fileobj.seek(0)
  451. self.assertEqual(fileobj.read(), expected)
  452. def test_dialect_apply(self):
  453. class testA(csv.excel):
  454. delimiter = "\t"
  455. class testB(csv.excel):
  456. delimiter = ":"
  457. class testC(csv.excel):
  458. delimiter = "|"
  459. class testUni(csv.excel):
  460. delimiter = "\u039B"
  461. class unspecified():
  462. # A class to pass as dialect but with no dialect attributes.
  463. pass
  464. csv.register_dialect('testC', testC)
  465. try:
  466. self.compare_dialect_123("1,2,3\r\n")
  467. self.compare_dialect_123("1,2,3\r\n", dialect=None)
  468. self.compare_dialect_123("1,2,3\r\n", dialect=unspecified)
  469. self.compare_dialect_123("1\t2\t3\r\n", testA)
  470. self.compare_dialect_123("1:2:3\r\n", dialect=testB())
  471. self.compare_dialect_123("1|2|3\r\n", dialect='testC')
  472. self.compare_dialect_123("1;2;3\r\n", dialect=testA,
  473. delimiter=';')
  474. self.compare_dialect_123("1\u039B2\u039B3\r\n",
  475. dialect=testUni)
  476. finally:
  477. csv.unregister_dialect('testC')
  478. def test_bad_dialect(self):
  479. # Unknown parameter
  480. self.assertRaises(TypeError, csv.reader, [], bad_attr = 0)
  481. # Bad values
  482. self.assertRaises(TypeError, csv.reader, [], delimiter = None)
  483. self.assertRaises(TypeError, csv.reader, [], quoting = -1)
  484. self.assertRaises(TypeError, csv.reader, [], quoting = 100)
  485. def test_copy(self):
  486. for name in csv.list_dialects():
  487. dialect = csv.get_dialect(name)
  488. self.assertRaises(TypeError, copy.copy, dialect)
  489. def test_pickle(self):
  490. for name in csv.list_dialects():
  491. dialect = csv.get_dialect(name)
  492. for proto in range(pickle.HIGHEST_PROTOCOL + 1):
  493. self.assertRaises(TypeError, pickle.dumps, dialect, proto)
  494. class TestCsvBase(unittest.TestCase):
  495. def readerAssertEqual(self, input, expected_result):
  496. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  497. fileobj.write(input)
  498. fileobj.seek(0)
  499. reader = csv.reader(fileobj, dialect = self.dialect)
  500. fields = list(reader)
  501. self.assertEqual(fields, expected_result)
  502. def writerAssertEqual(self, input, expected_result):
  503. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  504. writer = csv.writer(fileobj, dialect = self.dialect)
  505. writer.writerows(input)
  506. fileobj.seek(0)
  507. self.assertEqual(fileobj.read(), expected_result)
  508. class TestDialectExcel(TestCsvBase):
  509. dialect = 'excel'
  510. def test_single(self):
  511. self.readerAssertEqual('abc', [['abc']])
  512. def test_simple(self):
  513. self.readerAssertEqual('1,2,3,4,5', [['1','2','3','4','5']])
  514. def test_blankline(self):
  515. self.readerAssertEqual('', [])
  516. def test_empty_fields(self):
  517. self.readerAssertEqual(',', [['', '']])
  518. def test_singlequoted(self):
  519. self.readerAssertEqual('""', [['']])
  520. def test_singlequoted_left_empty(self):
  521. self.readerAssertEqual('"",', [['','']])
  522. def test_singlequoted_right_empty(self):
  523. self.readerAssertEqual(',""', [['','']])
  524. def test_single_quoted_quote(self):
  525. self.readerAssertEqual('""""', [['"']])
  526. def test_quoted_quotes(self):
  527. self.readerAssertEqual('""""""', [['""']])
  528. def test_inline_quote(self):
  529. self.readerAssertEqual('a""b', [['a""b']])
  530. def test_inline_quotes(self):
  531. self.readerAssertEqual('a"b"c', [['a"b"c']])
  532. def test_quotes_and_more(self):
  533. # Excel would never write a field containing '"a"b', but when
  534. # reading one, it will return 'ab'.
  535. self.readerAssertEqual('"a"b', [['ab']])
  536. def test_lone_quote(self):
  537. self.readerAssertEqual('a"b', [['a"b']])
  538. def test_quote_and_quote(self):
  539. # Excel would never write a field containing '"a" "b"', but when
  540. # reading one, it will return 'a "b"'.
  541. self.readerAssertEqual('"a" "b"', [['a "b"']])
  542. def test_space_and_quote(self):
  543. self.readerAssertEqual(' "a"', [[' "a"']])
  544. def test_quoted(self):
  545. self.readerAssertEqual('1,2,3,"I think, therefore I am",5,6',
  546. [['1', '2', '3',
  547. 'I think, therefore I am',
  548. '5', '6']])
  549. def test_quoted_quote(self):
  550. self.readerAssertEqual('1,2,3,"""I see,"" said the blind man","as he picked up his hammer and saw"',
  551. [['1', '2', '3',
  552. '"I see," said the blind man',
  553. 'as he picked up his hammer and saw']])
  554. def test_quoted_nl(self):
  555. input = '''\
  556. 1,2,3,"""I see,""
  557. said the blind man","as he picked up his
  558. hammer and saw"
  559. 9,8,7,6'''
  560. self.readerAssertEqual(input,
  561. [['1', '2', '3',
  562. '"I see,"\nsaid the blind man',
  563. 'as he picked up his\nhammer and saw'],
  564. ['9','8','7','6']])
  565. def test_dubious_quote(self):
  566. self.readerAssertEqual('12,12,1",', [['12', '12', '1"', '']])
  567. def test_null(self):
  568. self.writerAssertEqual([], '')
  569. def test_single_writer(self):
  570. self.writerAssertEqual([['abc']], 'abc\r\n')
  571. def test_simple_writer(self):
  572. self.writerAssertEqual([[1, 2, 'abc', 3, 4]], '1,2,abc,3,4\r\n')
  573. def test_quotes(self):
  574. self.writerAssertEqual([[1, 2, 'a"bc"', 3, 4]], '1,2,"a""bc""",3,4\r\n')
  575. def test_quote_fieldsep(self):
  576. self.writerAssertEqual([['abc,def']], '"abc,def"\r\n')
  577. def test_newlines(self):
  578. self.writerAssertEqual([[1, 2, 'a\nbc', 3, 4]], '1,2,"a\nbc",3,4\r\n')
  579. class EscapedExcel(csv.excel):
  580. quoting = csv.QUOTE_NONE
  581. escapechar = '\\'
  582. class TestEscapedExcel(TestCsvBase):
  583. dialect = EscapedExcel()
  584. def test_escape_fieldsep(self):
  585. self.writerAssertEqual([['abc,def']], 'abc\\,def\r\n')
  586. def test_read_escape_fieldsep(self):
  587. self.readerAssertEqual('abc\\,def\r\n', [['abc,def']])
  588. class TestDialectUnix(TestCsvBase):
  589. dialect = 'unix'
  590. def test_simple_writer(self):
  591. self.writerAssertEqual([[1, 'abc def', 'abc']], '"1","abc def","abc"\n')
  592. def test_simple_reader(self):
  593. self.readerAssertEqual('"1","abc def","abc"\n', [['1', 'abc def', 'abc']])
  594. class QuotedEscapedExcel(csv.excel):
  595. quoting = csv.QUOTE_NONNUMERIC
  596. escapechar = '\\'
  597. class TestQuotedEscapedExcel(TestCsvBase):
  598. dialect = QuotedEscapedExcel()
  599. def test_write_escape_fieldsep(self):
  600. self.writerAssertEqual([['abc,def']], '"abc,def"\r\n')
  601. def test_read_escape_fieldsep(self):
  602. self.readerAssertEqual('"abc\\,def"\r\n', [['abc,def']])
  603. class TestDictFields(unittest.TestCase):
  604. ### "long" means the row is longer than the number of fieldnames
  605. ### "short" means there are fewer elements in the row than fieldnames
  606. def test_writeheader_return_value(self):
  607. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  608. writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
  609. writeheader_return_value = writer.writeheader()
  610. self.assertEqual(writeheader_return_value, 10)
  611. def test_write_simple_dict(self):
  612. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  613. writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
  614. writer.writeheader()
  615. fileobj.seek(0)
  616. self.assertEqual(fileobj.readline(), "f1,f2,f3\r\n")
  617. writer.writerow({"f1": 10, "f3": "abc"})
  618. fileobj.seek(0)
  619. fileobj.readline() # header
  620. self.assertEqual(fileobj.read(), "10,,abc\r\n")
  621. def test_write_multiple_dict_rows(self):
  622. fileobj = StringIO()
  623. writer = csv.DictWriter(fileobj, fieldnames=["f1", "f2", "f3"])
  624. writer.writeheader()
  625. self.assertEqual(fileobj.getvalue(), "f1,f2,f3\r\n")
  626. writer.writerows([{"f1": 1, "f2": "abc", "f3": "f"},
  627. {"f1": 2, "f2": 5, "f3": "xyz"}])
  628. self.assertEqual(fileobj.getvalue(),
  629. "f1,f2,f3\r\n1,abc,f\r\n2,5,xyz\r\n")
  630. def test_write_no_fields(self):
  631. fileobj = StringIO()
  632. self.assertRaises(TypeError, csv.DictWriter, fileobj)
  633. def test_write_fields_not_in_fieldnames(self):
  634. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  635. writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
  636. # Of special note is the non-string key (issue 19449)
  637. with self.assertRaises(ValueError) as cx:
  638. writer.writerow({"f4": 10, "f2": "spam", 1: "abc"})
  639. exception = str(cx.exception)
  640. self.assertIn("fieldnames", exception)
  641. self.assertIn("'f4'", exception)
  642. self.assertNotIn("'f2'", exception)
  643. self.assertIn("1", exception)
  644. def test_typo_in_extrasaction_raises_error(self):
  645. fileobj = StringIO()
  646. self.assertRaises(ValueError, csv.DictWriter, fileobj, ['f1', 'f2'],
  647. extrasaction="raised")
  648. def test_write_field_not_in_field_names_raise(self):
  649. fileobj = StringIO()
  650. writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="raise")
  651. dictrow = {'f0': 0, 'f1': 1, 'f2': 2, 'f3': 3}
  652. self.assertRaises(ValueError, csv.DictWriter.writerow, writer, dictrow)
  653. def test_write_field_not_in_field_names_ignore(self):
  654. fileobj = StringIO()
  655. writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="ignore")
  656. dictrow = {'f0': 0, 'f1': 1, 'f2': 2, 'f3': 3}
  657. csv.DictWriter.writerow(writer, dictrow)
  658. self.assertEqual(fileobj.getvalue(), "1,2\r\n")
  659. def test_read_dict_fields(self):
  660. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  661. fileobj.write("1,2,abc\r\n")
  662. fileobj.seek(0)
  663. reader = csv.DictReader(fileobj,
  664. fieldnames=["f1", "f2", "f3"])
  665. self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
  666. def test_read_dict_no_fieldnames(self):
  667. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  668. fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
  669. fileobj.seek(0)
  670. reader = csv.DictReader(fileobj)
  671. self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
  672. self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
  673. # Two test cases to make sure existing ways of implicitly setting
  674. # fieldnames continue to work. Both arise from discussion in issue3436.
  675. def test_read_dict_fieldnames_from_file(self):
  676. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  677. fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
  678. fileobj.seek(0)
  679. reader = csv.DictReader(fileobj,
  680. fieldnames=next(csv.reader(fileobj)))
  681. self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
  682. self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
  683. def test_read_dict_fieldnames_chain(self):
  684. import itertools
  685. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  686. fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
  687. fileobj.seek(0)
  688. reader = csv.DictReader(fileobj)
  689. first = next(reader)
  690. for row in itertools.chain([first], reader):
  691. self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
  692. self.assertEqual(row, {"f1": '1', "f2": '2', "f3": 'abc'})
  693. def test_read_long(self):
  694. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  695. fileobj.write("1,2,abc,4,5,6\r\n")
  696. fileobj.seek(0)
  697. reader = csv.DictReader(fileobj,
  698. fieldnames=["f1", "f2"])
  699. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  700. None: ["abc", "4", "5", "6"]})
  701. def test_read_long_with_rest(self):
  702. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  703. fileobj.write("1,2,abc,4,5,6\r\n")
  704. fileobj.seek(0)
  705. reader = csv.DictReader(fileobj,
  706. fieldnames=["f1", "f2"], restkey="_rest")
  707. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  708. "_rest": ["abc", "4", "5", "6"]})
  709. def test_read_long_with_rest_no_fieldnames(self):
  710. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  711. fileobj.write("f1,f2\r\n1,2,abc,4,5,6\r\n")
  712. fileobj.seek(0)
  713. reader = csv.DictReader(fileobj, restkey="_rest")
  714. self.assertEqual(reader.fieldnames, ["f1", "f2"])
  715. self.assertEqual(next(reader), {"f1": '1', "f2": '2',
  716. "_rest": ["abc", "4", "5", "6"]})
  717. def test_read_short(self):
  718. with TemporaryFile("w+", encoding="utf-8") as fileobj:
  719. fileobj.write("1,2,abc,4,5,6\r\n1,2,abc\r\n")
  720. fileobj.seek(0)
  721. reader = csv.DictReader(fileobj,
  722. fieldnames="1 2 3 4 5 6".split(),
  723. restval="DEFAULT")
  724. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  725. "4": '4', "5": '5', "6": '6'})
  726. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  727. "4": 'DEFAULT', "5": 'DEFAULT',
  728. "6": 'DEFAULT'})
  729. def test_read_multi(self):
  730. sample = [
  731. '2147483648,43.0e12,17,abc,def\r\n',
  732. '147483648,43.0e2,17,abc,def\r\n',
  733. '47483648,43.0,170,abc,def\r\n'
  734. ]
  735. reader = csv.DictReader(sample,
  736. fieldnames="i1 float i2 s1 s2".split())
  737. self.assertEqual(next(reader), {"i1": '2147483648',
  738. "float": '43.0e12',
  739. "i2": '17',
  740. "s1": 'abc',
  741. "s2": 'def'})
  742. def test_read_with_blanks(self):
  743. reader = csv.DictReader(["1,2,abc,4,5,6\r\n","\r\n",
  744. "1,2,abc,4,5,6\r\n"],
  745. fieldnames="1 2 3 4 5 6".split())
  746. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  747. "4": '4', "5": '5', "6": '6'})
  748. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  749. "4": '4', "5": '5', "6": '6'})
  750. def test_read_semi_sep(self):
  751. reader = csv.DictReader(["1;2;abc;4;5;6\r\n"],
  752. fieldnames="1 2 3 4 5 6".split(),
  753. delimiter=';')
  754. self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
  755. "4": '4', "5": '5', "6": '6'})
  756. class TestArrayWrites(unittest.TestCase):
  757. def test_int_write(self):
  758. import array
  759. contents = [(20-i) for i in range(20)]
  760. a = array.array('i', contents)
  761. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  762. writer = csv.writer(fileobj, dialect="excel")
  763. writer.writerow(a)
  764. expected = ",".join([str(i) for i in a])+"\r\n"
  765. fileobj.seek(0)
  766. self.assertEqual(fileobj.read(), expected)
  767. def test_double_write(self):
  768. import array
  769. contents = [(20-i)*0.1 for i in range(20)]
  770. a = array.array('d', contents)
  771. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  772. writer = csv.writer(fileobj, dialect="excel")
  773. writer.writerow(a)
  774. expected = ",".join([str(i) for i in a])+"\r\n"
  775. fileobj.seek(0)
  776. self.assertEqual(fileobj.read(), expected)
  777. def test_float_write(self):
  778. import array
  779. contents = [(20-i)*0.1 for i in range(20)]
  780. a = array.array('f', contents)
  781. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  782. writer = csv.writer(fileobj, dialect="excel")
  783. writer.writerow(a)
  784. expected = ",".join([str(i) for i in a])+"\r\n"
  785. fileobj.seek(0)
  786. self.assertEqual(fileobj.read(), expected)
  787. def test_char_write(self):
  788. import array, string
  789. a = array.array('u', string.ascii_letters)
  790. with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
  791. writer = csv.writer(fileobj, dialect="excel")
  792. writer.writerow(a)
  793. expected = ",".join(a)+"\r\n"
  794. fileobj.seek(0)
  795. self.assertEqual(fileobj.read(), expected)
  796. class TestDialectValidity(unittest.TestCase):
  797. def test_quoting(self):
  798. class mydialect(csv.Dialect):
  799. delimiter = ";"
  800. escapechar = '\\'
  801. doublequote = False
  802. skipinitialspace = True
  803. lineterminator = '\r\n'
  804. quoting = csv.QUOTE_NONE
  805. d = mydialect()
  806. self.assertEqual(d.quoting, csv.QUOTE_NONE)
  807. mydialect.quoting = None
  808. self.assertRaises(csv.Error, mydialect)
  809. mydialect.doublequote = True
  810. mydialect.quoting = csv.QUOTE_ALL
  811. mydialect.quotechar = '"'
  812. d = mydialect()
  813. self.assertEqual(d.quoting, csv.QUOTE_ALL)
  814. self.assertEqual(d.quotechar, '"')
  815. self.assertTrue(d.doublequote)
  816. mydialect.quotechar = ""
  817. with self.assertRaises(csv.Error) as cm:
  818. mydialect()
  819. self.assertEqual(str(cm.exception),
  820. '"quotechar" must be a 1-character string')
  821. mydialect.quotechar = "''"
  822. with self.assertRaises(csv.Error) as cm:
  823. mydialect()
  824. self.assertEqual(str(cm.exception),
  825. '"quotechar" must be a 1-character string')
  826. mydialect.quotechar = 4
  827. with self.assertRaises(csv.Error) as cm:
  828. mydialect()
  829. self.assertEqual(str(cm.exception),
  830. '"quotechar" must be string or None, not int')
  831. def test_delimiter(self):
  832. class mydialect(csv.Dialect):
  833. delimiter = ";"
  834. escapechar = '\\'
  835. doublequote = False
  836. skipinitialspace = True
  837. lineterminator = '\r\n'
  838. quoting = csv.QUOTE_NONE
  839. d = mydialect()
  840. self.assertEqual(d.delimiter, ";")
  841. mydialect.delimiter = ":::"
  842. with self.assertRaises(csv.Error) as cm:
  843. mydialect()
  844. self.assertEqual(str(cm.exception),
  845. '"delimiter" must be a 1-character string')
  846. mydialect.delimiter = ""
  847. with self.assertRaises(csv.Error) as cm:
  848. mydialect()
  849. self.assertEqual(str(cm.exception),
  850. '"delimiter" must be a 1-character string')
  851. mydialect.delimiter = b","
  852. with self.assertRaises(csv.Error) as cm:
  853. mydialect()
  854. self.assertEqual(str(cm.exception),
  855. '"delimiter" must be string, not bytes')
  856. mydialect.delimiter = 4
  857. with self.assertRaises(csv.Error) as cm:
  858. mydialect()
  859. self.assertEqual(str(cm.exception),
  860. '"delimiter" must be string, not int')
  861. mydialect.delimiter = None
  862. with self.assertRaises(csv.Error) as cm:
  863. mydialect()
  864. self.assertEqual(str(cm.exception),
  865. '"delimiter" must be string, not NoneType')
  866. def test_escapechar(self):
  867. class mydialect(csv.Dialect):
  868. delimiter = ";"
  869. escapechar = '\\'
  870. doublequote = False
  871. skipinitialspace = True
  872. lineterminator = '\r\n'
  873. quoting = csv.QUOTE_NONE
  874. d = mydialect()
  875. self.assertEqual(d.escapechar, "\\")
  876. mydialect.escapechar = ""
  877. with self.assertRaisesRegex(csv.Error, '"escapechar" must be a 1-character string'):
  878. mydialect()
  879. mydialect.escapechar = "**"
  880. with self.assertRaisesRegex(csv.Error, '"escapechar" must be a 1-character string'):
  881. mydialect()
  882. mydialect.escapechar = b"*"
  883. with self.assertRaisesRegex(csv.Error, '"escapechar" must be string or None, not bytes'):
  884. mydialect()
  885. mydialect.escapechar = 4
  886. with self.assertRaisesRegex(csv.Error, '"escapechar" must be string or None, not int'):
  887. mydialect()
  888. def test_lineterminator(self):
  889. class mydialect(csv.Dialect):
  890. delimiter = ";"
  891. escapechar = '\\'
  892. doublequote = False
  893. skipinitialspace = True
  894. lineterminator = '\r\n'
  895. quoting = csv.QUOTE_NONE
  896. d = mydialect()
  897. self.assertEqual(d.lineterminator, '\r\n')
  898. mydialect.lineterminator = ":::"
  899. d = mydialect()
  900. self.assertEqual(d.lineterminator, ":::")
  901. mydialect.lineterminator = 4
  902. with self.assertRaises(csv.Error) as cm:
  903. mydialect()
  904. self.assertEqual(str(cm.exception),
  905. '"lineterminator" must be a string')
  906. def test_invalid_chars(self):
  907. def create_invalid(field_name, value):
  908. class mydialect(csv.Dialect):
  909. pass
  910. setattr(mydialect, field_name, value)
  911. d = mydialect()
  912. for field_name in ("delimiter", "escapechar", "quotechar"):
  913. with self.subTest(field_name=field_name):
  914. self.assertRaises(csv.Error, create_invalid, field_name, "")
  915. self.assertRaises(csv.Error, create_invalid, field_name, "abc")
  916. self.assertRaises(csv.Error, create_invalid, field_name, b'x')
  917. self.assertRaises(csv.Error, create_invalid, field_name, 5)
  918. class TestSniffer(unittest.TestCase):
  919. sample1 = """\
  920. Harry's, Arlington Heights, IL, 2/1/03, Kimi Hayes
  921. Shark City, Glendale Heights, IL, 12/28/02, Prezence
  922. Tommy's Place, Blue Island, IL, 12/28/02, Blue Sunday/White Crow
  923. Stonecutters Seafood and Chop House, Lemont, IL, 12/19/02, Week Back
  924. """
  925. sample2 = """\
  926. 'Harry''s':'Arlington Heights':'IL':'2/1/03':'Kimi Hayes'
  927. 'Shark City':'Glendale Heights':'IL':'12/28/02':'Prezence'
  928. 'Tommy''s Place':'Blue Island':'IL':'12/28/02':'Blue Sunday/White Crow'
  929. 'Stonecutters ''Seafood'' and Chop House':'Lemont':'IL':'12/19/02':'Week Back'
  930. """
  931. header1 = '''\
  932. "venue","city","state","date","performers"
  933. '''
  934. sample3 = '''\
  935. 05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
  936. 05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
  937. 05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
  938. '''
  939. sample4 = '''\
  940. 2147483648;43.0e12;17;abc;def
  941. 147483648;43.0e2;17;abc;def
  942. 47483648;43.0;170;abc;def
  943. '''
  944. sample5 = "aaa\tbbb\r\nAAA\t\r\nBBB\t\r\n"
  945. sample6 = "a|b|c\r\nd|e|f\r\n"
  946. sample7 = "'a'|'b'|'c'\r\n'd'|e|f\r\n"
  947. # Issue 18155: Use a delimiter that is a special char to regex:
  948. header2 = '''\
  949. "venue"+"city"+"state"+"date"+"performers"
  950. '''
  951. sample8 = """\
  952. Harry's+ Arlington Heights+ IL+ 2/1/03+ Kimi Hayes
  953. Shark City+ Glendale Heights+ IL+ 12/28/02+ Prezence
  954. Tommy's Place+ Blue Island+ IL+ 12/28/02+ Blue Sunday/White Crow
  955. Stonecutters Seafood and Chop House+ Lemont+ IL+ 12/19/02+ Week Back
  956. """
  957. sample9 = """\
  958. 'Harry''s'+ Arlington Heights'+ 'IL'+ '2/1/03'+ 'Kimi Hayes'
  959. 'Shark City'+ Glendale Heights'+' IL'+ '12/28/02'+ 'Prezence'
  960. 'Tommy''s Place'+ Blue Island'+ 'IL'+ '12/28/02'+ 'Blue Sunday/White Crow'
  961. 'Stonecutters ''Seafood'' and Chop House'+ 'Lemont'+ 'IL'+ '12/19/02'+ 'Week Back'
  962. """
  963. sample10 = dedent("""
  964. abc,def
  965. ghijkl,mno
  966. ghi,jkl
  967. """)
  968. sample11 = dedent("""
  969. abc,def
  970. ghijkl,mnop
  971. ghi,jkl
  972. """)
  973. sample12 = dedent(""""time","forces"
  974. 1,1.5
  975. 0.5,5+0j
  976. 0,0
  977. 1+1j,6
  978. """)
  979. sample13 = dedent(""""time","forces"
  980. 0,0
  981. 1,2
  982. a,b
  983. """)
  984. sample14 = """\
  985. abc\0def
  986. ghijkl\0mno
  987. ghi\0jkl
  988. """
  989. def test_issue43625(self):
  990. sniffer = csv.Sniffer()
  991. self.assertTrue(sniffer.has_header(self.sample12))
  992. self.assertFalse(sniffer.has_header(self.sample13))
  993. def test_has_header_strings(self):
  994. "More to document existing (unexpected?) behavior than anything else."
  995. sniffer = csv.Sniffer()
  996. self.assertFalse(sniffer.has_header(self.sample10))
  997. self.assertFalse(sniffer.has_header(self.sample11))
  998. def test_has_header(self):
  999. sniffer = csv.Sniffer()
  1000. self.assertIs(sniffer.has_header(self.sample1), False)
  1001. self.assertIs(sniffer.has_header(self.header1 + self.sample1), True)
  1002. def test_has_header_regex_special_delimiter(self):
  1003. sniffer = csv.Sniffer()
  1004. self.assertIs(sniffer.has_header(self.sample8), False)
  1005. self.assertIs(sniffer.has_header(self.header2 + self.sample8), True)
  1006. def test_guess_quote_and_delimiter(self):
  1007. sniffer = csv.Sniffer()
  1008. for header in (";'123;4';", "'123;4';", ";'123;4'", "'123;4'"):
  1009. with self.subTest(header):
  1010. dialect = sniffer.sniff(header, ",;")
  1011. self.assertEqual(dialect.delimiter, ';')
  1012. self.assertEqual(dialect.quotechar, "'")
  1013. self.assertIs(dialect.doublequote, False)
  1014. self.assertIs(dialect.skipinitialspace, False)
  1015. def test_sniff(self):
  1016. sniffer = csv.Sniffer()
  1017. dialect = sniffer.sniff(self.sample1)
  1018. self.assertEqual(dialect.delimiter, ",")
  1019. self.assertEqual(dialect.quotechar, '"')
  1020. self.assertIs(dialect.skipinitialspace, True)
  1021. dialect = sniffer.sniff(self.sample2)
  1022. self.assertEqual(dialect.delimiter, ":")
  1023. self.assertEqual(dialect.quotechar, "'")
  1024. self.assertIs(dialect.skipinitialspace, False)
  1025. def test_delimiters(self):
  1026. sniffer = csv.Sniffer()
  1027. dialect = sniffer.sniff(self.sample3)
  1028. # given that all three lines in sample3 are equal,
  1029. # I think that any character could have been 'guessed' as the
  1030. # delimiter, depending on dictionary order
  1031. self.assertIn(dialect.delimiter, self.sample3)
  1032. dialect = sniffer.sniff(self.sample3, delimiters="?,")
  1033. self.assertEqual(dialect.delimiter, "?")
  1034. dialect = sniffer.sniff(self.sample3, delimiters="/,")
  1035. self.assertEqual(dialect.delimiter, "/")
  1036. dialect = sniffer.sniff(self.sample4)
  1037. self.assertEqual(dialect.delimiter, ";")
  1038. dialect = sniffer.sniff(self.sample5)
  1039. self.assertEqual(dialect.delimiter, "\t")
  1040. dialect = sniffer.sniff(self.sample6)
  1041. self.assertEqual(dialect.delimiter, "|")
  1042. dialect = sniffer.sniff(self.sample7)
  1043. self.assertEqual(dialect.delimiter, "|")
  1044. self.assertEqual(dialect.quotechar, "'")
  1045. dialect = sniffer.sniff(self.sample8)
  1046. self.assertEqual(dialect.delimiter, '+')
  1047. dialect = sniffer.sniff(self.sample9)
  1048. self.assertEqual(dialect.delimiter, '+')
  1049. self.assertEqual(dialect.quotechar, "'")
  1050. dialect = sniffer.sniff(self.sample14)
  1051. self.assertEqual(dialect.delimiter, '\0')
  1052. def test_doublequote(self):
  1053. sniffer = csv.Sniffer()
  1054. dialect = sniffer.sniff(self.header1)
  1055. self.assertFalse(dialect.doublequote)
  1056. dialect = sniffer.sniff(self.header2)
  1057. self.assertFalse(dialect.doublequote)
  1058. dialect = sniffer.sniff(self.sample2)
  1059. self.assertTrue(dialect.doublequote)
  1060. dialect = sniffer.sniff(self.sample8)
  1061. self.assertFalse(dialect.doublequote)
  1062. dialect = sniffer.sniff(self.sample9)
  1063. self.assertTrue(dialect.doublequote)
  1064. class NUL:
  1065. def write(s, *args):
  1066. pass
  1067. writelines = write
  1068. @unittest.skipUnless(hasattr(sys, "gettotalrefcount"),
  1069. 'requires sys.gettotalrefcount()')
  1070. class TestLeaks(unittest.TestCase):
  1071. def test_create_read(self):
  1072. delta = 0
  1073. lastrc = sys.gettotalrefcount()
  1074. for i in range(20):
  1075. gc.collect()
  1076. self.assertEqual(gc.garbage, [])
  1077. rc = sys.gettotalrefcount()
  1078. csv.reader(["a,b,c\r\n"])
  1079. csv.reader(["a,b,c\r\n"])
  1080. csv.reader(["a,b,c\r\n"])
  1081. delta = rc-lastrc
  1082. lastrc = rc
  1083. # if csv.reader() leaks, last delta should be 3 or more
  1084. self.assertLess(delta, 3)
  1085. def test_create_write(self):
  1086. delta = 0
  1087. lastrc = sys.gettotalrefcount()
  1088. s = NUL()
  1089. for i in range(20):
  1090. gc.collect()
  1091. self.assertEqual(gc.garbage, [])
  1092. rc = sys.gettotalrefcount()
  1093. csv.writer(s)
  1094. csv.writer(s)
  1095. csv.writer(s)
  1096. delta = rc-lastrc
  1097. lastrc = rc
  1098. # if csv.writer() leaks, last delta should be 3 or more
  1099. self.assertLess(delta, 3)
  1100. def test_read(self):
  1101. delta = 0
  1102. rows = ["a,b,c\r\n"]*5
  1103. lastrc = sys.gettotalrefcount()
  1104. for i in range(20):
  1105. gc.collect()
  1106. self.assertEqual(gc.garbage, [])
  1107. rc = sys.gettotalrefcount()
  1108. rdr = csv.reader(rows)
  1109. for row in rdr:
  1110. pass
  1111. delta = rc-lastrc
  1112. lastrc = rc
  1113. # if reader leaks during read, delta should be 5 or more
  1114. self.assertLess(delta, 5)
  1115. def test_write(self):
  1116. delta = 0
  1117. rows = [[1,2,3]]*5
  1118. s = NUL()
  1119. lastrc = sys.gettotalrefcount()
  1120. for i in range(20):
  1121. gc.collect()
  1122. self.assertEqual(gc.garbage, [])
  1123. rc = sys.gettotalrefcount()
  1124. writer = csv.writer(s)
  1125. for row in rows:
  1126. writer.writerow(row)
  1127. delta = rc-lastrc
  1128. lastrc = rc
  1129. # if writer leaks during write, last delta should be 5 or more
  1130. self.assertLess(delta, 5)
  1131. class TestUnicode(unittest.TestCase):
  1132. names = ["Martin von Löwis",
  1133. "Marc André Lemburg",
  1134. "Guido van Rossum",
  1135. "François Pinard"]
  1136. def test_unicode_read(self):
  1137. with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
  1138. fileobj.write(",".join(self.names) + "\r\n")
  1139. fileobj.seek(0)
  1140. reader = csv.reader(fileobj)
  1141. self.assertEqual(list(reader), [self.names])
  1142. def test_unicode_write(self):
  1143. with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
  1144. writer = csv.writer(fileobj)
  1145. writer.writerow(self.names)
  1146. expected = ",".join(self.names)+"\r\n"
  1147. fileobj.seek(0)
  1148. self.assertEqual(fileobj.read(), expected)
  1149. class KeyOrderingTest(unittest.TestCase):
  1150. def test_ordering_for_the_dict_reader_and_writer(self):
  1151. resultset = set()
  1152. for keys in permutations("abcde"):
  1153. with TemporaryFile('w+', newline='', encoding="utf-8") as fileobject:
  1154. dw = csv.DictWriter(fileobject, keys)
  1155. dw.writeheader()
  1156. fileobject.seek(0)
  1157. dr = csv.DictReader(fileobject)
  1158. kt = tuple(dr.fieldnames)
  1159. self.assertEqual(keys, kt)
  1160. resultset.add(kt)
  1161. # Final sanity check: were all permutations unique?
  1162. self.assertEqual(len(resultset), 120, "Key ordering: some key permutations not collected (expected 120)")
  1163. def test_ordered_dict_reader(self):
  1164. data = dedent('''\
  1165. FirstName,LastName
  1166. Eric,Idle
  1167. Graham,Chapman,Over1,Over2
  1168. Under1
  1169. John,Cleese
  1170. ''').splitlines()
  1171. self.assertEqual(list(csv.DictReader(data)),
  1172. [OrderedDict([('FirstName', 'Eric'), ('LastName', 'Idle')]),
  1173. OrderedDict([('FirstName', 'Graham'), ('LastName', 'Chapman'),
  1174. (None, ['Over1', 'Over2'])]),
  1175. OrderedDict([('FirstName', 'Under1'), ('LastName', None)]),
  1176. OrderedDict([('FirstName', 'John'), ('LastName', 'Cleese')]),
  1177. ])
  1178. self.assertEqual(list(csv.DictReader(data, restkey='OtherInfo')),
  1179. [OrderedDict([('FirstName', 'Eric'), ('LastName', 'Idle')]),
  1180. OrderedDict([('FirstName', 'Graham'), ('LastName', 'Chapman'),
  1181. ('OtherInfo', ['Over1', 'Over2'])]),
  1182. OrderedDict([('FirstName', 'Under1'), ('LastName', None)]),
  1183. OrderedDict([('FirstName', 'John'), ('LastName', 'Cleese')]),
  1184. ])
  1185. del data[0] # Remove the header row
  1186. self.assertEqual(list(csv.DictReader(data, fieldnames=['fname', 'lname'])),
  1187. [OrderedDict([('fname', 'Eric'), ('lname', 'Idle')]),
  1188. OrderedDict([('fname', 'Graham'), ('lname', 'Chapman'),
  1189. (None, ['Over1', 'Over2'])]),
  1190. OrderedDict([('fname', 'Under1'), ('lname', None)]),
  1191. OrderedDict([('fname', 'John'), ('lname', 'Cleese')]),
  1192. ])
  1193. class MiscTestCase(unittest.TestCase):
  1194. def test__all__(self):
  1195. extra = {'__doc__', '__version__'}
  1196. support.check__all__(self, csv, ('csv', '_csv'), extra=extra)
  1197. def test_subclassable(self):
  1198. # issue 44089
  1199. class Foo(csv.Error): ...
  1200. if __name__ == '__main__':
  1201. unittest.main()