1379 lines
62 KiB
Python
1379 lines
62 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest import mock
|
|
|
|
import docforge.application as application_module
|
|
from docforge.application import CanonicalApplicationService, GenericCanonicalApplier
|
|
from docforge.changesets import ChangesetStore
|
|
from docforge.errors import DocForgeError
|
|
from docforge.project import Project
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIXTURES = ROOT / "tests" / "fixtures"
|
|
|
|
|
|
def _append_in_process(
|
|
root: str,
|
|
node_id: str,
|
|
expected_changeset_hash: str,
|
|
expected_content_hash: str,
|
|
start: Any,
|
|
result: Any,
|
|
) -> None:
|
|
"""Attempt one append in an independent process and report its stable outcome."""
|
|
|
|
start.wait()
|
|
try:
|
|
ChangesetStore(Project.open(root), "alpha-editor").propose_update(
|
|
changeset_id="simultaneous",
|
|
expected_changeset_hash=expected_changeset_hash,
|
|
node_id=node_id,
|
|
expected_content_hash=expected_content_hash,
|
|
metadata={"summary": f"Concurrent proposal for {node_id}."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Prove process-safe optimistic serialization.",
|
|
)
|
|
except DocForgeError as error:
|
|
result.send(error.code)
|
|
else:
|
|
result.send("ok")
|
|
finally:
|
|
result.close()
|
|
|
|
|
|
class DocForgeChangesetTests(unittest.TestCase):
|
|
def copy_fixture(self, destination: Path) -> Path:
|
|
root = destination / "alpha"
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
|
return root
|
|
|
|
def canonical_digest(self, root: Path) -> str:
|
|
paths = [
|
|
root / ".docforge" / "project.toml",
|
|
root / "POLICY.md",
|
|
*(root / "docs" / "content").glob("*"),
|
|
]
|
|
digest = hashlib.sha256()
|
|
for path in sorted((path for path in paths if path.is_file()), key=lambda item: str(item)):
|
|
digest.update(path.relative_to(root).as_posix().encode())
|
|
digest.update(path.read_bytes())
|
|
return digest.hexdigest()
|
|
|
|
@staticmethod
|
|
def node_hash(project: Project, node_id: str) -> str:
|
|
return next(node.content_hash for node in project.load().nodes if node.node_id == node_id)
|
|
|
|
@staticmethod
|
|
def new_metadata(*, family: str = "guide") -> dict[str, object]:
|
|
return {
|
|
"title": "New proposal node",
|
|
"family": family,
|
|
"authority": "proposal",
|
|
"status": "active",
|
|
"tags": ["proposal", "test"],
|
|
"summary": "Exercises an isolated structured creation proposal.",
|
|
}
|
|
|
|
def test_all_four_operations_validate_and_never_change_canonical_sources(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
|
|
create_root = self.copy_fixture(parent / "create")
|
|
create_before = self.canonical_digest(create_root)
|
|
create_store = ChangesetStore(Project.open(create_root), "alpha-editor")
|
|
created = create_store.create("create-node")
|
|
proposed = create_store.propose_create(
|
|
changeset_id="create-node",
|
|
expected_changeset_hash=created["changeset_hash"],
|
|
node_id="guide.new-node",
|
|
target_source="docs/content/new-node.md",
|
|
metadata=self.new_metadata(),
|
|
content="This content exists only inside the proposal.",
|
|
relationship_changes=[
|
|
{
|
|
"action": "add",
|
|
"source_id": "guide.new-node",
|
|
"relation": "depends_on",
|
|
"target_id": "guide.foundation",
|
|
}
|
|
],
|
|
rationale="Add one independently reviewable guide node.",
|
|
)
|
|
self.assertEqual(1, proposed["operation_count"])
|
|
self.assertEqual(create_before, self.canonical_digest(create_root))
|
|
self.assertFalse((create_root / "docs/content/new-node.md").exists())
|
|
create_diff = create_store.diff("create-node")
|
|
self.assertEqual("create", create_diff["changes"][0]["operation"])
|
|
self.assertIn("proposal", create_diff["changes"][0]["content_diff"])
|
|
self.assertEqual(create_diff, create_store.diff("create-node"))
|
|
|
|
update_root = self.copy_fixture(parent / "update")
|
|
update_before = self.canonical_digest(update_root)
|
|
update_project = Project.open(update_root)
|
|
update_store = ChangesetStore(update_project, "alpha-editor")
|
|
update_changeset = update_store.create("update-node")
|
|
updated = update_store.propose_update(
|
|
changeset_id="update-node",
|
|
expected_changeset_hash=update_changeset["changeset_hash"],
|
|
node_id="guide.workflow",
|
|
expected_content_hash=self.node_hash(update_project, "guide.workflow"),
|
|
metadata={"summary": "A proposed replacement workflow summary."},
|
|
content="Proposed workflow content that is not canonical yet.",
|
|
relationship_changes=[],
|
|
rationale="Replace the workflow explanation without applying it.",
|
|
)
|
|
self.assertEqual(3, updated["projected_node_count"])
|
|
update_diff = update_store.diff("update-node")["changes"][0]
|
|
self.assertEqual("update", update_diff["operation"])
|
|
self.assertIn("summary", update_diff["metadata"])
|
|
self.assertEqual(update_before, self.canonical_digest(update_root))
|
|
|
|
move_root = self.copy_fixture(parent / "move")
|
|
move_before = self.canonical_digest(move_root)
|
|
move_project = Project.open(move_root)
|
|
move_store = ChangesetStore(move_project, "alpha-editor")
|
|
move_changeset = move_store.create("move-node")
|
|
move_store.propose_move(
|
|
changeset_id="move-node",
|
|
expected_changeset_hash=move_changeset["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=self.node_hash(move_project, "guide.foundation"),
|
|
target_source="docs/content/foundation-moved.md",
|
|
rationale="Move the node while preserving its stable identity.",
|
|
)
|
|
move_diff = move_store.diff("move-node")["changes"][0]
|
|
self.assertEqual(
|
|
{
|
|
"before": "docs/content/foundation.md",
|
|
"after": "docs/content/foundation-moved.md",
|
|
},
|
|
move_diff["source"],
|
|
)
|
|
self.assertEqual(move_before, self.canonical_digest(move_root))
|
|
|
|
delete_root = self.copy_fixture(parent / "delete")
|
|
delete_before = self.canonical_digest(delete_root)
|
|
delete_project = Project.open(delete_root)
|
|
delete_store = ChangesetStore(delete_project, "alpha-editor")
|
|
delete_changeset = delete_store.create("delete-node")
|
|
deleted = delete_store.propose_delete(
|
|
changeset_id="delete-node",
|
|
expected_changeset_hash=delete_changeset["changeset_hash"],
|
|
node_id="proof.validation",
|
|
expected_content_hash=self.node_hash(delete_project, "proof.validation"),
|
|
relationship_changes=[
|
|
{
|
|
"action": "remove",
|
|
"source_id": "proof.validation",
|
|
"relation": "proves",
|
|
"target_id": "guide.workflow",
|
|
}
|
|
],
|
|
rationale="Remove the proof node and its owned relationship in one proposal.",
|
|
)
|
|
self.assertEqual(2, deleted["projected_node_count"])
|
|
self.assertEqual("delete", delete_store.diff("delete-node")["changes"][0]["operation"])
|
|
self.assertEqual(delete_before, self.canonical_digest(delete_root))
|
|
|
|
def test_hash_bound_application_writes_projection_and_refreshes_derived_state(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
hashes = {node.node_id: node.content_hash for node in project.load().nodes}
|
|
created = store.create("apply-all")
|
|
updated = store.propose_update(
|
|
changeset_id="apply-all",
|
|
expected_changeset_hash=created["changeset_hash"],
|
|
node_id="guide.workflow",
|
|
expected_content_hash=hashes["guide.workflow"],
|
|
metadata={"summary": "Applied through the canonical application service."},
|
|
content="The applied workflow is now canonical.",
|
|
relationship_changes=[],
|
|
rationale="Exercise a canonical update.",
|
|
)
|
|
added = store.propose_create(
|
|
changeset_id="apply-all",
|
|
expected_changeset_hash=updated["changeset_hash"],
|
|
node_id="guide.applied",
|
|
target_source="docs/content/applied.md",
|
|
metadata=self.new_metadata(),
|
|
content="This node was created by an approved changeset.",
|
|
relationship_changes=[],
|
|
rationale="Exercise a canonical creation.",
|
|
)
|
|
moved = store.propose_move(
|
|
changeset_id="apply-all",
|
|
expected_changeset_hash=added["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=hashes["guide.foundation"],
|
|
target_source="docs/content/foundation-moved.md",
|
|
rationale="Exercise a canonical move.",
|
|
)
|
|
final = store.propose_delete(
|
|
changeset_id="apply-all",
|
|
expected_changeset_hash=moved["changeset_hash"],
|
|
node_id="proof.validation",
|
|
expected_content_hash=hashes["proof.validation"],
|
|
relationship_changes=[
|
|
{
|
|
"action": "remove",
|
|
"source_id": "proof.validation",
|
|
"relation": "proves",
|
|
"target_id": "guide.workflow",
|
|
}
|
|
],
|
|
rationale="Exercise a canonical deletion.",
|
|
)
|
|
service = CanonicalApplicationService(
|
|
project,
|
|
applier_id="alpha-editor",
|
|
applier=GenericCanonicalApplier(project),
|
|
)
|
|
result = service.apply("apply-all", str(final["changeset_hash"]))
|
|
|
|
snapshot = project.load()
|
|
node_ids = {node.node_id for node in snapshot.nodes}
|
|
workflow = next(node for node in snapshot.nodes if node.node_id == "guide.workflow")
|
|
self.assertTrue(result["applied"])
|
|
self.assertEqual("applied", result["lifecycle"]["status"])
|
|
self.assertEqual("ok", result["derived_refresh"]["status"])
|
|
self.assertEqual(
|
|
[
|
|
"docs/content/applied.md",
|
|
"docs/content/foundation-moved.md",
|
|
"docs/content/foundation.md",
|
|
"docs/content/proof.toml",
|
|
"docs/content/workflow.md",
|
|
],
|
|
result["applied_sources"],
|
|
)
|
|
self.assertEqual({"guide.applied", "guide.foundation", "guide.workflow"}, node_ids)
|
|
self.assertEqual(
|
|
"Applied through the canonical application service.",
|
|
workflow.summary,
|
|
)
|
|
self.assertFalse((root / "docs/content/foundation.md").exists())
|
|
self.assertFalse((root / "docs/content/proof.toml").exists())
|
|
self.assertTrue((root / "docs/content/foundation-moved.md").is_file())
|
|
self.assertTrue((root / ".docforge/cache/index.sqlite3").is_file())
|
|
self.assertTrue((root / ".docforge/rendered/manual.html").is_file())
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "cannot be modified") as closed:
|
|
service.apply("apply-all", str(final["changeset_hash"]))
|
|
self.assertEqual("changeset_closed", closed.exception.code)
|
|
|
|
def test_canonical_update_exchange_preserves_a_raced_external_edit(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"update-race",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Approved summary."},
|
|
"rationale": "Exercise the atomic update boundary.",
|
|
}
|
|
],
|
|
)
|
|
proposal_path = root / ".docforge/changesets/update-race.json"
|
|
proposal_bytes = proposal_path.read_bytes()
|
|
target = root / "docs/content/workflow.md"
|
|
exchange = application_module.rename_exchange_between_at
|
|
raced = False
|
|
|
|
def race(
|
|
first_directory_fd: int,
|
|
first: str,
|
|
second_directory_fd: int,
|
|
second: str,
|
|
) -> None:
|
|
nonlocal raced
|
|
if second == target.name and not raced:
|
|
raced = True
|
|
target.write_bytes(target.read_bytes() + b"\nExternal edit at exchange.\n")
|
|
exchange(first_directory_fd, first, second_directory_fd, second)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.application.rename_exchange_between_at",
|
|
side_effect=race,
|
|
),
|
|
self.assertRaises(DocForgeError) as captured,
|
|
):
|
|
store.apply(
|
|
changeset_id="update-race",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertEqual("base_conflict", captured.exception.code)
|
|
self.assertIn("External edit at exchange.", target.read_text(encoding="utf-8"))
|
|
self.assertEqual(proposal_bytes, proposal_path.read_bytes())
|
|
self.assertFalse((root / ".docforge/changesets/.state/update-race.json").exists())
|
|
self.assertFalse(tuple(target.parent.glob(".docforge-apply-*")))
|
|
|
|
def test_canonical_create_and_delete_races_preserve_foreign_targets(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
|
|
create_root = self.copy_fixture(parent / "create")
|
|
create_project = Project.open(create_root)
|
|
create_store = ChangesetStore(create_project, "alpha-editor")
|
|
create = create_store.register(
|
|
"create-race",
|
|
[
|
|
{
|
|
"operation": "create",
|
|
"node_id": "guide.raced",
|
|
"target_source": "docs/content/raced.md",
|
|
"metadata": self.new_metadata(),
|
|
"content": "Approved new content.",
|
|
"rationale": "Exercise no-replace creation.",
|
|
}
|
|
],
|
|
)
|
|
create_target = create_root / "docs/content/raced.md"
|
|
real_link = application_module.os.link
|
|
appeared = False
|
|
|
|
def race_create(
|
|
source: str,
|
|
target: str,
|
|
*,
|
|
src_dir_fd: int,
|
|
dst_dir_fd: int,
|
|
follow_symlinks: bool,
|
|
) -> None:
|
|
nonlocal appeared
|
|
if target == create_target.name and not appeared:
|
|
appeared = True
|
|
create_target.write_bytes(b"foreign create target\n")
|
|
real_link(
|
|
source,
|
|
target,
|
|
src_dir_fd=src_dir_fd,
|
|
dst_dir_fd=dst_dir_fd,
|
|
follow_symlinks=follow_symlinks,
|
|
)
|
|
|
|
with (
|
|
mock.patch("docforge.application.os.link", side_effect=race_create),
|
|
self.assertRaises(DocForgeError) as create_error,
|
|
):
|
|
create_store.apply(
|
|
changeset_id="create-race",
|
|
expected_changeset_hash=str(create["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(create_project).apply,
|
|
)
|
|
self.assertEqual("base_conflict", create_error.exception.code)
|
|
self.assertEqual(b"foreign create target\n", create_target.read_bytes())
|
|
self.assertFalse(tuple(create_target.parent.glob(".docforge-apply-*")))
|
|
|
|
delete_root = self.copy_fixture(parent / "delete")
|
|
delete_project = Project.open(delete_root)
|
|
delete_store = ChangesetStore(delete_project, "alpha-editor")
|
|
delete = delete_store.register(
|
|
"delete-race",
|
|
[
|
|
{
|
|
"operation": "delete",
|
|
"node_id": "proof.validation",
|
|
"relationship_changes": [
|
|
{
|
|
"action": "remove",
|
|
"source_id": "proof.validation",
|
|
"relation": "proves",
|
|
"target_id": "guide.workflow",
|
|
}
|
|
],
|
|
"rationale": "Exercise atomic deletion.",
|
|
}
|
|
],
|
|
)
|
|
delete_target = delete_root / "docs/content/proof.toml"
|
|
exchange = application_module.rename_exchange_between_at
|
|
deleted_race = False
|
|
|
|
def race_delete(
|
|
first_directory_fd: int,
|
|
first: str,
|
|
second_directory_fd: int,
|
|
second: str,
|
|
) -> None:
|
|
nonlocal deleted_race
|
|
if second == delete_target.name and not deleted_race:
|
|
deleted_race = True
|
|
delete_target.write_bytes(
|
|
delete_target.read_bytes() + b"\n# foreign delete edit\n"
|
|
)
|
|
exchange(first_directory_fd, first, second_directory_fd, second)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.application.rename_exchange_between_at",
|
|
side_effect=race_delete,
|
|
),
|
|
self.assertRaises(DocForgeError) as delete_error,
|
|
):
|
|
delete_store.apply(
|
|
changeset_id="delete-race",
|
|
expected_changeset_hash=str(delete["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(delete_project).apply,
|
|
)
|
|
self.assertEqual("base_conflict", delete_error.exception.code)
|
|
self.assertIn("# foreign delete edit", delete_target.read_text(encoding="utf-8"))
|
|
self.assertFalse(tuple(delete_target.parent.glob(".docforge-apply-*")))
|
|
|
|
def test_delete_detach_race_restores_foreign_target_and_retains_original(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"delete-detach-race",
|
|
[
|
|
{
|
|
"operation": "delete",
|
|
"node_id": "proof.validation",
|
|
"relationship_changes": [
|
|
{
|
|
"action": "remove",
|
|
"source_id": "proof.validation",
|
|
"relation": "proves",
|
|
"target_id": "guide.workflow",
|
|
}
|
|
],
|
|
"rationale": "Race the final no-replace canonical detachment.",
|
|
}
|
|
],
|
|
)
|
|
target = root / "docs/content/proof.toml"
|
|
original = target.read_bytes()
|
|
move = application_module.rename_noreplace_between_at
|
|
raced = False
|
|
|
|
def race_detach(
|
|
source_directory_fd: int,
|
|
source: str,
|
|
target_directory_fd: int,
|
|
destination: str,
|
|
) -> bool:
|
|
nonlocal raced
|
|
if source == target.name and destination.startswith(".detached-") and not raced:
|
|
raced = True
|
|
replacement = target.with_name(".foreign-delete")
|
|
replacement.write_bytes(b"foreign replacement at delete detach\n")
|
|
os.replace(replacement, target)
|
|
return move(
|
|
source_directory_fd,
|
|
source,
|
|
target_directory_fd,
|
|
destination,
|
|
)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.application.rename_noreplace_between_at",
|
|
side_effect=race_detach,
|
|
),
|
|
self.assertRaises(DocForgeError) as captured,
|
|
):
|
|
store.apply(
|
|
changeset_id="delete-detach-race",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertTrue(raced)
|
|
self.assertEqual("application_recovery_required", captured.exception.code)
|
|
self.assertEqual(b"foreign replacement at delete detach\n", target.read_bytes())
|
|
conflicts = captured.exception.details["conflicts"]
|
|
retained = root / conflicts[0]["retained"]
|
|
self.assertEqual(original, retained.read_bytes())
|
|
self.assertFalse(
|
|
(root / ".docforge/changesets/.state/delete-detach-race.json").exists()
|
|
)
|
|
|
|
def test_create_rollback_detach_race_never_unlinks_foreign_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"create-rollback-detach-race",
|
|
[
|
|
{
|
|
"operation": "create",
|
|
"node_id": "guide.created",
|
|
"target_source": "docs/content/a-created.md",
|
|
"metadata": self.new_metadata(),
|
|
"content": "Approved content that publishes first.",
|
|
"rationale": "Exercise create rollback detachment.",
|
|
},
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Synthetic failing second publication."},
|
|
"rationale": "Trigger rollback after create publication.",
|
|
},
|
|
],
|
|
)
|
|
target = root / "docs/content/a-created.md"
|
|
publish = GenericCanonicalApplier._publish
|
|
move = application_module.rename_noreplace_between_at
|
|
publish_calls = 0
|
|
raced = False
|
|
|
|
def fail_second(
|
|
applier: GenericCanonicalApplier,
|
|
publication: Any,
|
|
) -> None:
|
|
nonlocal publish_calls
|
|
publish_calls += 1
|
|
if publish_calls == 1:
|
|
publish(applier, publication)
|
|
return
|
|
raise DocForgeError("application_failure", "Synthetic second publication failure")
|
|
|
|
def race_rollback_detach(
|
|
source_directory_fd: int,
|
|
source: str,
|
|
target_directory_fd: int,
|
|
destination: str,
|
|
) -> bool:
|
|
nonlocal raced
|
|
if source == target.name and destination.startswith(".detached-") and not raced:
|
|
raced = True
|
|
replacement = target.with_name(".foreign-create-rollback")
|
|
replacement.write_bytes(b"foreign replacement during create rollback\n")
|
|
os.replace(replacement, target)
|
|
return move(
|
|
source_directory_fd,
|
|
source,
|
|
target_directory_fd,
|
|
destination,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(
|
|
GenericCanonicalApplier,
|
|
"_publish",
|
|
autospec=True,
|
|
side_effect=fail_second,
|
|
),
|
|
mock.patch(
|
|
"docforge.application.rename_noreplace_between_at",
|
|
side_effect=race_rollback_detach,
|
|
),
|
|
self.assertRaises(DocForgeError) as captured,
|
|
):
|
|
store.apply(
|
|
changeset_id="create-rollback-detach-race",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertTrue(raced)
|
|
self.assertEqual("application_recovery_required", captured.exception.code)
|
|
self.assertEqual(
|
|
b"foreign replacement during create rollback\n",
|
|
target.read_bytes(),
|
|
)
|
|
self.assertFalse(
|
|
(root / ".docforge/changesets/.state/create-rollback-detach-race.json").exists()
|
|
)
|
|
|
|
def test_rollback_never_clobbers_a_foreign_edit_and_retains_original_bytes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"rollback-race",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.foundation",
|
|
"metadata": {"summary": "First approved update."},
|
|
"rationale": "Publish before the synthetic failure.",
|
|
},
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Second approved update."},
|
|
"rationale": "Trigger rollback after the first publication.",
|
|
},
|
|
],
|
|
)
|
|
first_target = root / "docs/content/foundation.md"
|
|
first_before = first_target.read_bytes()
|
|
publish = GenericCanonicalApplier._publish
|
|
calls = 0
|
|
|
|
def fail_after_foreign_edit(
|
|
applier: GenericCanonicalApplier,
|
|
publication: Any,
|
|
) -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
publish(applier, publication)
|
|
first_target.write_bytes(
|
|
first_target.read_bytes() + b"\nForeign edit after publication.\n"
|
|
)
|
|
return
|
|
raise DocForgeError("application_failure", "Synthetic second-target failure")
|
|
|
|
with (
|
|
mock.patch.object(
|
|
GenericCanonicalApplier,
|
|
"_publish",
|
|
autospec=True,
|
|
side_effect=fail_after_foreign_edit,
|
|
),
|
|
self.assertRaises(DocForgeError) as captured,
|
|
):
|
|
store.apply(
|
|
changeset_id="rollback-race",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertEqual("application_recovery_required", captured.exception.code)
|
|
conflicts = captured.exception.details["conflicts"]
|
|
self.assertEqual("target_or_backup_changed", conflicts[0]["reason"])
|
|
self.assertIn(
|
|
"Foreign edit after publication.",
|
|
first_target.read_text(encoding="utf-8"),
|
|
)
|
|
retained = root / conflicts[0]["retained"]
|
|
self.assertTrue(retained.is_file())
|
|
self.assertEqual(first_before, retained.read_bytes())
|
|
self.assertFalse((root / ".docforge/changesets/.state/rollback-race.json").exists())
|
|
|
|
def test_canonical_update_preserves_existing_file_mode(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
target = root / "docs/content/workflow.md"
|
|
target.chmod(0o6750)
|
|
before = target.stat()
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"mode",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Mode-preserving update."},
|
|
"rationale": "Preserve canonical file permissions.",
|
|
}
|
|
],
|
|
)
|
|
|
|
result = store.apply(
|
|
changeset_id="mode",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertTrue(result["applied"])
|
|
after = target.stat()
|
|
self.assertEqual(0o6750, stat.S_IMODE(after.st_mode))
|
|
self.assertEqual(before.st_uid, after.st_uid)
|
|
self.assertEqual(before.st_gid, after.st_gid)
|
|
self.assertEqual([], result["retained_recovery_files"])
|
|
self.assertEqual("clean", result["application_recovery"]["status"])
|
|
self.assertFalse(tuple((root / ".docforge/application").glob("transaction-*")))
|
|
|
|
def test_nested_creation_fsyncs_each_new_directory_and_parent_entry(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"nested-durable",
|
|
[
|
|
{
|
|
"operation": "create",
|
|
"node_id": "guide.nested",
|
|
"target_source": "docs/content/nested/deeper/guide.md",
|
|
"metadata": self.new_metadata(),
|
|
"content": "Nested canonical content.",
|
|
"rationale": "Prove durable nested-directory creation.",
|
|
}
|
|
],
|
|
)
|
|
real_fsync = os.fsync
|
|
fsynced_directories: set[Path] = set()
|
|
|
|
def record_fsync(descriptor: int) -> None:
|
|
try:
|
|
path = Path(os.readlink(f"/proc/self/fd/{descriptor}"))
|
|
if path.is_dir():
|
|
fsynced_directories.add(path)
|
|
except OSError:
|
|
pass
|
|
real_fsync(descriptor)
|
|
|
|
with mock.patch(
|
|
"docforge.application.os.fsync",
|
|
side_effect=record_fsync,
|
|
):
|
|
result = store.apply(
|
|
changeset_id="nested-durable",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertTrue(result["applied"])
|
|
for path in (
|
|
root / "docs/content",
|
|
root / "docs/content/nested",
|
|
root / "docs/content/nested/deeper",
|
|
):
|
|
self.assertIn(path, fsynced_directories)
|
|
|
|
def test_post_commit_cleanup_failure_closes_proposal_with_recovery_record(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
proposal = store.register(
|
|
"cleanup-recovery",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Committed despite private cleanup failure."},
|
|
"rationale": "Persist actionable post-commit recovery evidence.",
|
|
}
|
|
],
|
|
)
|
|
real_unlink = os.unlink
|
|
failed = False
|
|
|
|
def fail_private_cleanup(
|
|
path: str | bytes,
|
|
*,
|
|
dir_fd: int | None = None,
|
|
) -> None:
|
|
nonlocal failed
|
|
if (
|
|
isinstance(path, str)
|
|
and path.startswith("staged-")
|
|
and dir_fd is not None
|
|
and not failed
|
|
):
|
|
failed = True
|
|
raise PermissionError("synthetic private cleanup failure")
|
|
real_unlink(path, dir_fd=dir_fd)
|
|
|
|
with mock.patch(
|
|
"docforge.application.os.unlink",
|
|
side_effect=fail_private_cleanup,
|
|
):
|
|
result = store.apply(
|
|
changeset_id="cleanup-recovery",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
|
|
self.assertTrue(failed)
|
|
self.assertTrue(result["applied"])
|
|
self.assertEqual("applied", result["lifecycle"]["status"])
|
|
self.assertEqual(
|
|
"cleanup_required",
|
|
result["application_recovery"]["status"],
|
|
)
|
|
self.assertEqual(
|
|
"cleanup_required",
|
|
result["lifecycle"]["application_recovery"]["status"],
|
|
)
|
|
retained = result["application_recovery"]["retained"]
|
|
self.assertTrue(retained)
|
|
lifecycle_path = root / ".docforge/changesets/.state/cleanup-recovery.json"
|
|
lifecycle = json.loads(lifecycle_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
"cleanup_required",
|
|
lifecycle["application_recovery"]["status"],
|
|
)
|
|
with self.assertRaises(DocForgeError) as closed:
|
|
store.apply(
|
|
changeset_id="cleanup-recovery",
|
|
expected_changeset_hash=str(proposal["changeset_hash"]),
|
|
applier_id="alpha-editor",
|
|
application=GenericCanonicalApplier(project).apply,
|
|
)
|
|
self.assertEqual("changeset_closed", closed.exception.code)
|
|
|
|
def test_changeset_rollback_fsyncs_the_parent_directory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "proposal.json"
|
|
real_fsync = os.fsync
|
|
fsynced_modes: list[int] = []
|
|
|
|
def record_fsync(descriptor: int) -> None:
|
|
fsynced_modes.append(os.fstat(descriptor).st_mode)
|
|
real_fsync(descriptor)
|
|
|
|
for previous in (b"previous proposal\n", None):
|
|
with self.subTest(previous=previous):
|
|
target.write_bytes(b"replacement proposal\n")
|
|
fsynced_modes.clear()
|
|
|
|
with mock.patch(
|
|
"docforge.changesets.os.fsync",
|
|
side_effect=record_fsync,
|
|
):
|
|
ChangesetStore._restore(target, previous, root)
|
|
|
|
self.assertTrue(any(stat.S_ISDIR(mode) for mode in fsynced_modes))
|
|
if previous is None:
|
|
self.assertFalse(target.exists())
|
|
else:
|
|
self.assertEqual(previous, target.read_bytes())
|
|
|
|
def test_abandoned_proposal_releases_overlap_and_stale_work_remains_active(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
first = store.register(
|
|
"first",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.foundation",
|
|
"metadata": {"summary": "Abandoned proposal."},
|
|
"rationale": "Reserve then release this node.",
|
|
}
|
|
],
|
|
)
|
|
store.abandon(
|
|
"first",
|
|
str(first["changeset_hash"]),
|
|
"The proposal is no longer wanted.",
|
|
)
|
|
second = store.register(
|
|
"second",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.foundation",
|
|
"metadata": {"summary": "Replacement proposal."},
|
|
"rationale": "Verify terminal proposals release conflicts.",
|
|
}
|
|
],
|
|
)
|
|
workflow = root / "docs/content/workflow.md"
|
|
workflow.write_text(
|
|
workflow.read_text(encoding="utf-8") + "\nUnrelated current fact.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
active = store.list_changesets(include_history=False)
|
|
stale = store.list_changesets(include_history=False, status="stale")
|
|
|
|
self.assertEqual(0, active["count"])
|
|
self.assertEqual(["second"], [item["changeset_id"] for item in stale["changesets"]])
|
|
self.assertEqual("stale", stale["changesets"][0]["lifecycle"]["status"])
|
|
self.assertEqual("ready", second["lifecycle"])
|
|
|
|
def test_lifecycle_receipts_obey_the_changeset_size_limit(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
store = ChangesetStore(Project.open(root), "alpha-editor")
|
|
created = store.create("bounded-lifecycle")
|
|
|
|
with self.assertRaises(DocForgeError) as oversized:
|
|
store.abandon(
|
|
"bounded-lifecycle",
|
|
str(created["changeset_hash"]),
|
|
"x" * 100_001,
|
|
)
|
|
|
|
self.assertEqual("changeset_too_large", oversized.exception.code)
|
|
self.assertFalse((root / ".docforge/changesets/.state/bounded-lifecycle.json").exists())
|
|
|
|
def test_applied_receipt_survives_a_derived_refresh_failure(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
registered = ChangesetStore(project, "alpha-editor").register(
|
|
"degraded-refresh",
|
|
[
|
|
{
|
|
"operation": "update",
|
|
"node_id": "guide.workflow",
|
|
"metadata": {"summary": "Canonical even if refresh fails."},
|
|
"rationale": "Separate canonical success from disposable refresh.",
|
|
}
|
|
],
|
|
)
|
|
service = CanonicalApplicationService(
|
|
project,
|
|
applier_id="alpha-editor",
|
|
applier=GenericCanonicalApplier(project),
|
|
)
|
|
with mock.patch.object(
|
|
service.index,
|
|
"build",
|
|
side_effect=DocForgeError("index_failure", "Synthetic derived failure"),
|
|
):
|
|
result = service.apply(
|
|
"degraded-refresh",
|
|
str(registered["changeset_hash"]),
|
|
)
|
|
|
|
self.assertTrue(result["applied"])
|
|
self.assertEqual("applied", result["lifecycle"]["status"])
|
|
self.assertEqual("degraded", result["derived_refresh"]["status"])
|
|
self.assertEqual("index", result["derived_refresh"]["errors"][0]["component"])
|
|
with self.assertRaisesRegex(DocForgeError, "cannot be modified"):
|
|
service.apply(
|
|
"degraded-refresh",
|
|
str(registered["changeset_hash"]),
|
|
)
|
|
|
|
def test_relationship_only_update_is_hash_bound_and_does_not_rewrite_node(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
workflow = next(
|
|
node for node in project.load().nodes if node.node_id == "guide.workflow"
|
|
)
|
|
created = store.create("relationship-only")
|
|
proposed = store.propose_relationship_update(
|
|
changeset_id="relationship-only",
|
|
expected_changeset_hash=str(created["changeset_hash"]),
|
|
node_id="guide.workflow",
|
|
expected_content_hash=workflow.content_hash,
|
|
relationship_changes=[
|
|
{
|
|
"action": "remove",
|
|
"source_id": "guide.workflow",
|
|
"relation": "depends_on",
|
|
"target_id": "guide.foundation",
|
|
}
|
|
],
|
|
rationale="Queue one relationship correction without changing node content.",
|
|
)
|
|
|
|
change = store.diff("relationship-only")["changes"][0]
|
|
self.assertEqual("update", change["operation"])
|
|
self.assertEqual("", change["content_diff"])
|
|
self.assertEqual({}, change["metadata"])
|
|
self.assertEqual(1, proposed["projected_edge_count"])
|
|
with self.assertRaisesRegex(DocForgeError, "at least one"):
|
|
store.propose_relationship_update(
|
|
changeset_id="relationship-only",
|
|
expected_changeset_hash=str(proposed["changeset_hash"]),
|
|
node_id="guide.foundation",
|
|
expected_content_hash=self.node_hash(project, "guide.foundation"),
|
|
relationship_changes=[],
|
|
rationale="Reject an empty relationship operation.",
|
|
)
|
|
|
|
def test_optimistic_and_cross_changeset_conflicts_preserve_both_proposals(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
first = store.create("first")
|
|
second = store.create("second")
|
|
foundation_hash = self.node_hash(project, "guide.foundation")
|
|
first_result = store.propose_update(
|
|
changeset_id="first",
|
|
expected_changeset_hash=first["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=foundation_hash,
|
|
metadata={"summary": "First proposal owns this node."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Claim the foundation node in the first proposal.",
|
|
)
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "overlaps another proposal") as conflict:
|
|
store.propose_update(
|
|
changeset_id="second",
|
|
expected_changeset_hash=second["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=foundation_hash,
|
|
metadata={"summary": "Second proposal conflicts with the first."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Prove exact overlap detection.",
|
|
)
|
|
self.assertEqual("proposal_conflict", conflict.exception.code)
|
|
self.assertEqual("first", conflict.exception.details["conflicts"][0]["changeset_id"])
|
|
self.assertEqual(0, store.validate("second")["operation_count"])
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "changed after") as stale_hash:
|
|
store.propose_create(
|
|
changeset_id="first",
|
|
expected_changeset_hash=first["changeset_hash"],
|
|
node_id="guide.another",
|
|
target_source="docs/content/another.md",
|
|
metadata=self.new_metadata(),
|
|
content="Another proposal operation.",
|
|
relationship_changes=[],
|
|
rationale="Use an obsolete changeset hash.",
|
|
)
|
|
self.assertEqual("changeset_conflict", stale_hash.exception.code)
|
|
self.assertEqual(
|
|
first_result["changeset_hash"], store.validate("first")["changeset_hash"]
|
|
)
|
|
|
|
def test_simultaneous_process_appends_serialize_without_losing_an_operation(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
initial = ChangesetStore(project, "alpha-editor").create("simultaneous")
|
|
hashes = {
|
|
node.node_id: node.content_hash
|
|
for node in project.load().nodes
|
|
if node.node_id in {"guide.foundation", "guide.workflow"}
|
|
}
|
|
context = multiprocessing.get_context("spawn")
|
|
start = context.Event()
|
|
receivers: list[Any] = []
|
|
processes: list[multiprocessing.Process] = []
|
|
for node_id in ("guide.foundation", "guide.workflow"):
|
|
receiver, sender = context.Pipe(duplex=False)
|
|
process = context.Process(
|
|
target=_append_in_process,
|
|
args=(
|
|
str(root),
|
|
node_id,
|
|
initial["changeset_hash"],
|
|
hashes[node_id],
|
|
start,
|
|
sender,
|
|
),
|
|
)
|
|
process.start()
|
|
sender.close()
|
|
receivers.append(receiver)
|
|
processes.append(process)
|
|
start.set()
|
|
results = sorted(receiver.recv() for receiver in receivers)
|
|
for receiver in receivers:
|
|
receiver.close()
|
|
for process in processes:
|
|
process.join(timeout=10)
|
|
self.assertEqual(0, process.exitcode)
|
|
|
|
self.assertEqual(["changeset_conflict", "ok"], results)
|
|
validated = ChangesetStore(project).validate("simultaneous")
|
|
self.assertEqual(1, validated["operation_count"])
|
|
|
|
def test_configured_changeset_count_operation_and_size_limits_fail_atomically(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
|
|
count_root = self.copy_fixture(parent / "count")
|
|
descriptor = count_root / ".docforge/project.toml"
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8")
|
|
.replace("max_changesets = 20", "max_changesets = 1")
|
|
.replace("max_changeset_operations = 20", "max_changeset_operations = 1"),
|
|
encoding="utf-8",
|
|
)
|
|
count_project = Project.open(count_root)
|
|
count_store = ChangesetStore(count_project, "alpha-editor")
|
|
limited = count_store.create("limited")
|
|
with self.assertRaisesRegex(DocForgeError, "changeset limit") as count_error:
|
|
count_store.create("excess")
|
|
self.assertEqual("changeset_limit", count_error.exception.code)
|
|
self.assertFalse((count_root / ".docforge/changesets/excess.json").exists())
|
|
|
|
first = count_store.propose_update(
|
|
changeset_id="limited",
|
|
expected_changeset_hash=limited["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=self.node_hash(count_project, "guide.foundation"),
|
|
metadata={"summary": "The only permitted operation."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Reach the configured operation limit.",
|
|
)
|
|
limited_path = count_root / ".docforge/changesets/limited.json"
|
|
limited_bytes = limited_path.read_bytes()
|
|
with self.assertRaisesRegex(DocForgeError, "operation limit") as operation_error:
|
|
count_store.propose_update(
|
|
changeset_id="limited",
|
|
expected_changeset_hash=first["changeset_hash"],
|
|
node_id="guide.workflow",
|
|
expected_content_hash=self.node_hash(count_project, "guide.workflow"),
|
|
metadata={"summary": "This operation exceeds the limit."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Prove the configured operation cap.",
|
|
)
|
|
self.assertEqual("changeset_operation_limit", operation_error.exception.code)
|
|
self.assertEqual(limited_bytes, limited_path.read_bytes())
|
|
|
|
size_root = self.copy_fixture(parent / "size")
|
|
size_descriptor = size_root / ".docforge/project.toml"
|
|
size_descriptor.write_text(
|
|
size_descriptor.read_text(encoding="utf-8").replace(
|
|
"max_changeset_bytes = 100000", "max_changeset_bytes = 64"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
size_store = ChangesetStore(Project.open(size_root), "alpha-editor")
|
|
with self.assertRaisesRegex(DocForgeError, "size limit") as size_error:
|
|
size_store.create("too-large")
|
|
self.assertEqual("changeset_too_large", size_error.exception.code)
|
|
self.assertFalse((size_root / ".docforge/changesets/too-large.json").exists())
|
|
|
|
def test_stale_base_bad_content_hash_and_failed_delete_are_atomic(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
|
|
stale_root = self.copy_fixture(parent / "stale")
|
|
stale_project = Project.open(stale_root)
|
|
stale_store = ChangesetStore(stale_project, "alpha-editor")
|
|
stale = stale_store.create("stale")
|
|
stale_path = stale_root / ".docforge/changesets/stale.json"
|
|
stale_bytes = stale_path.read_bytes()
|
|
workflow = stale_root / "docs/content/workflow.md"
|
|
workflow.write_text(workflow.read_text() + "\nCanonical change.\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(DocForgeError, "Canonical project changed") as base:
|
|
stale_store.propose_move(
|
|
changeset_id="stale",
|
|
expected_changeset_hash=stale["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash=self.node_hash(
|
|
Project.open(stale_root), "guide.foundation"
|
|
),
|
|
target_source="docs/content/foundation-moved.md",
|
|
rationale="This proposal has an obsolete base.",
|
|
)
|
|
self.assertEqual("base_conflict", base.exception.code)
|
|
self.assertEqual(stale_bytes, stale_path.read_bytes())
|
|
self.assertEqual("stale", stale_store.inspect("stale")["base_state"])
|
|
self.assertEqual("stale", stale_store.list_changesets()["changesets"][0]["base_state"])
|
|
|
|
delete_root = self.copy_fixture(parent / "delete")
|
|
delete_project = Project.open(delete_root)
|
|
delete_store = ChangesetStore(delete_project, "alpha-editor")
|
|
delete = delete_store.create("delete")
|
|
delete_path = delete_root / ".docforge/changesets/delete.json"
|
|
delete_bytes = delete_path.read_bytes()
|
|
with self.assertRaisesRegex(DocForgeError, "every incident relationship") as edges:
|
|
delete_store.propose_delete(
|
|
changeset_id="delete",
|
|
expected_changeset_hash=delete["changeset_hash"],
|
|
node_id="proof.validation",
|
|
expected_content_hash=self.node_hash(delete_project, "proof.validation"),
|
|
relationship_changes=[],
|
|
rationale="An incomplete delete must not be stored.",
|
|
)
|
|
self.assertEqual("unresolved_relationships", edges.exception.code)
|
|
self.assertEqual(delete_bytes, delete_path.read_bytes())
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "changed after"):
|
|
delete_store.propose_update(
|
|
changeset_id="delete",
|
|
expected_changeset_hash=delete["changeset_hash"],
|
|
node_id="guide.foundation",
|
|
expected_content_hash="0" * 64,
|
|
metadata={"summary": "Wrong base hash."},
|
|
content=None,
|
|
relationship_changes=[],
|
|
rationale="Reject a mismatched node hash.",
|
|
)
|
|
self.assertEqual(delete_bytes, delete_path.read_bytes())
|
|
|
|
def test_writer_family_permissions_ownership_and_paths_fail_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
descriptor = root / ".docforge/project.toml"
|
|
descriptor.write_text(
|
|
descriptor.read_text().replace(
|
|
'families = ["guide", "proof"]', 'families = ["guide"]'
|
|
)
|
|
+ """
|
|
|
|
[[changesets.writers]]
|
|
id = "other-editor"
|
|
families = ["guide"]
|
|
operations = ["create", "update", "move", "delete"]
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
project = Project.open(root)
|
|
store = ChangesetStore(project, "alpha-editor")
|
|
changeset = store.create("permissions")
|
|
path = root / ".docforge/changesets/permissions.json"
|
|
before = path.read_bytes()
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "families") as family:
|
|
store.propose_create(
|
|
changeset_id="permissions",
|
|
expected_changeset_hash=changeset["changeset_hash"],
|
|
node_id="proof.new",
|
|
target_source="docs/content/proof-new.md",
|
|
metadata=self.new_metadata(family="proof"),
|
|
content="A forbidden family proposal.",
|
|
relationship_changes=[],
|
|
rationale="Prove family permission enforcement.",
|
|
)
|
|
self.assertEqual("family_forbidden", family.exception.code)
|
|
self.assertEqual(before, path.read_bytes())
|
|
|
|
other = ChangesetStore(project, "other-editor")
|
|
with self.assertRaisesRegex(DocForgeError, "does not own") as owner:
|
|
other.propose_create(
|
|
changeset_id="permissions",
|
|
expected_changeset_hash=changeset["changeset_hash"],
|
|
node_id="guide.other",
|
|
target_source="docs/content/other.md",
|
|
metadata=self.new_metadata(),
|
|
content="Another writer cannot append here.",
|
|
relationship_changes=[],
|
|
rationale="Prove immutable changeset ownership.",
|
|
)
|
|
self.assertEqual("changeset_owner_conflict", owner.exception.code)
|
|
|
|
with self.assertRaisesRegex(DocForgeError, "canonical content root"):
|
|
store.propose_create(
|
|
changeset_id="permissions",
|
|
expected_changeset_hash=changeset["changeset_hash"],
|
|
node_id="guide.escape",
|
|
target_source="outside.md",
|
|
metadata=self.new_metadata(),
|
|
content="This path is outside the content root.",
|
|
relationship_changes=[],
|
|
rationale="Prove target confinement.",
|
|
)
|
|
|
|
outside = Path(directory) / "outside.json"
|
|
outside.write_text(json.dumps({"secret": True}), encoding="utf-8")
|
|
symlink = root / ".docforge/changesets/symlink.json"
|
|
symlink.symlink_to(outside)
|
|
with self.assertRaisesRegex(DocForgeError, "symbolic links"):
|
|
store.validate("symlink")
|
|
self.assertEqual({"secret": True}, json.loads(outside.read_text()))
|
|
|
|
def test_changeset_configuration_overlap_and_invalid_permissions_are_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
overlap_root = self.copy_fixture(parent / "overlap")
|
|
overlap = overlap_root / ".docforge/project.toml"
|
|
overlap.write_text(
|
|
overlap.read_text().replace(
|
|
'root = ".docforge/changesets"', 'root = "docs/content/changesets"'
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "must not overlap"):
|
|
Project.open(overlap_root)
|
|
|
|
invalid_root = self.copy_fixture(parent / "invalid")
|
|
invalid = invalid_root / ".docforge/project.toml"
|
|
invalid.write_text(
|
|
invalid.read_text().replace(
|
|
'operations = ["create", "update", "move", "delete"]',
|
|
'operations = ["create", "execute"]',
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "operations are empty or invalid"):
|
|
Project.open(invalid_root)
|
|
|
|
symlink_root = self.copy_fixture(parent / "symlink")
|
|
outside = parent / "outside-changesets"
|
|
outside.mkdir()
|
|
(symlink_root / ".docforge/changesets").symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaisesRegex(DocForgeError, "outside the project root"):
|
|
Project.open(symlink_root)
|
|
|
|
def test_toml_anchors_and_mid_write_source_set_changes_fail_atomically(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
|
|
toml_root = self.copy_fixture(parent / "toml")
|
|
toml_store = ChangesetStore(Project.open(toml_root), "alpha-editor")
|
|
changeset = toml_store.create("toml-create")
|
|
path = toml_root / ".docforge/changesets/toml-create.json"
|
|
before = path.read_bytes()
|
|
with self.assertRaisesRegex(DocForgeError, "explicit source_anchor"):
|
|
toml_store.propose_create(
|
|
changeset_id="toml-create",
|
|
expected_changeset_hash=changeset["changeset_hash"],
|
|
node_id="proof.second",
|
|
target_source="docs/content/proof.toml",
|
|
metadata=self.new_metadata(family="proof"),
|
|
content="A TOML proposal needs an explicit anchor.",
|
|
relationship_changes=[],
|
|
rationale="Reject ambiguous TOML placement.",
|
|
)
|
|
self.assertEqual(before, path.read_bytes())
|
|
|
|
duplicate_anchor = {
|
|
**self.new_metadata(family="proof"),
|
|
"source_anchor": "node-1",
|
|
}
|
|
with self.assertRaisesRegex(DocForgeError, "anchors must be unique"):
|
|
toml_store.propose_create(
|
|
changeset_id="toml-create",
|
|
expected_changeset_hash=changeset["changeset_hash"],
|
|
node_id="proof.second",
|
|
target_source="docs/content/proof.toml",
|
|
metadata=duplicate_anchor,
|
|
content="A duplicate anchor is still ambiguous.",
|
|
relationship_changes=[],
|
|
rationale="Reject duplicate TOML anchors.",
|
|
)
|
|
self.assertEqual(before, path.read_bytes())
|
|
|
|
race_root = self.copy_fixture(parent / "race")
|
|
race_project = Project.open(race_root)
|
|
race_store = ChangesetStore(race_project, "alpha-editor")
|
|
sources = race_project.canonical_source_paths()
|
|
invented = race_root / "docs/content/invented.md"
|
|
with (
|
|
mock.patch.object(
|
|
race_project,
|
|
"canonical_source_paths",
|
|
side_effect=[sources, (*sources, invented)],
|
|
),
|
|
self.assertRaisesRegex(DocForgeError, "changed during changeset storage"),
|
|
):
|
|
race_store.create("source-race")
|
|
self.assertFalse((race_root / ".docforge/changesets/source-race.json").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|