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