205 lines
8.7 KiB
Python
205 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from docforge.projection_fragments import (
|
|
FRAGMENT_CACHE_DIRECTORY,
|
|
FRAGMENT_KEY_CONTRACT,
|
|
FRAGMENT_RECORD_CONTRACT,
|
|
FRAGMENT_SCHEMA_VERSION,
|
|
FragmentKey,
|
|
FragmentRecord,
|
|
ProjectionFragmentCache,
|
|
fragment_semantic_hash,
|
|
)
|
|
|
|
|
|
class ProjectionFragmentCacheTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name) / "project"
|
|
self.root.mkdir()
|
|
self.cache_root = self.root / ".docforge" / "cache"
|
|
self.cache = ProjectionFragmentCache(
|
|
self.root,
|
|
self.cache_root,
|
|
maximum_content_bytes=1_000,
|
|
)
|
|
self.key = self.make_key({"node": "guide.alpha", "summary": "Alpha"})
|
|
|
|
@staticmethod
|
|
def make_key(
|
|
semantics: object,
|
|
*,
|
|
renderer_version: str = "1",
|
|
component_version: str = "graph.node@1",
|
|
) -> FragmentKey:
|
|
return FragmentKey.create(
|
|
projection_kind="graph",
|
|
renderer_id="portable_graph_html",
|
|
renderer_version=renderer_version,
|
|
component_version=component_version,
|
|
semantic_input_hash=fragment_semantic_hash(semantics),
|
|
)
|
|
|
|
@property
|
|
def entry_path(self) -> Path:
|
|
return self.cache_root / FRAGMENT_CACHE_DIRECTORY / f"{self.key.key_id}.json"
|
|
|
|
def test_miss_hit_and_unchanged_reuse_are_exact(self) -> None:
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
with mock.patch(
|
|
"docforge.projection_fragments.atomic_replace_bytes_at",
|
|
wraps=__import__(
|
|
"docforge.projection_fragments",
|
|
fromlist=["atomic_replace_bytes_at"],
|
|
).atomic_replace_bytes_at,
|
|
) as atomic:
|
|
stored = self.cache.put(self.key, b"<li>Alpha</li>")
|
|
self.assertIsNotNone(stored)
|
|
self.assertEqual(1, atomic.call_count)
|
|
before = self.entry_path.stat()
|
|
repeated = self.cache.put(self.key, b"<li>Alpha</li>")
|
|
self.assertEqual(1, atomic.call_count)
|
|
self.assertEqual(stored, repeated)
|
|
self.assertEqual(stored, self.cache.get(self.key))
|
|
after = self.entry_path.stat()
|
|
self.assertEqual((before.st_dev, before.st_ino), (after.st_dev, after.st_ino))
|
|
|
|
def test_key_versions_and_complete_semantics_invalidate_independently(self) -> None:
|
|
self.assertIsNotNone(self.cache.put(self.key, b"alpha"))
|
|
variants = (
|
|
self.make_key(
|
|
{"node": "guide.alpha", "summary": "Alpha"},
|
|
renderer_version="2",
|
|
),
|
|
self.make_key(
|
|
{"node": "guide.alpha", "summary": "Alpha"},
|
|
component_version="graph.node@2",
|
|
),
|
|
self.make_key({"node": "guide.alpha", "summary": "Changed"}),
|
|
)
|
|
for variant in variants:
|
|
with self.subTest(key=variant):
|
|
self.assertNotEqual(self.key.key_id, variant.key_id)
|
|
self.assertIsNone(self.cache.get(variant))
|
|
|
|
def test_add_delete_and_reorder_change_sequence_semantics(self) -> None:
|
|
base = ["a", "b"]
|
|
added = ["a", "b", "c"]
|
|
deleted = ["a"]
|
|
reordered = ["b", "a"]
|
|
keys = [self.make_key(value) for value in (base, added, deleted, reordered)]
|
|
self.assertEqual(4, len({key.key_id for key in keys}))
|
|
self.assertEqual(
|
|
fragment_semantic_hash({"a": 1, "b": 2}),
|
|
fragment_semantic_hash({"b": 2, "a": 1}),
|
|
)
|
|
|
|
def test_corrupt_oversized_incompatible_and_foreign_entries_are_misses(self) -> None:
|
|
self.assertIsNotNone(self.cache.put(self.key, b"alpha"))
|
|
self.entry_path.write_bytes(b"{bad-json")
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
self.assertIsNotNone(self.cache.put(self.key, b"repaired"))
|
|
repaired = self.cache.get(self.key)
|
|
self.assertIsNotNone(repaired)
|
|
assert repaired is not None
|
|
self.assertEqual(b"repaired", repaired.content)
|
|
|
|
self.entry_path.write_bytes(b"x" * (self.cache.maximum_record_bytes + 1))
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
|
|
incompatible = FragmentRecord.create(self.key, b"alpha").as_dict()
|
|
incompatible["schema_version"] = FRAGMENT_SCHEMA_VERSION + 1
|
|
self.entry_path.write_bytes(
|
|
json.dumps(incompatible, sort_keys=True, separators=(",", ":")).encode()
|
|
)
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
|
|
foreign_key = self.make_key({"node": "foreign"})
|
|
foreign = FragmentRecord.create(foreign_key, b"foreign").to_bytes()
|
|
self.entry_path.write_bytes(foreign)
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
|
|
def test_symlinked_entry_and_cache_root_fail_closed_without_outside_writes(self) -> None:
|
|
self.entry_path.parent.mkdir(parents=True)
|
|
outside = Path(self.temporary.name) / "outside.json"
|
|
outside.write_bytes(FragmentRecord.create(self.key, b"outside").to_bytes())
|
|
self.entry_path.symlink_to(outside)
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
self.assertIsNone(self.cache.put(self.key, b"replacement"))
|
|
self.assertEqual(b"outside", FragmentRecord.from_bytes(outside.read_bytes()).content)
|
|
|
|
outside_cache = Path(self.temporary.name) / "outside-cache"
|
|
escaped = ProjectionFragmentCache(self.root, outside_cache)
|
|
self.assertIsNone(escaped.put(self.key, b"escaped"))
|
|
self.assertFalse(outside_cache.exists())
|
|
|
|
def test_keys_and_records_have_deterministic_path_free_serialization(self) -> None:
|
|
same_key = self.make_key({"summary": "Alpha", "node": "guide.alpha"})
|
|
self.assertEqual(self.key, same_key)
|
|
first = FragmentRecord.create(self.key, b"\x00fragment\xff")
|
|
second = FragmentRecord.create(same_key, b"\x00fragment\xff")
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(first.to_bytes(), second.to_bytes())
|
|
self.assertEqual(first, FragmentRecord.from_bytes(first.to_bytes()))
|
|
document = json.loads(first.to_bytes())
|
|
self.assertEqual(FRAGMENT_SCHEMA_VERSION, document["schema_version"])
|
|
self.assertEqual(FRAGMENT_RECORD_CONTRACT, document["contract"])
|
|
self.assertEqual(FRAGMENT_KEY_CONTRACT, document["key"]["contract"])
|
|
self.assertNotIn(str(self.root), first.to_bytes().decode("utf-8"))
|
|
self.assertEqual(first.byte_count, len(first.content))
|
|
self.assertEqual(64, len(first.content_sha256))
|
|
|
|
def test_cache_write_failure_is_a_miss_and_does_not_mutate_canonical_files(self) -> None:
|
|
canonical = self.root / "canonical.md"
|
|
canonical.write_text("canonical", encoding="utf-8")
|
|
with mock.patch(
|
|
"docforge.projection_fragments.atomic_replace_bytes_at",
|
|
side_effect=OSError("synthetic cache failure"),
|
|
):
|
|
self.assertIsNone(self.cache.put(self.key, b"fragment"))
|
|
self.assertIsNone(self.cache.get(self.key))
|
|
self.assertEqual("canonical", canonical.read_text(encoding="utf-8"))
|
|
|
|
def test_record_rejects_noncanonical_serialization_and_invalid_evidence(self) -> None:
|
|
record = FragmentRecord.create(self.key, b"alpha")
|
|
pretty = json.dumps(record.as_dict(), sort_keys=True, indent=2).encode()
|
|
with self.assertRaisesRegex(Exception, "canonically serialized"):
|
|
FragmentRecord.from_bytes(pretty)
|
|
tampered = record.as_dict()
|
|
tampered["byte_count"] = record.byte_count + 1
|
|
canonical = json.dumps(tampered, sort_keys=True, separators=(",", ":")).encode()
|
|
with self.assertRaisesRegex(Exception, "byte evidence"):
|
|
FragmentRecord.from_bytes(canonical)
|
|
|
|
def test_prune_retains_only_the_exact_bounded_active_inventory(self) -> None:
|
|
retained = self.key
|
|
stale = self.make_key({"node": "stale"})
|
|
self.assertIsNotNone(self.cache.put(retained, b"retained"))
|
|
self.assertIsNotNone(self.cache.put(stale, b"stale"))
|
|
foreign = self.entry_path.parent / "foreign.tmp"
|
|
foreign.write_bytes(b"foreign")
|
|
|
|
self.assertTrue(self.cache.prune((retained,)))
|
|
retained_record = self.cache.get(retained)
|
|
self.assertIsNotNone(retained_record)
|
|
assert retained_record is not None
|
|
self.assertEqual(b"retained", retained_record.content)
|
|
self.assertIsNone(self.cache.get(stale))
|
|
self.assertFalse(foreign.exists())
|
|
|
|
with mock.patch(
|
|
"docforge.projection_fragments.MAX_FRAGMENT_CACHE_BYTES",
|
|
1,
|
|
):
|
|
self.assertFalse(self.cache.prune((retained,)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|