snegov/cura-te-ipsum
1
0 Fork 0
Code Issues 20 Releases Wiki Activity

Correct retention before enabling automatic deletion (#8)

Browse Source 
* Correct retention before enabling automatic deletion

- Compute each retention tier's kept set independently from its own
  name range and union them, instead of carrying a single 'previous
  backup' comparison across tier boundaries - that carry-over was the
  actual cause of bug #32 (a coarser tier could evict a backup a finer
  tier, or the always-keep-latest rule, had already decided to retain)
- Group weekly retention by (ISO year, ISO week) instead of ISO week
  number alone, fixing bug #33 (same week number in different years,
  or late-December dates that belong to next year's ISO week 1, were
  treated as duplicates/collisions)
- Validate the computed retention plan (latest snapshot present, no
  phantom entries) before deleting anything
- Fix cli.py to run initiate_backup before cleanup_old_backups, and
  skip cleanup entirely if the backup raised - retention must only run
  once a new snapshot is durable and complete, and a failed backup must
  never trigger deletion of history
- Quarantine (rename) a deletion candidate before recursively removing
  it, reusing the already-hardened stale-staging-dir cleanup to finish
  the job if the process crashes mid-delete
- Remove both retention xfail markers (bugs #32, #33) now that the
  tests they guarded pass; update pre-existing tests whose expectations
  encoded the old bug (deleting the latest snapshot when it collided
  with the next-newest one in a coarser bucket)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address Copilot review: replace asserts with runtime checks, don't swallow quarantine-delete failures

- Retention-plan validation now raises BackupFailedError instead of
  using assert, which python -O would silently strip ahead of a
  destructive operation
- _quarantine_and_delete no longer passes ignore_errors=True to
  rmtree; a removal failure is now logged with the quarantine path so
  operators can find and retry it, instead of silently leaving a
  hidden directory consuming space

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Maxim Shulga <maxim.shulga@alludo.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
snegovClaude Sonnet 5Maxim Shulga
 authored and GitHub committed 2026-08-03 22:47:43 +00:00
1 parent 72a8fe2b59
commit fd613edad0
5 files changed
+209 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_backups.py
+55 -15
View File
@@ -105,7 +105,7 @@ class TestBackupCleanup:
"""Test threshold for keeping weekly backups"""
mock_datetime.now.return_value = datetime(2021, 11, 11)
backups = [
add_backup("20211111_0300"), # remove, not first weekly (Thu)
add_backup("20211111_0300"), # keep, latest snapshot always kept
add_backup("20211110_0300"), # remove, not first weekly (Wed)
add_backup("20211108_0100"), # keep, first weekly 2021-11-08 (Mon)
add_backup("20211107_2300"), # remove, not first weekly (Sun)
@@ -122,8 +122,8 @@ class TestBackupCleanup:
add_backup("20211002_0100"), # remove, older than 5 weeks
]
expected_backups = [backups[2], backups[4], backups[6],
backups[9], backups[12]]
expected_backups = [backups[0], backups[2], backups[4],
backups[6], backups[9], backups[12]]
run_cleanup(keep_weekly=5)
check_backups(expected_backups)
@@ -133,12 +133,12 @@ class TestBackupCleanup:
"""Test threshold for keeping weekly backups"""
mock_datetime.now.return_value = datetime(2021, 11, 11)
backups = [
add_backup("20211111_0300"), # remove, not first weekly (Thu)
add_backup("20211111_0300"), # keep, latest snapshot always kept
add_backup("20211110_0300"), # keep, first weekly (Wed)
add_backup("20211107_0100"), # remove, not first weekly (Sun)
add_backup("20211102_0100"), # keep, first weekly (Tue)
]
expected_backups = [backups[1], backups[3]]
expected_backups = [backups[0], backups[1], backups[3]]
run_cleanup(keep_weekly=5)
check_backups(expected_backups)
@@ -169,7 +169,7 @@ class TestBackupCleanup:
"""Test threshold for keeping yearly backups"""
mock_datetime.now.return_value = datetime(2021, 11, 11)
backups = [
add_backup("20211103_0300"), # remove, not first yearly in 2021
add_backup("20211103_0300"), # keep, latest snapshot always kept
add_backup("20210810_0000"), # remove, not first yearly in 2021
add_backup("20210716_0100"), # keep, first yearly in 2021
add_backup("20201216_0100"), # remove, not first yearly in 2020
@@ -181,7 +181,8 @@ class TestBackupCleanup:
add_backup("20171116_0100"), # remove, older than 3 years
add_backup("20171115_0100"), # remove, older than 3 years
]
expected_backups = [backups[2], backups[4], backups[6], backups[8]]
expected_backups = [backups[0], backups[2], backups[4], backups[6],
backups[8]]
run_cleanup(keep_yearly=3)
check_backups(expected_backups)
@@ -200,8 +201,6 @@ class TestBackupCleanup:
run_cleanup(keep_all=2, dry_run=True)
check_backups(backups)
@pytest.mark.xfail(reason="Bug #32: coarser retention tiers override finer "
"tiers, violating retention guarantees", strict=True)
@mock.patch(f"{bk.__name__}.datetime", wraps=datetime)
def test_combined_retention_policies(self, mock_datetime, add_backup,
run_cleanup, check_backups):
@@ -333,13 +332,12 @@ class TestBackupCleanup:
add_backup("20201228_1200"), # Mon, ISO 2020-W53
]
# Keep one week: should keep oldest backup from ISO week 53
expected = [backups[5]] # 20201228
# Keep one week: should keep oldest backup from ISO week 53, plus
# the latest snapshot overall (always kept)
expected = [backups[0], backups[5]] # 20210103, 20201228
run_cleanup(keep_weekly=1)
check_backups(expected)
@pytest.mark.xfail(reason="Bug #33: Weekly retention bug: same ISO week "
"number across years treated as duplicates", strict=True)
@mock.patch(f"{bk.__name__}.datetime", wraps=datetime)
def test_iso_week_number_across_years(self, mock_datetime, add_backup,
run_cleanup, check_backups):
@@ -383,8 +381,9 @@ class TestBackupCleanup:
add_backup("20250112_1200"), # Sun, ISO 2025-W02
]
# Keep weekly: oldest from week 1 (Dec 30) and week 2 (Jan 8)
expected = [backups[0], backups[4]]
# Keep weekly: oldest from week 1 (Dec 30) and week 2 (Jan 8), plus
# the latest snapshot overall (always kept)
expected = [backups[0], backups[4], backups[5]]
run_cleanup(keep_weekly=10)
check_backups(expected)
@@ -931,3 +930,44 @@ class TestDeletionBoundarySafety:
run_cleanup(keep_all=0)
assert not os.path.exists(old.path)
class TestRetentionPlan:
"""Tests for the independent-per-tier retention plan computation."""
def test_latest_snapshot_always_retained_even_if_grouped_with_next(
self, backup_dir, add_backup, run_cleanup
):
"""
Regression for the bug where the latest snapshot could be
evicted if it shared a weekly/monthly/yearly bucket with the
next-newest one, because retention carried a single "previous
backup" comparison across tier boundaries.
"""
latest = add_backup("20211111_0300")
add_backup("20211110_0300") # same ISO week as latest
add_backup("20211004_0100")
run_cleanup(keep_weekly=5)
assert os.path.isdir(latest.path)
def test_quarantined_dir_recovered_by_stale_staging_cleanup(
self, backup_dir, add_backup
):
"""
If a crash happens between quarantine (rename) and the recursive
delete, the quarantined directory must still get cleaned up on
the next startup via cleanup_stale_staging_dirs.
"""
add_backup("20220101_0000")
old = add_backup("20200101_0000")
quarantine_path = os.path.join(
str(backup_dir), f"{bk.STAGING_PREFIX}trash-{old.name}"
)
os.rename(old.path, quarantine_path)
bk.cleanup_stale_staging_dirs(str(backup_dir))
assert not os.path.exists(quarantine_path)
tests/test_cli.py
+35 -3
View File
@@ -466,12 +466,44 @@ class TestLockHandling:
assert result == 1
m_release.assert_called_once_with(str(backups_dir))
def test_skips_cleanup_when_backup_fails(self, tmp_path):
"""
A failed backup must prevent every retention deletion: if
initiate_backup raises, cleanup_old_backups must never run.
"""
backups_dir = tmp_path / "backups"
backups_dir.mkdir()
source = tmp_path / "src"
source.mkdir()
from curateipsum import backup
with mock.patch('sys.argv', ['cura-te-ipsum', '-b',
str(backups_dir), str(source)]):
with mock.patch('curateipsum.backup.set_backups_lock',
return_value=True):
with mock.patch(
'curateipsum.backup.cleanup_old_backups') \
as m_cleanup:
with mock.patch(
'curateipsum.backup.initiate_backup',
side_effect=backup.BackupFailedError("boom")):
with mock.patch(
'curateipsum.backup.release_backups_lock'):
cli.main()
m_cleanup.assert_not_called()
class TestBackupExecution:
"""Test backup execution flow."""
def test_calls_cleanup_before_initiate(self, tmp_path):
"""Should call cleanup_old_backups before initiate_backup."""
def test_calls_initiate_before_cleanup(self, tmp_path):
"""
Should call initiate_backup before cleanup_old_backups: retention
deletes old snapshots, so it must only run once the new snapshot
is durable, never before it.
"""
backups_dir = tmp_path / "backups"
backups_dir.mkdir()
source = tmp_path / "src"
@@ -497,7 +529,7 @@ class TestBackupExecution:
'curateipsum.backup.release_backups_lock'):
cli.main()
assert call_order == ['cleanup', 'initiate']
assert call_order == ['initiate', 'cleanup']
def test_passes_correct_arguments_to_initiate_backup(self, tmp_path):
"""Should pass all required arguments to initiate_backup."""