Skip to content

Commit a1f3470

Browse files
committed
fix: copy stream data through short reads
`stream_copy` stopped after any read shorter than `chunk_size`, truncating valid unbuffered streams. Continue until an empty chunk reports end of input, preserving the existing writer contract and byte count. Add binary/text short-read and empty-input regressions, plus a synchronized real POSIX pipe reproduction. The regressions fail before the fix; utility and typing modules pass with both CLI and Gix backends (126 passed, 39 skipped each). Applicable pre-commit checks pass. Assisted-by: OpenAI Codex (model: unknown, autonomous)
1 parent cfb222f commit a1f3470

2 files changed

Lines changed: 62 additions & 1 deletion

File tree

‎git/util.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,8 @@ def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr]
219219
"""Copy all data from the `source` stream into the `destination` stream in chunks
220220
of size `chunk_size`.
221221
222+
Keep reading after short reads until the source returns an empty chunk.
223+
222224
:return:
223225
Number of bytes written
224226
"""
@@ -227,7 +229,7 @@ def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr]
227229
chunk = source.read(chunk_size)
228230
destination.write(chunk)
229231
br += len(chunk)
230-
if len(chunk) < chunk_size:
232+
if not chunk:
231233
break
232234
# END reading output stream
233235
return br

‎test/test_util.py‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import contextlib
77
import errno
88
from datetime import datetime
9+
from io import BytesIO, StringIO
910
import os
1011
import pathlib
1112
import pickle
@@ -40,12 +41,70 @@
4041
is_cygwin_git,
4142
remove_password_if_present,
4243
rmtree,
44+
stream_copy,
4345
)
4446
from test.cleanup import cleanup_directory
4547
from test.lib.fixtures import writable_fixture
4648
from test.lib import TestBase, requires_symlinks, with_rw_repo
4749

4850

51+
@pytest.mark.parametrize(
52+
"parts, expected",
53+
[([b"ab", b"c", b"def"], b"abcdef"), (["ab", "c", "def"], "abcdef"), ([], b""), ([], "")],
54+
)
55+
def test_stream_copy_waits_for_an_empty_read(parts, expected):
56+
chunks = iter(parts)
57+
58+
class Reader:
59+
def read(self, size):
60+
chunk = next(chunks, expected[:0])
61+
assert len(chunk) <= size
62+
return chunk
63+
64+
destination = BytesIO() if isinstance(expected, bytes) else StringIO()
65+
assert stream_copy(Reader(), destination, chunk_size=4) == len(expected)
66+
assert destination.getvalue() == expected
67+
68+
69+
@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX unbuffered pipe reads")
70+
def test_stream_copy_keeps_reading_an_unbuffered_pipe():
71+
read_fd, write_fd = os.pipe()
72+
first_copied = threading.Event()
73+
errors = []
74+
75+
class Destination(BytesIO):
76+
def write(self, data):
77+
result = super().write(data)
78+
if data:
79+
first_copied.set()
80+
return result
81+
82+
def produce():
83+
try:
84+
os.write(write_fd, b"first-")
85+
if not first_copied.wait(5):
86+
raise RuntimeError("first pipe chunk was not copied")
87+
os.write(write_fd, b"second")
88+
except Exception as error:
89+
errors.append(error)
90+
finally:
91+
os.close(write_fd)
92+
93+
destination = Destination()
94+
with os.fdopen(read_fd, "rb", buffering=0) as source:
95+
thread = threading.Thread(target=produce)
96+
thread.start()
97+
try:
98+
copied = stream_copy(source, destination, chunk_size=1024)
99+
finally:
100+
first_copied.set()
101+
thread.join(5)
102+
assert not thread.is_alive()
103+
assert not errors
104+
assert copied == len(b"first-second")
105+
assert destination.getvalue() == b"first-second"
106+
107+
49108
@pytest.fixture
50109
def tmp_path(request):
51110
# Unreadable files and symlinks cannot be copied by native fixture tooling.

0 commit comments

Comments
 (0)