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_at raced = False def race(directory_fd: int, first: str, 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(directory_fd, first, second) with ( mock.patch( "docforge.application.rename_exchange_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_at deleted_race = False def race_delete(directory_fd: int, first: str, 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(directory_fd, first, second) with ( mock.patch( "docforge.application.rename_exchange_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_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(0o640) 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"]) self.assertEqual(0o640, stat.S_IMODE(target.stat().st_mode)) self.assertEqual([], result["retained_recovery_files"]) 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()