189 lines
7.4 KiB
Python
189 lines
7.4 KiB
Python
"""Shared canonical repository contract and a substitutable consumer test."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from gewu.adapters.git_repository import GitCanonicalRepository
|
|
from gewu.application.identity import Principal
|
|
from gewu.application.ports import (
|
|
CanonicalConflictError,
|
|
CanonicalRepository,
|
|
CanonicalRepositoryError,
|
|
validate_canonical_path,
|
|
)
|
|
from gewu.application.reading import CanonicalNodeReader
|
|
from gewu.domain.contracts import KnowledgeNode
|
|
from tests.domain.proposal_fixtures import NODE
|
|
|
|
|
|
class InMemoryCanonicalRepository:
|
|
"""Independent port implementation used to prove consumer substitution."""
|
|
|
|
def __init__(self) -> None:
|
|
initial = hashlib.sha256(b"memory-repository-initial").hexdigest()[:40]
|
|
self._head = initial
|
|
self._snapshots = {initial: {}}
|
|
self._parents: dict[str, str | None] = {initial: None}
|
|
self._changed: dict[str, frozenset[str]] = {initial: frozenset()}
|
|
|
|
def head(self) -> str:
|
|
return self._head
|
|
|
|
def read_file(self, revision: str, path: str) -> bytes | None:
|
|
validate_canonical_path(path)
|
|
try:
|
|
return self._snapshots[revision].get(path)
|
|
except KeyError as error:
|
|
raise CanonicalRepositoryError("Unknown canonical revision") from error
|
|
|
|
def list_files(self, revision: str, prefix: str) -> tuple[str, ...]:
|
|
prefix = prefix.rstrip("/")
|
|
validate_canonical_path(prefix)
|
|
try:
|
|
files = self._snapshots[revision]
|
|
except KeyError as error:
|
|
raise CanonicalRepositoryError("Unknown canonical revision") from error
|
|
return tuple(sorted(path for path in files if path.startswith(prefix)))
|
|
|
|
def publish_files(
|
|
self,
|
|
*,
|
|
expected_head: str,
|
|
files: Mapping[str, bytes],
|
|
message: str,
|
|
index_directory: Path,
|
|
) -> str:
|
|
del message, index_directory
|
|
if expected_head != self._head:
|
|
raise CanonicalConflictError("Canonical branch changed before publication")
|
|
if not files:
|
|
raise CanonicalRepositoryError("Canonical publication requires at least one file")
|
|
for path in files:
|
|
validate_canonical_path(path)
|
|
parent_files = self._snapshots[self._head]
|
|
changed = frozenset(path for path, data in files.items() if parent_files.get(path) != data)
|
|
digest = hashlib.sha256()
|
|
digest.update(self._head.encode())
|
|
for path, data in sorted(files.items()):
|
|
digest.update(path.encode())
|
|
digest.update(data)
|
|
revision = digest.hexdigest()[:40]
|
|
snapshot = {**parent_files, **files}
|
|
self._snapshots[revision] = snapshot
|
|
self._parents[revision] = self._head
|
|
self._changed[revision] = changed
|
|
self._head = revision
|
|
return revision
|
|
|
|
def introducing_commit(self, revision: str, path: str) -> str | None:
|
|
revisions = self.history(revision, path, 50)
|
|
return revisions[-1] if revisions else None
|
|
|
|
def history(self, revision: str, path: str, limit: int) -> tuple[str, ...]:
|
|
validate_canonical_path(path)
|
|
if not 1 <= limit <= 50:
|
|
raise CanonicalRepositoryError("Canonical history limit is invalid")
|
|
if revision not in self._snapshots:
|
|
raise CanonicalRepositoryError("Unknown canonical revision")
|
|
revisions: list[str] = []
|
|
cursor: str | None = revision
|
|
while cursor is not None and len(revisions) < limit:
|
|
if path in self._changed[cursor]:
|
|
revisions.append(cursor)
|
|
cursor = self._parents[cursor]
|
|
return tuple(revisions)
|
|
|
|
|
|
@pytest.fixture(params=("git", "memory"), ids=("git-adapter", "in-memory-substitute"))
|
|
def canonical_repository(request: pytest.FixtureRequest, tmp_path: Path) -> CanonicalRepository:
|
|
if request.param == "git":
|
|
return GitCanonicalRepository(tmp_path / "knowledge.git")
|
|
return InMemoryCanonicalRepository()
|
|
|
|
|
|
def test_canonical_repository_contract(
|
|
canonical_repository: CanonicalRepository,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
base = canonical_repository.head()
|
|
record = "projects/project-a/nodes/node-a.yaml"
|
|
source = "sources/source-a/revision-a.yaml"
|
|
first = canonical_repository.publish_files(
|
|
expected_head=base,
|
|
files={record: b"title: first\n", source: b"title: source\n"},
|
|
message="Initial records",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
assert canonical_repository.head() == first
|
|
assert canonical_repository.read_file(first, record) == b"title: first\n"
|
|
assert canonical_repository.list_files(first, "projects/") == (record,)
|
|
assert canonical_repository.read_file(first, "unrelated/user-file.txt") is None
|
|
assert canonical_repository.introducing_commit(first, record) == first
|
|
assert canonical_repository.history(first, record, 20) == (first,)
|
|
assert canonical_repository.introducing_commit(first, "projects/absent.yaml") is None
|
|
|
|
second = canonical_repository.publish_files(
|
|
expected_head=first,
|
|
files={record: b"title: second\n"},
|
|
message="Update record",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
assert canonical_repository.read_file(first, record) == b"title: first\n"
|
|
assert canonical_repository.read_file(second, record) == b"title: second\n"
|
|
assert canonical_repository.history(second, record, 20) == (second, first)
|
|
assert canonical_repository.introducing_commit(second, record) == first
|
|
with pytest.raises(CanonicalConflictError):
|
|
canonical_repository.publish_files(
|
|
expected_head=base,
|
|
files={record: b"stale\n"},
|
|
message="Stale update",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
assert canonical_repository.head() == second
|
|
with pytest.raises(CanonicalRepositoryError):
|
|
canonical_repository.read_file(second, "../outside.yaml")
|
|
with pytest.raises(CanonicalRepositoryError):
|
|
canonical_repository.publish_files(
|
|
expected_head=second,
|
|
files={"../outside.yaml": b"unsafe\n"},
|
|
message="Unsafe update",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
|
|
|
|
def test_node_reader_works_unchanged_with_substitute_repository(tmp_path: Path) -> None:
|
|
repository = InMemoryCanonicalRepository()
|
|
project_hash = hashlib.sha256(NODE.project_id.encode()).hexdigest()
|
|
node_hash = hashlib.sha256(NODE.node_id.encode()).hexdigest()
|
|
path = f"projects/{project_hash}/nodes/{node_hash}.yaml"
|
|
raw = yaml.safe_dump(NODE.model_dump(mode="json"), allow_unicode=True).encode()
|
|
revision = repository.publish_files(
|
|
expected_head=repository.head(),
|
|
files={path: raw},
|
|
message="Publish reader fixture",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
|
|
reader = CanonicalNodeReader(repository)
|
|
response = reader.get_node(
|
|
NODE.project_id,
|
|
NODE.node_id,
|
|
principal=Principal(
|
|
"reader",
|
|
"reader-key",
|
|
frozenset({"viewer"}),
|
|
frozenset({NODE.project_id}),
|
|
),
|
|
)
|
|
loaded = KnowledgeNode.model_validate_json(json.dumps(response.data[0]), strict=True)
|
|
assert loaded == NODE
|
|
assert response.canonical_revision == revision
|
|
assert response.citations == NODE.claims[0].evidence
|