diff --git a/src/backup/rotation.py b/src/backup/rotation.py new file mode 100644 index 0000000..545a56b --- /dev/null +++ b/src/backup/rotation.py @@ -0,0 +1,49 @@ +from collections.abc import Iterable +from datetime import datetime, timedelta + +# (période, horizon) pour les paliers à période fixe +_FIXED_TIERS = ( + (timedelta(hours=1), timedelta(hours=12)), # horaire / 12 h + (timedelta(hours=12), timedelta(hours=72)), # 12 h / 72 h + (timedelta(days=1), timedelta(days=21)), # quotidien / 21 j +) +_MONTHLY_HORIZON = 12 # mois calendaires + + +def _select_fixed(timestamps, now, period, horizon): + cutoff = now - horizon + buckets: dict[int, datetime] = {} + for t in timestamps: + if t > now or t < cutoff: + continue + idx = int((now - t) // period) + if idx not in buckets or t > buckets[idx]: + buckets[idx] = t + return set(buckets.values()) + + +def _month_index(d: datetime) -> int: + return d.year * 12 + (d.month - 1) + + +def _select_monthly(timestamps, now): + now_idx = _month_index(now) + buckets: dict[int, datetime] = {} + for t in timestamps: + if t > now: + continue + idx = _month_index(t) + if idx > now_idx or now_idx - idx >= _MONTHLY_HORIZON: + continue + if idx not in buckets or t > buckets[idx]: + buckets[idx] = t + return set(buckets.values()) + + +def select_retained(timestamps: Iterable[datetime], now: datetime) -> set[datetime]: + items = list(timestamps) + retained: set[datetime] = set() + for period, horizon in _FIXED_TIERS: + retained |= _select_fixed(items, now, period, horizon) + retained |= _select_monthly(items, now) + return retained diff --git a/tests/backup/test_rotation.py b/tests/backup/test_rotation.py new file mode 100644 index 0000000..849be23 --- /dev/null +++ b/tests/backup/test_rotation.py @@ -0,0 +1,52 @@ +from datetime import datetime, timedelta, timezone + +from src.backup.rotation import select_retained + +NOW = datetime(2026, 6, 24, 12, 0, 0, tzinfo=timezone.utc) + + +def test_most_recent_always_kept(): + ts = [NOW, NOW - timedelta(hours=2)] + assert NOW in select_retained(ts, NOW) + + +def test_recent_hours_kept_by_hourly_tier(): + t = NOW - timedelta(hours=3) + assert t in select_retained([t], NOW) + + +def test_two_backups_same_hour_keep_newest(): + older = NOW - timedelta(hours=2, minutes=50) + newer = NOW - timedelta(hours=2, minutes=10) + retained = select_retained([older, newer], NOW) + assert newer in retained + assert older not in retained + + +def test_daily_tier_keeps_old_daily_backup(): + t = NOW - timedelta(days=10) + assert t in select_retained([t], NOW) + + +def test_monthly_tier_keeps_one_per_calendar_month(): + older = NOW - timedelta(days=40) # mois M-2 ou M-1 selon calendrier + same_month_newer = older + timedelta(days=2) + retained = select_retained([older, same_month_newer], NOW) + # même mois calendaire -> seul le plus récent du mois est conservé + assert same_month_newer in retained + assert older not in retained + + +def test_keeps_backup_within_12_months(): + t = NOW - timedelta(days=300) + assert t in select_retained([t], NOW) + + +def test_drops_backup_older_than_12_months(): + t = NOW - timedelta(days=400) + assert t not in select_retained([t], NOW) + + +def test_future_timestamps_ignored(): + t = NOW + timedelta(hours=1) + assert t not in select_retained([t], NOW)