feat(backup): rotation multi-paliers (fonction pure)
Implémente la fonction select_retained() pour la rétention multi-paliers: - Paliers fixes: horaire/12h, 12h/72h, quotidien/21j - Palier mensuel: 12 mois calendaires - Fonction pure, pas de dépendances sur d'autres modules backup Fixes #89 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||||
@@ -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)
|
||||||
Reference in New Issue
Block a user