994 lines
43 KiB
Python
994 lines
43 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import contextlib
|
||
|
|
import io
|
||
|
|
import json
|
||
|
|
import shutil
|
||
|
|
import sqlite3
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from collections.abc import Callable
|
||
|
|
from dataclasses import replace
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest import mock
|
||
|
|
|
||
|
|
from jsonschema import Draft202012Validator
|
||
|
|
from mcp.shared.memory import create_connected_server_and_client_session
|
||
|
|
|
||
|
|
import docforge.generation_diff as generation_diff_module
|
||
|
|
from docforge.cli import main
|
||
|
|
from docforge.errors import DocForgeError
|
||
|
|
from docforge.generation_diff import (
|
||
|
|
GenerationDiffDraft,
|
||
|
|
finalize_generation_diff,
|
||
|
|
generation_diff_path,
|
||
|
|
publish_generation_diff,
|
||
|
|
validate_generation_diff_receipt,
|
||
|
|
)
|
||
|
|
from docforge.index import ProjectIndex
|
||
|
|
from docforge.mcp_server import DocForgeService, create_server
|
||
|
|
from docforge.models import ProjectDescriptor, ProjectSnapshot
|
||
|
|
from docforge.pagination import canonical_hash
|
||
|
|
from docforge.project import Project, project_root_fingerprint
|
||
|
|
from docforge.telemetry import request
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
FIXTURES = ROOT / "tests" / "fixtures"
|
||
|
|
GENERATION_DIFF_SCHEMA = json.loads(
|
||
|
|
(ROOT / "schemas" / "generation-diff.schema.json").read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
GENERATION_DIFF_PAGE_SCHEMA = json.loads(
|
||
|
|
(ROOT / "schemas" / "generation-diff-page.schema.json").read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
RESULT_SCHEMA = json.loads((ROOT / "schemas" / "result.schema.json").read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
|
||
|
|
class StaticProject:
|
||
|
|
"""Small legacy one-method project used to prove load-free status behavior."""
|
||
|
|
|
||
|
|
def __init__(self, snapshot: ProjectSnapshot) -> None:
|
||
|
|
self.descriptor: ProjectDescriptor = snapshot.descriptor
|
||
|
|
self.snapshot = snapshot
|
||
|
|
self.load_calls = 0
|
||
|
|
|
||
|
|
def load(self) -> ProjectSnapshot:
|
||
|
|
self.load_calls += 1
|
||
|
|
return self.snapshot
|
||
|
|
|
||
|
|
def canonical_source_paths(self) -> tuple[Path, ...]:
|
||
|
|
return ()
|
||
|
|
|
||
|
|
def validate_proposal(
|
||
|
|
self,
|
||
|
|
base: ProjectSnapshot,
|
||
|
|
projected: ProjectSnapshot,
|
||
|
|
operations: tuple[object, ...],
|
||
|
|
) -> None:
|
||
|
|
del base, projected, operations
|
||
|
|
|
||
|
|
|
||
|
|
class GenerationDiffTests(unittest.TestCase):
|
||
|
|
def copy_fixture(self, destination: Path) -> Path:
|
||
|
|
root = destination / "alpha"
|
||
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
||
|
|
shutil.rmtree(root / ".docforge" / "cache", ignore_errors=True)
|
||
|
|
return root
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def change_graph(root: Path, suffix: str = "changed") -> None:
|
||
|
|
foundation = root / "docs" / "content" / "foundation.md"
|
||
|
|
foundation.write_text(
|
||
|
|
foundation.read_text(encoding="utf-8").replace(
|
||
|
|
"Defines which Alpha files own documentation facts.",
|
||
|
|
f"Defines which Alpha files own documentation facts. {suffix}",
|
||
|
|
),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
proof = root / "docs" / "content" / "proof.toml"
|
||
|
|
proof.write_text(
|
||
|
|
proof.read_text(encoding="utf-8").replace(
|
||
|
|
'proves = ["guide.workflow"]',
|
||
|
|
'proves = ["guide.foundation"]',
|
||
|
|
),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def rehash_receipt(receipt: dict[str, object]) -> dict[str, object]:
|
||
|
|
items = receipt["items"]
|
||
|
|
if not isinstance(items, list):
|
||
|
|
raise AssertionError("receipt items are not a list")
|
||
|
|
for item in items:
|
||
|
|
if not isinstance(item, dict):
|
||
|
|
raise AssertionError("receipt item is not an object")
|
||
|
|
item["item_hash"] = canonical_hash(
|
||
|
|
{key: value for key, value in item.items() if key != "item_hash"}
|
||
|
|
)
|
||
|
|
retained_hash = canonical_hash([item["item_hash"] for item in items])
|
||
|
|
receipt["retained_collection_hash"] = retained_hash
|
||
|
|
if not receipt["details_truncated"]:
|
||
|
|
receipt["full_collection_hash"] = retained_hash
|
||
|
|
receipt["receipt_hash"] = canonical_hash(
|
||
|
|
{key: value for key, value in receipt.items() if key != "receipt_hash"}
|
||
|
|
)
|
||
|
|
return receipt
|
||
|
|
|
||
|
|
def transition_receipt(self, root: Path) -> tuple[Project, ProjectIndex, dict[str, object]]:
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
index.build()
|
||
|
|
receipt = json.loads(generation_diff_path(project.descriptor).read_text(encoding="utf-8"))
|
||
|
|
return project, index, receipt
|
||
|
|
|
||
|
|
def test_first_build_and_exact_transition_are_schema_valid(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
|
||
|
|
first = index.build()
|
||
|
|
first_receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
Draft202012Validator(GENERATION_DIFF_SCHEMA).validate(first_receipt)
|
||
|
|
self.assertTrue(
|
||
|
|
validate_generation_diff_receipt(
|
||
|
|
first_receipt,
|
||
|
|
descriptor=project.descriptor,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
self.assertEqual("baseline", first_receipt["kind"])
|
||
|
|
self.assertEqual("no_predecessor", first_receipt["reason"])
|
||
|
|
self.assertEqual(0, first_receipt["full_item_count"])
|
||
|
|
self.assertEqual("ok", first["status"])
|
||
|
|
|
||
|
|
self.change_graph(root)
|
||
|
|
second = index.build()
|
||
|
|
receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
Draft202012Validator(GENERATION_DIFF_SCHEMA).validate(receipt)
|
||
|
|
self.assertEqual("transition", receipt["kind"])
|
||
|
|
self.assertEqual(
|
||
|
|
{
|
||
|
|
"nodes_added": 0,
|
||
|
|
"nodes_removed": 0,
|
||
|
|
"nodes_changed": 2,
|
||
|
|
"edges_added": 1,
|
||
|
|
"edges_removed": 1,
|
||
|
|
"total_changes": 4,
|
||
|
|
},
|
||
|
|
receipt["summary"],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
[
|
||
|
|
("edge", "added"),
|
||
|
|
("edge", "removed"),
|
||
|
|
("node", "changed"),
|
||
|
|
("node", "changed"),
|
||
|
|
],
|
||
|
|
[(item["entity"], item["change"]) for item in receipt["items"]],
|
||
|
|
)
|
||
|
|
self.assertNotIn("logic_hash", json.dumps(receipt, sort_keys=True).casefold())
|
||
|
|
self.assertEqual("ok", second["status"])
|
||
|
|
|
||
|
|
status = ProjectIndex(project).generation_diff()
|
||
|
|
self.assertEqual("current", status["receipt_state"])
|
||
|
|
self.assertEqual(receipt["receipt_hash"], status["generation_diff"]["receipt_hash"])
|
||
|
|
|
||
|
|
def test_same_generation_reindex_preserves_latest_meaningful_transition(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
index.build()
|
||
|
|
before = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
|
||
|
|
index.build()
|
||
|
|
after = json.loads(generation_diff_path(project.descriptor).read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
self.assertEqual("transition", after["kind"])
|
||
|
|
self.assertEqual(before["from_generation"], after["from_generation"])
|
||
|
|
self.assertEqual(before["to_generation"], after["to_generation"])
|
||
|
|
self.assertEqual(before["summary"], after["summary"])
|
||
|
|
self.assertEqual(before["full_collection_hash"], after["full_collection_hash"])
|
||
|
|
self.assertNotEqual(before["index_signature"], after["index_signature"])
|
||
|
|
self.assertEqual(
|
||
|
|
["generation-diff.json"],
|
||
|
|
[path.name for path in project.descriptor.cache_root.glob("*generation-diff*")],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_receipt_item_and_byte_limits_preserve_exact_summary_hashes(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
descriptor = Project.open(root).descriptor
|
||
|
|
generation = {
|
||
|
|
"revision": "unversioned",
|
||
|
|
"source_hash": "1" * 64,
|
||
|
|
"node_count": 0,
|
||
|
|
"node_hash": "2" * 64,
|
||
|
|
"edge_count": 0,
|
||
|
|
"edge_hash": "3" * 64,
|
||
|
|
"index_schema_version": 3,
|
||
|
|
}
|
||
|
|
items: list[dict[str, object]] = []
|
||
|
|
for index in range(1_002):
|
||
|
|
item: dict[str, object] = {
|
||
|
|
"entity": "edge",
|
||
|
|
"change": "added",
|
||
|
|
"source_id": f"source-{index:04d}",
|
||
|
|
"relation": "relates_to",
|
||
|
|
"target_id": f"target-{index:04d}",
|
||
|
|
}
|
||
|
|
item["item_hash"] = canonical_hash(item)
|
||
|
|
items.append(item)
|
||
|
|
fields: dict[str, object] = {
|
||
|
|
"schema_version": 1,
|
||
|
|
"diff_semantics_version": 1,
|
||
|
|
"project_id": descriptor.project_id,
|
||
|
|
"project_root_fingerprint": project_root_fingerprint(descriptor.root),
|
||
|
|
"adapter": descriptor.adapter,
|
||
|
|
"kind": "transition",
|
||
|
|
"reason": None,
|
||
|
|
"from_generation": generation,
|
||
|
|
"to_generation": {**generation, "source_hash": "4" * 64},
|
||
|
|
"summary": {
|
||
|
|
"nodes_added": 0,
|
||
|
|
"nodes_removed": 0,
|
||
|
|
"nodes_changed": 0,
|
||
|
|
"edges_added": len(items),
|
||
|
|
"edges_removed": 0,
|
||
|
|
"total_changes": len(items),
|
||
|
|
},
|
||
|
|
"full_item_count": len(items),
|
||
|
|
"full_collection_hash": canonical_hash([item["item_hash"] for item in items]),
|
||
|
|
}
|
||
|
|
receipt = finalize_generation_diff(
|
||
|
|
GenerationDiffDraft(fields=fields, items=tuple(items)),
|
||
|
|
signature=(1, 2, 3, 4, 5),
|
||
|
|
)
|
||
|
|
Draft202012Validator(GENERATION_DIFF_SCHEMA).validate(receipt)
|
||
|
|
self.assertEqual(1_002, receipt["full_item_count"])
|
||
|
|
self.assertEqual(1_000, receipt["retained_item_count"])
|
||
|
|
self.assertEqual("receipt_item_limit", receipt["truncation_reason"])
|
||
|
|
self.assertNotEqual(
|
||
|
|
receipt["full_collection_hash"],
|
||
|
|
receipt["retained_collection_hash"],
|
||
|
|
)
|
||
|
|
|
||
|
|
huge = dict(items[0])
|
||
|
|
huge["source_id"] = "source-" + ("x" * 1_100_000)
|
||
|
|
huge["item_hash"] = canonical_hash(
|
||
|
|
{key: value for key, value in huge.items() if key != "item_hash"}
|
||
|
|
)
|
||
|
|
huge_fields = {
|
||
|
|
**fields,
|
||
|
|
"summary": {
|
||
|
|
**fields["summary"],
|
||
|
|
"edges_added": 1,
|
||
|
|
"total_changes": 1,
|
||
|
|
},
|
||
|
|
"full_item_count": 1,
|
||
|
|
"full_collection_hash": canonical_hash([huge["item_hash"]]),
|
||
|
|
}
|
||
|
|
byte_limited = finalize_generation_diff(
|
||
|
|
GenerationDiffDraft(fields=huge_fields, items=(huge,)),
|
||
|
|
signature=(1, 2, 3, 4, 5),
|
||
|
|
)
|
||
|
|
self.assertEqual(0, byte_limited["retained_item_count"])
|
||
|
|
self.assertEqual("receipt_byte_limit", byte_limited["truncation_reason"])
|
||
|
|
|
||
|
|
def test_receipt_runtime_and_schema_reject_malformed_semantics(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project, _, receipt = self.transition_receipt(root)
|
||
|
|
validator = Draft202012Validator(GENERATION_DIFF_SCHEMA)
|
||
|
|
|
||
|
|
malformed_fields = json.loads(json.dumps(receipt))
|
||
|
|
changed = next(
|
||
|
|
item
|
||
|
|
for item in malformed_fields["items"]
|
||
|
|
if item["entity"] == "node" and item["change"] == "changed"
|
||
|
|
)
|
||
|
|
changed["changed_fields"] = 1
|
||
|
|
self.rehash_receipt(malformed_fields)
|
||
|
|
self.assertFalse(
|
||
|
|
validate_generation_diff_receipt(
|
||
|
|
malformed_fields,
|
||
|
|
descriptor=project.descriptor,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
self.assertFalse(validator.is_valid(malformed_fields))
|
||
|
|
|
||
|
|
duplicate_fields = json.loads(json.dumps(receipt))
|
||
|
|
changed = next(
|
||
|
|
item
|
||
|
|
for item in duplicate_fields["items"]
|
||
|
|
if item["entity"] == "node" and item["change"] == "changed"
|
||
|
|
)
|
||
|
|
changed["changed_fields"] = ["content", "content"]
|
||
|
|
self.rehash_receipt(duplicate_fields)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(duplicate_fields))
|
||
|
|
self.assertFalse(validator.is_valid(duplicate_fields))
|
||
|
|
|
||
|
|
impossible_added = json.loads(json.dumps(receipt))
|
||
|
|
changed = next(
|
||
|
|
item
|
||
|
|
for item in impossible_added["items"]
|
||
|
|
if item["entity"] == "node" and item["change"] == "changed"
|
||
|
|
)
|
||
|
|
changed["change"] = "added"
|
||
|
|
changed["changed_fields"] = []
|
||
|
|
self.rehash_receipt(impossible_added)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(impossible_added))
|
||
|
|
self.assertFalse(validator.is_valid(impossible_added))
|
||
|
|
|
||
|
|
reordered = json.loads(json.dumps(receipt))
|
||
|
|
reordered["items"].reverse()
|
||
|
|
self.rehash_receipt(reordered)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(reordered))
|
||
|
|
|
||
|
|
same_generation = json.loads(json.dumps(receipt))
|
||
|
|
same_generation["from_generation"] = same_generation["to_generation"]
|
||
|
|
self.rehash_receipt(same_generation)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(same_generation))
|
||
|
|
|
||
|
|
impossible_truncation = json.loads(json.dumps(receipt))
|
||
|
|
impossible_truncation["items"].pop()
|
||
|
|
impossible_truncation["retained_item_count"] = len(impossible_truncation["items"])
|
||
|
|
impossible_truncation["details_truncated"] = True
|
||
|
|
impossible_truncation["truncation_reason"] = "receipt_item_limit"
|
||
|
|
self.rehash_receipt(impossible_truncation)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(impossible_truncation))
|
||
|
|
self.assertFalse(validator.is_valid(impossible_truncation))
|
||
|
|
|
||
|
|
baseline_with_changes = json.loads(json.dumps(receipt))
|
||
|
|
baseline_with_changes["kind"] = "baseline"
|
||
|
|
baseline_with_changes["reason"] = "no_meaningful_transition"
|
||
|
|
baseline_with_changes["from_generation"] = None
|
||
|
|
self.rehash_receipt(baseline_with_changes)
|
||
|
|
self.assertFalse(validate_generation_diff_receipt(baseline_with_changes))
|
||
|
|
self.assertFalse(validator.is_valid(baseline_with_changes))
|
||
|
|
|
||
|
|
def test_generation_collision_and_nondeterministic_build_fail_precommit(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
base = Project.open(root).load()
|
||
|
|
static = StaticProject(base)
|
||
|
|
index = ProjectIndex(static)
|
||
|
|
index.build()
|
||
|
|
original_index = index.path.read_bytes()
|
||
|
|
original_receipt = generation_diff_path(static.descriptor).read_bytes()
|
||
|
|
|
||
|
|
changed_node = replace(
|
||
|
|
base.nodes[0],
|
||
|
|
title="Different graph under the same generation",
|
||
|
|
)
|
||
|
|
static.snapshot = replace(
|
||
|
|
base,
|
||
|
|
nodes=(changed_node, *base.nodes[1:]),
|
||
|
|
)
|
||
|
|
with self.assertRaises(DocForgeError) as collision:
|
||
|
|
index.build()
|
||
|
|
self.assertEqual("generation_collision", collision.exception.code)
|
||
|
|
self.assertEqual(original_index, index.path.read_bytes())
|
||
|
|
self.assertEqual(
|
||
|
|
original_receipt,
|
||
|
|
generation_diff_path(static.descriptor).read_bytes(),
|
||
|
|
)
|
||
|
|
|
||
|
|
calls = 0
|
||
|
|
|
||
|
|
def unstable() -> ProjectSnapshot:
|
||
|
|
nonlocal calls
|
||
|
|
calls += 1
|
||
|
|
return (
|
||
|
|
base
|
||
|
|
if calls == 1
|
||
|
|
else replace(
|
||
|
|
base,
|
||
|
|
nodes=(changed_node, *base.nodes[1:]),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
static.snapshot = base
|
||
|
|
with (
|
||
|
|
mock.patch.object(static, "load", side_effect=unstable),
|
||
|
|
self.assertRaises(DocForgeError) as changed,
|
||
|
|
):
|
||
|
|
ProjectIndex(static).build()
|
||
|
|
self.assertEqual("source_changed", changed.exception.code)
|
||
|
|
|
||
|
|
def test_post_commit_receipt_failures_report_degraded_success(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
|
||
|
|
with (
|
||
|
|
mock.patch.object(
|
||
|
|
ProjectIndex,
|
||
|
|
"_write_attestation",
|
||
|
|
side_effect=OSError("attestation failed"),
|
||
|
|
),
|
||
|
|
mock.patch.object(
|
||
|
|
Project,
|
||
|
|
"record_generation",
|
||
|
|
side_effect=OSError("generation failed"),
|
||
|
|
),
|
||
|
|
mock.patch(
|
||
|
|
"docforge.index.publish_generation_diff",
|
||
|
|
side_effect=OSError("diff failed"),
|
||
|
|
),
|
||
|
|
):
|
||
|
|
result = index.build()
|
||
|
|
|
||
|
|
self.assertEqual("ok", result["status"])
|
||
|
|
self.assertEqual("degraded", result["publication"]["state"])
|
||
|
|
self.assertEqual("published", result["publication"]["index"])
|
||
|
|
self.assertEqual(
|
||
|
|
{"attestation", "source_generation", "generation_diff"},
|
||
|
|
{error["stage"] for error in result["publication"]["errors"]},
|
||
|
|
)
|
||
|
|
with contextlib.closing(__import__("sqlite3").connect(index.path)) as connection:
|
||
|
|
metadata = dict(connection.execute("SELECT key, value FROM metadata"))
|
||
|
|
self.assertEqual(project.load().source_hash, metadata["source_hash"])
|
||
|
|
|
||
|
|
def test_predecessor_requires_attestation_and_valid_generation_identity(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
index.attestation_path.unlink()
|
||
|
|
self.change_graph(root)
|
||
|
|
result = index.build()
|
||
|
|
receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
self.assertEqual("ok", result["status"])
|
||
|
|
self.assertEqual("baseline", receipt["kind"])
|
||
|
|
self.assertEqual("predecessor_unattested", receipt["reason"])
|
||
|
|
self.assertTrue(validate_generation_diff_receipt(receipt))
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
with contextlib.closing(sqlite3.connect(index.path)) as connection:
|
||
|
|
connection.execute("UPDATE metadata SET value = 'bad' WHERE key = 'source_hash'")
|
||
|
|
connection.commit()
|
||
|
|
index._write_attestation()
|
||
|
|
self.change_graph(root)
|
||
|
|
index.build()
|
||
|
|
receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
self.assertEqual("baseline", receipt["kind"])
|
||
|
|
self.assertEqual("predecessor_corrupt", receipt["reason"])
|
||
|
|
self.assertTrue(validate_generation_diff_receipt(receipt))
|
||
|
|
|
||
|
|
def test_sqlite_sidecars_refuse_precommit_publication(self) -> None:
|
||
|
|
for suffix in ("-wal", "-journal", "-shm"):
|
||
|
|
with self.subTest(suffix=suffix), tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
original_index = index.path.read_bytes()
|
||
|
|
original_receipt = generation_diff_path(project.descriptor).read_bytes()
|
||
|
|
Path(f"{index.path}{suffix}").write_bytes(b"unproven-sidecar")
|
||
|
|
self.change_graph(root)
|
||
|
|
with self.assertRaises(DocForgeError) as blocked:
|
||
|
|
index.build()
|
||
|
|
self.assertEqual("index_busy", blocked.exception.code)
|
||
|
|
self.assertEqual(original_index, index.path.read_bytes())
|
||
|
|
self.assertEqual(
|
||
|
|
original_receipt,
|
||
|
|
generation_diff_path(project.descriptor).read_bytes(),
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_live_wal_state_cannot_bypass_main_index_identity(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
original_receipt = generation_diff_path(project.descriptor).read_bytes()
|
||
|
|
connection = sqlite3.connect(index.path)
|
||
|
|
try:
|
||
|
|
self.assertEqual(
|
||
|
|
"wal",
|
||
|
|
connection.execute("PRAGMA journal_mode=WAL").fetchone()[0],
|
||
|
|
)
|
||
|
|
connection.execute(
|
||
|
|
"UPDATE metadata SET value = ? WHERE key = 'source_hash'",
|
||
|
|
("f" * 64,),
|
||
|
|
)
|
||
|
|
connection.commit()
|
||
|
|
self.assertTrue(Path(f"{index.path}-wal").exists())
|
||
|
|
main_file_after_wal = index.path.read_bytes()
|
||
|
|
with self.assertRaises(DocForgeError) as blocked:
|
||
|
|
index.build()
|
||
|
|
self.assertEqual("index_busy", blocked.exception.code)
|
||
|
|
self.assertEqual(main_file_after_wal, index.path.read_bytes())
|
||
|
|
self.assertEqual(
|
||
|
|
original_receipt,
|
||
|
|
generation_diff_path(project.descriptor).read_bytes(),
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
|
||
|
|
def test_sidecar_or_source_change_during_diff_preparation_aborts_precommit(self) -> None:
|
||
|
|
for mutation in ("source", "sidecar"):
|
||
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
original_index = index.path.read_bytes()
|
||
|
|
original_receipt = generation_diff_path(project.descriptor).read_bytes()
|
||
|
|
self.change_graph(root)
|
||
|
|
from docforge import index as index_module
|
||
|
|
|
||
|
|
real_prepare = index_module.prepare_generation_diff
|
||
|
|
|
||
|
|
def mutate_after_prepare(
|
||
|
|
*args: object,
|
||
|
|
_prepare: Callable[..., object] = real_prepare,
|
||
|
|
_mutation: str = mutation,
|
||
|
|
_root: Path = root,
|
||
|
|
_index: ProjectIndex = index,
|
||
|
|
**kwargs: object,
|
||
|
|
) -> object:
|
||
|
|
draft = _prepare(*args, **kwargs)
|
||
|
|
if _mutation == "source":
|
||
|
|
source = _root / "docs" / "content" / "workflow.md"
|
||
|
|
source.write_text(
|
||
|
|
source.read_text(encoding="utf-8") + "\nConcurrent change.\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
Path(f"{_index.path}-wal").write_bytes(b"appeared")
|
||
|
|
return draft
|
||
|
|
|
||
|
|
with (
|
||
|
|
mock.patch(
|
||
|
|
"docforge.index.prepare_generation_diff",
|
||
|
|
side_effect=mutate_after_prepare,
|
||
|
|
),
|
||
|
|
self.assertRaises(DocForgeError) as blocked,
|
||
|
|
):
|
||
|
|
index.build()
|
||
|
|
self.assertEqual(
|
||
|
|
"source_changed" if mutation == "source" else "index_busy",
|
||
|
|
blocked.exception.code,
|
||
|
|
)
|
||
|
|
self.assertEqual(original_index, index.path.read_bytes())
|
||
|
|
self.assertEqual(
|
||
|
|
original_receipt,
|
||
|
|
generation_diff_path(project.descriptor).read_bytes(),
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_cache_root_symlink_cannot_redirect_receipt_publication(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
cache_root = project.descriptor.cache_root
|
||
|
|
preserved = cache_root.with_name("preserved-cache")
|
||
|
|
outside = root / "outside-cache"
|
||
|
|
outside.mkdir()
|
||
|
|
cache_root.rename(preserved)
|
||
|
|
cache_root.symlink_to(outside, target_is_directory=True)
|
||
|
|
try:
|
||
|
|
with self.assertRaises(DocForgeError) as blocked:
|
||
|
|
publish_generation_diff(project.descriptor, receipt)
|
||
|
|
self.assertEqual("path_escape", blocked.exception.code)
|
||
|
|
self.assertFalse((outside / "generation-diff.json").exists())
|
||
|
|
finally:
|
||
|
|
cache_root.unlink()
|
||
|
|
preserved.rename(cache_root)
|
||
|
|
|
||
|
|
def test_post_commit_identity_and_durability_failures_are_degraded(self) -> None:
|
||
|
|
cases = (
|
||
|
|
("_fsync_cache_directory", OSError("fsync failed"), "index_directory_sync"),
|
||
|
|
(
|
||
|
|
"_published_index_signature",
|
||
|
|
OSError("signature failed"),
|
||
|
|
"index_identity",
|
||
|
|
),
|
||
|
|
)
|
||
|
|
for method, failure, stage in cases:
|
||
|
|
with self.subTest(method=method), tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
with mock.patch.object(ProjectIndex, method, side_effect=failure):
|
||
|
|
result = index.build()
|
||
|
|
self.assertEqual("ok", result["status"])
|
||
|
|
self.assertEqual("degraded", result["publication"]["state"])
|
||
|
|
self.assertIn(
|
||
|
|
stage,
|
||
|
|
{error["stage"] for error in result["publication"]["errors"]},
|
||
|
|
)
|
||
|
|
with contextlib.closing(sqlite3.connect(index.path)) as connection:
|
||
|
|
metadata = dict(connection.execute("SELECT key, value FROM metadata"))
|
||
|
|
self.assertEqual(project.load().source_hash, metadata["source_hash"])
|
||
|
|
|
||
|
|
def test_post_commit_receipts_fail_independently(self) -> None:
|
||
|
|
cases = (
|
||
|
|
("attestation", "docforge.index.ProjectIndex._write_attestation"),
|
||
|
|
("source_generation", "docforge.project.Project.record_generation"),
|
||
|
|
("generation_diff", "docforge.index.publish_generation_diff"),
|
||
|
|
)
|
||
|
|
for failed_receipt, target in cases:
|
||
|
|
with self.subTest(receipt=failed_receipt), tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
with mock.patch(target, side_effect=OSError("failed independently")):
|
||
|
|
result = index.build()
|
||
|
|
receipts = result["publication"]["receipts"]
|
||
|
|
self.assertEqual("unavailable", receipts[failed_receipt]["state"])
|
||
|
|
for name in {"attestation", "source_generation", "generation_diff"} - {
|
||
|
|
failed_receipt
|
||
|
|
}:
|
||
|
|
self.assertEqual("published", receipts[name]["state"])
|
||
|
|
|
||
|
|
def test_read_is_bounded_read_only_and_legacy_projects_do_not_load(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
receipt_path = generation_diff_path(project.descriptor)
|
||
|
|
original_receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
|
||
|
|
before = {
|
||
|
|
path: path.stat().st_mtime_ns
|
||
|
|
for path in project.descriptor.cache_root.iterdir()
|
||
|
|
if path.is_file()
|
||
|
|
}
|
||
|
|
with (
|
||
|
|
request("test", enabled=True) as collector,
|
||
|
|
mock.patch.object(project, "load", side_effect=AssertionError("loaded")),
|
||
|
|
mock.patch.object(index, "check", side_effect=AssertionError("checked")),
|
||
|
|
mock.patch.object(index, "build", side_effect=AssertionError("built")),
|
||
|
|
mock.patch.object(index, "synchronize", side_effect=AssertionError("synced")),
|
||
|
|
):
|
||
|
|
result = index.generation_diff()
|
||
|
|
self.assertEqual("current", result["receipt_state"])
|
||
|
|
self.assertIsNotNone(collector)
|
||
|
|
diagnostics = collector.as_dict(outcome="ok")
|
||
|
|
for counter in (
|
||
|
|
"project_loads",
|
||
|
|
"source_files_parsed",
|
||
|
|
"adapter_projection_loads",
|
||
|
|
"adapter_source_extractions",
|
||
|
|
"index_checks",
|
||
|
|
"index_synchronizations",
|
||
|
|
"index_builds",
|
||
|
|
):
|
||
|
|
self.assertEqual(0, diagnostics["counters"][counter])
|
||
|
|
self.assertGreaterEqual(diagnostics["counters"]["source_generation_checks"], 2)
|
||
|
|
self.assertEqual(
|
||
|
|
before,
|
||
|
|
{
|
||
|
|
path: path.stat().st_mtime_ns
|
||
|
|
for path in project.descriptor.cache_root.iterdir()
|
||
|
|
if path.is_file()
|
||
|
|
},
|
||
|
|
)
|
||
|
|
with mock.patch.object(
|
||
|
|
generation_diff_module,
|
||
|
|
"validate_generation_diff_receipt",
|
||
|
|
wraps=generation_diff_module.validate_generation_diff_receipt,
|
||
|
|
) as validated:
|
||
|
|
self.assertEqual("current", index.generation_diff()["receipt_state"])
|
||
|
|
self.assertEqual(1, validated.call_count)
|
||
|
|
|
||
|
|
receipt_path.write_text("{broken", encoding="utf-8")
|
||
|
|
corrupt_before = receipt_path.read_bytes()
|
||
|
|
corrupt = index.generation_diff()
|
||
|
|
self.assertEqual("unverified", corrupt["receipt_state"])
|
||
|
|
self.assertEqual("corrupt_receipt", corrupt["receipt_reason"])
|
||
|
|
self.assertEqual(corrupt_before, receipt_path.read_bytes())
|
||
|
|
|
||
|
|
foreign = {**original_receipt, "project_id": "foreign-project"}
|
||
|
|
foreign["receipt_hash"] = canonical_hash(
|
||
|
|
{key: value for key, value in foreign.items() if key != "receipt_hash"}
|
||
|
|
)
|
||
|
|
receipt_path.write_text(
|
||
|
|
json.dumps(foreign, sort_keys=True, indent=2) + "\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
foreign_result = index.generation_diff()
|
||
|
|
self.assertEqual("unverified", foreign_result["receipt_state"])
|
||
|
|
self.assertEqual("foreign_receipt", foreign_result["receipt_reason"])
|
||
|
|
|
||
|
|
outside = root / "foreign-generation-diff.json"
|
||
|
|
outside.write_text(
|
||
|
|
json.dumps(original_receipt, sort_keys=True, indent=2) + "\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
receipt_path.unlink()
|
||
|
|
receipt_path.symlink_to(outside)
|
||
|
|
unsafe = index.generation_diff()
|
||
|
|
self.assertEqual("unsafe", unsafe["receipt_state"])
|
||
|
|
self.assertEqual("unsafe_receipt", unsafe["receipt_reason"])
|
||
|
|
|
||
|
|
snapshot = project.load()
|
||
|
|
legacy = StaticProject(snapshot)
|
||
|
|
calls = legacy.load_calls
|
||
|
|
unknown = ProjectIndex(legacy).generation_diff()
|
||
|
|
self.assertEqual("unknown", unknown["receipt_state"])
|
||
|
|
self.assertEqual(calls, legacy.load_calls)
|
||
|
|
|
||
|
|
def test_cli_generation_diff_is_additive_and_paged(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
self.change_graph(root)
|
||
|
|
index.build()
|
||
|
|
output = io.StringIO()
|
||
|
|
with contextlib.redirect_stdout(output):
|
||
|
|
code = main(
|
||
|
|
[
|
||
|
|
"--project-root",
|
||
|
|
str(root),
|
||
|
|
"generation-diff",
|
||
|
|
"--limit",
|
||
|
|
"1",
|
||
|
|
]
|
||
|
|
)
|
||
|
|
result = json.loads(output.getvalue())
|
||
|
|
self.assertEqual(0, code)
|
||
|
|
Draft202012Validator(RESULT_SCHEMA).validate(result)
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).validate(
|
||
|
|
{
|
||
|
|
"generation_diff": result["generation_diff"],
|
||
|
|
"pagination": result["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
self.assertEqual("generation-diff.items", result["pagination"]["kind"])
|
||
|
|
self.assertEqual(1, result["pagination"]["returned_count"])
|
||
|
|
self.assertTrue(result["pagination"]["has_more"])
|
||
|
|
self.assertNotIn("next_cursor", result)
|
||
|
|
self.assertNotIn("pagination", result["generation_diff"])
|
||
|
|
self.assertEqual(
|
||
|
|
{
|
||
|
|
"page_schema_version",
|
||
|
|
"receipt_header",
|
||
|
|
"items",
|
||
|
|
"omissions",
|
||
|
|
"page_hash",
|
||
|
|
},
|
||
|
|
set(result["generation_diff"]),
|
||
|
|
)
|
||
|
|
page = result["generation_diff"]
|
||
|
|
self.assertIn("stored_receipt_hash", page["receipt_header"])
|
||
|
|
self.assertNotIn("receipt_hash", page["receipt_header"])
|
||
|
|
self.assertEqual(
|
||
|
|
page["page_hash"],
|
||
|
|
canonical_hash(
|
||
|
|
{
|
||
|
|
"page_schema_version": 1,
|
||
|
|
"receipt_state": result["receipt_state"],
|
||
|
|
"receipt_header": page["receipt_header"],
|
||
|
|
"pagination": result["pagination"],
|
||
|
|
"items": page["items"],
|
||
|
|
"omissions": page["omissions"],
|
||
|
|
}
|
||
|
|
),
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
result["pagination"]["returned_count"],
|
||
|
|
len(page["items"]) + len(page["omissions"]),
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
result["pagination"]["total_count"],
|
||
|
|
page["receipt_header"]["retained_item_count"],
|
||
|
|
)
|
||
|
|
|
||
|
|
malformed_page = json.loads(json.dumps(result))
|
||
|
|
stored_receipt = json.loads(
|
||
|
|
generation_diff_path(project.descriptor).read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
node = next(item for item in stored_receipt["items"] if item["entity"] == "node")
|
||
|
|
node["change"] = "added"
|
||
|
|
malformed_page["generation_diff"]["items"] = [node]
|
||
|
|
self.assertFalse(
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).is_valid(
|
||
|
|
{
|
||
|
|
"generation_diff": malformed_page["generation_diff"],
|
||
|
|
"pagination": malformed_page["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
malformed_header = json.loads(json.dumps(result))
|
||
|
|
header = malformed_header["generation_diff"]["receipt_header"]
|
||
|
|
header["kind"] = "baseline"
|
||
|
|
header["reason"] = "no_meaningful_transition"
|
||
|
|
header["from_generation"] = None
|
||
|
|
self.assertFalse(
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).is_valid(
|
||
|
|
{
|
||
|
|
"generation_diff": malformed_header["generation_diff"],
|
||
|
|
"pagination": malformed_header["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
contradictory_pagination = json.loads(json.dumps(result))
|
||
|
|
contradictory_pagination["pagination"]["has_more"] = False
|
||
|
|
self.assertFalse(
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).is_valid(
|
||
|
|
{
|
||
|
|
"generation_diff": contradictory_pagination["generation_diff"],
|
||
|
|
"pagination": contradictory_pagination["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
)
|
||
|
|
missing_cursor = json.loads(json.dumps(result))
|
||
|
|
missing_cursor["pagination"]["next_cursor"] = None
|
||
|
|
self.assertFalse(
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).is_valid(
|
||
|
|
{
|
||
|
|
"generation_diff": missing_cursor["generation_diff"],
|
||
|
|
"pagination": missing_cursor["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class GenerationDiffMcpTests(unittest.IsolatedAsyncioTestCase):
|
||
|
|
def copy_fixture(self, destination: Path) -> Path:
|
||
|
|
root = destination / "alpha"
|
||
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
||
|
|
shutil.rmtree(root / ".docforge" / "cache", ignore_errors=True)
|
||
|
|
return root
|
||
|
|
|
||
|
|
async def test_mcp_surface_paginates_and_cursors_bind_the_receipt(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
GenerationDiffTests.change_graph(root)
|
||
|
|
index.build()
|
||
|
|
|
||
|
|
async with create_connected_server_and_client_session(
|
||
|
|
create_server(root),
|
||
|
|
raise_exceptions=True,
|
||
|
|
) as session:
|
||
|
|
tools = {tool.name: tool for tool in (await session.list_tools()).tools}
|
||
|
|
schema = tools["docforge_get_generation_diff"].inputSchema
|
||
|
|
self.assertEqual({"limit", "cursor"}, set(schema["properties"]))
|
||
|
|
self.assertEqual([], schema.get("required", []))
|
||
|
|
|
||
|
|
first = await session.call_tool(
|
||
|
|
"docforge_get_generation_diff",
|
||
|
|
{"limit": 1},
|
||
|
|
)
|
||
|
|
first_result = first.structuredContent
|
||
|
|
Draft202012Validator(RESULT_SCHEMA).validate(first_result)
|
||
|
|
Draft202012Validator(GENERATION_DIFF_PAGE_SCHEMA).validate(
|
||
|
|
{
|
||
|
|
"generation_diff": first_result["generation_diff"],
|
||
|
|
"pagination": first_result["pagination"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
self.assertEqual("current", first_result["receipt_state"])
|
||
|
|
self.assertEqual(1, first_result["pagination"]["returned_count"])
|
||
|
|
cursor = first_result["pagination"]["next_cursor"]
|
||
|
|
self.assertIsInstance(cursor, str)
|
||
|
|
self.assertLess(len(cursor), 1_000)
|
||
|
|
self.assertNotIn("next_cursor", first_result)
|
||
|
|
|
||
|
|
second = await session.call_tool(
|
||
|
|
"docforge_get_generation_diff",
|
||
|
|
{"limit": 2, "cursor": cursor},
|
||
|
|
)
|
||
|
|
self.assertEqual(2, second.structuredContent["pagination"]["returned_count"])
|
||
|
|
|
||
|
|
foundation = root / "docs" / "content" / "foundation.md"
|
||
|
|
foundation.write_text(
|
||
|
|
foundation.read_text(encoding="utf-8") + "\nAnother transition.\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
ProjectIndex(Project.open(root)).build()
|
||
|
|
stale = await session.call_tool(
|
||
|
|
"docforge_get_generation_diff",
|
||
|
|
{"limit": 1, "cursor": cursor},
|
||
|
|
)
|
||
|
|
self.assertEqual("stale_cursor", stale.structuredContent["error"]["code"])
|
||
|
|
Draft202012Validator(RESULT_SCHEMA).validate(stale.structuredContent)
|
||
|
|
|
||
|
|
def test_service_diagnostics_drop_before_primary_page(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
descriptor_path = root / ".docforge" / "project.toml"
|
||
|
|
descriptor_path.write_text(
|
||
|
|
descriptor_path.read_text(encoding="utf-8").replace(
|
||
|
|
"max_results = 20",
|
||
|
|
"max_results = 20\nmax_tool_output_chars = 2100",
|
||
|
|
),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
project = Project.open(root)
|
||
|
|
ProjectIndex(project).build()
|
||
|
|
service = DocForgeService(
|
||
|
|
project,
|
||
|
|
diagnostics=True,
|
||
|
|
capability_mode_name="read",
|
||
|
|
)
|
||
|
|
result = service.generation_diff()
|
||
|
|
self.assertEqual("ok", result["status"])
|
||
|
|
self.assertNotIn("diagnostics", result)
|
||
|
|
self.assertLessEqual(service._encoded_length(result), 2_100)
|
||
|
|
|
||
|
|
def test_page_sizing_uses_logarithmic_response_encodes(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = self.copy_fixture(Path(directory))
|
||
|
|
descriptor_path = root / ".docforge" / "project.toml"
|
||
|
|
descriptor_path.write_text(
|
||
|
|
descriptor_path.read_text(encoding="utf-8").replace(
|
||
|
|
"max_results = 20",
|
||
|
|
"max_results = 1000",
|
||
|
|
),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
project = Project.open(root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index.build()
|
||
|
|
stored = index.generation_diff()
|
||
|
|
receipt = dict(stored["generation_diff"])
|
||
|
|
receipt["items"] = [
|
||
|
|
{
|
||
|
|
"item_hash": canonical_hash({"ordinal": ordinal}),
|
||
|
|
"payload": "x" * 500,
|
||
|
|
}
|
||
|
|
for ordinal in range(1_000)
|
||
|
|
]
|
||
|
|
service = DocForgeService(
|
||
|
|
project,
|
||
|
|
capability_mode_name="read",
|
||
|
|
)
|
||
|
|
with (
|
||
|
|
mock.patch.object(
|
||
|
|
service.index,
|
||
|
|
"generation_diff",
|
||
|
|
return_value={**stored, "generation_diff": receipt},
|
||
|
|
),
|
||
|
|
mock.patch.object(
|
||
|
|
service,
|
||
|
|
"_encoded_length",
|
||
|
|
wraps=service._encoded_length,
|
||
|
|
) as encoded,
|
||
|
|
):
|
||
|
|
result = service.generation_diff(limit=1_000)
|
||
|
|
self.assertEqual("ok", result["status"])
|
||
|
|
self.assertGreater(result["pagination"]["returned_count"], 0)
|
||
|
|
self.assertLess(result["pagination"]["returned_count"], 1_000)
|
||
|
|
self.assertLessEqual(encoded.call_count, 15)
|