Lib/test/test_csv.py

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