Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_stream_csv(self):
table = [["A", "B", "C"], [1, 2, 3], [4, 5, 6]]
dialect = SimpleDialect(delimiter=";", quotechar="", escapechar="")
with self.subTest(name="simple"):
self._stream_test(table, dialect)
table = [["A,0", "B", "C"], [1, 2, 3], [4, 5, 6]]
dialect = SimpleDialect(delimiter=",", quotechar="", escapechar="\\")
with self.subTest(name="escaped"):
self._stream_test(table, dialect)
table = [["A,0", "B", "C"], [1, 2, 3], [4, 5, 6]]
dialect = SimpleDialect(delimiter=",", quotechar='"', escapechar="")
with self.subTest(name="quoted"):
self._stream_test(table, dialect)
table = [['a"A,0"b', "B", "C"], [1, 2, 3], [4, 5, 6]]
dialect = SimpleDialect(delimiter=",", quotechar='"', escapechar="")
with self.subTest(name="double"):
def _write_error_test(self, exc, fields, **kwargs):
with tempfile.TemporaryFile("w+", newline="", prefix="ccsv_") as fp:
writer = clevercsv.writer(fp, **kwargs)
with self.assertRaises(exc):
writer.writerow(fields)
fp.seek(0)
self.assertEqual(fp.read(), "")
def _df_test(self, table, dialect, **kwargs):
tmpfd, tmpfname = tempfile.mkstemp(prefix="ccsv_", suffix=".csv")
tmpid = os.fdopen(tmpfd, "w")
w = writer(tmpid, dialect=dialect)
w.writerows(table)
tmpid.close()
exp_df = pd.DataFrame.from_records(table[1:], columns=table[0])
df = wrappers.csv2df(tmpfname)
try:
self.assertTrue(df.equals(exp_df))
finally:
os.unlink(tmpfname)
def _df_test(self, table, dialect, **kwargs):
tmpfd, tmpfname = tempfile.mkstemp(prefix="ccsv_", suffix=".csv")
tmpid = os.fdopen(tmpfd, "w")
w = writer(tmpid, dialect=dialect)
w.writerows(table)
tmpid.close()
exp_df = pd.DataFrame.from_records(table[1:], columns=table[0])
df = wrappers.csv2df(tmpfname)
try:
self.assertTrue(df.equals(exp_df))
finally:
os.unlink(tmpfname)
def test_read_bigfield(self):
limit = clevercsv.field_size_limit()
try:
size = 500
bigstring = "X" * size
bigline = "%s,%s" % (bigstring, bigstring)
self._read_test([bigline], [[bigstring, bigstring]])
clevercsv.field_size_limit(size)
self._read_test([bigline], [[bigstring, bigstring]])
self.assertEqual(clevercsv.field_size_limit(), size)
clevercsv.field_size_limit(size=-1)
self.assertRaises(clevercsv.Error, self._read_test, [bigline], [])
self.assertRaises(TypeError, clevercsv.field_size_limit, None)
self.assertRaises(TypeError, clevercsv.field_size_limit, 1, None)
finally:
clevercsv.field_size_limit(limit)
def test_read_bigfield(self):
limit = clevercsv.field_size_limit()
try:
size = 500
bigstring = "X" * size
bigline = "%s,%s" % (bigstring, bigstring)
self._read_test([bigline], [[bigstring, bigstring]])
clevercsv.field_size_limit(size)
self._read_test([bigline], [[bigstring, bigstring]])
self.assertEqual(clevercsv.field_size_limit(), size)
clevercsv.field_size_limit(size=-1)
self.assertRaises(clevercsv.Error, self._read_test, [bigline], [])
self.assertRaises(TypeError, clevercsv.field_size_limit, None)
self.assertRaises(TypeError, clevercsv.field_size_limit, 1, None)
finally:
clevercsv.field_size_limit(limit)
def test_read_bigfield(self):
limit = clevercsv.field_size_limit()
try:
size = 500
bigstring = "X" * size
bigline = "%s,%s" % (bigstring, bigstring)
self._read_test([bigline], [[bigstring, bigstring]])
clevercsv.field_size_limit(size)
self._read_test([bigline], [[bigstring, bigstring]])
self.assertEqual(clevercsv.field_size_limit(), size)
clevercsv.field_size_limit(size=-1)
self.assertRaises(clevercsv.Error, self._read_test, [bigline], [])
self.assertRaises(TypeError, clevercsv.field_size_limit, None)
self.assertRaises(TypeError, clevercsv.field_size_limit, 1, None)
finally:
clevercsv.field_size_limit(limit)
def _write_test(self, table, expected, dialect="excel", transpose=False):
tmpfd, tmpfname = tempfile.mkstemp(prefix="ccsv_", suffix=".csv")
wrappers.write_table(
table, tmpfname, dialect=dialect, transpose=transpose
)
with open(tmpfname, "r") as fp:
data = fp.read()
try:
self.assertEqual(data, expected)
finally:
os.close(tmpfd)
os.unlink(tmpfname)
def _df_test(self, table, dialect, **kwargs):
tmpfd, tmpfname = tempfile.mkstemp(prefix="ccsv_", suffix=".csv")
tmpid = os.fdopen(tmpfd, "w")
w = writer(tmpid, dialect=dialect)
w.writerows(table)
tmpid.close()
exp_df = pd.DataFrame.from_records(table[1:], columns=table[0])
df = wrappers.csv2df(tmpfname)
try:
self.assertTrue(df.equals(exp_df))
finally:
os.unlink(tmpfname)
def _df_test(self, table, dialect, **kwargs):
tmpfd, tmpfname = tempfile.mkstemp(prefix="ccsv_", suffix=".csv")
tmpid = os.fdopen(tmpfd, "w")
w = writer(tmpid, dialect=dialect)
w.writerows(table)
tmpid.close()
exp_df = pd.DataFrame.from_records(table[1:], columns=table[0])
df = wrappers.csv2df(tmpfname)
try:
self.assertTrue(df.equals(exp_df))
finally:
os.unlink(tmpfname)