From 3150bc5399537772dcd617fa261f20980d967af8 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Wed, 31 Dec 2025 09:02:34 +0100 Subject: [PATCH] test(e2e): add docker-based end-to-end coverage for --backups-root and --force-keep - Run E2E suite via unittest discovery inside the container - Add E2E test for --id mode with real filesystem + fake dirval - Add E2E test for --all + --force-keep to ensure latest backups are skipped https://chatgpt.com/share/6954d89e-bf08-800f-be4a-5d237d190ddd --- README.md | 1 + src/cleanback/__main__.py | 27 ++++- tests/e2e/Dockerfile.e2e | 2 +- tests/e2e/__init__.py | 0 tests/e2e/test_e2e_force_keep.py | 163 +++++++++++++++++++++++++++++++ tests/unit/__init__.py | 0 tests/unit/test_main.py | 30 ++++++ 7 files changed, 219 insertions(+), 4 deletions(-) create mode 100644 tests/e2e/__init__.py create mode 100644 tests/e2e/test_e2e_force_keep.py create mode 100644 tests/unit/__init__.py diff --git a/README.md b/README.md index f183f6a..5b2a3a8 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ Scans: | `--workers ` | Parallel workers (default: CPU count, min 2) | | `--timeout ` | Per-directory validation timeout (float supported, default: 300.0) | | `--yes` | Non-interactive mode: delete failures automatically | +| `--force-keep ` | In `--all` mode: skip the last *n* backup folders (default: 0) | --- diff --git a/src/cleanback/__main__.py b/src/cleanback/__main__.py index f73a565..0f8b67f 100755 --- a/src/cleanback/__main__.py +++ b/src/cleanback/__main__.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """ Cleanup Failed Docker Backups — parallel validator (using dirval) +with optional "keep last N backups" behavior in --all mode. Validates backup subdirectories under: - //backup-docker-to-local (when --id is used) @@ -40,7 +41,7 @@ class ValidationResult: def discover_target_subdirs( - backups_root: Path, backup_id: Optional[str], all_mode: bool + backups_root: Path, backup_id: Optional[str], all_mode: bool, force_keep: int ) -> List[Path]: """ Return a list of subdirectories to validate: @@ -48,12 +49,23 @@ def discover_target_subdirs( - If --all: for each /* that has backup-docker-to-local, include its subdirs """ targets: List[Path] = [] + if force_keep < 0: + raise ValueError("--force-keep must be >= 0") if not backups_root.is_dir(): raise FileNotFoundError(f"Backups root does not exist: {backups_root}") if all_mode: - for backup_folder in sorted(p for p in backups_root.iterdir() if p.is_dir()): + backup_folders = sorted(p for p in backups_root.iterdir() if p.is_dir()) + + # Skip the last N backup folders (by sorted name order). + # This is intentionally simple: timestamp-like folder names sort correctly. + if force_keep: + if len(backup_folders) <= force_keep: + return [] + backup_folders = backup_folders[:-force_keep] + + for backup_folder in backup_folders: candidate = backup_folder / "backup-docker-to-local" if candidate.is_dir(): targets.extend(sorted([p for p in candidate.iterdir() if p.is_dir()])) @@ -241,6 +253,12 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: action="store_true", help="Do not prompt; delete failing directories automatically.", ) + parser.add_argument( + "--force-keep", + type=int, + default=0, + help="In --all mode: keep (skip) the last N backup folders under --backups-root (default: 0).", + ) return parser.parse_args(argv) @@ -249,7 +267,10 @@ def main(argv: Optional[List[str]] = None) -> int: try: subdirs = discover_target_subdirs( - args.backups_root, args.backup_id, bool(args.all_mode) + args.backups_root, + args.backup_id, + bool(args.all_mode), + int(args.force_keep), ) except Exception as e: print(f"ERROR: {e}", file=sys.stderr) diff --git a/tests/e2e/Dockerfile.e2e b/tests/e2e/Dockerfile.e2e index 079d65d..c04e3c0 100644 --- a/tests/e2e/Dockerfile.e2e +++ b/tests/e2e/Dockerfile.e2e @@ -14,4 +14,4 @@ RUN python -m pip install -U pip \ RUN mkdir -p /Backups # Run E2E unittest -CMD ["python", "-m", "unittest", "-v", "tests.e2e.test_e2e_docker"] +CMD ["python", "-m", "unittest", "discover", "-v", "-s", "tests/e2e", "-p", "test_*.py"] diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/test_e2e_force_keep.py b/tests/e2e/test_e2e_force_keep.py new file mode 100644 index 0000000..814ebc6 --- /dev/null +++ b/tests/e2e/test_e2e_force_keep.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +import os +import subprocess +import tempfile +import unittest +from pathlib import Path + + +FAKE_TIMEOUT_SLEEP = 0.3 +SHORT_TIMEOUT = "0.1" + +FAKE_DIRVAL = f"""#!/usr/bin/env python3 +import sys, time, argparse, pathlib + +def main(): + p = argparse.ArgumentParser() + p.add_argument("path") + p.add_argument("--validate", action="store_true") + args = p.parse_args() + + d = pathlib.Path(args.path) + name = d.name.lower() + + if "timeout" in name: + time.sleep({FAKE_TIMEOUT_SLEEP}) + print("Simulated long run...") + return 0 + + if (d / "VALID").exists(): + print("ok") + return 0 + + print("failed") + return 1 + +if __name__ == "__main__": + sys.exit(main()) +""" + + +class CleanbackE2EForceKeepTests(unittest.TestCase): + """ + E2E test that validates --force-keep in --all mode. + It creates two backup folders directly under /Backups so --all can find them: + /Backups/-01/backup-docker-to-local/{good,bad} + /Backups/-02/backup-docker-to-local/{good,bad} + With --force-keep 1, the last (sorted) backup folder (-02) is skipped. + """ + + def setUp(self): + self.backups_root = Path("/Backups") + self.backups_root.mkdir(parents=True, exist_ok=True) + + # Unique prefix to avoid collisions across runs + self.prefix = f"E2EKEEP-{os.getpid()}" + + # Create fake `dirval` executable on disk (real file, real chmod) + self.bin_dir = Path(tempfile.mkdtemp(prefix="cleanback-bin-")) + self.dirval = self.bin_dir / "dirval" + self.dirval.write_text(FAKE_DIRVAL, encoding="utf-8") + self.dirval.chmod(0o755) + + # Two backup folders directly under /Backups (so --all can discover them) + self.b1 = self.backups_root / f"{self.prefix}-01" / "backup-docker-to-local" + self.b2 = self.backups_root / f"{self.prefix}-02" / "backup-docker-to-local" + self.b1.mkdir(parents=True, exist_ok=True) + self.b2.mkdir(parents=True, exist_ok=True) + + # Within each: good + bad + self.b1_good = self.b1 / "good" + self.b1_bad = self.b1 / "bad" + self.b2_good = self.b2 / "good" + self.b2_bad = self.b2 / "bad" + + for p in (self.b1_good, self.b1_bad, self.b2_good, self.b2_bad): + p.mkdir(parents=True, exist_ok=True) + + # Mark goods as valid + (self.b1_good / "VALID").write_text("1", encoding="utf-8") + (self.b2_good / "VALID").write_text("1", encoding="utf-8") + + # Convenience for teardown + self.created_roots = [ + self.backups_root / f"{self.prefix}-01", + self.backups_root / f"{self.prefix}-02", + ] + + def tearDown(self): + # Cleanup created backup folders + for root in self.created_roots: + try: + if root.exists(): + for p in sorted(root.rglob("*"), reverse=True): + try: + if p.is_dir(): + p.rmdir() + else: + p.unlink() + except Exception: + pass + try: + root.rmdir() + except Exception: + pass + except Exception: + pass + + # Cleanup temp bin dir + try: + if self.bin_dir.exists(): + for p in sorted(self.bin_dir.rglob("*"), reverse=True): + try: + if p.is_dir(): + p.rmdir() + else: + p.unlink() + except Exception: + pass + try: + self.bin_dir.rmdir() + except Exception: + pass + except Exception: + pass + + def test_all_mode_force_keep_skips_last_backup_folder(self): + env = os.environ.copy() + env["PATH"] = f"{self.bin_dir}:{env.get('PATH', '')}" + + cmd = [ + "python", + "-m", + "cleanback", + "--backups-root", + "/Backups", + "--all", + "--force-keep", + "1", + "--dirval-cmd", + "dirval", + "--workers", + "4", + "--timeout", + SHORT_TIMEOUT, + "--yes", + ] + proc = subprocess.run(cmd, text=True, capture_output=True, env=env) + + self.assertEqual(proc.returncode, 0, msg=proc.stderr or proc.stdout) + + # First backup folder (-01) should be processed: bad removed, good kept + self.assertTrue(self.b1_good.exists(), "b1 good should remain") + self.assertFalse(self.b1_bad.exists(), "b1 bad should be deleted") + + # Last backup folder (-02) should be skipped entirely: both remain + self.assertTrue(self.b2_good.exists(), "b2 good should remain (skipped)") + self.assertTrue(self.b2_bad.exists(), "b2 bad should remain (skipped)") + + self.assertIn("Summary:", proc.stdout) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 640a6c9..9c6f2d3 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -153,6 +153,36 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase): self.assertFalse(self.timeoutC.exists()) self.assertTrue(self.goodX.exists()) self.assertFalse(self.badY.exists()) + + def test_all_mode_force_keep_skips_last_backup_folder(self): + # Given backup folders: ID1, ID2 (sorted) + # --force-keep 1 should skip ID2 completely. + rc, out, err, _ = self.run_main( + [ + "--backups-root", + str(self.backups_root), + "--all", + "--force-keep", + "1", + "--dirval-cmd", + str(self.dirval), + "--workers", + "4", + "--timeout", + SHORT_TIMEOUT, + "--yes", + ] + ) + self.assertEqual(rc, 0, msg=err or out) + + # ID1 should be processed + self.assertTrue(self.goodA.exists()) + self.assertFalse(self.badB.exists()) + self.assertFalse(self.timeoutC.exists()) + + # ID2 should be untouched + self.assertTrue(self.goodX.exists()) + self.assertTrue(self.badY.exists()) def test_dirval_missing_errors(self): rc, out, err, _ = self.run_main(