backupmanager_ynh/sources/tests/test_retention.py

"""Tests unitaires pour les algorithmes de rétention (fonctions pures, sans Flask)."""
from datetime import datetime, timedelta


def days_ago(n):
    return datetime.utcnow() - timedelta(days=n)


# ---------------------------------------------------------------------------
# _retention_count
# ---------------------------------------------------------------------------

class TestRetentionCount:
    def test_supprime_les_plus_anciens(self):
        from retention import _retention_count
        archives = [
            "a_20260101.tar", "a_20260102.tar", "a_20260103.tar",
            "a_20260104.tar", "a_20260105.tar",
        ]
        assert _retention_count(archives, 3) == ["a_20260101.tar", "a_20260102.tar"]

    def test_moins_que_n_garde_tout(self):
        from retention import _retention_count
        archives = ["a_20260101.tar", "a_20260102.tar"]
        assert _retention_count(archives, 5) == []

    def test_exactement_n_garde_tout(self):
        from retention import _retention_count
        archives = ["a_20260101.tar", "a_20260102.tar", "a_20260103.tar"]
        assert _retention_count(archives, 3) == []

    def test_liste_vide(self):
        from retention import _retention_count
        assert _retention_count([], 5) == []

    def test_keep_1_supprime_tous_sauf_dernier(self):
        from retention import _retention_count
        archives = ["a_20260101.tar", "a_20260102.tar", "a_20260103.tar"]
        assert _retention_count(archives, 1) == ["a_20260101.tar", "a_20260102.tar"]


# ---------------------------------------------------------------------------
# _retention_daily
# ---------------------------------------------------------------------------

class TestRetentionDaily:
    def test_supprime_archives_trop_anciennes(self):
        from retention import _retention_daily
        archives = [
            f"a_{days_ago(35).strftime('%Y%m%d')}.tar",
            f"a_{days_ago(31).strftime('%Y%m%d')}.tar",
            f"a_{days_ago(10).strftime('%Y%m%d')}.tar",
            f"a_{days_ago(1).strftime('%Y%m%d')}.tar",
        ]
        result = _retention_daily(archives, 30)
        assert archives[0] in result
        assert archives[1] in result
        assert archives[2] not in result
        assert archives[3] not in result

    def test_deduplique_meme_jour(self):
        from retention import _retention_daily
        today = days_ago(0).strftime("%Y%m%d")
        a1 = f"a_{today}.tar"
        a2 = f"a_{today}_2.tar"
        archives = sorted([a1, a2])
        result = _retention_daily(archives, 30)
        assert len(result) == 1

    def test_garde_une_par_jour_dans_fenetre(self):
        from retention import _retention_daily
        archives = [f"a_{days_ago(i).strftime('%Y%m%d')}.tar" for i in range(9, -1, -1)]
        assert _retention_daily(archives, 30) == []

    def test_liste_vide(self):
        from retention import _retention_daily
        assert _retention_daily([], 30) == []

    def test_toutes_hors_fenetre(self):
        from retention import _retention_daily
        archives = [
            f"a_{days_ago(60).strftime('%Y%m%d')}.tar",
            f"a_{days_ago(45).strftime('%Y%m%d')}.tar",
        ]
        result = _retention_daily(archives, 30)
        assert set(result) == set(archives)


# ---------------------------------------------------------------------------
# _retention_gfs
# ---------------------------------------------------------------------------

class TestRetentionGFS:
    def test_liste_vide(self):
        from retention import _retention_gfs
        assert _retention_gfs([], {"daily": 7, "weekly": 4, "monthly": 12}) == []

    def test_tous_dans_fenetre_daily_gardes(self):
        from retention import _retention_gfs
        archives = [f"a_2026050{i}.tar" for i in range(1, 8)]  # 7 archives
        assert _retention_gfs(archives, {"daily": 7, "weekly": 2, "monthly": 2}) == []

    def test_daily_supprime_au_dela_de_n(self):
        from retention import _retention_gfs
        # 15 archives consécutives (mai 01-15), daily=7
        # Le mensuel et hebdo tombent dans la fenêtre daily pour cet exemple
        archives = [f"a_202605{i:02d}.tar" for i in range(1, 16)]
        to_delete = _retention_gfs(archives, {"daily": 7, "weekly": 1, "monthly": 1})
        # Les 7 plus récentes (mai 9-15) sont gardées par daily
        for i in range(9, 16):
            assert f"a_202605{i:02d}.tar" not in to_delete

    def test_monthly_sauve_archive_ancienne(self):
        from retention import _retention_gfs
        # Archives sur 5 mois distincts
        archives = [
            "a_20260115.tar",  # janvier (plus vieux)
            "a_20260215.tar",  # février
            "a_20260315.tar",  # mars
            "a_20260415.tar",  # avril
            "a_20260510.tar",  # mai
            "a_20260511.tar",  # mai (plus récent)
        ]
        # daily=2 garde mai11+mai10, weekly=1 garde semaine de mai11,
        # monthly=3 garde mai→mai11, avr→avr15, mars→mars15
        # Supprimés : jan15, fév15
        to_delete = _retention_gfs(archives, {"daily": 2, "weekly": 1, "monthly": 3})
        assert "a_20260115.tar" in to_delete
        assert "a_20260215.tar" in to_delete
        assert "a_20260315.tar" not in to_delete
        assert "a_20260415.tar" not in to_delete
        assert "a_20260510.tar" not in to_delete
        assert "a_20260511.tar" not in to_delete
        assert len(to_delete) == 2

    def test_weekly_un_seul_representant_par_semaine(self):
        from retention import _retention_gfs
        # Semaine 19 (mai 04-10) + semaine 20 (mai 11) — vérifiées ISO
        # 2026-05-04 est un lundi (début de semaine 19)
        # 2026-05-11 est un lundi (début de semaine 20)
        archives = [
            "a_20260504.tar", "a_20260505.tar", "a_20260506.tar",
            "a_20260507.tar", "a_20260508.tar", "a_20260509.tar",
            "a_20260510.tar", "a_20260511.tar",
        ]
        # daily=3 : mai11, mai10, mai09
        # weekly=2 : sem20→mai11, sem19→mai10 (déjà en daily)
        # monthly=1 : mai→mai11
        # Keepers : {mai09, mai10, mai11}
        # Supprimés : {mai04, mai05, mai06, mai07, mai08}
        to_delete = _retention_gfs(archives, {"daily": 3, "weekly": 2, "monthly": 1})
        assert "a_20260511.tar" not in to_delete
        assert "a_20260510.tar" not in to_delete
        assert "a_20260509.tar" not in to_delete
        assert "a_20260504.tar" in to_delete
        assert "a_20260508.tar" in to_delete
        assert len(to_delete) == 5

    def test_une_archive_peut_couvrir_plusieurs_niveaux(self):
        from retention import _retention_gfs
        # Une seule archive : elle doit être conservée même avec daily=1, weekly=1, monthly=1
        archives = ["a_20260511.tar"]
        assert _retention_gfs(archives, {"daily": 1, "weekly": 1, "monthly": 1}) == []

    def test_defaults_utilises_si_config_vide(self):
        from retention import _retention_gfs
        # Config vide → defaults daily=7, weekly=4, monthly=12
        # Avec 3 archives, tout doit être gardé
        archives = ["a_20260509.tar", "a_20260510.tar", "a_20260511.tar"]
        assert _retention_gfs(archives, {}) == []