fix(checkpoints): report legacy archive deletion failures
This commit is contained in:
@@ -180,6 +180,9 @@ def cmd_clear_legacy(args: argparse.Namespace) -> int:
|
||||
|
||||
result = clear_legacy()
|
||||
print(f"Deleted {result['deleted']} archive(s), reclaimed {_fmt_bytes(result['bytes_freed'])}.")
|
||||
if result["errors"]:
|
||||
print(f"Failed to delete {result['errors']} archive(s). See logs for details.")
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
@@ -26,6 +26,27 @@ def _prune_result(**kwargs) -> dict:
|
||||
return result
|
||||
|
||||
|
||||
def test_clear_legacy_returns_error_when_an_archive_cannot_be_deleted(monkeypatch, capsys):
|
||||
import hermes_cli.checkpoints as checkpoints_cli
|
||||
import tools.checkpoint_manager as ckpt_mgr
|
||||
|
||||
monkeypatch.setattr(
|
||||
ckpt_mgr,
|
||||
"store_status",
|
||||
lambda: {"legacy_archives": [{"name": "legacy-read-only", "size_bytes": 1}]},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
ckpt_mgr,
|
||||
"clear_legacy",
|
||||
lambda: {"deleted": 0, "errors": 1, "bytes_freed": 0},
|
||||
)
|
||||
|
||||
rc = checkpoints_cli.cmd_clear_legacy(_ns(force=True))
|
||||
|
||||
assert rc == 2
|
||||
assert "Failed to delete 1 archive(s)." in capsys.readouterr().out
|
||||
|
||||
|
||||
_V2_ORPHAN_ONLY_STATUS = {
|
||||
"projects": [],
|
||||
"pre_v2_projects": [],
|
||||
@@ -122,4 +143,3 @@ def test_empty_preview_binds_empty_allowlist(monkeypatch, capsys):
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1187,6 +1187,50 @@ class TestClearFunctions:
|
||||
# Store preserved
|
||||
assert (base / "store" / "HEAD").exists()
|
||||
|
||||
def test_clear_legacy_reports_partial_deletion_failures(self, tmp_path, monkeypatch):
|
||||
base = tmp_path / "checkpoints"
|
||||
failed = base / "legacy-failed"
|
||||
deleted = base / "legacy-deleted"
|
||||
for archive in (failed, deleted):
|
||||
archive.mkdir(parents=True)
|
||||
(archive / "data").write_bytes(b"x")
|
||||
|
||||
real_rmtree = shutil.rmtree
|
||||
|
||||
def fail_one(path, *args, **kwargs):
|
||||
if Path(path) == failed:
|
||||
raise OSError("read-only file")
|
||||
return real_rmtree(path, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr("tools.checkpoint_manager.shutil.rmtree", fail_one)
|
||||
|
||||
result = clear_legacy(base)
|
||||
|
||||
assert result["deleted"] == 1
|
||||
assert result["errors"] == 1
|
||||
assert not deleted.exists()
|
||||
assert failed.exists()
|
||||
|
||||
def test_clear_legacy_reports_all_deletion_failures(self, tmp_path, monkeypatch):
|
||||
base = tmp_path / "checkpoints"
|
||||
archives = [base / "legacy-first", base / "legacy-second"]
|
||||
for archive in archives:
|
||||
archive.mkdir(parents=True)
|
||||
|
||||
def always_fail(*args, **kwargs):
|
||||
raise OSError("read-only file")
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tools.checkpoint_manager.shutil.rmtree",
|
||||
always_fail,
|
||||
)
|
||||
|
||||
result = clear_legacy(base)
|
||||
|
||||
assert result["deleted"] == 0
|
||||
assert result["errors"] == 2
|
||||
assert all(archive.exists() for archive in archives)
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Orphan pruning must not act on an unreachable volume
|
||||
|
||||
@@ -1254,9 +1254,9 @@ def clear_all(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:
|
||||
|
||||
|
||||
def clear_legacy(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:
|
||||
"""Delete all ``legacy-*`` archive directories. Returns ``{"bytes_freed": N, "deleted": count}``."""
|
||||
"""Delete all ``legacy-*`` archive directories and report any failures."""
|
||||
base = checkpoint_base or _resolve_checkpoint_base()
|
||||
out = {"bytes_freed": 0, "deleted": 0}
|
||||
out = {"bytes_freed": 0, "deleted": 0, "errors": 0}
|
||||
if not base.exists():
|
||||
return out
|
||||
for child in _legacy_archives(base):
|
||||
|
||||
Reference in New Issue
Block a user