Add a progress bar for rsync

On a terminal, run rsync with --info=progress2 and draw its updates as a
progress bar on stderr. The ETA is estimated locally, since rsync reports
the elapsed time when a file finishes. Disable it with --no-progress.
This commit is contained in:
Fuxino
2026-10-14 09:46:50 +02:00
parent d0fb5b65fb
commit 7cd90f8d11
4 changed files with 152 additions and 7 deletions
+50 -2
View File
@@ -53,8 +53,9 @@ def out(tmp_path: Path) -> Path:
def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: Optional[list[str]] = None,
remove_before: bool = False) -> sb.Backup:
backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before)
remove_before: bool = False, progress: bool = False) -> sb.Backup:
backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before,
progress=progress)
assert backup.check_params() == 0
@@ -187,6 +188,16 @@ class TestRsync:
assert run_backup(out, [str(src)], keep=0) == 0
assert backups(out) == ['2026-10-14_12-00-00']
def test_progress(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture,
capsys: pytest.CaptureFixture[str]) -> None:
assert run_backup(out, [str(src)], progress=True) == 0
assert '100%' in capsys.readouterr().err
# Only the summary is logged, not the progress updates
assert 'rsync: sent' in caplog.text
assert 'rsync: total size' in caplog.text
assert '%' not in caplog.text
def test_cli(self, src: Path, out: Path, tmp_path: Path, run_cli: Callable[..., int]) -> None:
config_file = str(tmp_path / 'missing.conf')
@@ -219,6 +230,43 @@ class TestReadRecords:
assert list(sb._read_records(stream)) == ['file', '', 'namé', '�']
class TestProgressBar:
@pytest.fixture(autouse=True)
def terminal(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((81, 24)))
# 30 s elapsed at each update
monkeypatch.setattr(sb, 'default_timer', iter([0.0, 30.0]).__next__)
def test_update(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar()
assert bar.update(' 2.00M 25% 1.47MB/s 0:00:02 (xfr#1, ir-chk=2/4)')
assert capsys.readouterr().err == \
'\r[#######-----------------------] 25% 2.00M 1.47MB/s ETA 0:01:30 (scanning)\033[K'
def test_without_human_readable(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar()
assert bar.update(' 4,000,000 50% 2.97MB/s 0:00:01 (xfr#2, to-chk=1/4)')
assert capsys.readouterr().err.endswith('] 50% 4,000,000 2.97MB/s ETA 0:00:30\033[K')
def test_not_progress(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar()
for record in ['home/user/file 50% 1:00:00', 'sent 1 bytes received 2 bytes', 'total size is 3']:
assert not bar.update(record)
bar.clear()
assert capsys.readouterr().err == ''
def test_narrow_terminal(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((20, 24)))
assert sb._ProgressBar().update(' 6.00M 100% 1.48MB/s 0:00:03 (xfr#3, to-chk=0/4)')
# No room for the bar
assert capsys.readouterr().err == '\r100% 6.00M 1.48MB\033[K'
class FakePopen:
"""Stands in for rsync: creates the destination and returns a chosen return code"""