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')