From d0fb5b65fb56b864179a251b9c78e4afa2268ff7 Mon Sep 17 00:00:00 2001 From: Fuxino Date: Wed, 14 Oct 2026 09:42:37 +0200 Subject: [PATCH] Read rsync output while it runs Read the output as it arrives, split on newlines and carriage returns, instead of all at once with communicate(). This is needed for the progress bar. Invalid UTF-8 in the output is replaced instead of raising. --- simple_backup/simple_backup.py | 31 ++++++++++++++++++++++++------- tests/test_local.py | 30 ++++++++++++++++++++++++++++-- tests/test_remote.py | 6 ++++-- 3 files changed, 56 insertions(+), 11 deletions(-) diff --git a/simple_backup/simple_backup.py b/simple_backup/simple_backup.py index 19e69b2..e73fb98 100755 --- a/simple_backup/simple_backup.py +++ b/simple_backup/simple_backup.py @@ -18,7 +18,9 @@ from __future__ import annotations import sys import os import pwd -from typing import Callable, List, Optional, ParamSpec, TypedDict, TypeVar, Union +import io +import re +from typing import Callable, Iterator, List, Optional, ParamSpec, TypedDict, TypeVar import warnings from functools import wraps from shutil import rmtree, which @@ -647,24 +649,24 @@ class Backup: args.append('--rsync-path=sudo rsync') with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False) as p: - output: Union[bytes, List[str]] - output, _ = p.communicate() + assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase) + p.stdin.close() + + output = [r for r in _read_records(p.stdout) if r != ''] try: del os.environ['SSHPASS'] except KeyError: pass - returncode = p.returncode - - output = output.decode("utf-8").split('\n') + returncode = p.wait() if returncode == 0: if self._verbose: logger.info('rsync: %s', output) else: - logger.info('rsync: %s', output[-3]) logger.info('rsync: %s', output[-2]) + logger.info('rsync: %s', output[-1]) else: self._returncode_log(returncode) @@ -709,6 +711,21 @@ class Backup: return 0 +def _read_records(stream: io.BufferedIOBase) -> Iterator[str]: + """Yield rsync output as it arrives, split on '\\n' and '\\r' (which ends progress updates)""" + + buffer = b'' + + while chunk := stream.read1(65536): + *records, buffer = re.split(rb'[\r\n]', buffer + chunk) + + for r in records: + yield r.decode('utf-8', errors='replace') + + if buffer != b'': + yield buffer.decode('utf-8', errors='replace') + + def _detect_user() -> Optional[str]: if os.geteuid() == 0: return os.getenv('SUDO_USER') or os.getenv('DOAS_USER') diff --git a/tests/test_local.py b/tests/test_local.py index 83a284a..0f2da3c 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -1,5 +1,6 @@ """Tests for local backups, with real rsync runs and with a fake Popen""" +import io import os from datetime import datetime, timedelta from pathlib import Path @@ -195,6 +196,29 @@ class TestRsync: assert backups(out) == ['2026-10-14_11-00-00'] +class ChunkStream(io.BufferedIOBase): + """Returns the given chunks one per read1() call, like a pipe""" + + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + + def read1(self, size: int = -1) -> bytes: + return self.chunks.pop(0) if self.chunks else b'' + + +class TestReadRecords: + def test_split_on_newline_and_cr(self) -> None: + stream = ChunkStream([b'file\n\r 1.00M 50%\r 2.00M 100%\n\nsent 1 bytes\ntotal']) + + assert list(sb._read_records(stream)) == ['file', '', ' 1.00M 50%', ' 2.00M 100%', '', 'sent 1 bytes', + 'total'] + + def test_records_across_chunks(self) -> None: + stream = ChunkStream([b'fi', b'le\r', b'\nna', 'mé\n'.encode(), b'\xff\n']) + + assert list(sb._read_records(stream)) == ['file', '', 'namé', '�'] + + class FakePopen: """Stands in for rsync: creates the destination and returns a chosen return code""" @@ -204,6 +228,8 @@ class FakePopen: def __init__(self, args: list[str], **kwargs: Any) -> None: FakePopen.calls.append(args) os.makedirs(args[args.index('/') + 1]) + self.stdin = io.BytesIO() + self.stdout = io.BytesIO(b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n') def __enter__(self) -> 'FakePopen': return self @@ -211,8 +237,8 @@ class FakePopen: def __exit__(self, *args: Any) -> None: pass - def communicate(self) -> tuple[bytes, None]: - return b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n', None + def wait(self) -> int: + return FakePopen.returncode @pytest.fixture diff --git a/tests/test_remote.py b/tests/test_remote.py index 9e3e7fb..18cc314 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -257,6 +257,8 @@ class RecordPopen: def __init__(self, args: list[str], **kwargs: Any) -> None: RecordPopen.calls.append(args) + self.stdin = io.BytesIO() + self.stdout = io.BytesIO(b'sent 1 bytes received 2 bytes\ntotal size is 3\n') def __enter__(self) -> 'RecordPopen': return self @@ -264,8 +266,8 @@ class RecordPopen: def __exit__(self, *args: Any) -> None: pass - def communicate(self) -> tuple[bytes, None]: - return b'sent 1 bytes received 2 bytes\ntotal size is 3\n', None + def wait(self) -> int: + return RecordPopen.returncode @pytest.mark.usefixtures('connect')