Release version 1.2.1

- Change --force-keep semantics to skip the last N subdirectories inside each
  backup-docker-to-local folder
- Remove old behavior that skipped entire backup folders in --all mode
- Update CLI help text to reflect the new behavior
- Align unit and E2E tests with timestamp-based force-keep semantics

https://chatgpt.com/share/695d1ed9-44f0-800f-a236-e903c61036cb

CHANGELOG.md

@@ -1,3 +1,13 @@
+## [1.2.1] - 2026-01-06
+
+* Fixed: --force-keep now applies to timestamp subdirectories inside each backup-docker-to-local folder instead of skipping entire backup folders.
+
+
+## [1.2.0] - 2025-12-31
+
+* Adds a force keep N option to all mode to skip the most recent backups during cleanup, with Docker based E2E tests ensuring the latest backups are preserved.
+
+
 ## [1.1.0] - 2025-12-31
 
 * The backups directory is now configurable via --backups-root instead of being hardcoded to /Backups.

README.md

@@ -100,6 +100,7 @@ Scans:
 | `--workers <n>`      | Parallel workers (default: CPU count, min 2)                       |
 | `--timeout <sec>`    | Per-directory validation timeout (float supported, default: 300.0) |
 | `--yes`              | Non-interactive mode: delete failures automatically                |
+| `--force-keep <n>`   | In `--all` mode: skip the last *n* backup folders (default: 0)      |
 
 ---
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cleanback"
-version = "1.1.0"
+version = "1.2.1"
 description = "Cleanup Failed Docker Backups — parallel validator (using dirval)"
 readme = "README.md"
 requires-python = ">=3.8"

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:
 - <BACKUPS_ROOT>/<ID>/backup-docker-to-local          (when --id is used)
@@ -39,31 +40,57 @@ class ValidationResult:
     stdout: str
 
 
+def _sorted_timestamp_subdirs(path: Path) -> List[Path]:
+    # Timestamp-like folder names sort correctly lexicographically.
+    # We keep it simple: sort by name.
+    return sorted([p for p in path.iterdir() if p.is_dir()], key=lambda p: p.name)
+
+
+def _apply_force_keep(subdirs: List[Path], force_keep: int) -> List[Path]:
+    if force_keep <= 0:
+        return subdirs
+    if len(subdirs) <= force_keep:
+        return []
+    return subdirs[:-force_keep]
+
+
 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:
       - If backup_id is given: <root>/<id>/backup-docker-to-local/* (dirs only)
       - If --all: for each <root>/* that has backup-docker-to-local, include its subdirs
+    force_keep:
+      - Skips the last N timestamp subdirectories inside each backup-docker-to-local folder.
     """
     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()],
+            key=lambda p: p.name,
+        )
+        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()]))
+                subdirs = _sorted_timestamp_subdirs(candidate)
+                subdirs = _apply_force_keep(subdirs, force_keep)
+                targets.extend(subdirs)
     else:
         if not backup_id:
             raise ValueError("Either --id or --all must be provided.")
         base = backups_root / backup_id / "backup-docker-to-local"
         if not base.is_dir():
             raise FileNotFoundError(f"Directory does not exist: {base}")
-        targets = sorted([p for p in base.iterdir() if p.is_dir()])
+        subdirs = _sorted_timestamp_subdirs(base)
+        subdirs = _apply_force_keep(subdirs, force_keep)
+        targets = subdirs
 
     return targets
 
@@ -241,6 +268,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="Keep (skip) the last N timestamp subdirectories inside each backup-docker-to-local folder (default: 0).",
+    )
     return parser.parse_args(argv)
 
 
@@ -249,7 +282,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)

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"]

tests/e2e/test_e2e_force_keep.py

@@ -0,0 +1,171 @@
+#!/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.
+
+    The current behavior is:
+    - In --all mode, cleanback discovers each /Backups/<ID>/backup-docker-to-local/*
+    - Within each backup-docker-to-local folder, subdirs are sorted by name
+    - With --force-keep N, the last N subdirs in that folder are skipped (kept)
+
+    This test creates two backup folders under /Backups so --all can find them:
+      /Backups/<prefix>-01/backup-docker-to-local/{good,bad}
+      /Backups/<prefix>-02/backup-docker-to-local/{good,bad}
+
+    With --force-keep 1:
+    - In each folder, "good" is the last (sorted) and is skipped (kept)
+    - "bad" is processed and deleted
+    """
+
+    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_timestamp_subdir_per_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)
+
+        # In each folder, sorted subdirs are: bad, good -> good is skipped, bad is processed
+        self.assertTrue(self.b1_good.exists(), "b1 good should remain (skipped)")
+        self.assertFalse(self.b1_bad.exists(), "b1 bad should be deleted")
+
+        self.assertTrue(self.b2_good.exists(), "b2 good should remain (skipped)")
+        self.assertFalse(self.b2_bad.exists(), "b2 bad should be deleted")
+
+        self.assertIn("Summary:", proc.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/unit/test_main.py

@@ -154,6 +154,36 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase):
         self.assertTrue(self.goodX.exists())
         self.assertFalse(self.badY.exists())
 
+    def test_all_mode_force_keep_skips_last_timestamp_subdir_per_backup_folder(self):
+        # Subdirs are sorted by name.
+        # --force-keep 1 skips the last subdir inside each backup-docker-to-local folder.
+        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 sorted: badB, goodA, timeoutC -> timeoutC is skipped, others processed
+        self.assertTrue(self.goodA.exists(), "goodA should remain")
+        self.assertFalse(self.badB.exists(), "badB should be deleted")
+        self.assertTrue(self.timeoutC.exists(), "timeoutC should be skipped (kept)")
+
+        # ID2 sorted: badY, goodX -> goodX is skipped, badY processed
+        self.assertTrue(self.goodX.exists(), "goodX should be skipped (kept)")
+        self.assertFalse(self.badY.exists(), "badY should be processed and deleted")
+
     def test_dirval_missing_errors(self):
         rc, out, err, _ = self.run_main(
             [