DocForge2/tests/test_adapter_sdk.py

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()