diff --git a/src/cleanback/__main__.py b/src/cleanback/__main__.py index 0f8b67f..c2d1b88 100755 --- a/src/cleanback/__main__.py +++ b/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: //backup-docker-to-local/* (dirs only) - If --all: for each /* 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) diff --git a/tests/e2e/test_e2e_force_keep.py b/tests/e2e/test_e2e_force_keep.py index 814ebc6..42a37e5 100644 --- a/tests/e2e/test_e2e_force_keep.py +++ b/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//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/-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. + + 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 (-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 (-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) diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 9c6f2d3..ceaf9c4 100644 --- a/tests/unit/test_main.py +++ b/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(