From a1f3470e95d1005ca5f45d5f173c64f3b2b959eb Mon Sep 17 00:00:00 2001 From: Gaven Lan Ren Date: Wed, 28 Oct 2026 14:13:04 -0400 Subject: [PATCH] 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) --- git/util.py | 4 +++- test/test_util.py | 59 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/git/util.py b/git/util.py index 2d4cc7c23..6e43d9a24 100644 --- a/git/util.py +++ b/git/util.py @@ -219,6 +219,8 @@ def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr] """Copy all data from the `source` stream into the `destination` stream in chunks of size `chunk_size`. + Keep reading after short reads until the source returns an empty chunk. + :return: Number of bytes written """ @@ -227,7 +229,7 @@ def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr] chunk = source.read(chunk_size) destination.write(chunk) br += len(chunk) - if len(chunk) < chunk_size: + if not chunk: break # END reading output stream return br diff --git a/test/test_util.py b/test/test_util.py index acbff17b7..0ada556a8 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -6,6 +6,7 @@ import contextlib import errno from datetime import datetime +from io import BytesIO, StringIO import os import pathlib import pickle @@ -40,12 +41,70 @@ is_cygwin_git, remove_password_if_present, rmtree, + stream_copy, ) from test.cleanup import cleanup_directory from test.lib.fixtures import writable_fixture from test.lib import TestBase, requires_symlinks, with_rw_repo +@pytest.mark.parametrize( + "parts, expected", + [([b"ab", b"c", b"def"], b"abcdef"), (["ab", "c", "def"], "abcdef"), ([], b""), ([], "")], +) +def test_stream_copy_waits_for_an_empty_read(parts, expected): + chunks = iter(parts) + + class Reader: + def read(self, size): + chunk = next(chunks, expected[:0]) + assert len(chunk) <= size + return chunk + + destination = BytesIO() if isinstance(expected, bytes) else StringIO() + assert stream_copy(Reader(), destination, chunk_size=4) == len(expected) + assert destination.getvalue() == expected + + +@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX unbuffered pipe reads") +def test_stream_copy_keeps_reading_an_unbuffered_pipe(): + read_fd, write_fd = os.pipe() + first_copied = threading.Event() + errors = [] + + class Destination(BytesIO): + def write(self, data): + result = super().write(data) + if data: + first_copied.set() + return result + + def produce(): + try: + os.write(write_fd, b"first-") + if not first_copied.wait(5): + raise RuntimeError("first pipe chunk was not copied") + os.write(write_fd, b"second") + except Exception as error: + errors.append(error) + finally: + os.close(write_fd) + + destination = Destination() + with os.fdopen(read_fd, "rb", buffering=0) as source: + thread = threading.Thread(target=produce) + thread.start() + try: + copied = stream_copy(source, destination, chunk_size=1024) + finally: + first_copied.set() + thread.join(5) + assert not thread.is_alive() + assert not errors + assert copied == len(b"first-second") + assert destination.getvalue() == b"first-second" + + @pytest.fixture def tmp_path(request): # Unreadable files and symlinks cannot be copied by native fixture tooling.