import os import pytest from machine_backup.logs import CommandResult from machine_backup.modules import postgres as pg_module from machine_backup.modules.base import ActionState, ModuleStatus from machine_backup.modules.postgres import PostgresModule from conftest import make_runner DB_LIST = "advmfr_hr\nelevaspeed\npostgres\n" @pytest.fixture def fake_pg_bins(monkeypatch): monkeypatch.setattr(pg_module, "_find_pg_bin", lambda tool: f"/fake/bin/{tool}") def _module(responses=None, **kw): runner = make_runner(responses) return PostgresModule(runner=runner, **kw), runner def test_check_reports_missing_binaries(monkeypatch): monkeypatch.setattr(pg_module, "_find_pg_bin", lambda tool: None) problems = PostgresModule().check() assert any("pg_dump" in p.message for p in problems) assert all(p.fatal for p in problems) def test_check_reports_unreachable_server(fake_pg_bins): mod, _ = _module( {"psql": CommandResult(returncode=2, stderr="connection refused")} ) problems = mod.check() assert any("connection refused" in p.message for p in problems) assert all(p.fatal for p in problems) def test_check_passes_when_everything_is_up(fake_pg_bins): mod, _ = _module({"psql": CommandResult(returncode=0, stdout="1\n")}) assert mod.check() == [] def _dumped_databases(runner) -> list[str]: """Bases effectivement passees a pg_dump (dernier argument de la commande).""" return [ c[-1] for c in runner.calls if os.path.basename(c[0]) == "pg_dump" ] def test_collect_excludes_templates_and_dumps_globals(tmp_path, fake_pg_bins): mod, runner = _module({"psql": CommandResult(returncode=0, stdout=DB_LIST)}) result = mod.collect(tmp_path / "postgres") assert any( os.path.basename(c[0]) == "pg_dumpall" and "--globals-only" in c for c in runner.calls ) assert _dumped_databases(runner) == ["advmfr_hr", "elevaspeed", "postgres"] assert all("-Fc" in c for c in runner.calls if os.path.basename(c[0]) == "pg_dump") assert result.status is ModuleStatus.OK def test_collect_marks_every_artifact_sensitive(tmp_path, fake_pg_bins): mod, _ = _module({"psql": CommandResult(returncode=0, stdout=DB_LIST)}) dest = tmp_path / "postgres" # Le runner factice n'ecrit rien : on simule les fichiers produits. dest.mkdir(parents=True) (dest / "globals.sql").write_text("CREATE ROLE elevaspeed;") (dest / "advmfr_hr.dump").write_bytes(b"dump") result = mod.collect(dest) assert result.artifacts assert all(a.sensitive for a in result.artifacts) def test_collect_reports_progress(tmp_path, fake_pg_bins, recording_progress): mod, _ = _module({"psql": CommandResult(returncode=0, stdout=DB_LIST)}) mod.collect(tmp_path / "postgres", progress=recording_progress) # Trois bases + les globals. assert recording_progress.total == 4 assert "advmfr_hr" in recording_progress.steps def test_collect_continues_after_one_failed_dump(tmp_path, fake_pg_bins): responses = { "psql": CommandResult(returncode=0, stdout=DB_LIST), "pg_dump": CommandResult(returncode=1, stderr="permission denied"), } mod, runner = _module(responses) result = mod.collect(tmp_path / "postgres") assert result.status is ModuleStatus.FAILED assert "advmfr_hr" in (result.error or "") # Les autres bases ont quand meme ete tentees. assert _dumped_databases(runner) == ["advmfr_hr", "elevaspeed", "postgres"] def test_plan_restore_classification(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "globals.sql").write_text("CREATE ROLE elevaspeed;") (src / "advmfr_hr.dump").write_bytes(b"d") (src / "nouvelle.dump").write_bytes(b"d") mod, _ = _module({"psql": CommandResult(returncode=0, stdout="advmfr_hr\npostgres\n")}) actions = mod.plan_restore(src) by_target = {a.target: a for a in actions} assert by_target["__globals__"].state is ActionState.ADDITIVE assert by_target["advmfr_hr"].state is ActionState.DESTRUCTIVE assert "existe" in by_target["advmfr_hr"].reason assert by_target["nouvelle"].state is ActionState.ADDITIVE def test_plan_restore_without_globals(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "solo.dump").write_bytes(b"d") mod, _ = _module({"psql": CommandResult(returncode=0, stdout="postgres\n")}) actions = mod.plan_restore(src) assert [a.target for a in actions] == ["solo"] def test_plan_restore_raises_when_server_unreachable(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "x.dump").write_bytes(b"d") mod, _ = _module({"psql": CommandResult(returncode=2, stderr="down")}) with pytest.raises(RuntimeError, match="serveur"): mod.plan_restore(src) def test_restore_skips_destructive_without_force(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "advmfr_hr.dump").write_bytes(b"d") mod, runner = _module({"psql": CommandResult(returncode=0, stdout="advmfr_hr\n")}) actions = mod.plan_restore(src) mod.restore(src, actions, force=False) joined = [" ".join(c) for c in runner.calls] assert not any("dropdb" in c for c in joined) assert not any("pg_restore" in c for c in joined) def test_restore_force_drops_recreates_then_restores(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "advmfr_hr.dump").write_bytes(b"d") mod, runner = _module({"psql": CommandResult(returncode=0, stdout="advmfr_hr\n")}) actions = mod.plan_restore(src) mod.restore(src, actions, force=True) order = [os.path.basename(c[0]) for c in runner.calls] assert order.index("dropdb") < order.index("createdb") assert order.index("createdb") < order.index("pg_restore") def test_restore_additive_creates_then_restores(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "nouvelle.dump").write_bytes(b"d") mod, runner = _module({"psql": CommandResult(returncode=0, stdout="postgres\n")}) actions = mod.plan_restore(src) mod.restore(src, actions, force=False) order = [os.path.basename(c[0]) for c in runner.calls] assert "dropdb" not in order assert order.index("createdb") < order.index("pg_restore") def test_restore_replays_globals_first(tmp_path, fake_pg_bins): src = tmp_path / "postgres" src.mkdir() (src / "globals.sql").write_text("CREATE ROLE x;") (src / "nouvelle.dump").write_bytes(b"d") mod, runner = _module({"psql": CommandResult(returncode=0, stdout="postgres\n")}) actions = mod.plan_restore(src) mod.restore(src, actions, force=False) names = [os.path.basename(c[0]) for c in runner.calls] # psql (globals) doit preceder createdb. assert names.index("psql") < names.index("createdb") # --- integration : serveur PostgreSQL local reel ------------------------- def _server_available() -> bool: from machine_backup.logs import run_logged psql = pg_module._find_pg_bin("psql") if psql is None: return False try: return run_logged([psql, "-X", "-At", "-c", "SELECT 1", "postgres"]).returncode == 0 except OSError: return False @pytest.mark.skipif(not _server_available(), reason="pas de serveur PostgreSQL local") def test_integration_dumps_a_disposable_database(tmp_path): """Teste les primitives sur une base jetable. On evite volontairement `collect()` complet : il dumperait toutes les bases reelles de la machine, donc des donnees clients, dans un repertoire de test. """ from machine_backup.logs import run_logged name = f"mb_test_{os.getpid()}" createdb = pg_module._find_pg_bin("createdb") dropdb = pg_module._find_pg_bin("dropdb") run_logged([createdb, name], check=True) try: mod = PostgresModule() assert name in mod._list_databases() dest = tmp_path / "postgres" dest.mkdir() mod._dump_globals(dest) mod._dump_database(name, dest) assert (dest / "globals.sql").stat().st_size > 0 assert (dest / f"{name}.dump").stat().st_size > 0 finally: run_logged([dropdb, "--if-exists", name])