|
4 | 4 | from datetime import datetime
|
5 | 5 | import collections
|
6 | 6 | import pytest
|
| 7 | +import gzip |
| 8 | +import bz2 |
| 9 | +import lzma |
7 | 10 |
|
8 | 11 | import numpy as np
|
9 | 12 | import pandas as pd
|
10 | 13 |
|
11 |
| -from pandas import Series, DataFrame, compat |
| 14 | +from pandas import Series, DataFrame |
12 | 15 |
|
13 | 16 | from pandas.compat import StringIO, u
|
14 | 17 | from pandas.util.testing import (assert_series_equal, assert_almost_equal,
|
@@ -139,67 +142,34 @@ def test_to_csv_path_is_none(self):
|
139 | 142 | csv_str = s.to_csv(path=None)
|
140 | 143 | assert isinstance(csv_str, str)
|
141 | 144 |
|
142 |
| - def test_to_csv_compression_gzip(self): |
| 145 | + @pytest.mark.parametrize('compression, open_func', [ |
| 146 | + ('gzip', gzip.open), |
| 147 | + ('bz2', bz2.BZ2File), |
| 148 | + (pytest.param('xz', lzma.open, marks=td.skip_if_no_lzma)), |
| 149 | + ]) |
| 150 | + def test_to_csv_compression(self, compression, open_func): |
143 | 151 |
|
144 | 152 | s = Series([0.123456, 0.234567, 0.567567], index=['A', 'B', 'C'],
|
145 | 153 | name='X')
|
146 | 154 |
|
147 | 155 | with ensure_clean() as filename:
|
148 | 156 |
|
149 |
| - s.to_csv(filename, compression="gzip", header=True) |
| 157 | + s.to_csv(filename, compression=compression, header=True) |
150 | 158 |
|
151 | 159 | # test the round trip - to_csv -> read_csv
|
152 |
| - rs = pd.read_csv(filename, compression="gzip", index_col=0, |
| 160 | + rs = pd.read_csv(filename, compression=compression, index_col=0, |
153 | 161 | squeeze=True)
|
154 | 162 | assert_series_equal(s, rs)
|
155 | 163 |
|
156 |
| - # explicitly make sure file is gziped |
157 |
| - import gzip |
158 |
| - f = gzip.open(filename, 'rb') |
| 164 | + # explicitly make sure file is compressed |
| 165 | + f = open_func(filename, 'rb') |
159 | 166 | text = f.read().decode('utf8')
|
160 |
| - f.close() |
161 |
| - assert s.name in text |
162 |
| - |
163 |
| - def test_to_csv_compression_bz2(self): |
164 |
| - |
165 |
| - s = Series([0.123456, 0.234567, 0.567567], index=['A', 'B', 'C'], |
166 |
| - name='X') |
167 |
| - |
168 |
| - with ensure_clean() as filename: |
169 |
| - |
170 |
| - s.to_csv(filename, compression="bz2", header=True) |
171 |
| - |
172 |
| - # test the round trip - to_csv -> read_csv |
173 |
| - rs = pd.read_csv(filename, compression="bz2", index_col=0, |
174 |
| - squeeze=True) |
175 |
| - assert_series_equal(s, rs) |
176 |
| - |
177 |
| - # explicitly make sure file is bz2ed |
178 |
| - import bz2 |
179 |
| - f = bz2.BZ2File(filename, 'rb') |
180 |
| - text = f.read().decode('utf8') |
181 |
| - f.close() |
182 | 167 | assert s.name in text
|
183 | 168 |
|
184 |
| - @td.skip_if_no_lzma |
185 |
| - def test_to_csv_compression_xz(self): |
186 |
| - |
187 |
| - s = Series([0.123456, 0.234567, 0.567567], index=['A', 'B', 'C'], |
188 |
| - name='X') |
189 |
| - |
190 |
| - with ensure_clean() as filename: |
191 |
| - |
192 |
| - s.to_csv(filename, compression="xz", header=True) |
193 |
| - |
194 |
| - # test the round trip - to_csv -> read_csv |
195 |
| - rs = pd.read_csv(filename, compression="xz", index_col=0, |
196 |
| - squeeze=True) |
197 |
| - assert_series_equal(s, rs) |
| 169 | + f = open_func(filename, 'rb') |
| 170 | + assert_series_equal(s, pd.read_csv(f, index_col=0, |
| 171 | + squeeze=True)) |
198 | 172 |
|
199 |
| - # explicitly make sure file is xzipped |
200 |
| - lzma = compat.import_lzma() |
201 |
| - f = lzma.open(filename, 'rb') |
202 |
| - assert_series_equal(s, pd.read_csv(f, index_col=0, squeeze=True)) |
203 | 173 | f.close()
|
204 | 174 |
|
205 | 175 |
|
|
0 commit comments