261 lines
9.1 KiB
Python
261 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
import docforge.adapter_sdk as sdk
|
|
from docforge.errors import DocForgeError
|
|
|
|
|
|
class ReferenceIncrementalAdapter:
|
|
"""Small public-SDK-only adapter used to prove the conformance entry point."""
|
|
|
|
def __init__(self, root: Path) -> None:
|
|
self.root = root
|
|
|
|
@staticmethod
|
|
def _hash(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
def _node(self) -> sdk.AdapterNode:
|
|
content = "A deterministic reference function."
|
|
return sdk.AdapterNode(
|
|
sdk.Node(
|
|
node_id="reference.run",
|
|
title="Reference run",
|
|
family="source",
|
|
authority="derived",
|
|
status="current",
|
|
tags=("python",),
|
|
summary=content,
|
|
content=content,
|
|
source_path="src/reference.py",
|
|
source_anchor="L1",
|
|
content_hash=self._hash(content),
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _logic() -> sdk.LogicProjection:
|
|
return sdk.LogicProjection(
|
|
owner_node_id="reference.run",
|
|
source_id="source.reference",
|
|
nodes=(
|
|
sdk.LogicNode("entry", "entry", "Enter run", "L1"),
|
|
sdk.LogicNode("return", "return", "Return", "L2"),
|
|
),
|
|
edges=(sdk.LogicEdge("entry", "return", "return", "RETURN", 0),),
|
|
)
|
|
|
|
def load_manifest(self) -> sdk.AdapterManifest:
|
|
fingerprint = self._hash("def run(): return True\n")
|
|
return sdk.AdapterManifest(
|
|
project_id="sdk-reference",
|
|
title="SDK reference",
|
|
adapter_id="python-reference",
|
|
adapter_version="1",
|
|
root=self.root,
|
|
revision=fingerprint[:12],
|
|
source_hash=fingerprint,
|
|
families=("source",),
|
|
allowed_relations=(),
|
|
sources=(
|
|
sdk.AdapterSource(
|
|
source_id="source.reference",
|
|
source_path="src/reference.py",
|
|
fingerprint=fingerprint,
|
|
extractor_version="python-reference@1",
|
|
),
|
|
),
|
|
estimated_nodes=1,
|
|
)
|
|
|
|
def extract_source(self, source: sdk.AdapterSource) -> sdk.AdapterSourceProjection:
|
|
return sdk.AdapterSourceProjection(
|
|
source_id=source.source_id,
|
|
fingerprint=source.fingerprint,
|
|
nodes=(self._node(),),
|
|
edges=(),
|
|
logic=(self._logic(),),
|
|
)
|
|
|
|
def load_projection(self) -> sdk.AdapterProjection:
|
|
manifest = self.load_manifest()
|
|
return sdk.AdapterProjection(
|
|
project_id=manifest.project_id,
|
|
title=manifest.title,
|
|
adapter_id=manifest.adapter_id,
|
|
adapter_version=manifest.adapter_version,
|
|
root=manifest.root,
|
|
revision=manifest.revision,
|
|
source_hash=manifest.source_hash,
|
|
nodes=(self._node(),),
|
|
edges=(),
|
|
)
|
|
|
|
def load_complete_assembly(self) -> sdk.AdapterAssembly:
|
|
return sdk.AdapterAssembly(
|
|
projection=self.load_projection(),
|
|
logic=(self._logic(),),
|
|
)
|
|
|
|
|
|
class LegacyAdapter:
|
|
def __init__(self, projection: sdk.AdapterProjection) -> None:
|
|
self.projection = projection
|
|
|
|
def load_projection(self) -> sdk.AdapterProjection:
|
|
return self.projection
|
|
|
|
|
|
class LogicWithoutCompleteAdapter:
|
|
def __init__(self, root: Path) -> None:
|
|
self.reference = ReferenceIncrementalAdapter(root)
|
|
|
|
def load_manifest(self) -> sdk.AdapterManifest:
|
|
return self.reference.load_manifest()
|
|
|
|
def extract_source(self, source: sdk.AdapterSource) -> sdk.AdapterSourceProjection:
|
|
return self.reference.extract_source(source)
|
|
|
|
def load_projection(self) -> sdk.AdapterProjection:
|
|
return self.reference.load_projection()
|
|
|
|
|
|
class NondeterministicCompleteAdapter(ReferenceIncrementalAdapter):
|
|
def __init__(self, root: Path) -> None:
|
|
super().__init__(root)
|
|
self.complete_calls = 0
|
|
|
|
def load_complete_assembly(self) -> sdk.AdapterAssembly:
|
|
self.complete_calls += 1
|
|
assembly = super().load_complete_assembly()
|
|
if self.complete_calls % 2 == 0:
|
|
logic = assembly.logic[0]
|
|
return replace(
|
|
assembly,
|
|
logic=(
|
|
replace(
|
|
logic,
|
|
nodes=(
|
|
logic.nodes[0],
|
|
replace(logic.nodes[1], label="Unstable return"),
|
|
),
|
|
),
|
|
),
|
|
)
|
|
return assembly
|
|
|
|
|
|
class AdapterSdkTests(unittest.TestCase):
|
|
def test_public_facade_contains_the_complete_authoring_surface(self) -> None:
|
|
expected = {
|
|
"AdapterAssembly",
|
|
"AdapterConformanceReport",
|
|
"AdapterManifest",
|
|
"AdapterNode",
|
|
"AdapterProjection",
|
|
"AdapterSource",
|
|
"AdapterSourceProjection",
|
|
"CompleteAdapterAssemblyLoader",
|
|
"Edge",
|
|
"LogicEdge",
|
|
"LogicNode",
|
|
"LogicProjection",
|
|
"Node",
|
|
"verify_adapter_conformance",
|
|
}
|
|
|
|
self.assertLessEqual(expected, set(sdk.__all__))
|
|
for name in expected:
|
|
with self.subTest(name=name):
|
|
self.assertTrue(hasattr(sdk, name))
|
|
|
|
def test_conformance_proves_deterministic_complete_graph_and_logic_parity(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
report = sdk.verify_adapter_conformance(
|
|
ReferenceIncrementalAdapter(root),
|
|
cache_root=root / ".cache" / "conformance",
|
|
)
|
|
|
|
self.assertEqual(1, report.schema_version)
|
|
self.assertTrue(report.incremental)
|
|
self.assertEqual(1, report.node_count)
|
|
self.assertEqual(0, report.edge_count)
|
|
self.assertEqual(1, report.logic_projection_count)
|
|
self.assertEqual(64, len(report.assembly_hash))
|
|
self.assertEqual(report.assembly_hash, report.as_dict()["assembly_hash"])
|
|
|
|
def test_conformance_keeps_load_projection_only_adapters_first_class(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
content = "Legacy adapter content."
|
|
projection = sdk.AdapterProjection(
|
|
project_id="legacy-sdk",
|
|
title="Legacy SDK",
|
|
adapter_id="legacy",
|
|
adapter_version="1",
|
|
root=root,
|
|
revision="legacy",
|
|
source_hash=hashlib.sha256(content.encode()).hexdigest(),
|
|
nodes=(
|
|
sdk.AdapterNode(
|
|
sdk.Node(
|
|
node_id="legacy.node",
|
|
title="Legacy node",
|
|
family="guide",
|
|
authority="authoritative",
|
|
status="current",
|
|
tags=("legacy",),
|
|
summary=content,
|
|
content=content,
|
|
source_path="docs/legacy.md",
|
|
source_anchor=None,
|
|
content_hash=hashlib.sha256(content.encode()).hexdigest(),
|
|
)
|
|
),
|
|
),
|
|
edges=(),
|
|
)
|
|
|
|
report = sdk.verify_adapter_conformance(
|
|
LegacyAdapter(projection),
|
|
cache_root=root / ".cache" / "legacy-conformance",
|
|
)
|
|
|
|
self.assertFalse(report.incremental)
|
|
self.assertEqual(0, report.logic_projection_count)
|
|
self.assertEqual(projection.identity(), projection.identity())
|
|
|
|
def test_conformance_requires_a_complete_oracle_for_incremental_logic(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
|
|
with self.assertRaises(DocForgeError) as captured:
|
|
sdk.verify_adapter_conformance(
|
|
LogicWithoutCompleteAdapter(root),
|
|
cache_root=root / ".cache" / "missing-logic-oracle",
|
|
)
|
|
|
|
self.assertEqual("complete_logic_oracle_required", captured.exception.code)
|
|
|
|
def test_conformance_rejects_nondeterministic_complete_logic(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
|
|
with self.assertRaises(DocForgeError) as captured:
|
|
sdk.verify_adapter_conformance(
|
|
NondeterministicCompleteAdapter(root),
|
|
cache_root=root / ".cache" / "nondeterministic",
|
|
)
|
|
|
|
self.assertEqual("nondeterministic_adapter", captured.exception.code)
|
|
self.assertEqual(["logic"], captured.exception.details["fields"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|