241 lines
9.1 KiB
Python
241 lines
9.1 KiB
Python
"""Tests for Mnemosyne snapshot (point-in-time backup/restore) feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nexus.mnemosyne.archive import MnemosyneArchive
|
|
from nexus.mnemosyne.ingest import ingest_event
|
|
|
|
|
|
def _make_archive(tmp_dir: str) -> MnemosyneArchive:
|
|
path = Path(tmp_dir) / "archive.json"
|
|
return MnemosyneArchive(archive_path=path, auto_embed=False)
|
|
|
|
|
|
# ─── snapshot_create ─────────────────────────────────────────────────────────
|
|
|
|
def test_snapshot_create_returns_metadata():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Alpha", content="First entry", topics=["a"])
|
|
ingest_event(archive, title="Beta", content="Second entry", topics=["b"])
|
|
|
|
result = archive.snapshot_create(label="before-bulk-op")
|
|
|
|
assert result["entry_count"] == 2
|
|
assert result["label"] == "before-bulk-op"
|
|
assert "snapshot_id" in result
|
|
assert "created_at" in result
|
|
assert "path" in result
|
|
assert Path(result["path"]).exists()
|
|
|
|
|
|
def test_snapshot_create_no_label():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Gamma", content="Third entry", topics=[])
|
|
|
|
result = archive.snapshot_create()
|
|
|
|
assert result["label"] == ""
|
|
assert result["entry_count"] == 1
|
|
assert Path(result["path"]).exists()
|
|
|
|
|
|
def test_snapshot_file_contains_entries():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
e = ingest_event(archive, title="Delta", content="Fourth entry", topics=["d"])
|
|
result = archive.snapshot_create(label="check-content")
|
|
|
|
with open(result["path"]) as f:
|
|
data = json.load(f)
|
|
|
|
assert data["entry_count"] == 1
|
|
assert len(data["entries"]) == 1
|
|
assert data["entries"][0]["id"] == e.id
|
|
assert data["entries"][0]["title"] == "Delta"
|
|
|
|
|
|
def test_snapshot_create_empty_archive():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
result = archive.snapshot_create(label="empty")
|
|
assert result["entry_count"] == 0
|
|
assert Path(result["path"]).exists()
|
|
|
|
|
|
# ─── snapshot_list ───────────────────────────────────────────────────────────
|
|
|
|
def test_snapshot_list_empty():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
assert archive.snapshot_list() == []
|
|
|
|
|
|
def test_snapshot_list_returns_all():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="One", content="c1", topics=[])
|
|
archive.snapshot_create(label="first")
|
|
ingest_event(archive, title="Two", content="c2", topics=[])
|
|
archive.snapshot_create(label="second")
|
|
|
|
snapshots = archive.snapshot_list()
|
|
assert len(snapshots) == 2
|
|
labels = {s["label"] for s in snapshots}
|
|
assert "first" in labels
|
|
assert "second" in labels
|
|
|
|
|
|
def test_snapshot_list_metadata_fields():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
archive.snapshot_create(label="meta-check")
|
|
snapshots = archive.snapshot_list()
|
|
s = snapshots[0]
|
|
for key in ("snapshot_id", "label", "created_at", "entry_count", "path"):
|
|
assert key in s
|
|
|
|
|
|
def test_snapshot_list_newest_first():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
archive.snapshot_create(label="a")
|
|
archive.snapshot_create(label="b")
|
|
snapshots = archive.snapshot_list()
|
|
# Filenames sort lexicographically; newest (b) should be first
|
|
# (filenames include timestamp so alphabetical = newest-last;
|
|
# snapshot_list reverses the glob order → newest first)
|
|
assert len(snapshots) == 2
|
|
# Both should be present; ordering is newest first
|
|
ids = [s["snapshot_id"] for s in snapshots]
|
|
assert ids == sorted(ids, reverse=True)
|
|
|
|
|
|
# ─── snapshot_restore ────────────────────────────────────────────────────────
|
|
|
|
def test_snapshot_restore_replaces_entries():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Kept", content="original content", topics=["orig"])
|
|
snap = archive.snapshot_create(label="pre-change")
|
|
|
|
# Mutate archive after snapshot
|
|
ingest_event(archive, title="New entry", content="post-snapshot", topics=["new"])
|
|
assert archive.count == 2
|
|
|
|
result = archive.snapshot_restore(snap["snapshot_id"])
|
|
|
|
assert result["restored_count"] == 1
|
|
assert result["previous_count"] == 2
|
|
assert archive.count == 1
|
|
entry = list(archive._entries.values())[0]
|
|
assert entry.title == "Kept"
|
|
|
|
|
|
def test_snapshot_restore_persists_to_disk():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "archive.json"
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Persisted", content="should survive reload", topics=[])
|
|
snap = archive.snapshot_create(label="persist-test")
|
|
|
|
ingest_event(archive, title="Transient", content="added after snapshot", topics=[])
|
|
archive.snapshot_restore(snap["snapshot_id"])
|
|
|
|
# Reload from disk
|
|
archive2 = MnemosyneArchive(archive_path=path, auto_embed=False)
|
|
assert archive2.count == 1
|
|
assert list(archive2._entries.values())[0].title == "Persisted"
|
|
|
|
|
|
def test_snapshot_restore_missing_raises():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
with pytest.raises(FileNotFoundError):
|
|
archive.snapshot_restore("nonexistent_snapshot_id")
|
|
|
|
|
|
# ─── snapshot_diff ───────────────────────────────────────────────────────────
|
|
|
|
def test_snapshot_diff_no_changes():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Stable", content="unchanged content", topics=[])
|
|
snap = archive.snapshot_create(label="baseline")
|
|
|
|
diff = archive.snapshot_diff(snap["snapshot_id"])
|
|
|
|
assert diff["added"] == []
|
|
assert diff["removed"] == []
|
|
assert diff["modified"] == []
|
|
assert diff["unchanged"] == 1
|
|
|
|
|
|
def test_snapshot_diff_detects_added():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
ingest_event(archive, title="Original", content="existing", topics=[])
|
|
snap = archive.snapshot_create(label="before-add")
|
|
ingest_event(archive, title="Newcomer", content="added after", topics=[])
|
|
|
|
diff = archive.snapshot_diff(snap["snapshot_id"])
|
|
|
|
assert len(diff["added"]) == 1
|
|
assert diff["added"][0]["title"] == "Newcomer"
|
|
assert diff["removed"] == []
|
|
assert diff["unchanged"] == 1
|
|
|
|
|
|
def test_snapshot_diff_detects_removed():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
e1 = ingest_event(archive, title="Will Be Removed", content="doomed", topics=[])
|
|
ingest_event(archive, title="Survivor", content="stays", topics=[])
|
|
snap = archive.snapshot_create(label="pre-removal")
|
|
archive.remove(e1.id)
|
|
|
|
diff = archive.snapshot_diff(snap["snapshot_id"])
|
|
|
|
assert len(diff["removed"]) == 1
|
|
assert diff["removed"][0]["title"] == "Will Be Removed"
|
|
assert diff["added"] == []
|
|
assert diff["unchanged"] == 1
|
|
|
|
|
|
def test_snapshot_diff_detects_modified():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
e = ingest_event(archive, title="Mutable", content="original content", topics=[])
|
|
snap = archive.snapshot_create(label="pre-edit")
|
|
archive.update_entry(e.id, content="updated content", auto_link=False)
|
|
|
|
diff = archive.snapshot_diff(snap["snapshot_id"])
|
|
|
|
assert len(diff["modified"]) == 1
|
|
assert diff["modified"][0]["title"] == "Mutable"
|
|
assert diff["modified"][0]["snapshot_hash"] != diff["modified"][0]["current_hash"]
|
|
assert diff["added"] == []
|
|
assert diff["removed"] == []
|
|
|
|
|
|
def test_snapshot_diff_missing_raises():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
with pytest.raises(FileNotFoundError):
|
|
archive.snapshot_diff("no_such_snapshot")
|
|
|
|
|
|
def test_snapshot_diff_includes_snapshot_id():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
archive = _make_archive(tmp)
|
|
snap = archive.snapshot_create(label="id-check")
|
|
diff = archive.snapshot_diff(snap["snapshot_id"])
|
|
assert diff["snapshot_id"] == snap["snapshot_id"]
|