6 Commits

Author SHA1 Message Date
c02ae86207 Release version 1.3.0 2026-01-06 17:25:23 +01:00
838286c54e feat: make cleanup production-safe by separating invalid backups from infra errors
- Delete only truly invalid backups (dirval rc=1)
- Treat timeouts and missing dirval as infrastructure errors
- Never auto-delete backups affected by timeouts
- Return exit code 1 on infrastructure problems
- Update unit and E2E tests to reflect new safety semantics
- Align README with new deletion and exit-code behavior

https://chatgpt.com/share/695d36f6-0000-800f-98e7-f88a798d6e91
2026-01-06 17:23:05 +01:00
9e67392bd6 Release version 1.2.1 2026-01-06 15:42:05 +01:00
f402cea6f2 fix: apply --force-keep to timestamp subdirectories instead of backup folders
- 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
2026-01-06 15:40:16 +01:00
20a850ee21 Release version 1.2.0 2025-12-31 09:03:38 +01:00
3150bc5399 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
2025-12-31 09:02:34 +01:00
10 changed files with 390 additions and 66 deletions

View File

@@ -1,3 +1,18 @@
## [1.3.0] - 2026-01-06
* Cleanup is now production-safe: only invalid backups are deleted; timeouts no longer trigger automatic removal.
## [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 ## [1.1.0] - 2025-12-31
* The backups directory is now configurable via --backups-root instead of being hardcoded to /Backups. * The backups directory is now configurable via --backups-root instead of being hardcoded to /Backups.

104
README.md
View File

@@ -7,21 +7,24 @@
**Repository:** https://github.com/kevinveenbirkenbach/cleanup-failed-backups **Repository:** https://github.com/kevinveenbirkenbach/cleanup-failed-backups
`cleanback` validates and (optionally) cleans up **failed Docker backup directories**. `cleanback` validates and (optionally) cleans up **failed Docker backup directories** in a **production-safe** way.
It scans backup folders under a configurable backups root (e.g. `/Backups`), uses `dirval` to validate each subdirectory, and lets you delete the ones that fail validation.
Validation runs **in parallel** for performance; deletions are controlled and can be **interactive** or **fully automatic**. It scans backup folders under a configurable backups root (for example `/Backups`), uses `dirval` to validate each backup subdirectory, and removes **only those backups that are confirmed to be invalid**.
Validation runs **in parallel** for performance; deletions are **explicitly controlled** and can be interactive or fully automated.
--- ---
## ✨ Highlights ## ✨ Highlights
- **Parallel validation** of backup subdirectories - **Parallel validation** of backup subdirectories
- Uses **`dirval`** (directory-validator) via CLI - Uses **`dirval`** (directory validator) via CLI
- **Interactive** or **non-interactive** deletion flow (`--yes`) - **Safe deletion model**: only truly invalid backups are removed
- **Interactive** or **non-interactive** cleanup (`--yes`)
- Supports validating a single backup **ID** or **all** backups - Supports validating a single backup **ID** or **all** backups
- Clear **exit code semantics** for CI and system services
- Clean **Python package** with `pyproject.toml` - Clean **Python package** with `pyproject.toml`
- **Unit + Docker-based E2E tests** - **Unit tests** and **Docker-based E2E tests**
--- ---
@@ -31,7 +34,7 @@ Validation runs **in parallel** for performance; deletions are controlled and ca
```bash ```bash
pip install cleanback pip install cleanback
```` ```
This installs: This installs:
@@ -51,7 +54,7 @@ pip install -e .
## 🔧 Requirements ## 🔧 Requirements
* Python **3.8+** * Python **3.8+**
* Access to the backups root directory tree (e.g. `/Backups`) * Read/write access to the backups root directory tree (e.g. `/Backups`)
* `dirval` (installed automatically via pip dependency) * `dirval` (installed automatically via pip dependency)
--- ---
@@ -66,6 +69,8 @@ After installation, the command is:
cleanback cleanback
``` ```
---
### Validate a single backup ID ### Validate a single backup ID
```bash ```bash
@@ -78,6 +83,8 @@ Validates directories under:
/Backups/<ID>/backup-docker-to-local/* /Backups/<ID>/backup-docker-to-local/*
``` ```
---
### Validate all backups ### Validate all backups
```bash ```bash
@@ -92,49 +99,78 @@ Scans:
--- ---
### Common options ## ⚙️ Common options
| Option | Description | | Option | Description |
| -------------------- | ------------------------------------------------------------------ | | -------------------- | ------------------------------------------------------------------------------------- |
| `--dirval-cmd <cmd>` | Path or name of `dirval` executable (default: `dirval`) | | `--dirval-cmd <cmd>` | Path or name of `dirval` executable (default: `dirval`) |
| `--workers <n>` | Parallel workers (default: CPU count, min 2) | | `--workers <n>` | Number of parallel validator workers (default: CPU count, minimum 2) |
| `--timeout <sec>` | Per-directory validation timeout (float supported, default: 300.0) | | `--timeout <sec>` | Per-directory validation timeout in seconds (float supported, default: `300.0`) |
| `--yes` | Non-interactive mode: delete failures automatically | | `--yes` | Non-interactive mode: automatically delete **invalid** backups (dirval rc=1 only) |
| `--force-keep <n>` | In `--all` mode: skip the last *n* timestamp subdirectories inside each backup folder |
> **Note:** Backups affected by timeouts or infrastructure errors are **never deleted automatically**, even when `--yes` is used.
--- ---
### Examples ## 🧪 Examples
```bash ```bash
# Validate a single backup and prompt on failures # Validate a single backup and prompt before deleting invalid ones
cleanback --backups-root /Backups --id 2024-09-01T12-00-00 cleanback --backups-root /Backups --id 2024-09-01T12-00-00
# Validate everything with 8 workers and auto-delete failures
cleanback --backups-root /Backups --all --workers 8 --yes
# Use a custom dirval binary and short timeout
cleanback --backups-root /Backups --all --dirval-cmd /usr/local/bin/dirval --timeout 5.0
``` ```
--- ```bash
# Validate all backups and automatically delete invalid ones
## 🧪 Tests cleanback --backups-root /Backups --all --workers 8 --yes
```
### Run all tests
```bash ```bash
make test # Use a custom dirval binary and a short timeout (testing only)
cleanback \
--backups-root /Backups \
--all \
--dirval-cmd /usr/local/bin/dirval \
--timeout 5.0
``` ```
--- ---
## 🔒 Safety & Design Notes ## 🔒 Safety & Design Notes
* **No host filesystem is modified** during tests * **Validation and deletion are strictly separated**
(E2E tests run exclusively inside Docker) * Only backups explicitly marked **invalid by `dirval`** are eligible for deletion
* Deletions are **explicitly confirmed** unless `--yes` is used * **Timeouts and infrastructure errors are NOT treated as invalid backups**
* Timeouts are treated as **validation failures** * Backups affected by timeouts are **never deleted automatically**
* Validation and deletion phases are **clearly separated** * Infrastructure problems (timeouts, missing `dirval`) cause a **non-zero exit code**
* Deletions require confirmation unless `--yes` is specified
* Tests never touch the host filesystem (E2E tests run inside Docker only)
This design makes `cleanback` safe for unattended operation on production systems.
---
## 🚦 Exit codes
`cleanback` uses exit codes to clearly distinguish between backup issues and infrastructure problems:
| Exit code | Meaning |
| --------- | ------------------------------------------------------------------ |
| `0` | All backups valid, or invalid backups were successfully removed |
| `1` | Validation infrastructure problem (e.g. timeout, missing `dirval`) |
| `2` | CLI usage or configuration error |
This makes the tool suitable for **CI pipelines**, **systemd services**, and other automation.
---
## 🧪 Tests
Run all tests (unit + Docker-based E2E):
```bash
make test
```
--- ---

View File

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

View File

@@ -1,6 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Cleanup Failed Docker Backups — parallel validator (using dirval) Cleanup Failed Docker Backups — parallel validator (using dirval)
with optional "keep last N backups" behavior in --all mode.
Validates backup subdirectories under: Validates backup subdirectories under:
- <BACKUPS_ROOT>/<ID>/backup-docker-to-local (when --id is used) - <BACKUPS_ROOT>/<ID>/backup-docker-to-local (when --id is used)
@@ -39,31 +40,57 @@ class ValidationResult:
stdout: str 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( 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]: ) -> List[Path]:
""" """
Return a list of subdirectories to validate: Return a list of subdirectories to validate:
- If backup_id is given: <root>/<id>/backup-docker-to-local/* (dirs only) - 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 - 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] = [] targets: List[Path] = []
if force_keep < 0:
raise ValueError("--force-keep must be >= 0")
if not backups_root.is_dir(): if not backups_root.is_dir():
raise FileNotFoundError(f"Backups root does not exist: {backups_root}") raise FileNotFoundError(f"Backups root does not exist: {backups_root}")
if all_mode: 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" candidate = backup_folder / "backup-docker-to-local"
if candidate.is_dir(): 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: else:
if not backup_id: if not backup_id:
raise ValueError("Either --id or --all must be provided.") raise ValueError("Either --id or --all must be provided.")
base = backups_root / backup_id / "backup-docker-to-local" base = backups_root / backup_id / "backup-docker-to-local"
if not base.is_dir(): if not base.is_dir():
raise FileNotFoundError(f"Directory does not exist: {base}") 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 return targets
@@ -241,15 +268,37 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
action="store_true", action="store_true",
help="Do not prompt; delete failing directories automatically.", 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) return parser.parse_args(argv)
def _is_timeout(res: ValidationResult) -> bool:
return res.returncode == 124 or "timed out" in (res.stderr or "").lower()
def _is_dirval_missing(res: ValidationResult) -> bool:
return res.returncode == 127 or "not found" in (res.stderr or "").lower()
def _is_invalid(res: ValidationResult) -> bool:
# dirval: 0 = ok, 1 = invalid, others = infra errors (timeout/missing/etc.)
return res.returncode == 1
def main(argv: Optional[List[str]] = None) -> int: def main(argv: Optional[List[str]] = None) -> int:
args = parse_args(argv) args = parse_args(argv)
try: try:
subdirs = discover_target_subdirs( 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: except Exception as e:
print(f"ERROR: {e}", file=sys.stderr) print(f"ERROR: {e}", file=sys.stderr)
@@ -260,18 +309,43 @@ def main(argv: Optional[List[str]] = None) -> int:
return 0 return 0
results = parallel_validate(subdirs, args.dirval_cmd, args.workers, args.timeout) results = parallel_validate(subdirs, args.dirval_cmd, args.workers, args.timeout)
failures = [r for r in results if not r.ok]
if not failures: invalids = [r for r in results if _is_invalid(r)]
timeouts = [r for r in results if _is_timeout(r)]
missing = [r for r in results if _is_dirval_missing(r)]
deleted = 0
if invalids:
print(f"\n{len(invalids)} directory(ies) are invalid (dirval rc=1).")
deleted = process_deletions(invalids, assume_yes=args.yes)
ok_count = sum(1 for r in results if r.ok)
if timeouts or missing:
print("\nERROR: validation infrastructure problem detected.")
if timeouts:
print(f"- timeouts: {len(timeouts)} (will NOT delete these)")
for r in timeouts[:10]:
print(f" timeout: {r.subdir}")
if len(timeouts) > 10:
print(f" ... (+{len(timeouts) - 10} more)")
if missing:
print(f"- dirval missing: {len(missing)} (will NOT delete these)")
for r in missing[:10]:
print(f" missing: {r.subdir}")
if len(missing) > 10:
print(f" ... (+{len(missing) - 10} more)")
print(
f"\nSummary: deleted={deleted}, invalid={len(invalids)}, ok={ok_count}, timeouts={len(timeouts)}, missing={len(missing)}"
)
return 1
if not invalids:
print("\nAll directories validated successfully. No action required.") print("\nAll directories validated successfully. No action required.")
return 0 return 0
print(f"\n{len(failures)} directory(ies) failed validation.") print(f"\nSummary: deleted={deleted}, invalid={len(invalids)}, ok={ok_count}")
deleted = process_deletions(failures, assume_yes=args.yes)
kept = len(failures) - deleted
print(
f"\nSummary: deleted={deleted}, kept={kept}, ok={len(results) - len(failures)}"
)
return 0 return 0

View File

@@ -14,4 +14,4 @@ RUN python -m pip install -U pip \
RUN mkdir -p /Backups RUN mkdir -p /Backups
# Run E2E unittest # 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"]

0
tests/e2e/__init__.py Normal file
View File

View File

@@ -121,13 +121,6 @@ class CleanbackE2EDockerTests(unittest.TestCase):
env["PATH"] = f"{self.bin_dir}:{env.get('PATH', '')}" env["PATH"] = f"{self.bin_dir}:{env.get('PATH', '')}"
# Run: python -m cleanback --id <ID> --yes # Run: python -m cleanback --id <ID> --yes
# We must point BACKUPS_ROOT to our run_root. Easiest: set /Backups = run_root
# But code currently has BACKUPS_ROOT = /Backups constant.
#
# Therefore, we create our test tree under /Backups (done above) and pass --id
# relative to that structure by using run_root/<ID>. To do that, we make
# run_root the direct child under /Backups, then we pass the composite id:
# "<run-folder>/<ID>".
composite_id = f"{self.run_root.name}/{self.backup_id}" composite_id = f"{self.run_root.name}/{self.backup_id}"
cmd = [ cmd = [
@@ -148,14 +141,19 @@ class CleanbackE2EDockerTests(unittest.TestCase):
] ]
proc = subprocess.run(cmd, text=True, capture_output=True, env=env) proc = subprocess.run(cmd, text=True, capture_output=True, env=env)
self.assertEqual(proc.returncode, 0, msg=proc.stderr or proc.stdout) # New behavior:
# - invalid dirs are deleted and do NOT cause failure
# - timeouts are treated as infrastructure problems -> exit code 1 and NOT deleted
self.assertEqual(proc.returncode, 1, msg=proc.stderr or proc.stdout)
self.assertTrue(self.good.exists(), "good should remain") self.assertTrue(self.good.exists(), "good should remain")
self.assertFalse(self.bad.exists(), "bad should be deleted") self.assertFalse(self.bad.exists(), "bad should be deleted")
self.assertFalse( self.assertTrue(
self.timeout.exists(), self.timeout.exists(),
"timeout should be deleted (timeout treated as failure)", "timeout should NOT be deleted (timeouts are infrastructure problems)",
) )
self.assertIn("Summary:", proc.stdout) self.assertIn("Summary:", proc.stdout)
self.assertIn("validation infrastructure problem", proc.stdout.lower())
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -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)

0
tests/unit/__init__.py Normal file
View File

View File

@@ -123,12 +123,12 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase):
"--yes", "--yes",
] ]
) )
self.assertEqual(rc, 0, msg=err or out) self.assertEqual(rc, 1, msg=err or out)
self.assertTrue(self.goodA.exists(), "goodA should remain") self.assertTrue(self.goodA.exists(), "goodA should remain")
self.assertFalse(self.badB.exists(), "badB should be deleted") self.assertFalse(self.badB.exists(), "badB should be deleted")
self.assertFalse( self.assertTrue(
self.timeoutC.exists(), self.timeoutC.exists(),
"timeoutC should be deleted (timeout treated as failure)", "timeoutC should NOT be deleted (timeout is infra error)",
) )
self.assertIn("Summary:", out) self.assertIn("Summary:", out)
@@ -147,13 +147,43 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase):
"--yes", "--yes",
] ]
) )
self.assertEqual(rc, 0, msg=err or out) self.assertEqual(rc, 1, msg=err or out)
self.assertTrue(self.goodA.exists()) self.assertTrue(self.goodA.exists())
self.assertFalse(self.badB.exists()) self.assertFalse(self.badB.exists())
self.assertFalse(self.timeoutC.exists()) self.assertTrue(self.timeoutC.exists())
self.assertTrue(self.goodX.exists()) self.assertTrue(self.goodX.exists())
self.assertFalse(self.badY.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): def test_dirval_missing_errors(self):
rc, out, err, _ = self.run_main( rc, out, err, _ = self.run_main(
[ [
@@ -168,8 +198,8 @@ class CleanupBackupsUsingDirvalTests(unittest.TestCase):
"--yes", "--yes",
] ]
) )
self.assertEqual(rc, 0, msg=err or out) self.assertEqual(rc, 1, msg=err or out)
self.assertIn("dirval not found", out + err) self.assertIn("dirval missing", out + err)
def test_no_targets_message(self): def test_no_targets_message(self):
empty = self.backups_root / "EMPTY" / "backup-docker-to-local" empty = self.backups_root / "EMPTY" / "backup-docker-to-local"