mirror of
https://github.com/python/cpython.git
synced 2026-10-29 10:55:16 +08:00
Serhiy StorchakaandClaude Opus 5
a60343ed17
Cache in the dialect a 128-bit set of ASCII characters which need quoting or escaping (delimiter, quotechar, escapechar, '\r', '\n' and characters of lineterminator) and a flag whether any of them is non-ASCII. Testing a character is now one bit test instead of five comparisons and a call to PyUnicode_FindChar(). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2109 lines
87 KiB
Python
2109 lines
87 KiB
Python
# Copyright (C) 2001 Python Software Foundation
|
||
# csv package unit tests
|
||
|
||
import copy
|
||
import sys
|
||
import unittest
|
||
from io import StringIO
|
||
from tempfile import TemporaryFile
|
||
import csv
|
||
import gc
|
||
import pickle
|
||
from test import support
|
||
from test.support import cpython_only, import_helper, check_disallow_instantiation
|
||
from test.support.import_helper import ensure_lazy_imports
|
||
from itertools import permutations
|
||
from textwrap import dedent
|
||
from collections import OrderedDict
|
||
|
||
|
||
class BadIterable:
|
||
def __iter__(self):
|
||
raise OSError
|
||
|
||
|
||
class Test_Csv(unittest.TestCase):
|
||
"""
|
||
Test the underlying C csv parser in ways that are not appropriate
|
||
from the high level interface. Further tests of this nature are done
|
||
in TestDialectRegistry.
|
||
"""
|
||
def _test_arg_valid(self, ctor, arg):
|
||
ctor(arg)
|
||
self.assertRaises(TypeError, ctor)
|
||
self.assertRaises(TypeError, ctor, None)
|
||
self.assertRaises(TypeError, ctor, arg, bad_attr=0)
|
||
self.assertRaises(TypeError, ctor, arg, delimiter='')
|
||
self.assertRaises(TypeError, ctor, arg, escapechar='')
|
||
self.assertRaises(TypeError, ctor, arg, quotechar='')
|
||
self.assertRaises(TypeError, ctor, arg, delimiter='^^')
|
||
self.assertRaises(TypeError, ctor, arg, escapechar='^^')
|
||
self.assertRaises(TypeError, ctor, arg, quotechar='^^')
|
||
self.assertRaises(csv.Error, ctor, arg, 'foo')
|
||
self.assertRaises(TypeError, ctor, arg, delimiter=None)
|
||
self.assertRaises(TypeError, ctor, arg, delimiter=1)
|
||
self.assertRaises(TypeError, ctor, arg, escapechar=1)
|
||
self.assertRaises(TypeError, ctor, arg, quotechar=1)
|
||
self.assertRaises(TypeError, ctor, arg, lineterminator=None)
|
||
self.assertRaises(TypeError, ctor, arg, lineterminator=1)
|
||
self.assertRaises(TypeError, ctor, arg, quoting=None)
|
||
self.assertRaises(TypeError, ctor, arg,
|
||
quoting=csv.QUOTE_ALL, quotechar='')
|
||
self.assertRaises(TypeError, ctor, arg,
|
||
quoting=csv.QUOTE_ALL, quotechar=None)
|
||
self.assertRaises(TypeError, ctor, arg,
|
||
quoting=csv.QUOTE_NONE, quotechar='')
|
||
self.assertRaises(ValueError, ctor, arg, delimiter='\n')
|
||
self.assertRaises(ValueError, ctor, arg, escapechar='\n')
|
||
self.assertRaises(ValueError, ctor, arg, quotechar='\n')
|
||
self.assertRaises(ValueError, ctor, arg, delimiter='\r')
|
||
self.assertRaises(ValueError, ctor, arg, escapechar='\r')
|
||
self.assertRaises(ValueError, ctor, arg, quotechar='\r')
|
||
ctor(arg, delimiter=' ')
|
||
ctor(arg, escapechar=' ')
|
||
ctor(arg, quotechar=' ')
|
||
ctor(arg, delimiter='\t', skipinitialspace=True)
|
||
ctor(arg, escapechar='\t', skipinitialspace=True)
|
||
ctor(arg, quotechar='\t', skipinitialspace=True)
|
||
ctor(arg, delimiter=' ', skipinitialspace=True)
|
||
self.assertRaises(ValueError, ctor, arg,
|
||
escapechar=' ', skipinitialspace=True)
|
||
self.assertRaises(ValueError, ctor, arg,
|
||
quotechar=' ', skipinitialspace=True)
|
||
ctor(arg, delimiter='^')
|
||
ctor(arg, escapechar='^')
|
||
ctor(arg, quotechar='^')
|
||
self.assertRaises(ValueError, ctor, arg, delimiter='^', escapechar='^')
|
||
self.assertRaises(ValueError, ctor, arg, delimiter='^', quotechar='^')
|
||
self.assertRaises(ValueError, ctor, arg, escapechar='^', quotechar='^')
|
||
ctor(arg, delimiter='\x85')
|
||
ctor(arg, escapechar='\x85')
|
||
ctor(arg, quotechar='\x85')
|
||
ctor(arg, lineterminator='\x85')
|
||
self.assertRaises(ValueError, ctor, arg,
|
||
delimiter='\x85', lineterminator='\x85')
|
||
self.assertRaises(ValueError, ctor, arg,
|
||
escapechar='\x85', lineterminator='\x85')
|
||
self.assertRaises(ValueError, ctor, arg,
|
||
quotechar='\x85', lineterminator='\x85')
|
||
|
||
def test_reader_arg_valid(self):
|
||
self._test_arg_valid(csv.reader, [])
|
||
self.assertRaises(OSError, csv.reader, BadIterable())
|
||
|
||
def test_writer_arg_valid(self):
|
||
self._test_arg_valid(csv.writer, StringIO())
|
||
class BadWriter:
|
||
@property
|
||
def write(self):
|
||
raise OSError
|
||
self.assertRaises(OSError, csv.writer, BadWriter())
|
||
|
||
def _test_default_attrs(self, ctor, *args):
|
||
obj = ctor(*args)
|
||
# Check defaults
|
||
self.assertEqual(obj.dialect.delimiter, ',')
|
||
self.assertIs(obj.dialect.doublequote, True)
|
||
self.assertEqual(obj.dialect.escapechar, None)
|
||
self.assertEqual(obj.dialect.lineterminator, "\r\n")
|
||
self.assertEqual(obj.dialect.quotechar, '"')
|
||
self.assertEqual(obj.dialect.quoting, csv.QUOTE_MINIMAL)
|
||
self.assertIs(obj.dialect.skipinitialspace, False)
|
||
self.assertIs(obj.dialect.strict, False)
|
||
# Try deleting or changing attributes (they are read-only)
|
||
self.assertRaises(AttributeError, delattr, obj.dialect, 'delimiter')
|
||
self.assertRaises(AttributeError, setattr, obj.dialect, 'delimiter', ':')
|
||
self.assertRaises(AttributeError, delattr, obj.dialect, 'quoting')
|
||
self.assertRaises(AttributeError, setattr, obj.dialect,
|
||
'quoting', None)
|
||
|
||
def test_reader_attrs(self):
|
||
self._test_default_attrs(csv.reader, [])
|
||
|
||
def test_writer_attrs(self):
|
||
self._test_default_attrs(csv.writer, StringIO())
|
||
|
||
def _test_kw_attrs(self, ctor, *args):
|
||
# Now try with alternate options
|
||
kwargs = dict(delimiter=':', doublequote=False, escapechar='\\',
|
||
lineterminator='\r', quotechar='*',
|
||
quoting=csv.QUOTE_NONE, skipinitialspace=True,
|
||
strict=True)
|
||
obj = ctor(*args, **kwargs)
|
||
self.assertEqual(obj.dialect.delimiter, ':')
|
||
self.assertIs(obj.dialect.doublequote, False)
|
||
self.assertEqual(obj.dialect.escapechar, '\\')
|
||
self.assertEqual(obj.dialect.lineterminator, "\r")
|
||
self.assertEqual(obj.dialect.quotechar, '*')
|
||
self.assertEqual(obj.dialect.quoting, csv.QUOTE_NONE)
|
||
self.assertIs(obj.dialect.skipinitialspace, True)
|
||
self.assertIs(obj.dialect.strict, True)
|
||
|
||
def test_reader_kw_attrs(self):
|
||
self._test_kw_attrs(csv.reader, [])
|
||
|
||
def test_writer_kw_attrs(self):
|
||
self._test_kw_attrs(csv.writer, StringIO())
|
||
|
||
def _test_dialect_attrs(self, ctor, *args):
|
||
# Now try with dialect-derived options
|
||
class dialect:
|
||
delimiter='-'
|
||
doublequote=False
|
||
escapechar='^'
|
||
lineterminator='$'
|
||
quotechar='#'
|
||
quoting=csv.QUOTE_ALL
|
||
skipinitialspace=True
|
||
strict=False
|
||
args = args + (dialect,)
|
||
obj = ctor(*args)
|
||
self.assertEqual(obj.dialect.delimiter, '-')
|
||
self.assertIs(obj.dialect.doublequote, False)
|
||
self.assertEqual(obj.dialect.escapechar, '^')
|
||
self.assertEqual(obj.dialect.lineterminator, "$")
|
||
self.assertEqual(obj.dialect.quotechar, '#')
|
||
self.assertEqual(obj.dialect.quoting, csv.QUOTE_ALL)
|
||
self.assertIs(obj.dialect.skipinitialspace, True)
|
||
self.assertIs(obj.dialect.strict, False)
|
||
|
||
def test_reader_dialect_attrs(self):
|
||
self._test_dialect_attrs(csv.reader, [])
|
||
|
||
def test_writer_dialect_attrs(self):
|
||
self._test_dialect_attrs(csv.writer, StringIO())
|
||
|
||
|
||
def _write_test(self, fields, expect, **kwargs):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, **kwargs)
|
||
writer.writerow(fields)
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(),
|
||
expect + writer.dialect.lineterminator)
|
||
|
||
def _write_error_test(self, exc, fields, **kwargs):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, **kwargs)
|
||
with self.assertRaises(exc):
|
||
writer.writerow(fields)
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), '')
|
||
|
||
def test_write_arg_valid(self):
|
||
self._write_error_test(csv.Error, None)
|
||
# Check that exceptions are passed up the chain
|
||
self._write_error_test(OSError, BadIterable())
|
||
class BadList:
|
||
def __len__(self):
|
||
return 10
|
||
def __getitem__(self, i):
|
||
if i > 2:
|
||
raise OSError
|
||
self._write_error_test(OSError, BadList())
|
||
class BadItem:
|
||
def __str__(self):
|
||
raise OSError
|
||
self._write_error_test(OSError, [BadItem()])
|
||
def test_write_bigfield(self):
|
||
# This exercises the buffer realloc functionality
|
||
bigstring = 'X' * 50000
|
||
self._write_test([bigstring,bigstring], '%s,%s' % \
|
||
(bigstring, bigstring))
|
||
|
||
def test_write_quoting(self):
|
||
self._write_test(['a',1,'p,q'], 'a,1,"p,q"')
|
||
self._write_error_test(csv.Error, ['a',1,'p,q'],
|
||
quoting = csv.QUOTE_NONE)
|
||
self._write_test(['a',1,'p,q'], 'a,1,"p,q"',
|
||
quoting = csv.QUOTE_MINIMAL)
|
||
self._write_test(['a',1,'p,q'], '"a",1,"p,q"',
|
||
quoting = csv.QUOTE_NONNUMERIC)
|
||
self._write_test(['a',1,'p,q'], '"a","1","p,q"',
|
||
quoting = csv.QUOTE_ALL)
|
||
self._write_test(['a\nb',1], '"a\nb","1"',
|
||
quoting = csv.QUOTE_ALL)
|
||
self._write_test(['a','',None,1], '"a","",,1',
|
||
quoting = csv.QUOTE_STRINGS)
|
||
self._write_test(['a','',None,1], '"a","",,"1"',
|
||
quoting = csv.QUOTE_NOTNULL)
|
||
# FULLWIDTH QUOTATION MARK
|
||
self._write_test(['a', 1, 'p,q', 'r＂s', 'x！y'],
|
||
'a,1,＂p,q＂,＂r＂＂s＂,x！y',
|
||
quotechar='＂')
|
||
|
||
def test_write_delimiter(self):
|
||
self._write_test(['a', 1, 'p,q', 'x;y'], 'a,1,"p,q",x;y')
|
||
self._write_test(['a', 1, 'p;q', 'x,y'], 'a;1;"p;q";x,y', delimiter=';')
|
||
self._write_test(['a', 1, 'p\0q', 'x,y'], 'a\x001\0"p\0q"\0x,y',
|
||
delimiter='\0')
|
||
self._write_test(['a', 1, 'p🍌q', 'x🍍y'], 'a🍌1🍌"p🍌q"🍌x🍍y',
|
||
delimiter='🍌')
|
||
|
||
def test_write_escape(self):
|
||
self._write_test(['a',1,'p,q'], 'a,1,"p,q"',
|
||
escapechar='\\')
|
||
self._write_error_test(csv.Error, ['a',1,'p,"q"'],
|
||
escapechar=None, doublequote=False)
|
||
self._write_test(['a',1,'p,"q"'], 'a,1,"p,\\"q\\""',
|
||
escapechar='\\', doublequote = False)
|
||
self._write_test(['"'], '""""',
|
||
escapechar='\\', quoting = csv.QUOTE_MINIMAL)
|
||
self._write_test(['"'], '\\"',
|
||
escapechar='\\', quoting = csv.QUOTE_MINIMAL,
|
||
doublequote = False)
|
||
self._write_test(['"'], '\\"',
|
||
escapechar='\\', quoting = csv.QUOTE_NONE)
|
||
self._write_test(['a',1,'p,q'], 'a,1,p\\,q',
|
||
escapechar='\\', quoting = csv.QUOTE_NONE)
|
||
self._write_test(['\\', 'a'], '\\\\,a',
|
||
escapechar='\\', quoting=csv.QUOTE_NONE)
|
||
self._write_test(['\\', 'a'], '\\\\,a',
|
||
escapechar='\\', quoting=csv.QUOTE_MINIMAL)
|
||
self._write_test(['\\', 'a'], '"\\\\","a"',
|
||
escapechar='\\', quoting=csv.QUOTE_ALL)
|
||
self._write_test(['\\ ', 'a'], '\\\\ ,a',
|
||
escapechar='\\', quoting=csv.QUOTE_MINIMAL)
|
||
self._write_test(['\\,', 'a'], '\\\\\\,,a',
|
||
escapechar='\\', quoting=csv.QUOTE_NONE)
|
||
self._write_test([',\\', 'a'], '",\\\\",a',
|
||
escapechar='\\', quoting=csv.QUOTE_MINIMAL)
|
||
self._write_test(['C\\', '6', '7', 'X"'], 'C\\\\,6,7,"X"""',
|
||
escapechar='\\', quoting=csv.QUOTE_MINIMAL)
|
||
# SYMBOL FOR ESCAPE
|
||
self._write_test(['a', 1, 'p,q', 'r\u241bs', 'x\u241ay'],
|
||
'a,1,p\u241b,q,r\u241b\u241bs,x\u241ay',
|
||
escapechar='\u241b', quoting=csv.QUOTE_NONE)
|
||
|
||
def test_write_lineterminator(self):
|
||
for lineterminator in ('\r\n', '\n', '\r', '!@#', '\0',
|
||
'\x85', '\u2028', '\U0001f600'):
|
||
with self.subTest(lineterminator=lineterminator):
|
||
with StringIO() as sio:
|
||
writer = csv.writer(sio, lineterminator=lineterminator)
|
||
writer.writerow(['a', 'b'])
|
||
writer.writerow([1, 2])
|
||
writer.writerow(['\r', '\n'])
|
||
writer.writerow([f'a{lineterminator[-1]}b', 'c'])
|
||
self.assertEqual(sio.getvalue(),
|
||
f'a,b{lineterminator}'
|
||
f'1,2{lineterminator}'
|
||
f'"\r","\n"{lineterminator}'
|
||
f'"a{lineterminator[-1]}b",c{lineterminator}')
|
||
|
||
def test_write_iterable(self):
|
||
self._write_test(iter(['a', 1, 'p,q']), 'a,1,"p,q"')
|
||
self._write_test(iter(['a', 1, None]), 'a,1,')
|
||
self._write_test(iter([]), '')
|
||
self._write_test(iter([None]), '""')
|
||
self._write_error_test(csv.Error, iter([None]), quoting=csv.QUOTE_NONE)
|
||
self._write_test(iter([None, None]), ',')
|
||
|
||
def test_writerows(self):
|
||
class BrokenFile:
|
||
def write(self, buf):
|
||
raise OSError
|
||
writer = csv.writer(BrokenFile())
|
||
self.assertRaises(OSError, writer.writerows, [['a']])
|
||
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
self.assertRaises(TypeError, writer.writerows, None)
|
||
writer.writerows([['a', 'b'], ['c', 'd']])
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), "a,b\r\nc,d\r\n")
|
||
|
||
def test_writerows_with_none(self):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
writer.writerows([['a', None], [None, 'd']])
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), "a,\r\n,d\r\n")
|
||
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
writer.writerows([[None], ['a']])
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), '""\r\na\r\n')
|
||
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
writer.writerows([['a'], [None]])
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), 'a\r\n""\r\n')
|
||
|
||
|
||
def test_write_empty_fields(self):
|
||
self._write_test((), '')
|
||
self._write_test([''], '""')
|
||
self._write_error_test(csv.Error, [''], quoting=csv.QUOTE_NONE)
|
||
self._write_test([''], '""', quoting=csv.QUOTE_STRINGS)
|
||
self._write_test([''], '""', quoting=csv.QUOTE_NOTNULL)
|
||
self._write_test([None], '""')
|
||
self._write_error_test(csv.Error, [None], quoting=csv.QUOTE_NONE)
|
||
self._write_error_test(csv.Error, [None], quoting=csv.QUOTE_STRINGS)
|
||
self._write_error_test(csv.Error, [None], quoting=csv.QUOTE_NOTNULL)
|
||
self._write_test(['', ''], ',')
|
||
self._write_test([None, None], ',')
|
||
|
||
def test_write_empty_fields_space_delimiter(self):
|
||
self._write_test([''], '""', delimiter=' ', skipinitialspace=False)
|
||
self._write_test([''], '""', delimiter=' ', skipinitialspace=True)
|
||
self._write_test([None], '""', delimiter=' ', skipinitialspace=False)
|
||
self._write_test([None], '""', delimiter=' ', skipinitialspace=True)
|
||
|
||
self._write_test(['', ''], ' ', delimiter=' ', skipinitialspace=False)
|
||
self._write_test(['', ''], '"" ""', delimiter=' ', skipinitialspace=True)
|
||
self._write_test([None, None], ' ', delimiter=' ', skipinitialspace=False)
|
||
self._write_test([None, None], '"" ""', delimiter=' ', skipinitialspace=True)
|
||
|
||
self._write_test(['', ''], ' ', delimiter=' ', skipinitialspace=False,
|
||
quoting=csv.QUOTE_NONE)
|
||
self._write_error_test(csv.Error, ['', ''],
|
||
delimiter=' ', skipinitialspace=True,
|
||
quoting=csv.QUOTE_NONE)
|
||
for quoting in csv.QUOTE_STRINGS, csv.QUOTE_NOTNULL:
|
||
self._write_test(['', ''], '"" ""', delimiter=' ', skipinitialspace=False,
|
||
quoting=quoting)
|
||
self._write_test(['', ''], '"" ""', delimiter=' ', skipinitialspace=True,
|
||
quoting=quoting)
|
||
|
||
for quoting in csv.QUOTE_NONE, csv.QUOTE_STRINGS, csv.QUOTE_NOTNULL:
|
||
self._write_test([None, None], ' ', delimiter=' ', skipinitialspace=False,
|
||
quoting=quoting)
|
||
self._write_error_test(csv.Error, [None, None],
|
||
delimiter=' ', skipinitialspace=True,
|
||
quoting=quoting)
|
||
|
||
def test_writerows_errors(self):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
self.assertRaises(TypeError, writer.writerows, None)
|
||
self.assertRaises(OSError, writer.writerows, BadIterable())
|
||
|
||
def _read_test(self, input, expect, **kwargs):
|
||
reader = csv.reader(input, **kwargs)
|
||
result = list(reader)
|
||
self.assertEqual(result, expect)
|
||
|
||
def test_read_oddinputs(self):
|
||
self._read_test([], [])
|
||
self._read_test([''], [[]])
|
||
self.assertRaises(csv.Error, self._read_test,
|
||
['"ab"c'], None, strict = 1)
|
||
self._read_test(['"ab"c'], [['abc']], doublequote = 0)
|
||
|
||
self.assertRaises(csv.Error, self._read_test,
|
||
[b'abc'], None)
|
||
|
||
def test_read_eol(self):
|
||
self._read_test(['a,b', 'c,d'], [['a','b'], ['c','d']])
|
||
self._read_test(['a,b\n', 'c,d\n'], [['a','b'], ['c','d']])
|
||
self._read_test(['a,b\r\n', 'c,d\r\n'], [['a','b'], ['c','d']])
|
||
self._read_test(['a,b\r', 'c,d\r'], [['a','b'], ['c','d']])
|
||
|
||
errmsg = "with newline=''"
|
||
with self.assertRaisesRegex(csv.Error, errmsg):
|
||
next(csv.reader(['a,b\rc,d']))
|
||
with self.assertRaisesRegex(csv.Error, errmsg):
|
||
next(csv.reader(['a,b\nc,d']))
|
||
with self.assertRaisesRegex(csv.Error, errmsg):
|
||
next(csv.reader(['a,b\r\nc,d']))
|
||
|
||
def test_read_eof(self):
|
||
self._read_test(['a,"'], [['a', '']])
|
||
self._read_test(['"a'], [['a']])
|
||
self._read_test(['^'], [['\n']], escapechar='^')
|
||
self.assertRaises(csv.Error, self._read_test, ['a,"'], [], strict=True)
|
||
self.assertRaises(csv.Error, self._read_test, ['"a'], [], strict=True)
|
||
self.assertRaises(csv.Error, self._read_test,
|
||
['^'], [], escapechar='^', strict=True)
|
||
|
||
def test_read_nul(self):
|
||
self._read_test(['\0'], [['\0']])
|
||
self._read_test(['a,\0b,c'], [['a', '\0b', 'c']])
|
||
self._read_test(['a,b\0,c'], [['a', 'b\0', 'c']])
|
||
self._read_test(['a,b\\\0,c'], [['a', 'b\0', 'c']], escapechar='\\')
|
||
self._read_test(['a,"\0b",c'], [['a', '\0b', 'c']])
|
||
|
||
def test_read_delimiter(self):
|
||
self._read_test(['a,b,c'], [['a', 'b', 'c']])
|
||
self._read_test(['a;b;c'], [['a', 'b', 'c']], delimiter=';')
|
||
self._read_test(['a\0b\0c'], [['a', 'b', 'c']], delimiter='\0')
|
||
|
||
def test_read_escape(self):
|
||
self._read_test(['a,\\b,c'], [['a', 'b', 'c']], escapechar='\\')
|
||
self._read_test(['a,b\\,c'], [['a', 'b,c']], escapechar='\\')
|
||
self._read_test(['a,"b\\,c"'], [['a', 'b,c']], escapechar='\\')
|
||
self._read_test(['a,"b,\\c"'], [['a', 'b,c']], escapechar='\\')
|
||
self._read_test(['a,"b,c\\""'], [['a', 'b,c"']], escapechar='\\')
|
||
self._read_test(['a,"b,c"\\'], [['a', 'b,c\\']], escapechar='\\')
|
||
self._read_test(['a,^b,c'], [['a', 'b', 'c']], escapechar='^')
|
||
self._read_test(['a,\0b,c'], [['a', 'b', 'c']], escapechar='\0')
|
||
self._read_test(['a,\\b,c'], [['a', '\\b', 'c']], escapechar=None)
|
||
self._read_test(['a,\\b,c'], [['a', '\\b', 'c']])
|
||
|
||
def test_read_quoting(self):
|
||
self._read_test(['1,",3,",5'], [['1', ',3,', '5']])
|
||
self._read_test(['1,",3,",5'], [['1', '"', '3', '"', '5']],
|
||
quotechar=None, escapechar='\\')
|
||
self._read_test(['1,",3,",5'], [['1', '"', '3', '"', '5']],
|
||
quoting=csv.QUOTE_NONE, escapechar='\\')
|
||
# will this fail where locale uses comma for decimals?
|
||
self._read_test([',3,"5",7.3, 9'], [['', 3, '5', 7.3, 9]],
|
||
quoting=csv.QUOTE_NONNUMERIC)
|
||
self._read_test([',3,"5",7.3, 9'], [[None, '3', '5', '7.3', ' 9']],
|
||
quoting=csv.QUOTE_NOTNULL)
|
||
self._read_test([',3,"5",7.3, 9'], [[None, 3, '5', 7.3, 9]],
|
||
quoting=csv.QUOTE_STRINGS)
|
||
|
||
self._read_test([',,"",'], [['', '', '', '']])
|
||
self._read_test([',,"",'], [['', '', '', '']],
|
||
quoting=csv.QUOTE_NONNUMERIC)
|
||
self._read_test([',,"",'], [[None, None, '', None]],
|
||
quoting=csv.QUOTE_NOTNULL)
|
||
self._read_test([',,"",'], [[None, None, '', None]],
|
||
quoting=csv.QUOTE_STRINGS)
|
||
|
||
self._read_test(['"a\nb", 7'], [['a\nb', ' 7']])
|
||
self.assertRaises(ValueError, self._read_test,
|
||
['abc,3'], [[]],
|
||
quoting=csv.QUOTE_NONNUMERIC)
|
||
self.assertRaises(ValueError, self._read_test,
|
||
['abc,3'], [[]],
|
||
quoting=csv.QUOTE_STRINGS)
|
||
self._read_test(['1,@,3,@,5'], [['1', ',3,', '5']], quotechar='@')
|
||
self._read_test(['1,\0,3,\0,5'], [['1', ',3,', '5']], quotechar='\0')
|
||
self._read_test(['1\\.5,\\.5,.5'], [[1.5, 0.5, 0.5]],
|
||
quoting=csv.QUOTE_NONNUMERIC, escapechar='\\')
|
||
self._read_test(['1\\.5,\\.5,"\\.5"'], [[1.5, 0.5, ".5"]],
|
||
quoting=csv.QUOTE_STRINGS, escapechar='\\')
|
||
|
||
def test_read_skipinitialspace(self):
|
||
self._read_test(['no space, space, spaces,\ttab'],
|
||
[['no space', 'space', 'spaces', '\ttab']],
|
||
skipinitialspace=True)
|
||
self._read_test([' , , '],
|
||
[['', '', '']],
|
||
skipinitialspace=True)
|
||
self._read_test([' , , '],
|
||
[[None, None, None]],
|
||
skipinitialspace=True, quoting=csv.QUOTE_NOTNULL)
|
||
self._read_test([' , , '],
|
||
[[None, None, None]],
|
||
skipinitialspace=True, quoting=csv.QUOTE_STRINGS)
|
||
|
||
def test_read_space_delimiter(self):
|
||
self._read_test(['a b', ' a ', ' ', ''],
|
||
[['a', '', '', 'b'], ['', '', 'a', '', ''], ['', '', ''], []],
|
||
delimiter=' ', skipinitialspace=False)
|
||
self._read_test(['a b', ' a ', ' ', ''],
|
||
[['a', 'b'], ['a', ''], [''], []],
|
||
delimiter=' ', skipinitialspace=True)
|
||
|
||
def test_read_bigfield(self):
|
||
# This exercises the buffer realloc functionality and field size
|
||
# limits.
|
||
limit = csv.field_size_limit()
|
||
try:
|
||
size = 50000
|
||
bigstring = 'X' * size
|
||
bigline = '%s,%s' % (bigstring, bigstring)
|
||
self._read_test([bigline], [[bigstring, bigstring]])
|
||
csv.field_size_limit(size)
|
||
self._read_test([bigline], [[bigstring, bigstring]])
|
||
self.assertEqual(csv.field_size_limit(), size)
|
||
csv.field_size_limit(size-1)
|
||
self.assertRaises(csv.Error, self._read_test, [bigline], [])
|
||
self.assertRaises(TypeError, csv.field_size_limit, None)
|
||
self.assertRaises(TypeError, csv.field_size_limit, 1, None)
|
||
finally:
|
||
csv.field_size_limit(limit)
|
||
|
||
def test_read_linenum(self):
|
||
r = csv.reader(['line,1', 'line,2', 'line,3'])
|
||
self.assertEqual(r.line_num, 0)
|
||
next(r)
|
||
self.assertEqual(r.line_num, 1)
|
||
next(r)
|
||
self.assertEqual(r.line_num, 2)
|
||
next(r)
|
||
self.assertEqual(r.line_num, 3)
|
||
self.assertRaises(StopIteration, next, r)
|
||
self.assertEqual(r.line_num, 3)
|
||
|
||
def test_roundtrip_quoteed_newlines(self):
|
||
rows = [
|
||
['\na', 'b\nc', 'd\n'],
|
||
['\re', 'f\rg', 'h\r'],
|
||
['\r\ni', 'j\r\nk', 'l\r\n'],
|
||
['\n\rm', 'n\n\ro', 'p\n\r'],
|
||
['\r\rq', 'r\r\rs', 't\r\r'],
|
||
['\n\nu', 'v\n\nw', 'x\n\n'],
|
||
]
|
||
for lineterminator in '\r\n', '\n', '\r':
|
||
with self.subTest(lineterminator=lineterminator):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, lineterminator=lineterminator)
|
||
writer.writerows(rows)
|
||
fileobj.seek(0)
|
||
for i, row in enumerate(csv.reader(fileobj)):
|
||
self.assertEqual(row, rows[i])
|
||
|
||
def test_roundtrip_escaped_unquoted_newlines(self):
|
||
rows = [
|
||
['\na', 'b\nc', 'd\n'],
|
||
['\re', 'f\rg', 'h\r'],
|
||
['\r\ni', 'j\r\nk', 'l\r\n'],
|
||
['\n\rm', 'n\n\ro', 'p\n\r'],
|
||
['\r\rq', 'r\r\rs', 't\r\r'],
|
||
['\n\nu', 'v\n\nw', 'x\n\n'],
|
||
]
|
||
for lineterminator in '\r\n', '\n', '\r':
|
||
with self.subTest(lineterminator=lineterminator):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, lineterminator=lineterminator,
|
||
quoting=csv.QUOTE_NONE, escapechar="\\")
|
||
writer.writerows(rows)
|
||
fileobj.seek(0)
|
||
for i, row in enumerate(csv.reader(fileobj,
|
||
quoting=csv.QUOTE_NONE,
|
||
escapechar="\\")):
|
||
self.assertEqual(row, rows[i])
|
||
|
||
|
||
def test_reader_reentrant_iterator(self):
|
||
# gh-145105: re-entering the reader from the iterator must not crash.
|
||
class ReentrantIter:
|
||
def __init__(self):
|
||
self.reader = None
|
||
self.n = 0
|
||
def __iter__(self):
|
||
return self
|
||
def __next__(self):
|
||
self.n += 1
|
||
if self.n == 1:
|
||
try:
|
||
next(self.reader)
|
||
except StopIteration:
|
||
pass
|
||
return "a,b"
|
||
if self.n == 2:
|
||
return "x"
|
||
raise StopIteration
|
||
|
||
it = ReentrantIter()
|
||
reader = csv.reader(it)
|
||
it.reader = reader
|
||
with self.assertRaises(csv.Error):
|
||
next(reader)
|
||
|
||
|
||
class TestDialectRegistry(unittest.TestCase):
|
||
def test_registry_badargs(self):
|
||
self.assertRaises(TypeError, csv.list_dialects, None)
|
||
self.assertRaises(TypeError, csv.get_dialect)
|
||
self.assertRaises(csv.Error, csv.get_dialect, None)
|
||
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
|
||
self.assertRaises(TypeError, csv.unregister_dialect)
|
||
self.assertRaises(csv.Error, csv.unregister_dialect, None)
|
||
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
|
||
self.assertRaises(TypeError, csv.register_dialect, None)
|
||
self.assertRaises(TypeError, csv.register_dialect, None, None)
|
||
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
|
||
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
|
||
badargument=None)
|
||
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
|
||
quoting=None)
|
||
self.assertRaises(TypeError, csv.register_dialect, [])
|
||
|
||
def test_registry(self):
|
||
class myexceltsv(csv.excel):
|
||
delimiter = "\t"
|
||
name = "myexceltsv"
|
||
expected_dialects = csv.list_dialects() + [name]
|
||
expected_dialects.sort()
|
||
csv.register_dialect(name, myexceltsv)
|
||
self.addCleanup(csv.unregister_dialect, name)
|
||
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
|
||
got_dialects = sorted(csv.list_dialects())
|
||
self.assertEqual(expected_dialects, got_dialects)
|
||
|
||
def test_register_kwargs(self):
|
||
name = 'fedcba'
|
||
csv.register_dialect(name, delimiter=';')
|
||
self.addCleanup(csv.unregister_dialect, name)
|
||
self.assertEqual(csv.get_dialect(name).delimiter, ';')
|
||
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
|
||
|
||
def test_register_kwargs_override(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = "\t"
|
||
quotechar = '"'
|
||
doublequote = True
|
||
skipinitialspace = False
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_MINIMAL
|
||
|
||
name = 'test_dialect'
|
||
csv.register_dialect(name, mydialect,
|
||
delimiter=';',
|
||
quotechar="'",
|
||
doublequote=False,
|
||
skipinitialspace=True,
|
||
lineterminator='\n',
|
||
quoting=csv.QUOTE_ALL)
|
||
self.addCleanup(csv.unregister_dialect, name)
|
||
|
||
# Ensure that kwargs do override attributes of a dialect class:
|
||
dialect = csv.get_dialect(name)
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
self.assertEqual(dialect.doublequote, False)
|
||
self.assertEqual(dialect.skipinitialspace, True)
|
||
self.assertEqual(dialect.lineterminator, '\n')
|
||
self.assertEqual(dialect.quoting, csv.QUOTE_ALL)
|
||
|
||
def test_incomplete_dialect(self):
|
||
class myexceltsv(csv.Dialect):
|
||
delimiter = "\t"
|
||
self.assertRaises(csv.Error, myexceltsv)
|
||
|
||
def test_space_dialect(self):
|
||
class space(csv.excel):
|
||
delimiter = " "
|
||
quoting = csv.QUOTE_NONE
|
||
escapechar = "\\"
|
||
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("abc def\nc1ccccc1 benzene\n")
|
||
fileobj.seek(0)
|
||
reader = csv.reader(fileobj, dialect=space())
|
||
self.assertEqual(next(reader), ["abc", "", "", "def"])
|
||
self.assertEqual(next(reader), ["c1ccccc1", "benzene"])
|
||
|
||
def compare_dialect_123(self, expected, *writeargs, **kwwriteargs):
|
||
|
||
with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
|
||
|
||
writer = csv.writer(fileobj, *writeargs, **kwwriteargs)
|
||
writer.writerow([1,2,3])
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
def test_dialect_apply(self):
|
||
class testA(csv.excel):
|
||
delimiter = "\t"
|
||
class testB(csv.excel):
|
||
delimiter = ":"
|
||
class testC(csv.excel):
|
||
delimiter = "|"
|
||
class testUni(csv.excel):
|
||
delimiter = "\u039B"
|
||
|
||
class unspecified():
|
||
# A class to pass as dialect but with no dialect attributes.
|
||
pass
|
||
|
||
csv.register_dialect('testC', testC)
|
||
try:
|
||
self.compare_dialect_123("1,2,3\r\n")
|
||
self.compare_dialect_123("1,2,3\r\n", dialect=None)
|
||
self.compare_dialect_123("1,2,3\r\n", dialect=unspecified)
|
||
self.compare_dialect_123("1\t2\t3\r\n", testA)
|
||
self.compare_dialect_123("1:2:3\r\n", dialect=testB())
|
||
self.compare_dialect_123("1|2|3\r\n", dialect='testC')
|
||
self.compare_dialect_123("1;2;3\r\n", dialect=testA,
|
||
delimiter=';')
|
||
self.compare_dialect_123("1\u039B2\u039B3\r\n",
|
||
dialect=testUni)
|
||
|
||
finally:
|
||
csv.unregister_dialect('testC')
|
||
|
||
def test_copy(self):
|
||
for name in csv.list_dialects():
|
||
dialect = csv.get_dialect(name)
|
||
self.assertRaises(TypeError, copy.copy, dialect)
|
||
|
||
def test_replace(self):
|
||
dialect = csv.get_dialect('excel')
|
||
new = copy.replace(dialect, delimiter=';', strict=True)
|
||
self.assertIsInstance(new, type(dialect))
|
||
self.assertEqual(new.delimiter, ';')
|
||
self.assertTrue(new.strict)
|
||
# Not replaced parameters are inherited from the original dialect.
|
||
self.assertEqual(new.quotechar, dialect.quotechar)
|
||
self.assertEqual(new.escapechar, dialect.escapechar)
|
||
self.assertEqual(new.lineterminator, dialect.lineterminator)
|
||
self.assertEqual(new.quoting, dialect.quoting)
|
||
self.assertEqual(new.doublequote, dialect.doublequote)
|
||
self.assertEqual(new.skipinitialspace, dialect.skipinitialspace)
|
||
# The original dialect is left unchanged.
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertFalse(dialect.strict)
|
||
self.assertEqual(list(csv.reader(['a;b'], new)), [['a', 'b']])
|
||
|
||
self.assertIs(copy.replace(dialect), dialect)
|
||
self.assertRaises(TypeError, copy.replace, dialect, delimeter=';')
|
||
self.assertRaises(TypeError, copy.replace, dialect, delimiter=';;')
|
||
self.assertRaises(TypeError, dialect.__replace__, dialect)
|
||
|
||
def test_replace_dialect_subclass(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ";"
|
||
quotechar = '"'
|
||
doublequote = False
|
||
skipinitialspace = True
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_ALL
|
||
|
||
dialect = mydialect()
|
||
new = copy.replace(dialect, delimiter=':', quoting=csv.QUOTE_MINIMAL)
|
||
self.assertIsInstance(new, mydialect)
|
||
self.assertEqual(new.delimiter, ':')
|
||
self.assertEqual(new.quoting, csv.QUOTE_MINIMAL)
|
||
# Not replaced parameters are inherited from the original dialect.
|
||
self.assertEqual(new.quotechar, '"')
|
||
self.assertEqual(new.escapechar, None)
|
||
self.assertEqual(new.lineterminator, '\r\n')
|
||
self.assertFalse(new.doublequote)
|
||
self.assertTrue(new.skipinitialspace)
|
||
# The original dialect is left unchanged.
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
self.assertEqual(dialect.quoting, csv.QUOTE_ALL)
|
||
self.assertEqual(list(csv.reader(['a:b'], new)), [['a', 'b']])
|
||
# "strict" is supported even if it is not set on the class.
|
||
self.assertTrue(copy.replace(dialect, strict=True).strict)
|
||
|
||
with self.assertRaises(csv.Error):
|
||
copy.replace(dialect, delimiter='::')
|
||
with self.assertRaisesRegex(TypeError, "'delimeter'"):
|
||
copy.replace(dialect, delimeter=':')
|
||
|
||
def test_pickle(self):
|
||
for name in csv.list_dialects():
|
||
dialect = csv.get_dialect(name)
|
||
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
|
||
self.assertRaises(TypeError, pickle.dumps, dialect, proto)
|
||
|
||
class TestCsvBase(unittest.TestCase):
|
||
def readerAssertEqual(self, input, expected_result):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
fileobj.write(input)
|
||
fileobj.seek(0)
|
||
reader = csv.reader(fileobj, dialect = self.dialect)
|
||
fields = list(reader)
|
||
self.assertEqual(fields, expected_result)
|
||
|
||
def writerAssertEqual(self, input, expected_result):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, dialect = self.dialect)
|
||
writer.writerows(input)
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected_result)
|
||
|
||
class TestDialectExcel(TestCsvBase):
|
||
dialect = 'excel'
|
||
|
||
def test_single(self):
|
||
self.readerAssertEqual('abc', [['abc']])
|
||
|
||
def test_simple(self):
|
||
self.readerAssertEqual('1,2,3,4,5', [['1','2','3','4','5']])
|
||
|
||
def test_blankline(self):
|
||
self.readerAssertEqual('', [])
|
||
|
||
def test_empty_fields(self):
|
||
self.readerAssertEqual(',', [['', '']])
|
||
|
||
def test_singlequoted(self):
|
||
self.readerAssertEqual('""', [['']])
|
||
|
||
def test_singlequoted_left_empty(self):
|
||
self.readerAssertEqual('"",', [['','']])
|
||
|
||
def test_singlequoted_right_empty(self):
|
||
self.readerAssertEqual(',""', [['','']])
|
||
|
||
def test_single_quoted_quote(self):
|
||
self.readerAssertEqual('""""', [['"']])
|
||
|
||
def test_quoted_quotes(self):
|
||
self.readerAssertEqual('""""""', [['""']])
|
||
|
||
def test_inline_quote(self):
|
||
self.readerAssertEqual('a""b', [['a""b']])
|
||
|
||
def test_inline_quotes(self):
|
||
self.readerAssertEqual('a"b"c', [['a"b"c']])
|
||
|
||
def test_quotes_and_more(self):
|
||
# Excel would never write a field containing '"a"b', but when
|
||
# reading one, it will return 'ab'.
|
||
self.readerAssertEqual('"a"b', [['ab']])
|
||
|
||
def test_lone_quote(self):
|
||
self.readerAssertEqual('a"b', [['a"b']])
|
||
|
||
def test_quote_and_quote(self):
|
||
# Excel would never write a field containing '"a" "b"', but when
|
||
# reading one, it will return 'a "b"'.
|
||
self.readerAssertEqual('"a" "b"', [['a "b"']])
|
||
|
||
def test_space_and_quote(self):
|
||
self.readerAssertEqual(' "a"', [[' "a"']])
|
||
|
||
def test_quoted(self):
|
||
self.readerAssertEqual('1,2,3,"I think, therefore I am",5,6',
|
||
[['1', '2', '3',
|
||
'I think, therefore I am',
|
||
'5', '6']])
|
||
|
||
def test_quoted_quote(self):
|
||
self.readerAssertEqual('1,2,3,"""I see,"" said the blind man","as he picked up his hammer and saw"',
|
||
[['1', '2', '3',
|
||
'"I see," said the blind man',
|
||
'as he picked up his hammer and saw']])
|
||
|
||
def test_quoted_nl(self):
|
||
input = '''\
|
||
1,2,3,"""I see,""
|
||
said the blind man","as he picked up his
|
||
hammer and saw"
|
||
9,8,7,6'''
|
||
self.readerAssertEqual(input,
|
||
[['1', '2', '3',
|
||
'"I see,"\nsaid the blind man',
|
||
'as he picked up his\nhammer and saw'],
|
||
['9','8','7','6']])
|
||
|
||
def test_dubious_quote(self):
|
||
self.readerAssertEqual('12,12,1",', [['12', '12', '1"', '']])
|
||
|
||
def test_null(self):
|
||
self.writerAssertEqual([], '')
|
||
|
||
def test_single_writer(self):
|
||
self.writerAssertEqual([['abc']], 'abc\r\n')
|
||
|
||
def test_simple_writer(self):
|
||
self.writerAssertEqual([[1, 2, 'abc', 3, 4]], '1,2,abc,3,4\r\n')
|
||
|
||
def test_quotes(self):
|
||
self.writerAssertEqual([[1, 2, 'a"bc"', 3, 4]], '1,2,"a""bc""",3,4\r\n')
|
||
|
||
def test_quote_fieldsep(self):
|
||
self.writerAssertEqual([['abc,def']], '"abc,def"\r\n')
|
||
|
||
def test_newlines(self):
|
||
self.writerAssertEqual([[1, 2, 'a\nbc', 3, 4]], '1,2,"a\nbc",3,4\r\n')
|
||
|
||
class EscapedExcel(csv.excel):
|
||
quoting = csv.QUOTE_NONE
|
||
escapechar = '\\'
|
||
|
||
class TestEscapedExcel(TestCsvBase):
|
||
dialect = EscapedExcel()
|
||
|
||
def test_escape_fieldsep(self):
|
||
self.writerAssertEqual([['abc,def']], 'abc\\,def\r\n')
|
||
|
||
def test_read_escape_fieldsep(self):
|
||
self.readerAssertEqual('abc\\,def\r\n', [['abc,def']])
|
||
|
||
class TestDialectUnix(TestCsvBase):
|
||
dialect = 'unix'
|
||
|
||
def test_simple_writer(self):
|
||
self.writerAssertEqual([[1, 'abc def', 'abc']], '"1","abc def","abc"\n')
|
||
|
||
def test_simple_reader(self):
|
||
self.readerAssertEqual('"1","abc def","abc"\n', [['1', 'abc def', 'abc']])
|
||
|
||
class QuotedEscapedExcel(csv.excel):
|
||
quoting = csv.QUOTE_NONNUMERIC
|
||
escapechar = '\\'
|
||
|
||
class TestQuotedEscapedExcel(TestCsvBase):
|
||
dialect = QuotedEscapedExcel()
|
||
|
||
def test_write_escape_fieldsep(self):
|
||
self.writerAssertEqual([['abc,def']], '"abc,def"\r\n')
|
||
|
||
def test_read_escape_fieldsep(self):
|
||
self.readerAssertEqual('"abc\\,def"\r\n', [['abc,def']])
|
||
|
||
class TestDictFields(unittest.TestCase):
|
||
### "long" means the row is longer than the number of fieldnames
|
||
### "short" means there are fewer elements in the row than fieldnames
|
||
def test_writeheader_return_value(self):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
|
||
writeheader_return_value = writer.writeheader()
|
||
self.assertEqual(writeheader_return_value, 10)
|
||
|
||
def test_write_simple_dict(self):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
|
||
writer.writeheader()
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.readline(), "f1,f2,f3\r\n")
|
||
writer.writerow({"f1": 10, "f3": "abc"})
|
||
fileobj.seek(0)
|
||
fileobj.readline() # header
|
||
self.assertEqual(fileobj.read(), "10,,abc\r\n")
|
||
|
||
def test_write_multiple_dict_rows(self):
|
||
fileobj = StringIO()
|
||
writer = csv.DictWriter(fileobj, fieldnames=["f1", "f2", "f3"])
|
||
writer.writeheader()
|
||
self.assertEqual(fileobj.getvalue(), "f1,f2,f3\r\n")
|
||
writer.writerows([{"f1": 1, "f2": "abc", "f3": "f"},
|
||
{"f1": 2, "f2": 5, "f3": "xyz"}])
|
||
self.assertEqual(fileobj.getvalue(),
|
||
"f1,f2,f3\r\n1,abc,f\r\n2,5,xyz\r\n")
|
||
|
||
def test_write_no_fields(self):
|
||
fileobj = StringIO()
|
||
self.assertRaises(TypeError, csv.DictWriter, fileobj)
|
||
|
||
def test_write_fields_not_in_fieldnames(self):
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.DictWriter(fileobj, fieldnames = ["f1", "f2", "f3"])
|
||
# Of special note is the non-string key (issue 19449)
|
||
with self.assertRaises(ValueError) as cx:
|
||
writer.writerow({"f4": 10, "f2": "spam", 1: "abc"})
|
||
exception = str(cx.exception)
|
||
self.assertIn("fieldnames", exception)
|
||
self.assertIn("'f4'", exception)
|
||
self.assertNotIn("'f2'", exception)
|
||
self.assertIn("1", exception)
|
||
|
||
def test_typo_in_extrasaction_raises_error(self):
|
||
fileobj = StringIO()
|
||
self.assertRaises(ValueError, csv.DictWriter, fileobj, ['f1', 'f2'],
|
||
extrasaction="raised")
|
||
|
||
def test_write_field_not_in_field_names_raise(self):
|
||
fileobj = StringIO()
|
||
writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="raise")
|
||
dictrow = {'f0': 0, 'f1': 1, 'f2': 2, 'f3': 3}
|
||
self.assertRaises(ValueError, csv.DictWriter.writerow, writer, dictrow)
|
||
|
||
# see bpo-44512 (differently cased 'raise' should not result in 'ignore')
|
||
writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="RAISE")
|
||
self.assertRaises(ValueError, csv.DictWriter.writerow, writer, dictrow)
|
||
|
||
def test_write_field_not_in_field_names_ignore(self):
|
||
fileobj = StringIO()
|
||
writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="ignore")
|
||
dictrow = {'f0': 0, 'f1': 1, 'f2': 2, 'f3': 3}
|
||
csv.DictWriter.writerow(writer, dictrow)
|
||
self.assertEqual(fileobj.getvalue(), "1,2\r\n")
|
||
|
||
# bpo-44512
|
||
writer = csv.DictWriter(fileobj, ['f1', 'f2'], extrasaction="IGNORE")
|
||
csv.DictWriter.writerow(writer, dictrow)
|
||
|
||
def test_dict_reader_fieldnames_accepts_iter(self):
|
||
fieldnames = ["a", "b", "c"]
|
||
f = StringIO()
|
||
reader = csv.DictReader(f, iter(fieldnames))
|
||
self.assertEqual(reader.fieldnames, fieldnames)
|
||
|
||
def test_dict_reader_fieldnames_accepts_list(self):
|
||
fieldnames = ["a", "b", "c"]
|
||
f = StringIO()
|
||
reader = csv.DictReader(f, fieldnames)
|
||
self.assertEqual(reader.fieldnames, fieldnames)
|
||
|
||
def test_dict_reader_set_fieldnames(self):
|
||
fieldnames = ["a", "b", "c"]
|
||
f = StringIO()
|
||
reader = csv.DictReader(f)
|
||
self.assertIsNone(reader.fieldnames)
|
||
reader.fieldnames = fieldnames
|
||
self.assertEqual(reader.fieldnames, fieldnames)
|
||
|
||
def test_dict_writer_fieldnames_rejects_iter(self):
|
||
fieldnames = ["a", "b", "c"]
|
||
f = StringIO()
|
||
writer = csv.DictWriter(f, iter(fieldnames))
|
||
self.assertEqual(writer.fieldnames, fieldnames)
|
||
|
||
def test_dict_writer_fieldnames_accepts_list(self):
|
||
fieldnames = ["a", "b", "c"]
|
||
f = StringIO()
|
||
writer = csv.DictWriter(f, fieldnames)
|
||
self.assertEqual(writer.fieldnames, fieldnames)
|
||
|
||
def test_dict_reader_fieldnames_is_optional(self):
|
||
f = StringIO()
|
||
reader = csv.DictReader(f, fieldnames=None)
|
||
self.assertIsNone(reader.fieldnames)
|
||
|
||
def test_read_dict_fields(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("1,2,abc\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj,
|
||
fieldnames=["f1", "f2", "f3"])
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
|
||
|
||
def test_read_dict_no_fieldnames(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj)
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
|
||
self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
|
||
|
||
# Two test cases to make sure existing ways of implicitly setting
|
||
# fieldnames continue to work. Both arise from discussion in issue3436.
|
||
def test_read_dict_fieldnames_from_file(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj,
|
||
fieldnames=next(csv.reader(fileobj)))
|
||
self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2', "f3": 'abc'})
|
||
|
||
def test_read_dict_fieldnames_chain(self):
|
||
import itertools
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("f1,f2,f3\r\n1,2,abc\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj)
|
||
first = next(reader)
|
||
for row in itertools.chain([first], reader):
|
||
self.assertEqual(reader.fieldnames, ["f1", "f2", "f3"])
|
||
self.assertEqual(row, {"f1": '1', "f2": '2', "f3": 'abc'})
|
||
|
||
def test_read_long(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("1,2,abc,4,5,6\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj,
|
||
fieldnames=["f1", "f2"])
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2',
|
||
None: ["abc", "4", "5", "6"]})
|
||
|
||
def test_read_long_with_rest(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("1,2,abc,4,5,6\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj,
|
||
fieldnames=["f1", "f2"], restkey="_rest")
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2',
|
||
"_rest": ["abc", "4", "5", "6"]})
|
||
|
||
def test_read_long_with_rest_no_fieldnames(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("f1,f2\r\n1,2,abc,4,5,6\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj, restkey="_rest")
|
||
self.assertEqual(reader.fieldnames, ["f1", "f2"])
|
||
self.assertEqual(next(reader), {"f1": '1', "f2": '2',
|
||
"_rest": ["abc", "4", "5", "6"]})
|
||
|
||
def test_read_short(self):
|
||
with TemporaryFile("w+", encoding="utf-8") as fileobj:
|
||
fileobj.write("1,2,abc,4,5,6\r\n1,2,abc\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.DictReader(fileobj,
|
||
fieldnames="1 2 3 4 5 6".split(),
|
||
restval="DEFAULT")
|
||
self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
|
||
"4": '4', "5": '5', "6": '6'})
|
||
self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
|
||
"4": 'DEFAULT', "5": 'DEFAULT',
|
||
"6": 'DEFAULT'})
|
||
|
||
def test_read_multi(self):
|
||
sample = [
|
||
'2147483648,43.0e12,17,abc,def\r\n',
|
||
'147483648,43.0e2,17,abc,def\r\n',
|
||
'47483648,43.0,170,abc,def\r\n'
|
||
]
|
||
|
||
reader = csv.DictReader(sample,
|
||
fieldnames="i1 float i2 s1 s2".split())
|
||
self.assertEqual(next(reader), {"i1": '2147483648',
|
||
"float": '43.0e12',
|
||
"i2": '17',
|
||
"s1": 'abc',
|
||
"s2": 'def'})
|
||
|
||
def test_read_with_blanks(self):
|
||
reader = csv.DictReader(["1,2,abc,4,5,6\r\n","\r\n",
|
||
"1,2,abc,4,5,6\r\n"],
|
||
fieldnames="1 2 3 4 5 6".split())
|
||
self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
|
||
"4": '4', "5": '5', "6": '6'})
|
||
self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
|
||
"4": '4', "5": '5', "6": '6'})
|
||
|
||
def test_read_semi_sep(self):
|
||
reader = csv.DictReader(["1;2;abc;4;5;6\r\n"],
|
||
fieldnames="1 2 3 4 5 6".split(),
|
||
delimiter=';')
|
||
self.assertEqual(next(reader), {"1": '1', "2": '2', "3": 'abc',
|
||
"4": '4', "5": '5', "6": '6'})
|
||
|
||
class TestArrayWrites(unittest.TestCase):
|
||
def test_int_write(self):
|
||
import array
|
||
contents = [(20-i) for i in range(20)]
|
||
a = array.array('i', contents)
|
||
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, dialect="excel")
|
||
writer.writerow(a)
|
||
expected = ",".join([str(i) for i in a])+"\r\n"
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
def test_double_write(self):
|
||
import array
|
||
contents = [(20-i)*0.1 for i in range(20)]
|
||
a = array.array('d', contents)
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, dialect="excel")
|
||
writer.writerow(a)
|
||
expected = ",".join([str(i) for i in a])+"\r\n"
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
def test_float_write(self):
|
||
import array
|
||
contents = [(20-i)*0.1 for i in range(20)]
|
||
a = array.array('f', contents)
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, dialect="excel")
|
||
writer.writerow(a)
|
||
expected = ",".join([str(i) for i in a])+"\r\n"
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
def test_char_write(self):
|
||
import array, string
|
||
a = array.array('w', string.ascii_letters)
|
||
|
||
with TemporaryFile("w+", encoding="utf-8", newline='') as fileobj:
|
||
writer = csv.writer(fileobj, dialect="excel")
|
||
writer.writerow(a)
|
||
expected = ",".join(a)+"\r\n"
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
class TestDialectValidity(unittest.TestCase):
|
||
def test_quoting(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ";"
|
||
escapechar = '\\'
|
||
doublequote = False
|
||
skipinitialspace = True
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_NONE
|
||
d = mydialect()
|
||
self.assertEqual(d.quoting, csv.QUOTE_NONE)
|
||
|
||
mydialect.quoting = None
|
||
self.assertRaises(csv.Error, mydialect)
|
||
|
||
mydialect.quoting = 42
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'bad "quoting" value')
|
||
|
||
mydialect.doublequote = True
|
||
mydialect.quoting = csv.QUOTE_ALL
|
||
mydialect.quotechar = '"'
|
||
d = mydialect()
|
||
self.assertEqual(d.quoting, csv.QUOTE_ALL)
|
||
self.assertEqual(d.quotechar, '"')
|
||
self.assertTrue(d.doublequote)
|
||
|
||
mydialect.quotechar = ""
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"quotechar" must be a unicode character or None, '
|
||
'not a string of length 0')
|
||
|
||
mydialect.quotechar = "''"
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"quotechar" must be a unicode character or None, '
|
||
'not a string of length 2')
|
||
|
||
mydialect.quotechar = 4
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"quotechar" must be a unicode character or None, '
|
||
'not int')
|
||
|
||
def test_delimiter(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ";"
|
||
escapechar = '\\'
|
||
doublequote = False
|
||
skipinitialspace = True
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_NONE
|
||
d = mydialect()
|
||
self.assertEqual(d.delimiter, ";")
|
||
|
||
mydialect.delimiter = ":::"
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"delimiter" must be a unicode character, '
|
||
'not a string of length 3')
|
||
|
||
mydialect.delimiter = ""
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"delimiter" must be a unicode character, not a string of length 0')
|
||
|
||
mydialect.delimiter = b","
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"delimiter" must be a unicode character, not bytes')
|
||
|
||
mydialect.delimiter = 4
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"delimiter" must be a unicode character, not int')
|
||
|
||
mydialect.delimiter = None
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"delimiter" must be a unicode character, not NoneType')
|
||
|
||
def test_escapechar(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ";"
|
||
escapechar = '\\'
|
||
doublequote = False
|
||
skipinitialspace = True
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_NONE
|
||
d = mydialect()
|
||
self.assertEqual(d.escapechar, "\\")
|
||
|
||
mydialect.escapechar = ""
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"escapechar" must be a unicode character or None, '
|
||
'not a string of length 0')
|
||
|
||
mydialect.escapechar = "**"
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"escapechar" must be a unicode character or None, '
|
||
'not a string of length 2')
|
||
|
||
mydialect.escapechar = b"*"
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"escapechar" must be a unicode character or None, '
|
||
'not bytes')
|
||
|
||
mydialect.escapechar = 4
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"escapechar" must be a unicode character or None, '
|
||
'not int')
|
||
|
||
def test_lineterminator(self):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ";"
|
||
escapechar = '\\'
|
||
doublequote = False
|
||
skipinitialspace = True
|
||
lineterminator = '\r\n'
|
||
quoting = csv.QUOTE_NONE
|
||
d = mydialect()
|
||
self.assertEqual(d.lineterminator, '\r\n')
|
||
|
||
mydialect.lineterminator = ":::"
|
||
d = mydialect()
|
||
self.assertEqual(d.lineterminator, ":::")
|
||
|
||
mydialect.lineterminator = 4
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"lineterminator" must be a string, not int')
|
||
|
||
mydialect.lineterminator = None
|
||
with self.assertRaises(csv.Error) as cm:
|
||
mydialect()
|
||
self.assertEqual(str(cm.exception),
|
||
'"lineterminator" must be a string, not NoneType')
|
||
|
||
def test_invalid_chars(self):
|
||
def create_invalid(field_name, value, **kwargs):
|
||
class mydialect(csv.Dialect):
|
||
delimiter = ','
|
||
quoting = csv.QUOTE_ALL
|
||
quotechar = '"'
|
||
lineterminator = '\r\n'
|
||
setattr(mydialect, field_name, value)
|
||
for field_name, value in kwargs.items():
|
||
setattr(mydialect, field_name, value)
|
||
d = mydialect()
|
||
|
||
for field_name in ("delimiter", "escapechar", "quotechar"):
|
||
with self.subTest(field_name=field_name):
|
||
self.assertRaises(csv.Error, create_invalid, field_name, "")
|
||
self.assertRaises(csv.Error, create_invalid, field_name, "abc")
|
||
self.assertRaises(csv.Error, create_invalid, field_name, b'x')
|
||
self.assertRaises(csv.Error, create_invalid, field_name, 5)
|
||
self.assertRaises(ValueError, create_invalid, field_name, "\n")
|
||
self.assertRaises(ValueError, create_invalid, field_name, "\r")
|
||
if field_name != "delimiter":
|
||
self.assertRaises(ValueError, create_invalid, field_name, " ",
|
||
skipinitialspace=True)
|
||
|
||
def test_dialect_getattr_non_attribute_error_propagates(self):
|
||
# gh-145966: non-AttributeError exceptions raised by __getattr__
|
||
# during dialect attribute lookup must propagate, not be silenced.
|
||
class BadDialect:
|
||
def __getattr__(self, name):
|
||
raise RuntimeError("boom")
|
||
|
||
with self.assertRaises(RuntimeError):
|
||
csv.reader([], dialect=BadDialect())
|
||
|
||
with self.assertRaises(RuntimeError):
|
||
csv.writer(StringIO(), dialect=BadDialect())
|
||
|
||
|
||
class TestSniffer(unittest.TestCase):
|
||
sample1 = """\
|
||
Harry's, Arlington Heights, IL, 2/1/03, Kimi Hayes
|
||
Shark City, Glendale Heights, IL, 12/28/02, Prezence
|
||
Tommy's Place, Blue Island, IL, 12/28/02, Blue Sunday/White Crow
|
||
Stonecutters Seafood and Chop House, Lemont, IL, 12/19/02, Week Back
|
||
"""
|
||
sample2 = """\
|
||
'Harry''s':'Arlington Heights':'IL':'2/1/03':'Kimi Hayes'
|
||
'Shark City':'Glendale Heights':'IL':'12/28/02':'Prezence'
|
||
'Tommy''s Place':'Blue Island':'IL':'12/28/02':'Blue Sunday/White Crow'
|
||
'Stonecutters ''Seafood'' and Chop House':'Lemont':'IL':'12/19/02':'Week Back'
|
||
"""
|
||
header1 = '''\
|
||
"venue","city","state","date","performers"
|
||
'''
|
||
sample3 = '''\
|
||
05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
|
||
05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
|
||
05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03
|
||
'''
|
||
|
||
sample4 = '''\
|
||
2147483648;43.0e12;17;abc;def
|
||
147483648;43.0e2;17;abc;def
|
||
47483648;43.0;170;abc;def
|
||
'''
|
||
|
||
sample5 = "aaa\tbbb\r\nAAA\t\r\nBBB\t\r\n"
|
||
sample6 = "a|b|c\r\nd|e|f\r\n"
|
||
sample7 = "'a'|'b'|'c'\r\n'd'|e|f\r\n"
|
||
|
||
# Issue 18155: Use a delimiter that is a special char to regex:
|
||
|
||
header2 = '''\
|
||
"venue"+"city"+"state"+"date"+"performers"
|
||
'''
|
||
sample8 = """\
|
||
Harry's+ Arlington Heights+ IL+ 2/1/03+ Kimi Hayes
|
||
Shark City+ Glendale Heights+ IL+ 12/28/02+ Prezence
|
||
Tommy's Place+ Blue Island+ IL+ 12/28/02+ Blue Sunday/White Crow
|
||
Stonecutters Seafood and Chop House+ Lemont+ IL+ 12/19/02+ Week Back
|
||
"""
|
||
sample9 = """\
|
||
'Harry''s'+ Arlington Heights'+ 'IL'+ '2/1/03'+ 'Kimi Hayes'
|
||
'Shark City'+ Glendale Heights'+' IL'+ '12/28/02'+ 'Prezence'
|
||
'Tommy''s Place'+ Blue Island'+ 'IL'+ '12/28/02'+ 'Blue Sunday/White Crow'
|
||
'Stonecutters ''Seafood'' and Chop House'+ 'Lemont'+ 'IL'+ '12/19/02'+ 'Week Back'
|
||
"""
|
||
|
||
sample10 = dedent("""
|
||
abc,def
|
||
ghijkl,mno
|
||
ghi,jkl
|
||
""")
|
||
|
||
sample11 = dedent("""
|
||
abc,def
|
||
ghijkl,mnop
|
||
ghi,jkl
|
||
""")
|
||
|
||
sample12 = dedent(""""time","forces"
|
||
1,1.5
|
||
0.5,5+0j
|
||
0,0
|
||
1+1j,6
|
||
""")
|
||
|
||
sample13 = dedent(""""time","forces"
|
||
0,0
|
||
1,2
|
||
a,b
|
||
""")
|
||
|
||
sample14 = """\
|
||
abc\0def
|
||
ghijkl\0mno
|
||
ghi\0jkl
|
||
"""
|
||
|
||
sample15 = "\n\n\n"
|
||
sample16 = "abc\ndef\nghi"
|
||
|
||
sample17 = ["letter,offset"]
|
||
sample17.extend(f"{chr(ord('a') + i)},{i}" for i in range(20))
|
||
sample17.append("v,twenty_one") # 'u' was skipped
|
||
sample17 = '\n'.join(sample17)
|
||
|
||
sample18 = ["letter,offset"]
|
||
sample18.extend(f"{chr(ord('a') + i)},{i}" for i in range(21))
|
||
sample18.append("v,twenty_one") # 'u' was not skipped
|
||
sample18 = '\n'.join(sample18)
|
||
|
||
sample19 = ('time,title\r\n'
|
||
'2020-10-01,"Pocket - Save news, videos, stories and more"\r\n')
|
||
|
||
def test_issue43625(self):
|
||
sniffer = csv.Sniffer()
|
||
self.assertTrue(sniffer.has_header(self.sample12))
|
||
self.assertFalse(sniffer.has_header(self.sample13))
|
||
|
||
def test_has_header_strings(self):
|
||
"More to document existing (unexpected?) behavior than anything else."
|
||
sniffer = csv.Sniffer()
|
||
self.assertFalse(sniffer.has_header(self.sample10))
|
||
self.assertFalse(sniffer.has_header(self.sample11))
|
||
|
||
def test_has_header(self):
|
||
sniffer = csv.Sniffer()
|
||
self.assertIs(sniffer.has_header(self.sample1), False)
|
||
self.assertIs(sniffer.has_header(self.header1 + self.sample1), True)
|
||
|
||
def test_has_header_regex_special_delimiter(self):
|
||
sniffer = csv.Sniffer()
|
||
self.assertIs(sniffer.has_header(self.sample8), False)
|
||
self.assertIs(sniffer.has_header(self.header2 + self.sample8), True)
|
||
|
||
def test_has_header_checks_20_rows(self):
|
||
sniffer = csv.Sniffer()
|
||
self.assertFalse(sniffer.has_header(self.sample17))
|
||
self.assertTrue(sniffer.has_header(self.sample18))
|
||
|
||
def test_guess_quote_and_delimiter(self):
|
||
sniffer = csv.Sniffer()
|
||
for header in (";'123;4';", "'123;4';", ";'123;4'"):
|
||
with self.subTest(header):
|
||
dialect = sniffer.sniff(header, ",;")
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
self.assertIs(dialect.doublequote, False)
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
# A single quoted field is a single column without a delimiter.
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, "'123;4'", ",;")
|
||
|
||
def test_sniff(self):
|
||
sniffer = csv.Sniffer()
|
||
dialect = sniffer.sniff(self.sample1)
|
||
self.assertEqual(dialect.delimiter, ",")
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertIs(dialect.skipinitialspace, True)
|
||
|
||
dialect = sniffer.sniff(self.sample2)
|
||
self.assertEqual(dialect.delimiter, ":")
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
|
||
def test_delimiters(self):
|
||
sniffer = csv.Sniffer()
|
||
dialect = sniffer.sniff(self.sample3)
|
||
# given that all three lines in sample3 are equal,
|
||
# I think that any character could have been 'guessed' as the
|
||
# delimiter, depending on dictionary order
|
||
self.assertIn(dialect.delimiter, self.sample3)
|
||
dialect = sniffer.sniff(self.sample3, delimiters="?,")
|
||
self.assertEqual(dialect.delimiter, "?")
|
||
dialect = sniffer.sniff(self.sample3, delimiters="/,")
|
||
self.assertEqual(dialect.delimiter, "/")
|
||
dialect = sniffer.sniff(self.sample4)
|
||
self.assertEqual(dialect.delimiter, ";")
|
||
dialect = sniffer.sniff(self.sample5)
|
||
self.assertEqual(dialect.delimiter, "\t")
|
||
dialect = sniffer.sniff(self.sample6)
|
||
self.assertEqual(dialect.delimiter, "|")
|
||
dialect = sniffer.sniff(self.sample7)
|
||
self.assertEqual(dialect.delimiter, "|")
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
dialect = sniffer.sniff(self.sample8)
|
||
self.assertEqual(dialect.delimiter, '+')
|
||
dialect = sniffer.sniff(self.sample9)
|
||
self.assertEqual(dialect.delimiter, '+')
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
dialect = sniffer.sniff(self.sample14)
|
||
self.assertEqual(dialect.delimiter, '\0')
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, self.sample15)
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, self.sample16)
|
||
dialect = sniffer.sniff(self.sample19)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
|
||
def test_sniff_escapechar(self):
|
||
# gh-83273: escaped delimiters make the delimiter frequencies
|
||
# inconsistent, but the escape character can be detected by trial
|
||
# parsing.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'ab,cd\\,ef\ngh\\,ij,kl\nmn,op\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.escapechar, '\\')
|
||
self.assertIs(dialect.doublequote, False)
|
||
self.assertEqual(list(csv.reader(StringIO(sample), dialect)),
|
||
[['ab', 'cd,ef'], ['gh,ij', 'kl'], ['mn', 'op']])
|
||
# No escape character in the sample -- none is detected.
|
||
dialect = sniffer.sniff('ab,cd\ngh,ij\n')
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertIsNone(dialect.escapechar)
|
||
|
||
def test_sniff_quoted_rows_among_unquoted(self):
|
||
# Rows which happen to consist of a single quoted or unquoted
|
||
# field must not be mistaken for a single column of quoted fields.
|
||
sniffer = csv.Sniffer()
|
||
sample = '"header line"\na|b\nc|d\ne|f\n'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, '|')
|
||
# Even if an unterminated quote breaks the quoted parse.
|
||
sample = '"header"\na|"b c\nd|e\nf|g\n'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, '|')
|
||
|
||
def test_sniff_single_column_quoted(self):
|
||
# gh-98820: a sample consisting of a single column of quoted fields
|
||
# has no delimiter to detect, even if the quoted content contains
|
||
# characters which could pass for one. It also used to take
|
||
# quadratic time before failing.
|
||
sniffer = csv.Sniffer()
|
||
sample = '\n'.join('"%02d-%02d-%02d"' % (i, i, i) for i in range(50))
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, sample)
|
||
# Even if none of the requested delimiters occurs in the sample.
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, sample, ',:|\t')
|
||
# A mix of quoted and unquoted single-field rows.
|
||
sample = '"abc"\ndef\n"ghi"\njkl\n' * 10
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, sample)
|
||
# Even if the quoted fields contain delimiter characters.
|
||
sample = '"a,b"\ncd\n' * 20
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, sample)
|
||
# Even if the content of the quoted fields parses consistently
|
||
# with the other quote character.
|
||
sample = '''"a-'b'-c"\n"d-'e'-f"\n''' * 10
|
||
self.assertRaisesRegex(csv.Error, "Could not determine delimiter",
|
||
sniffer.sniff, sample)
|
||
|
||
def test_sniff_non_ascii_delimiter(self):
|
||
# gh-111820: an explicitly requested delimiter can be non-ASCII.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'aaa\xacbbb\xacccc\nddd\xaceee\xacfff\nggg\xachhh\xaciii\n'
|
||
dialect = sniffer.sniff(sample, delimiters='\xac')
|
||
self.assertEqual(dialect.delimiter, '\xac')
|
||
|
||
def test_sniff_preamble(self):
|
||
# A preamble (title or comment lines) before the data must not
|
||
# prevent detection, even if it is larger than the part of the
|
||
# sample which is parsed first. It is enough for the data rows
|
||
# to slightly outnumber the preamble lines.
|
||
sniffer = csv.Sniffer()
|
||
for n in 3, 24, 80, 320:
|
||
preamble = ''.join(f'Comment line {i:05}\n' for i in range(n))
|
||
for ndata in n + 2, 2 * n + 5:
|
||
with self.subTest(preamble_lines=n, data_lines=ndata):
|
||
sample = preamble + 'aaa,bbb,ccc\n' * ndata
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, ',')
|
||
|
||
def test_sniff_line_boundary_characters(self):
|
||
# Only '\r', '\n' and '\r\n' are row separators; characters like
|
||
# '\x1c' or '\x85', which str.splitlines() treats as line
|
||
# boundaries, are ordinary data for the reader -- '\x1c' can
|
||
# even be the delimiter.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'aa\x1cbb\ncc\x1cdd\nee\x1cff\n'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, '\x1c')
|
||
sample = 'a\x85b,c\nd,e\nf,g\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(next(csv.reader(StringIO(sample), dialect)),
|
||
['a\x85b', 'c'])
|
||
|
||
def test_sniff_delimiter_in_quoted_field(self):
|
||
# gh-97611: a delimiter inside a quoted field should not win over
|
||
# the delimiter which actually separates the fields.
|
||
sniffer = csv.Sniffer()
|
||
sample = (
|
||
'Surname;First Name;Year of birth\n'
|
||
'"\tDoe;Jane"\t;1971\n'
|
||
)
|
||
dialect = sniffer.sniff(sample, delimiters=',;\t')
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
sample = (
|
||
'Surname;First Name;Year of birth\n'
|
||
'"\t";"\t";1971\n'
|
||
'"Le Trec";"Mary Ann";1486\n'
|
||
)
|
||
dialect = sniffer.sniff(sample, delimiters=',;\t')
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
|
||
def test_sniff_embedded_lists(self):
|
||
# gh-119123: commas in bracketed lists adjacent to quotes should
|
||
# not be mistaken for the delimiter.
|
||
sniffer = csv.Sniffer()
|
||
sample = (
|
||
"id;is_sort;cost;group;merge\n"
|
||
"1;True;62.25;['345'];UNKNOWN\n"
|
||
"2;True;54.00;['235'];UNKNOWN\n"
|
||
"3;True;237.00;['567', '568'];UNKNOWN\n"
|
||
"4;True;46.50;['112', '112'];UNKNOWN\n"
|
||
)
|
||
dialect = sniffer.sniff(sample, delimiters=',;')
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
|
||
def test_sniff_space_adjacent_to_quotes(self):
|
||
# gh-88843: a space adjacent to stray quotes should not be
|
||
# detected as the delimiter.
|
||
sniffer = csv.Sniffer()
|
||
sample = "a|b\nc| 'd\ne|' f"
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, '|')
|
||
|
||
def test_sniff_crlf_lineterminator(self):
|
||
# gh-103925: a quote at the end of a \r\n-terminated line.
|
||
sniffer = csv.Sniffer()
|
||
sample = (
|
||
'Timestamp,URL,Title\r\n'
|
||
'2020-10-01 17:17:37+08:00,'
|
||
'https://www.mozilla.org/en-US/firefox/welcome/2/,'
|
||
'"Pocket - Save news, videos, stories and more"\r\n'
|
||
)
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertEqual(dialect.lineterminator, '\r\n')
|
||
|
||
def test_sniff_lineterminator(self):
|
||
sniffer = csv.Sniffer()
|
||
for lineterminator in '\r\n', '\n', '\r':
|
||
with self.subTest(lineterminator=lineterminator):
|
||
sample = lineterminator.join(['a,b,c', 'd,e,f', 'g,h,i', ''])
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.lineterminator, lineterminator)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
# The majority wins.
|
||
sample = 'a,b,c\nd,e,f\r\ng,h,i\n'
|
||
self.assertEqual(sniffer.sniff(sample).lineterminator, '\n')
|
||
sample = 'a,b,c\r\nd,e,f\ng,h,i\r\n'
|
||
self.assertEqual(sniffer.sniff(sample).lineterminator, '\r\n')
|
||
# A line break inside a quoted field is counted too, but it is
|
||
# outvoted by the real ones.
|
||
sample = 'a,"x\ny",c\r\nd,e,f\r\ng,h,i\r\n'
|
||
self.assertEqual(sniffer.sniff(sample).lineterminator, '\r\n')
|
||
|
||
def test_sniff_lineterminator_tie(self):
|
||
# A tie is broken in the order '\r\n', '\n', '\r'.
|
||
sniffer = csv.Sniffer()
|
||
for sample, lineterminator in (
|
||
('a,b,c\nd,e,f\r\n', '\r\n'),
|
||
('a,b,c\r\nd,e,f\ng,h,i\rj,k,l', '\r\n'),
|
||
('a,b,c\nd,e,f\rg,h,i', '\n'),
|
||
# A sample without a complete line is a tie of zeros.
|
||
('a,b,c', '\r\n'),
|
||
):
|
||
with self.subTest(sample=sample):
|
||
self.assertEqual(sniffer.sniff(sample).lineterminator,
|
||
lineterminator)
|
||
|
||
def test_sniff_excel_tab_with_quotes(self):
|
||
# gh-62029: tab-delimited data with a quoted field containing
|
||
# spaces and doubled quotes.
|
||
sniffer = csv.Sniffer()
|
||
sample = ('foo\tbar\t"baz ""quoted"" here"\tspam eggs\n'
|
||
'ham\teggs\tx y\tz w\n')
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, '\t')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertIs(dialect.doublequote, True)
|
||
|
||
def test_sniff_truncated_sample(self):
|
||
# A sample cut off in the middle of a quoted field should not
|
||
# spoil the detection.
|
||
sniffer = csv.Sniffer()
|
||
sample = '"a,a";"b"\n"c";"d"\n"e";"f,f\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ';')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
# Cut off in the middle of the last row.
|
||
sample = 'a,b,c\nd,e,f\ng,h,i\nj,k'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, ',')
|
||
# Cut off in the middle of an escaped sequence.
|
||
sample = 'ab,cd\\,ef\ngh\\,ij,kl\nmn,op\nqr,st\\'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.escapechar, '\\')
|
||
# Cut off at a line end in the middle of a quoted field.
|
||
sample = '"a","b"\n"c","d"\n"e","multi\nline'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, ',')
|
||
# An unclosed quote consumes everything to the end of the sample,
|
||
# but the rows before it still count.
|
||
sample = 'a|b\nc|d\ne|f\ng|"h\ni|j\nk|l'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, '|')
|
||
|
||
def test_sniff_skipinitialspace_quoted(self):
|
||
sniffer = csv.Sniffer()
|
||
sample = "'a': 'b': 'c'\n'd': 'e': 'f'\n"
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ':')
|
||
self.assertEqual(dialect.quotechar, "'")
|
||
self.assertIs(dialect.skipinitialspace, True)
|
||
|
||
def test_sniff_skipinitialspace_quoted_fields(self):
|
||
# A quote is only a quote at the very start of a field, so not
|
||
# skipping the space splits the quoted field.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'a, "b,c"\nd,e\nf,g\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertIs(dialect.skipinitialspace, True)
|
||
self.assertEqual(next(csv.reader(StringIO(sample), dialect)),
|
||
['a', 'b,c'])
|
||
|
||
# But without a delimiter inside the quotes nothing is split,
|
||
# so only the padding counts.
|
||
sample = 'a, "b"\nd,e\nf,g\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
self.assertEqual(next(csv.reader(StringIO(sample), dialect)),
|
||
['a', ' "b"'])
|
||
|
||
def test_sniff_skipinitialspace_data(self):
|
||
# The spaces are a part of the data if only some fields
|
||
# are padded.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'a, b\nc,d\ne, f\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
self.assertEqual(next(csv.reader(StringIO(sample), dialect)),
|
||
['a', ' b'])
|
||
|
||
def test_sniff_skipinitialspace_not_skipped(self):
|
||
# A quoted or escaped space is not skipped, so it is not
|
||
# an evidence of the padding.
|
||
sniffer = csv.Sniffer()
|
||
sample = 'a," b"\nc, d\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.quotechar, '"')
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
self.assertEqual(list(csv.reader(StringIO(sample), dialect)),
|
||
[['a', ' b'], ['c', ' d']])
|
||
|
||
# The escaped delimiter forces the escapechar detection.
|
||
sample = 'a,\\ b\\,c\nd, e\n'
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ',')
|
||
self.assertEqual(dialect.escapechar, '\\')
|
||
self.assertIs(dialect.skipinitialspace, False)
|
||
self.assertEqual(list(csv.reader(StringIO(sample), dialect)),
|
||
[['a', ' b,c'], ['d', ' e']])
|
||
|
||
def test_sniff_regex_backtracking(self):
|
||
# gh-109638: this artificial sample used to take minutes.
|
||
sniffer = csv.Sniffer()
|
||
sample = '"",' * 100 + '"' * 100 + '0' + '"' * 100 + '0'
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, ',')
|
||
|
||
def test_doublequote(self):
|
||
sniffer = csv.Sniffer()
|
||
dialect = sniffer.sniff(self.header1)
|
||
self.assertFalse(dialect.doublequote)
|
||
dialect = sniffer.sniff(self.header2)
|
||
self.assertFalse(dialect.doublequote)
|
||
dialect = sniffer.sniff(self.sample2)
|
||
self.assertTrue(dialect.doublequote)
|
||
dialect = sniffer.sniff(self.sample8)
|
||
self.assertFalse(dialect.doublequote)
|
||
dialect = sniffer.sniff(self.sample9)
|
||
self.assertTrue(dialect.doublequote)
|
||
# A doubled quote character in an unquoted field or an empty
|
||
# quoted field is not evidence of doubling.
|
||
dialect = sniffer.sniff('"x",a""b\n"y",c\n')
|
||
self.assertIs(dialect.doublequote, False)
|
||
dialect = sniffer.sniff('a,"",c\nd,"",f\n')
|
||
self.assertIs(dialect.doublequote, False)
|
||
# A doubled quote character inside a quoted field is.
|
||
dialect = sniffer.sniff('"a""b",c\n"d",e\n')
|
||
self.assertIs(dialect.doublequote, True)
|
||
|
||
def test_guess_delimiter_crlf_not_chosen(self):
|
||
# Ensure that we pick the real delimiter ("|") over "\r" in a tie.
|
||
sniffer = csv.Sniffer()
|
||
sample = "a|b\r\nc|d\r\ne|f\r\n"
|
||
self.assertEqual(sniffer.sniff(sample).delimiter, "|")
|
||
self.assertNotEqual(sniffer.sniff(sample).delimiter, "\r")
|
||
|
||
def test_zero_mode_tie_order_independence(self):
|
||
sniffer = csv.Sniffer()
|
||
# ":" appears in half the rows (1, 0, 1, 0) - a tie between
|
||
# 0 and 1 per line.
|
||
# "," appears once every row (true delimiter).
|
||
#
|
||
# Even if the zero-frequency bucket is appended vs. inserted, the tie
|
||
# yields an adjusted score of 0, so ":" should not be promoted and
|
||
# "," must be selected.
|
||
sample = (
|
||
"a,b:c\n"
|
||
"d,e\n"
|
||
"f,g:c\n"
|
||
"h,i\n"
|
||
)
|
||
dialect = sniffer.sniff(sample)
|
||
self.assertEqual(dialect.delimiter, ",")
|
||
|
||
def test_zero_mode_tie_order_comma_first(self):
|
||
sniffer = csv.Sniffer()
|
||
pattern = (
|
||
"a,b\n"
|
||
"c:d\n"
|
||
"e,f\n"
|
||
"g:h\n"
|
||
)
|
||
sample = pattern * 10
|
||
with self.assertRaisesRegex(csv.Error, "Could not determine delimiter"):
|
||
sniffer.sniff(sample)
|
||
|
||
def test_zero_mode_tie_order_colon_first(self):
|
||
sniffer = csv.Sniffer()
|
||
pattern = (
|
||
"a:b\n"
|
||
"c,d\n"
|
||
"e:f\n"
|
||
"g,h\n"
|
||
)
|
||
sample = pattern * 10
|
||
with self.assertRaisesRegex(csv.Error, "Could not determine delimiter"):
|
||
sniffer.sniff(sample)
|
||
|
||
|
||
class NUL:
|
||
def write(s, *args):
|
||
pass
|
||
writelines = write
|
||
|
||
@unittest.skipUnless(hasattr(sys, "gettotalrefcount"),
|
||
'requires sys.gettotalrefcount()')
|
||
class TestLeaks(unittest.TestCase):
|
||
def test_create_read(self):
|
||
delta = 0
|
||
lastrc = sys.gettotalrefcount()
|
||
for i in range(20):
|
||
gc.collect()
|
||
self.assertEqual(gc.garbage, [])
|
||
rc = sys.gettotalrefcount()
|
||
csv.reader(["a,b,c\r\n"])
|
||
csv.reader(["a,b,c\r\n"])
|
||
csv.reader(["a,b,c\r\n"])
|
||
delta = rc-lastrc
|
||
lastrc = rc
|
||
# if csv.reader() leaks, last delta should be 3 or more
|
||
self.assertLess(delta, 3)
|
||
|
||
def test_create_write(self):
|
||
delta = 0
|
||
lastrc = sys.gettotalrefcount()
|
||
s = NUL()
|
||
for i in range(20):
|
||
gc.collect()
|
||
self.assertEqual(gc.garbage, [])
|
||
rc = sys.gettotalrefcount()
|
||
csv.writer(s)
|
||
csv.writer(s)
|
||
csv.writer(s)
|
||
delta = rc-lastrc
|
||
lastrc = rc
|
||
# if csv.writer() leaks, last delta should be 3 or more
|
||
self.assertLess(delta, 3)
|
||
|
||
def test_read(self):
|
||
delta = 0
|
||
rows = ["a,b,c\r\n"]*5
|
||
lastrc = sys.gettotalrefcount()
|
||
for i in range(20):
|
||
gc.collect()
|
||
self.assertEqual(gc.garbage, [])
|
||
rc = sys.gettotalrefcount()
|
||
rdr = csv.reader(rows)
|
||
for row in rdr:
|
||
pass
|
||
delta = rc-lastrc
|
||
lastrc = rc
|
||
# if reader leaks during read, delta should be 5 or more
|
||
self.assertLess(delta, 5)
|
||
|
||
def test_write(self):
|
||
delta = 0
|
||
rows = [[1,2,3]]*5
|
||
s = NUL()
|
||
lastrc = sys.gettotalrefcount()
|
||
for i in range(20):
|
||
gc.collect()
|
||
self.assertEqual(gc.garbage, [])
|
||
rc = sys.gettotalrefcount()
|
||
writer = csv.writer(s)
|
||
for row in rows:
|
||
writer.writerow(row)
|
||
delta = rc-lastrc
|
||
lastrc = rc
|
||
# if writer leaks during write, last delta should be 5 or more
|
||
self.assertLess(delta, 5)
|
||
|
||
class TestUnicode(unittest.TestCase):
|
||
|
||
names = ["Martin von Löwis",
|
||
"Marc André Lemburg",
|
||
"Guido van Rossum",
|
||
"François Pinard"]
|
||
|
||
def test_unicode_read(self):
|
||
with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
|
||
fileobj.write(",".join(self.names) + "\r\n")
|
||
fileobj.seek(0)
|
||
reader = csv.reader(fileobj)
|
||
self.assertEqual(list(reader), [self.names])
|
||
|
||
|
||
def test_unicode_write(self):
|
||
with TemporaryFile("w+", newline='', encoding="utf-8") as fileobj:
|
||
writer = csv.writer(fileobj)
|
||
writer.writerow(self.names)
|
||
expected = ",".join(self.names)+"\r\n"
|
||
fileobj.seek(0)
|
||
self.assertEqual(fileobj.read(), expected)
|
||
|
||
class KeyOrderingTest(unittest.TestCase):
|
||
|
||
def test_ordering_for_the_dict_reader_and_writer(self):
|
||
resultset = set()
|
||
for keys in permutations("abcde"):
|
||
with TemporaryFile('w+', newline='', encoding="utf-8") as fileobject:
|
||
dw = csv.DictWriter(fileobject, keys)
|
||
dw.writeheader()
|
||
fileobject.seek(0)
|
||
dr = csv.DictReader(fileobject)
|
||
kt = tuple(dr.fieldnames)
|
||
self.assertEqual(keys, kt)
|
||
resultset.add(kt)
|
||
# Final sanity check: were all permutations unique?
|
||
self.assertEqual(len(resultset), 120, "Key ordering: some key permutations not collected (expected 120)")
|
||
|
||
def test_ordered_dict_reader(self):
|
||
data = dedent('''\
|
||
FirstName,LastName
|
||
Eric,Idle
|
||
Graham,Chapman,Over1,Over2
|
||
|
||
Under1
|
||
John,Cleese
|
||
''').splitlines()
|
||
|
||
self.assertEqual(list(csv.DictReader(data)),
|
||
[OrderedDict([('FirstName', 'Eric'), ('LastName', 'Idle')]),
|
||
OrderedDict([('FirstName', 'Graham'), ('LastName', 'Chapman'),
|
||
(None, ['Over1', 'Over2'])]),
|
||
OrderedDict([('FirstName', 'Under1'), ('LastName', None)]),
|
||
OrderedDict([('FirstName', 'John'), ('LastName', 'Cleese')]),
|
||
])
|
||
|
||
self.assertEqual(list(csv.DictReader(data, restkey='OtherInfo')),
|
||
[OrderedDict([('FirstName', 'Eric'), ('LastName', 'Idle')]),
|
||
OrderedDict([('FirstName', 'Graham'), ('LastName', 'Chapman'),
|
||
('OtherInfo', ['Over1', 'Over2'])]),
|
||
OrderedDict([('FirstName', 'Under1'), ('LastName', None)]),
|
||
OrderedDict([('FirstName', 'John'), ('LastName', 'Cleese')]),
|
||
])
|
||
|
||
del data[0] # Remove the header row
|
||
self.assertEqual(list(csv.DictReader(data, fieldnames=['fname', 'lname'])),
|
||
[OrderedDict([('fname', 'Eric'), ('lname', 'Idle')]),
|
||
OrderedDict([('fname', 'Graham'), ('lname', 'Chapman'),
|
||
(None, ['Over1', 'Over2'])]),
|
||
OrderedDict([('fname', 'Under1'), ('lname', None)]),
|
||
OrderedDict([('fname', 'John'), ('lname', 'Cleese')]),
|
||
])
|
||
|
||
|
||
class MiscTestCase(unittest.TestCase):
|
||
def test__all__(self):
|
||
support.check__all__(self, csv, ('csv', '_csv'))
|
||
|
||
@cpython_only
|
||
def test_lazy_import(self):
|
||
ensure_lazy_imports("csv", {"re"})
|
||
|
||
def test_subclassable(self):
|
||
# issue 44089
|
||
class Foo(csv.Error): ...
|
||
|
||
@support.cpython_only
|
||
def test_disallow_instantiation(self):
|
||
_csv = import_helper.import_module("_csv")
|
||
for tp in _csv.Reader, _csv.Writer:
|
||
with self.subTest(tp=tp):
|
||
check_disallow_instantiation(self, tp)
|
||
|
||
|
||
class TestModule(unittest.TestCase):
|
||
def test_deprecated__version__(self):
|
||
with self.assertWarnsRegex(
|
||
DeprecationWarning,
|
||
"'__version__' is deprecated and slated for removal in Python 3.20",
|
||
) as cm:
|
||
getattr(csv, "__version__")
|
||
self.assertEqual(cm.filename, __file__)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|