From e313258e51b045845e7653683961c568f195757f Mon Sep 17 00:00:00 2001 From: Gaven Lan Ren Date: Wed, 28 Oct 2026 14:13:04 -0400 Subject: [PATCH 1/2] 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 565e3565d..8e77533fb 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 45c870777..421697bbd 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -13,6 +13,7 @@ import threading import time from datetime import datetime +from io import BytesIO, StringIO from unittest import SkipTest, mock import ddt @@ -41,6 +42,7 @@ join_path, remove_password_if_present, rmtree, + stream_copy, ) from test.cleanup import cleanup_directory from test.lib import TestBase, requires_symlinks, with_rw_repo @@ -67,6 +69,63 @@ def test_join_path_keeps_later_tokens_relative(parts, expected): assert join_path(*parts) == expected +@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. From eb613ec3dd5cb0010844ce5b15dc12b6bcf6466f Mon Sep 17 00:00:00 2001 From: Byron Date: Thu, 29 Oct 2026 21:57:55 +0100 Subject: [PATCH 2/2] review - simplify test (remove threads) --- test/test_util.py | 32 +++++++------------------------- 1 file changed, 7 insertions(+), 25 deletions(-) diff --git a/test/test_util.py b/test/test_util.py index 421697bbd..bfde35aea 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -90,38 +90,20 @@ def read(self, size): @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() + if data and not writer.closed: + # Make the second chunk available only after the first short read is copied. + writer.write(b"second") + writer.close() 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 + with os.fdopen(read_fd, "rb", buffering=0) as source, os.fdopen(write_fd, "wb", buffering=0) as writer: + writer.write(b"first-") + copied = stream_copy(source, destination, chunk_size=1024) assert copied == len(b"first-second") assert destination.getvalue() == b"first-second"