(nbytes, file_offset)
| 138 | @pytest.mark.parametrize("nbytes", (-1, 0, 1, 5, 100)) |
| 139 | @pytest.mark.parametrize("file_offset", (-1, 0, 5, 100)) |
| 140 | def test_python_file_get_stream(nbytes, file_offset): |
| 141 | |
| 142 | data = b'data1data2data3data4data5' |
| 143 | |
| 144 | f = pa.PythonFile(BytesIO(data), mode='r') |
| 145 | |
| 146 | # negative nbytes or offsets don't make sense here, raise ValueError |
| 147 | if nbytes < 0 or file_offset < 0: |
| 148 | with pytest.raises(pa.ArrowInvalid, |
| 149 | match="should be a positive value"): |
| 150 | f.get_stream(file_offset=file_offset, nbytes=nbytes) |
| 151 | f.close() |
| 152 | return |
| 153 | else: |
| 154 | stream = f.get_stream(file_offset=file_offset, nbytes=nbytes) |
| 155 | |
| 156 | # Subsequent calls to 'read' should match behavior if same |
| 157 | # data passed to BytesIO where get_stream should handle if |
| 158 | # nbytes/file_offset results in no bytes b/c out of bounds. |
| 159 | start = min(file_offset, len(data)) |
| 160 | end = min(file_offset + nbytes, len(data)) |
| 161 | buf = BytesIO(data[start:end]) |
| 162 | |
| 163 | # read some chunks |
| 164 | assert stream.read(nbytes=4) == buf.read(4) |
| 165 | assert stream.read(nbytes=6) == buf.read(6) |
| 166 | |
| 167 | # Read to end of each stream |
| 168 | assert stream.read() == buf.read() |
| 169 | |
| 170 | # Try reading past the stream |
| 171 | n = len(data) * 2 |
| 172 | assert stream.read(n) == buf.read(n) |
| 173 | |
| 174 | # NativeFile[CInputStream] is not seekable |
| 175 | with pytest.raises(OSError, match="seekable"): |
| 176 | stream.seek(0) |
| 177 | |
| 178 | stream.close() |
| 179 | assert stream.closed |
| 180 | |
| 181 | |
| 182 | def test_python_file_read_at(): |
nothing calls this directly
no test coverage detected