377 lines
15 KiB
Python
Executable File
377 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from delete_ready_episode import (
|
|
DeleteCollisionError,
|
|
DeleteSafetyError,
|
|
delete_ready_episode,
|
|
)
|
|
|
|
|
|
class DeleteReadyEpisodeTests(unittest.TestCase):
|
|
def _roots(self, root: Path) -> tuple[Path, Path, Path]:
|
|
base = root / "tg3_data_collection"
|
|
ready = base / "ready"
|
|
deleting = base / "deleting"
|
|
active = base / "active"
|
|
failed = base / "failed"
|
|
for path in (ready, active, failed):
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
(active / "ACTIVE_SENTINEL").write_text("active", encoding="utf-8")
|
|
(failed / "FAILED_SENTINEL").write_text("failed", encoding="utf-8")
|
|
return ready, deleting, base
|
|
|
|
def _episode(
|
|
self,
|
|
ready: Path,
|
|
name: str = "episode_20260810T120000000_deadbeef",
|
|
*,
|
|
state: str = "complete",
|
|
) -> tuple[Path, str]:
|
|
episode = ready / name
|
|
(episode / "bag").mkdir(parents=True)
|
|
(episode / "bag" / "bag_0.mcap").write_bytes(b"mcap")
|
|
manifest = {"state": state, "episode_id": name, "files": []}
|
|
raw = (json.dumps(manifest, sort_keys=True) + "\n").encode("utf-8")
|
|
(episode / "manifest.json").write_bytes(raw)
|
|
(episode / "READY").write_text("ready\n", encoding="ascii")
|
|
return episode, hashlib.sha256(raw).hexdigest()
|
|
|
|
def _authorization(
|
|
self,
|
|
base: Path,
|
|
episode: str,
|
|
digest: str,
|
|
*,
|
|
payload_digest: str | None = None,
|
|
) -> Path:
|
|
ledger = base / "delete_ledger"
|
|
ledger.mkdir(exist_ok=True)
|
|
tombstone = f"{episode}.{digest}.deleting"
|
|
path = ledger / f"{episode}.{digest}.authorized.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"state": "AUTHORIZED_DELETE",
|
|
"episode_id": episode,
|
|
"manifest_sha256": payload_digest or digest,
|
|
"tombstone": tombstone,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
def test_compare_rename_delete_and_absent_retry_are_idempotent(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
name = episode.name
|
|
result = delete_ready_episode(
|
|
name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(result["state"], "deleted")
|
|
self.assertFalse(episode.exists())
|
|
self.assertEqual(
|
|
delete_ready_episode(
|
|
name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)["state"],
|
|
"already_absent",
|
|
)
|
|
self.assertEqual(
|
|
(base / "active" / "ACTIVE_SENTINEL").read_text(), "active"
|
|
)
|
|
self.assertEqual(
|
|
(base / "failed" / "FAILED_SENTINEL").read_text(), "failed"
|
|
)
|
|
|
|
def test_resume_matching_tombstone_after_lost_ack(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
deleting.mkdir()
|
|
tombstone = deleting / f"{episode.name}.{digest}.deleting"
|
|
episode.rename(tombstone)
|
|
result = delete_ready_episode(
|
|
"episode_20260810T120000000_deadbeef",
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(result["state"], "resumed_delete")
|
|
self.assertFalse(tombstone.exists())
|
|
|
|
def test_partial_tree_without_manifest_resumes_from_durable_ledger(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
tombstone = deleting / f"{episode.name}.{digest}.deleting"
|
|
ledger = (
|
|
base
|
|
/ "delete_ledger"
|
|
/ f"{episode.name}.{digest}.authorized.json"
|
|
)
|
|
|
|
def interrupt_after_control_files_removed(
|
|
_parent_fd: int, _name: str
|
|
) -> None:
|
|
(tombstone / "READY").unlink()
|
|
(tombstone / "manifest.json").unlink()
|
|
raise RuntimeError("injected power loss during recursive delete")
|
|
|
|
with mock.patch(
|
|
"delete_ready_episode._remove_tree_at",
|
|
side_effect=interrupt_after_control_files_removed,
|
|
):
|
|
with self.assertRaises(RuntimeError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(tombstone.is_dir())
|
|
self.assertFalse((tombstone / "manifest.json").exists())
|
|
self.assertTrue(ledger.is_file())
|
|
|
|
result = delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(result["state"], "resumed_delete")
|
|
self.assertFalse(tombstone.exists())
|
|
self.assertFalse(ledger.exists())
|
|
|
|
def test_partial_tombstone_without_ledger_is_refused(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
deleting.mkdir()
|
|
tombstone = deleting / f"{episode.name}.{digest}.deleting"
|
|
episode.rename(tombstone)
|
|
(tombstone / "manifest.json").unlink()
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(tombstone.exists())
|
|
|
|
def test_wrong_digest_or_symlink_ledger_is_refused(self) -> None:
|
|
for mode in ("wrong_digest", "symlink"):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
ready, deleting, base = self._roots(root)
|
|
episode, digest = self._episode(ready)
|
|
deleting.mkdir()
|
|
tombstone = deleting / f"{episode.name}.{digest}.deleting"
|
|
episode.rename(tombstone)
|
|
(tombstone / "manifest.json").unlink()
|
|
if mode == "wrong_digest":
|
|
ledger = self._authorization(
|
|
base, episode.name, digest, payload_digest="f" * 64
|
|
)
|
|
else:
|
|
ledger_root = base / "delete_ledger"
|
|
ledger_root.mkdir()
|
|
ledger = (
|
|
ledger_root
|
|
/ f"{episode.name}.{digest}.authorized.json"
|
|
)
|
|
outside = root / "outside"
|
|
outside.write_text("keep", encoding="utf-8")
|
|
ledger.symlink_to(outside)
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(tombstone.exists())
|
|
self.assertTrue(ledger.exists())
|
|
|
|
def test_ledger_only_after_tree_removal_is_finalized_idempotently(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, base = self._roots(Path(temporary))
|
|
deleting.mkdir()
|
|
episode = "episode_20260810T120000000_deadbeef"
|
|
digest = "a" * 64
|
|
ledger = self._authorization(base, episode, digest)
|
|
result = delete_ready_episode(
|
|
episode,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(result["state"], "already_absent")
|
|
self.assertFalse(ledger.exists())
|
|
|
|
def test_conflicting_ledger_digest_refuses_ready_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
collision = self._authorization(base, episode.name, "b" * 64)
|
|
with self.assertRaises(DeleteCollisionError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(episode.exists())
|
|
self.assertTrue(collision.exists())
|
|
|
|
def test_wrong_manifest_digest_refuses_and_preserves_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, _digest = self._episode(ready)
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
"0" * 64,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(episode.exists())
|
|
|
|
def test_noncomplete_manifest_refuses_and_preserves_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready, state="failed")
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(episode.exists())
|
|
|
|
def test_unsafe_episode_and_digest_are_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
for episode, digest in (("../active", "0" * 64), ("safe", "BAD")):
|
|
with self.subTest(episode=episode, digest=digest):
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
|
|
def test_symlink_episode_is_refused_without_touching_outside(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
ready, deleting, _base = self._roots(root)
|
|
outside = root / "outside"
|
|
outside.mkdir()
|
|
sentinel = outside / "sentinel"
|
|
sentinel.write_text("keep", encoding="utf-8")
|
|
name = "episode_20260810T120000000_deadbeef"
|
|
(ready / name).symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
name,
|
|
"0" * 64,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(sentinel.read_text(), "keep")
|
|
|
|
def test_symlink_manifest_ready_or_nested_entry_is_refused(self) -> None:
|
|
for target in ("manifest", "ready", "nested"):
|
|
with self.subTest(target=target), tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
ready, deleting, _base = self._roots(root)
|
|
episode, digest = self._episode(ready)
|
|
outside = root / "outside"
|
|
outside.write_text("keep", encoding="utf-8")
|
|
if target == "manifest":
|
|
(episode / "manifest.json").unlink()
|
|
(episode / "manifest.json").symlink_to(outside)
|
|
elif target == "ready":
|
|
(episode / "READY").unlink()
|
|
(episode / "READY").symlink_to(outside)
|
|
else:
|
|
(episode / "nested_link").symlink_to(outside)
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertEqual(outside.read_text(), "keep")
|
|
|
|
def test_conflicting_tombstone_refuses_without_moving_ready(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
deleting.mkdir()
|
|
collision = deleting / f"{episode.name}.{'1' * 64}.deleting"
|
|
collision.mkdir()
|
|
with self.assertRaises(DeleteCollisionError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(episode.exists())
|
|
self.assertTrue(collision.exists())
|
|
|
|
def test_ready_and_same_tombstone_collision_refuses_both(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
ready, deleting, _base = self._roots(Path(temporary))
|
|
episode, digest = self._episode(ready)
|
|
deleting.mkdir()
|
|
tombstone = deleting / f"{episode.name}.{digest}.deleting"
|
|
tombstone.mkdir()
|
|
with self.assertRaises(DeleteCollisionError):
|
|
delete_ready_episode(
|
|
episode.name,
|
|
digest,
|
|
ready_root=ready,
|
|
deleting_root=deleting,
|
|
)
|
|
self.assertTrue(episode.exists())
|
|
self.assertTrue(tombstone.exists())
|
|
|
|
def test_wrong_parent_relationship_is_refused(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
ready, _deleting, _base = self._roots(root)
|
|
elsewhere = root / "elsewhere" / "deleting"
|
|
elsewhere.mkdir(parents=True)
|
|
with self.assertRaises(DeleteSafetyError):
|
|
delete_ready_episode(
|
|
"episode_safe",
|
|
"0" * 64,
|
|
ready_root=ready,
|
|
deleting_root=elsewhere,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|