(tmpdir)
| 2203 | |
| 2204 | @pytest.mark.gzip |
| 2205 | def test_output_stream_python_file(tmpdir): |
| 2206 | data = b"some test data\n" * 10 + b"eof\n" |
| 2207 | |
| 2208 | def check_data(data, **kwargs): |
| 2209 | # XXX cannot use BytesIO because stream.close() is necessary |
| 2210 | # to finish writing compressed data, but it will also close the |
| 2211 | # underlying BytesIO |
| 2212 | fn = str(tmpdir / 'output_stream_file') |
| 2213 | with open(fn, 'wb') as f: |
| 2214 | with pa.output_stream(f, **kwargs) as stream: |
| 2215 | stream.write(data) |
| 2216 | with open(fn, 'rb') as f: |
| 2217 | return f.read() |
| 2218 | |
| 2219 | assert check_data(data) == data |
| 2220 | assert gzip.decompress(check_data(data, compression='gzip')) == data |
| 2221 | |
| 2222 | |
| 2223 | def test_output_stream_errors(tmpdir): |
nothing calls this directly
no test coverage detected