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,8 @@
+## [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.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cleanback"
-version = "1.2.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

@@ -40,6 +40,20 @@ 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, force_keep: int
 ) -> List[Path]:
@@ -47,6 +61,8 @@ def discover_target_subdirs(
     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:
@@ -56,26 +72,25 @@ def discover_target_subdirs(
         raise FileNotFoundError(f"Backups root does not exist: {backups_root}")
 
     if all_mode:
-        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]
-
+        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
 
@@ -257,7 +272,7 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
         "--force-keep",
         type=int,
         default=0,
-        help="In --all mode: keep (skip) the last N backup folders under --backups-root (default: 0).",
+        help="Keep (skip) the last N timestamp subdirectories inside each backup-docker-to-local folder (default: 0).",
     )
     return parser.parse_args(argv)
 

tests/e2e/test_e2e_force_keep.py

@@ -41,10 +41,19 @@ if __name__ == "__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:
+
+    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, the last (sorted) backup folder (<prefix>-02) is skipped.
+
+    With --force-keep 1:
+    - In each folder, "good" is the last (sorted) and is skipped (kept)
+    - "bad" is processed and deleted
     """
 
     def setUp(self):
@@ -123,7 +132,7 @@ class CleanbackE2EForceKeepTests(unittest.TestCase):
         except Exception:
             pass
 
-    def test_all_mode_force_keep_skips_last_backup_folder(self):
+    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', '')}"
 
@@ -148,13 +157,12 @@ class CleanbackE2EForceKeepTests(unittest.TestCase):
 
         self.assertEqual(proc.returncode, 0, msg=proc.stderr or proc.stdout)
 
-        # First backup folder (<prefix>-01) should be processed: bad removed, good kept
-        self.assertTrue(self.b1_good.exists(), "b1 good should remain")
+        # 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")
 
-        # Last backup folder (<prefix>-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.assertFalse(self.b2_bad.exists(), "b2 bad should be deleted")
 
         self.assertIn("Summary:", proc.stdout)
 

tests/unit/test_main.py

@@ -153,10 +153,10 @@ 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.
+
+    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",
@@ -175,14 +175,14 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase):
         )
         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())
+        # 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 should be untouched
-        self.assertTrue(self.goodX.exists())
-        self.assertTrue(self.badY.exists())
+        # 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(