Skip to content

TST: added ensure_clean file cleanup context manager for file based test #3103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 20, 2013
Merged

TST: added ensure_clean file cleanup context manager for file based test #3103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
357 changes: 177 additions & 180 deletions pandas/io/tests/test_excel.py
View file
Open in desktop

Large diffs are not rendered by default.

139 changes: 55 additions & 84 deletions pandas/io/tests/test_parsers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@
from pandas.io.parsers import (read_csv, read_table, read_fwf,
TextFileReader, TextParser)
from pandas.util.testing import (assert_almost_equal,
assert_series_equal, network)
assert_series_equal,
network,
ensure_clean)
import pandas.util.testing as tm
import pandas as pd

Expand Down Expand Up @@ -1372,29 +1374,25 @@ def test_utf16_bom_skiprows(self):

path = '__%s__.csv' % tm.rands(10)

for sep, dat in [('\t', data), (',', data2)]:
for enc in ['utf-16', 'utf-16le', 'utf-16be']:
bytes = dat.encode(enc)
with open(path, 'wb') as f:
f.write(bytes)

s = BytesIO(dat.encode('utf-8'))
if py3compat.PY3:
# somewhat False since the code never sees bytes
from io import TextIOWrapper
s = TextIOWrapper(s, encoding='utf-8')

result = self.read_csv(path, encoding=enc, skiprows=2,
sep=sep)
expected = self.read_csv(s, encoding='utf-8', skiprows=2,
sep=sep)

tm.assert_frame_equal(result, expected)

try:
os.remove(path)
except os.error:
pass
with ensure_clean(path) as path:
for sep, dat in [('\t', data), (',', data2)]:
for enc in ['utf-16', 'utf-16le', 'utf-16be']:
bytes = dat.encode(enc)
with open(path, 'wb') as f:
f.write(bytes)

s = BytesIO(dat.encode('utf-8'))
if py3compat.PY3:
# somewhat False since the code never sees bytes
from io import TextIOWrapper
s = TextIOWrapper(s, encoding='utf-8')

result = self.read_csv(path, encoding=enc, skiprows=2,
sep=sep)
expected = self.read_csv(s, encoding='utf-8', skiprows=2,
sep=sep)

tm.assert_frame_equal(result, expected)

def test_utf16_example(self):
path = os.path.join(self.dirpath, 'utf16_ex.txt')
Expand Down Expand Up @@ -1722,32 +1720,27 @@ def test_iteration_open_handle(self):
if PY3:
raise nose.SkipTest

with open('__foo__.txt', 'wb') as f:
f.write('AAA\nBBB\nCCC\nDDD\nEEE\nFFF\nGGG')
with ensure_clean() as path:
with open(path, 'wb') as f:
f.write('AAA\nBBB\nCCC\nDDD\nEEE\nFFF\nGGG')

with open('__foo__.txt', 'rb') as f:
for line in f:
if 'CCC' in line:
break
with open(path, 'rb') as f:
for line in f:
if 'CCC' in line:
break

try:
read_table(f, squeeze=True, header=None, engine='c')
except Exception:
pass
else:
raise ValueError('this should not happen')

result = read_table(f, squeeze=True, header=None,
engine='python')
try:
read_table(f, squeeze=True, header=None, engine='c')
except Exception:
pass
else:
raise ValueError('this should not happen')

expected = Series(['DDD', 'EEE', 'FFF', 'GGG'])
tm.assert_series_equal(result, expected)

try:
os.remove('__foo__.txt')
except os.error:
pass
result = read_table(f, squeeze=True, header=None,
engine='python')

expected = Series(['DDD', 'EEE', 'FFF', 'GGG'])
tm.assert_series_equal(result, expected)

class TestCParserHighMemory(ParserTests, unittest.TestCase):

Expand Down Expand Up @@ -1924,41 +1917,30 @@ def test_decompression(self):
data = open(self.csv1, 'rb').read()
expected = self.read_csv(self.csv1)

try:
tmp = gzip.GzipFile('__tmp__', mode='wb')
with ensure_clean() as path:
tmp = gzip.GzipFile(path, mode='wb')
tmp.write(data)
tmp.close()

result = self.read_csv('__tmp__', compression='gzip')
result = self.read_csv(path, compression='gzip')
tm.assert_frame_equal(result, expected)

result = self.read_csv(open('__tmp__', 'rb'), compression='gzip')
result = self.read_csv(open(path, 'rb'), compression='gzip')
tm.assert_frame_equal(result, expected)
finally:
# try:
# os.remove('__tmp__')
# except:
# pass
pass

try:
tmp = bz2.BZ2File('__tmp__', mode='wb')
with ensure_clean() as path:
tmp = bz2.BZ2File(path, mode='wb')
tmp.write(data)
tmp.close()

result = self.read_csv('__tmp__', compression='bz2')
result = self.read_csv(path, compression='bz2')
tm.assert_frame_equal(result, expected)

# result = self.read_csv(open('__tmp__', 'rb'), compression='bz2')
# result = self.read_csv(open(path, 'rb'), compression='bz2')
# tm.assert_frame_equal(result, expected)

self.assertRaises(ValueError, self.read_csv,
'__tmp__', compression='bz3')
finally:
try:
os.remove('__tmp__')
except:
pass
path, compression='bz3')

def test_decompression_regex_sep(self):
try:
Expand All @@ -1971,35 +1953,24 @@ def test_decompression_regex_sep(self):
data = data.replace(b',', b'::')
expected = self.read_csv(self.csv1)

try:
tmp = gzip.GzipFile('__tmp__', mode='wb')
with ensure_clean() as path:
tmp = gzip.GzipFile(path, mode='wb')
tmp.write(data)
tmp.close()

result = self.read_csv('__tmp__', sep='::', compression='gzip')
result = self.read_csv(path, sep='::', compression='gzip')
tm.assert_frame_equal(result, expected)
finally:
# try:
# os.remove('__tmp__')
# except:
# pass
pass

try:
tmp = bz2.BZ2File('__tmp__', mode='wb')
with ensure_clean() as path:
tmp = bz2.BZ2File(path, mode='wb')
tmp.write(data)
tmp.close()

result = self.read_csv('__tmp__', sep='::', compression='bz2')
result = self.read_csv(path, sep='::', compression='bz2')
tm.assert_frame_equal(result, expected)

self.assertRaises(ValueError, self.read_csv,
'__tmp__', compression='bz3')
finally:
try:
os.remove('__tmp__')
except:
pass
path, compression='bz3')

def test_memory_map(self):
# it works!
Expand Down
Loading