backupmanager_ynh/sources/tests/test_keep_local_archive.py

"""Tests pour l'option 'conserver l'archive en local' (job.keep_local_archive)."""
import pytest


def _make_job(app, keep_local_archive, with_destination):
    from db import db, Job, JobDestination, Destination

    dest = None
    if with_destination:
        dest = Destination(name="d1", host="h", port=22, user="u",
                            remote_path="/backups", key_name="k", enabled=True)
        db.session.add(dest)
        db.session.flush()

    job = Job(name="j1", type="custom_dir",
              config_json='{"source_path": "/opt/app", "excludes": [], "restore": {}}',
              cron_expr="", retention_mode="count", retention_value=5,
              enabled=True, keep_local_archive=keep_local_archive)
    db.session.add(job)
    db.session.flush()
    if dest:
        db.session.add(JobDestination(job_id=job.id, dest_type="ssh", dest_id=dest.id))
    db.session.commit()
    return job.id


@pytest.fixture(autouse=True)
def _stub_side_effects(monkeypatch):
    """Neutralise tout ce qui touche au disque/réseau réel."""
    import jobs.custom_dir
    import jobs.utils
    import retention
    import notifications

    monkeypatch.setattr(jobs.custom_dir, "backup_custom_dir",
                         lambda job, instance, backup_dir: ("archive_20260101", "log", False))
    monkeypatch.setattr(jobs.utils, "sudo_getsize", lambda path: 123)
    monkeypatch.setattr(retention, "apply_retention", lambda job, name, backup_dir: ([], []))
    monkeypatch.setattr(retention, "apply_ssh_retention", lambda job, dest, data_dir: [])
    monkeypatch.setattr(notifications, "send_job_notification", lambda run, job: None)


class TestKeepLocalArchive:
    def test_conserve_si_aucune_destination(self, app, monkeypatch):
        import jobs.utils
        called = []
        monkeypatch.setattr(jobs.utils, "sudo_rm_archive", lambda *a, **k: called.append(a))

        with app.app_context():
            from jobs.ynh_backup import execute_job
            from db import db, Run
            job_id = _make_job(app, keep_local_archive=False, with_destination=False)
            execute_job(job_id)
            run = Run.query.filter_by(job_id=job_id).first()
            assert not called
            assert "aucune destination configurée" in run.log_text

    def test_supprime_si_transfert_reussi(self, app, monkeypatch):
        import jobs.utils
        import jobs.transfer
        called = []
        monkeypatch.setattr(jobs.transfer, "transfer_archive", lambda *a, **k: "ok")
        monkeypatch.setattr(jobs.utils, "sudo_rm_archive", lambda *a, **k: called.append(a))

        with app.app_context():
            from jobs.ynh_backup import execute_job
            from db import db, Run
            job_id = _make_job(app, keep_local_archive=False, with_destination=True)
            execute_job(job_id)
            run = Run.query.filter_by(job_id=job_id).first()
            assert called == [("archive_20260101", app.config["YUNOHOST_BACKUP_DIR"])]
            assert "Archive locale supprimée" in run.log_text

    def test_conserve_si_transfert_echoue(self, app, monkeypatch):
        import jobs.utils
        import jobs.transfer
        called = []
        monkeypatch.setattr(jobs.transfer, "transfer_archive",
                             lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")))
        monkeypatch.setattr(jobs.utils, "sudo_rm_archive", lambda *a, **k: called.append(a))

        with app.app_context():
            from jobs.ynh_backup import execute_job
            from db import db, Run
            job_id = _make_job(app, keep_local_archive=False, with_destination=True)
            execute_job(job_id)
            run = Run.query.filter_by(job_id=job_id).first()
            assert not called
            assert "au moins un transfert a échoué" in run.log_text

    def test_conserve_par_defaut(self, app, monkeypatch):
        import jobs.utils
        import jobs.transfer
        called = []
        monkeypatch.setattr(jobs.transfer, "transfer_archive", lambda *a, **k: "ok")
        monkeypatch.setattr(jobs.utils, "sudo_rm_archive", lambda *a, **k: called.append(a))

        with app.app_context():
            from jobs.ynh_backup import execute_job
            job_id = _make_job(app, keep_local_archive=True, with_destination=True)
            execute_job(job_id)
            assert not called