(tmpdir)
| 2178 | |
| 2179 | @pytest.mark.gzip |
| 2180 | def test_output_stream_python_file(tmpdir): |
| 2181 | data = b"some test data\n" * 10 + b"eof\n" |
| 2182 | |
| 2183 | def check_data(data, **kwargs): |
| 2184 | # XXX cannot use BytesIO because stream.close() is necessary |
| 2185 | # to finish writing compressed data, but it will also close the |
| 2186 | # underlying BytesIO |
| 2187 | fn = str(tmpdir / 'output_stream_file') |
| 2188 | with open(fn, 'wb') as f: |
| 2189 | with pa.output_stream(f, **kwargs) as stream: |
| 2190 | stream.write(data) |
| 2191 | with open(fn, 'rb') as f: |
| 2192 | return f.read() |
| 2193 | |
| 2194 | assert check_data(data) == data |
| 2195 | assert gzip.decompress(check_data(data, compression='gzip')) == data |
| 2196 | |
| 2197 | |
| 2198 | def test_output_stream_errors(tmpdir): |
nothing calls this directly
no test coverage detected