(tmpdir)
| 1639 | |
| 1640 | @pytest.mark.gzip |
| 1641 | def test_compressed_input_openfile(tmpdir): |
| 1642 | if not Codec.is_available("gzip"): |
| 1643 | pytest.skip("gzip support is not built") |
| 1644 | |
| 1645 | data = b"some test data\n" * 10 + b"eof\n" |
| 1646 | fn = str(tmpdir / "test_compressed_input_openfile.gz") |
| 1647 | with gzip.open(fn, "wb") as f: |
| 1648 | f.write(data) |
| 1649 | |
| 1650 | with pa.CompressedInputStream(fn, "gzip") as compressed: |
| 1651 | buf = compressed.read_buffer() |
| 1652 | assert buf.to_pybytes() == data |
| 1653 | assert compressed.closed |
| 1654 | |
| 1655 | with pa.CompressedInputStream(pathlib.Path(fn), "gzip") as compressed: |
| 1656 | buf = compressed.read_buffer() |
| 1657 | assert buf.to_pybytes() == data |
| 1658 | assert compressed.closed |
| 1659 | |
| 1660 | f = open(fn, "rb") |
| 1661 | with pa.CompressedInputStream(f, "gzip") as compressed: |
| 1662 | buf = compressed.read_buffer() |
| 1663 | assert buf.to_pybytes() == data |
| 1664 | assert f.closed |
| 1665 | |
| 1666 | |
| 1667 | def check_compressed_concatenated(data, fn, compression): |
nothing calls this directly
no test coverage detected