DocForge2/tests/test_core.py

from __future__ import annotations

import contextlib
import io
import json
import shutil
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from docforge.cli import main  # noqa: E402
from docforge.context import compile_context  # noqa: E402
from docforge.errors import DocForgeError  # noqa: E402
from docforge.index import ProjectIndex  # noqa: E402
from docforge.project import Project  # noqa: E402

FIXTURES = ROOT / "tests" / "fixtures"


class DocForgeCoreTests(unittest.TestCase):
    def copy_fixture(self, name: str, destination: Path) -> Path:
        root = destination / name
        shutil.copytree(FIXTURES / name, root)
        return root

    def test_two_projects_load_distinct_confined_graphs(self) -> None:
        alpha = Project.open(FIXTURES / "alpha").load()
        beta = Project.open(FIXTURES / "beta").load()

        self.assertEqual("alpha-docs", alpha.descriptor.project_id)
        self.assertEqual(
            ["guide.foundation", "guide.workflow", "proof.validation"],
            [node.node_id for node in alpha.nodes],
        )
        self.assertEqual(["research.question"], [node.node_id for node in beta.nodes])
        self.assertNotEqual(alpha.source_hash, beta.source_hash)
        self.assertNotIn("research.question", {node.node_id for node in alpha.nodes})

    def test_missing_revision_tool_does_not_block_project_loading(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            with mock.patch("docforge.project.subprocess.run", side_effect=OSError):
                snapshot = Project.open(root).load()

            self.assertEqual("unversioned", snapshot.revision)

    def test_descriptor_rejects_parent_and_absolute_paths(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            descriptor = root / ".docforge" / "project.toml"
            original = descriptor.read_text(encoding="utf-8")
            for unsafe in ("../outside", "/tmp/outside"):
                with self.subTest(unsafe=unsafe):
                    descriptor.write_text(
                        original.replace(
                            'content_roots = ["docs/content"]', f'content_roots = ["{unsafe}"]'
                        ),
                        encoding="utf-8",
                    )
                    with self.assertRaisesRegex(DocForgeError, "inside the project root"):
                        Project.open(root)

    def test_descriptor_rejects_symbolic_link_escape(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            root = self.copy_fixture("alpha", parent)
            outside = parent / "outside"
            outside.mkdir()
            link = root / "escaped"
            link.symlink_to(outside, target_is_directory=True)
            descriptor = root / ".docforge" / "project.toml"
            descriptor.write_text(
                descriptor.read_text(encoding="utf-8").replace(
                    'content_roots = ["docs/content"]', 'content_roots = ["escaped"]'
                ),
                encoding="utf-8",
            )

            with self.assertRaisesRegex(DocForgeError, "outside the project root"):
                Project.open(root)

    def test_descriptor_rejects_unknown_fields_cache_overlap_and_mid_session_change(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            descriptor = root / ".docforge" / "project.toml"
            original = descriptor.read_text(encoding="utf-8")
            descriptor.write_text(original + "\nunknown_setting = true\n", encoding="utf-8")
            with self.assertRaisesRegex(DocForgeError, "unknown fields"):
                Project.open(root)

            descriptor.write_text(
                original.replace(
                    'cache_root = ".docforge/cache"', 'cache_root = "docs/content/cache"'
                ).replace(
                    'index = ".docforge/cache/index.sqlite3"',
                    'index = "docs/content/cache/index.sqlite3"',
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "must not overlap"):
                Project.open(root)

            descriptor.write_text(original, encoding="utf-8")
            project = Project.open(root)
            descriptor.write_text(original + "\n", encoding="utf-8")
            with self.assertRaisesRegex(DocForgeError, "changed after"):
                project.load()

            with self.assertRaisesRegex(DocForgeError, "does not exist"):
                Project.open(root / "missing")

    def test_duplicate_nodes_broken_edges_and_dependency_cycles_fail(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            content = root / "docs" / "content"
            duplicate = content / "duplicate.md"
            duplicate.write_text((content / "foundation.md").read_text(), encoding="utf-8")
            with self.assertRaisesRegex(DocForgeError, "unique"):
                Project.open(root).load()
            duplicate.unlink()

            workflow = content / "workflow.md"
            original = workflow.read_text(encoding="utf-8")
            workflow.write_text(
                original.replace(
                    'depends_on = ["guide.foundation"]', 'depends_on = ["missing.node"]'
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "missing nodes"):
                Project.open(root).load()
            workflow.write_text(original, encoding="utf-8")

            foundation = content / "foundation.md"
            foundation.write_text(
                foundation.read_text(encoding="utf-8").replace(
                    'summary = "Defines which Alpha files own documentation facts."',
                    'summary = "Defines which Alpha files own documentation facts."\n'
                    'depends_on = ["guide.workflow"]',
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "cycle"):
                Project.open(root).load()

    def test_index_build_is_repeatable_and_validates_every_row(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            first = index.build()
            second = index.build()
            validated = index.check()

            for key in ("source_hash", "node_hash", "node_count", "edge_hash", "edge_count"):
                self.assertEqual(first[key], second[key])
                self.assertEqual(first[key], validated[key])
            self.assertEqual(3, validated["node_count"])
            self.assertEqual(2, validated["edge_count"])

    def test_index_rejects_tampered_rows_and_another_project_cache(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            alpha_root = self.copy_fixture("alpha", parent)
            beta_root = self.copy_fixture("beta", parent)
            alpha = ProjectIndex(Project.open(alpha_root))
            alpha.build()

            with contextlib.closing(sqlite3.connect(alpha.path)) as connection:
                connection.execute(
                    "UPDATE nodes SET title = 'Tampered' WHERE node_id = 'guide.workflow'"
                )
                connection.commit()
            with self.assertRaisesRegex(DocForgeError, "do not match"):
                alpha.check()

            alpha.build()
            beta = ProjectIndex(Project.open(beta_root))
            beta.path.parent.mkdir(parents=True)
            shutil.copy2(alpha.path, beta.path)
            with self.assertRaisesRegex(DocForgeError, "does not match"):
                beta.check()

    def test_stale_source_fails_closed_and_failed_rebuild_preserves_index(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            index_bytes = index.path.read_bytes()
            workflow = root / "docs" / "content" / "workflow.md"
            workflow.write_text(
                workflow.read_text() + "\nChanged after indexing.\n", encoding="utf-8"
            )

            with self.assertRaisesRegex(DocForgeError, "does not match"):
                index.check()
            workflow.write_text("invalid", encoding="utf-8")
            with self.assertRaises(DocForgeError):
                index.build()
            self.assertEqual(index_bytes, index.path.read_bytes())

    def test_lookup_search_filter_and_traversal_are_deterministic(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()

            node = index.get_node("guide.workflow")["node"]
            self.assertEqual("Editing workflow", node["title"])
            search = index.search("canonical nodes", limit=10)
            self.assertEqual(["guide.workflow"], [item["node_id"] for item in search["results"]])
            filtered = index.filter_nodes(family="proof", tag="validation")
            self.assertEqual(
                ["proof.validation"], [item["node_id"] for item in filtered["results"]]
            )
            dependencies = index.dependencies("guide.workflow", depth=2)
            self.assertEqual(
                ["guide.foundation"], [item["node_id"] for item in dependencies["results"]]
            )
            backlinks = index.backlinks("guide.workflow")
            self.assertEqual(
                ["proof.validation"], [edge["source_id"] for edge in backlinks["edges"]]
            )
            impact = index.impact("guide.foundation", depth=2)
            self.assertEqual(
                ["guide.workflow", "proof.validation"],
                [item["node_id"] for item in impact["results"]],
            )

    def test_query_rechecks_source_identity_before_returning(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            checked = index.build()
            changed = {**checked, "source_hash": "0" * 64}

            with (
                mock.patch.object(index, "check", side_effect=[checked, changed]),
                self.assertRaisesRegex(DocForgeError, "changed during the query"),
            ):
                index.get_node("guide.workflow")

    def test_source_set_change_during_load_fails_closed(self) -> None:
        project = Project.open(FIXTURES / "alpha")
        sources = project.canonical_source_paths()
        invented = project.descriptor.root / "docs" / "content" / "invented.md"
        with (
            mock.patch.object(
                project, "canonical_source_paths", side_effect=[sources, (*sources, invented)]
            ),
            self.assertRaisesRegex(DocForgeError, "source set changed"),
        ):
            project.load()

    def test_context_is_bounded_cited_deterministic_and_reports_omissions(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            first = compile_context(index, "active", budget=180)
            second = compile_context(index, "active", budget=180)

            self.assertEqual(first, second)
            self.assertLessEqual(first["estimated_tokens"], 180)
            self.assertEqual("guide.workflow", first["entries"][0]["node_id"])
            self.assertEqual("required by profile", first["entries"][0]["reason"])
            self.assertTrue(first["entries"][0]["source_path"])
            self.assertTrue(first["entries"][0]["content_hash"])
            self.assertTrue(first["omissions"])

            with self.assertRaisesRegex(DocForgeError, "required node"):
                compile_context(index, "active", budget=10)

    def test_cli_json_is_repeatable_and_project_scoped(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("beta", Path(directory))
            outputs: list[str] = []
            for _ in range(2):
                stream = io.StringIO()
                with contextlib.redirect_stdout(stream):
                    self.assertEqual(0, main(["--project-root", str(root), "validate"]))
                outputs.append(stream.getvalue())
            self.assertEqual(outputs[0], outputs[1])
            result = json.loads(outputs[0])
            self.assertEqual("beta-notes", result["project_id"])
            self.assertEqual(1, result["node_count"])


if __name__ == "__main__":
    unittest.main()