44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
import pytest
|
|
|
|
from tht.corpus.models import CorpusManifest
|
|
from tht.corpus.store import CorpusStore, UnsafeCorpusPath
|
|
|
|
|
|
def test_publish_switches_active_atomically_and_resolves_materialized_files(tmp_path):
|
|
store = CorpusStore(tmp_path / "corpus")
|
|
generation = store.stage(CorpusManifest(), {})
|
|
seen = []
|
|
store._replace = lambda source, target: (seen.append(source.read_text()), source.replace(target))
|
|
published = store.publish(generation)
|
|
assert published == generation
|
|
assert store.active_generation() == generation
|
|
assert seen == [generation + "\n"]
|
|
|
|
|
|
def test_active_manifest_is_a_consistent_reader_snapshot(tmp_path):
|
|
store = CorpusStore(tmp_path / "corpus")
|
|
first = store.stage(CorpusManifest(metadata={"name": "first"}), {})
|
|
second = store.stage(CorpusManifest(metadata={"name": "second"}), {})
|
|
store.publish(first)
|
|
snapshot = store.active_manifest()
|
|
store.publish(second)
|
|
assert snapshot.metadata["name"] == "first"
|
|
assert store.active_manifest().metadata["name"] == "second"
|
|
|
|
|
|
def test_store_rejects_symlinked_generation_root(tmp_path):
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
root = tmp_path / "corpus"
|
|
root.symlink_to(outside, target_is_directory=True)
|
|
with pytest.raises(UnsafeCorpusPath):
|
|
CorpusStore(root)
|
|
|
|
|
|
def test_active_pointer_cannot_escape_generation_root(tmp_path):
|
|
store = CorpusStore(tmp_path / "corpus")
|
|
store.root.mkdir(parents=True, exist_ok=True)
|
|
store.active_path.write_text("../outside\n")
|
|
with pytest.raises(UnsafeCorpusPath):
|
|
store.active_manifest()
|