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