Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion b2sdk/_internal/stream/range.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ def read(self, size=None):
:rtype: bytes
"""
remaining = max(0, self.length - self.relative_pos)
if size is None:
if size is None or size < 0:
to_read = remaining
else:
to_read = min(size, remaining)
Expand Down
1 change: 1 addition & 0 deletions changelog.d/+range-negative-read.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Keep negative-size reads of `RangeOfInputStream` within the selected range, including empty and exhausted ranges.
42 changes: 42 additions & 0 deletions test/unit/stream/test_range.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
######################################################################
#
# File: test/unit/stream/test_range.py
#
# Copyright 2026 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
import io

import pytest

from b2sdk.v3 import RangeOfInputStream


@pytest.mark.parametrize('size', [-1, -2, None, 99])
@pytest.mark.parametrize('already_read', [0, 1, 3])
def test_read_all_stays_within_range(size, already_read):
with RangeOfInputStream(io.BytesIO(b'0123456789'), offset=2, length=3) as stream:
assert stream.read(already_read) == b'234'[:already_read]
assert stream.read(size) == b'234'[already_read:]
assert stream.tell() == 3
assert stream.read(size) == b''
assert stream.tell() == 3


def test_negative_read_of_empty_range_does_not_consume_source():
with RangeOfInputStream(io.BytesIO(b'0123456789'), offset=2, length=0) as stream:
assert stream.read(-1) == b''
assert stream.tell() == 0
assert stream.stream.tell() == 2


def test_zero_and_bounded_reads_preserve_position():
with RangeOfInputStream(io.BytesIO(b'0123456789'), offset=2, length=3) as stream:
assert stream.read(0) == b''
assert stream.tell() == 0
assert stream.read(2) == b'23'
assert stream.tell() == 2
assert stream.read(2) == b'4'
assert stream.tell() == 3