DocForge2/tests/test_rendering.py

from __future__ import annotations

import contextlib
import hashlib
import io
import json
import os
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from docforge.changesets import ChangesetStore
from docforge.cli import main
from docforge.errors import DocForgeError
from docforge.project import Project
from docforge.rendering import RenderService

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"


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

    @staticmethod
    def project_content_hash(root: Path) -> str:
        digest = hashlib.sha256()
        paths = [
            root / ".docforge/project.toml",
            root / "POLICY.md",
            *(root / "docs/content").glob("*"),
            *(root / "docs/templates").glob("*"),
        ]
        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("utf-8"))
            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)

    def test_declared_render_is_repeatable_and_status_detects_stale_output(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            service = RenderService(project)
            missing = service.status()
            self.assertTrue(missing["configured"])
            self.assertEqual("stale", missing["state"])
            self.assertEqual("missing", missing["outputs"][0]["state"])

            first = service.render("manual")
            projection_receipt = first["output"]["projection_receipt"]
            self.assertIsInstance(projection_receipt, dict)
            assert isinstance(projection_receipt, dict)
            self.assertEqual("manual", projection_receipt["kind"])
            self.assertEqual(
                first["output"]["actual_output_hash"],
                projection_receipt["artifacts"][0]["sha256"],
            )
            output = root / ".docforge/rendered/manual.html"
            first_bytes = output.read_bytes()
            second = service.render("manual")
            self.assertEqual(
                first["output"]["render_identity"], second["output"]["render_identity"]
            )
            self.assertEqual(
                first["output"]["actual_output_hash"], second["output"]["actual_output_hash"]
            )
            self.assertEqual(first_bytes, output.read_bytes())
            self.assertEqual("current", service.status("manual")["state"])

            workflow = root / "docs/content/workflow.md"
            workflow.write_text(
                workflow.read_text(encoding="utf-8") + "\nA new canonical sentence.\n",
                encoding="utf-8",
            )
            stale = service.status("manual")
            self.assertEqual("stale", stale["state"])
            self.assertEqual("stale", stale["outputs"][0]["state"])
            self.assertEqual(first_bytes, output.read_bytes())

    def test_warm_render_status_uses_only_publication_receipts(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            rendered = RenderService(Project.open(root)).render("manual")
            self.assertEqual("current", rendered["receipt"]["state"])

            project = Project.open(root)
            service = RenderService(project)
            with (
                mock.patch.object(
                    project,
                    "load",
                    side_effect=AssertionError("receipt status must not load canonical source"),
                ),
                mock.patch.object(
                    service,
                    "_prepare",
                    side_effect=AssertionError("receipt status must not render"),
                ),
            ):
                current = service.status("manual")
            self.assertEqual("current", current["state"])
            self.assertEqual("receipt", current["verification"])
            self.assertEqual("current", current["outputs"][0]["state"])

            output = root / ".docforge/rendered/manual.html"
            output.write_bytes(output.read_bytes() + b"\n")
            changed_output = service.status("manual")
            self.assertEqual("stale", changed_output["state"])
            self.assertEqual("output_changed", changed_output["outputs"][0]["reason"])

            RenderService(Project.open(root)).render("manual")
            template = root / "docs/templates/manual.html"
            template.write_text(
                template.read_text(encoding="utf-8") + "\n",
                encoding="utf-8",
            )
            changed_template = service.status("manual")
            self.assertEqual("template_changed", changed_template["outputs"][0]["reason"])

    def test_render_receipt_failures_are_degraded_after_output_publication(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            service = RenderService(Project.open(root))
            with mock.patch.object(
                service,
                "_publish_receipt",
                side_effect=DocForgeError(
                    "render_receipt_failure",
                    "Synthetic receipt failure",
                ),
            ):
                result = service.render("manual")

            self.assertEqual("degraded", result["state"])
            self.assertEqual("published", result["publication"])
            self.assertEqual("failed", result["receipt"]["state"])
            self.assertTrue((root / ".docforge/rendered/manual.html").is_file())

    def test_render_receipt_refuses_post_render_input_and_output_changes(self) -> None:
        for changed in ("template", "output", "source"):
            with self.subTest(changed=changed), tempfile.TemporaryDirectory() as directory:
                root = self.copy_fixture("alpha", Path(directory))
                service = RenderService(Project.open(root))
                publish = service._publish_receipt

                def mutate_then_publish(
                    snapshot,
                    view,
                    prepared,
                    *,
                    changed_kind=changed,
                    project_root=root,
                    publish_receipt=publish,
                ):
                    if changed_kind == "template":
                        target = project_root / "docs/templates/manual.html"
                    elif changed_kind == "output":
                        target = project_root / ".docforge/rendered/manual.html"
                    else:
                        target = project_root / "docs/content/workflow.md"
                    target.write_bytes(target.read_bytes() + b"\nChanged before receipt.\n")
                    return publish_receipt(snapshot, view, prepared)

                with mock.patch.object(
                    service,
                    "_publish_receipt",
                    side_effect=mutate_then_publish,
                ):
                    result = service.render("manual")

                self.assertEqual("degraded", result["state"])
                self.assertEqual("published", result["publication"])
                self.assertNotEqual("current", service.status("manual")["state"])
                self.assertEqual("stale", service.deep_status("manual")["state"])

    def test_missing_and_corrupt_render_receipts_are_conservative(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            service = RenderService(Project.open(root))
            service.render("manual")
            receipt = root / ".docforge/cache/render-receipts/manual.json"

            receipt.unlink()
            missing = service.status("manual")
            self.assertEqual("unverified", missing["outputs"][0]["state"])
            self.assertEqual("receipt_missing", missing["outputs"][0]["reason"])

            receipt.write_text("{not-json", encoding="utf-8")
            corrupt = service.status("manual")
            self.assertEqual("unverified", corrupt["outputs"][0]["state"])
            self.assertEqual("receipt_corrupt", corrupt["outputs"][0]["reason"])

    def test_render_receipt_schema_and_renderer_version_fail_closed(self) -> None:
        for mutation in (
            "missing_hash",
            "renderer_version",
            "file_identity",
            "projection_artifact",
        ):
            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
                root = self.copy_fixture("alpha", Path(directory))
                service = RenderService(Project.open(root))
                service.render("manual")
                receipt_path = root / ".docforge/cache/render-receipts/manual.json"
                receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
                if mutation == "missing_hash":
                    receipt.pop("output_hash")
                elif mutation == "renderer_version":
                    receipt["renderer_version"] = "obsolete"
                elif mutation == "projection_artifact":
                    receipt["projection_receipt"]["artifacts"][0]["sha256"] = "0" * 64
                else:
                    receipt["output_file"].pop("ctime_ns")
                receipt_path.write_text(
                    json.dumps(receipt, sort_keys=True, indent=2) + "\n",
                    encoding="utf-8",
                )

                status = service.status("manual")
                self.assertEqual("unverified", status["outputs"][0]["state"])
                self.assertEqual(
                    "foreign_or_incompatible_receipt",
                    status["outputs"][0]["reason"],
                )

    def test_render_status_detects_change_between_bounded_captures(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            service = RenderService(Project.open(root))
            service.render("manual")
            receipt_status = service._receipt_status
            calls = 0

            def mutate_between_captures(descriptor, view, current_state):
                nonlocal calls
                calls += 1
                if calls == 2:
                    output = root / ".docforge/rendered/manual.html"
                    output.write_bytes(output.read_bytes() + b"\n")
                return receipt_status(descriptor, view, current_state)

            with mock.patch.object(
                service,
                "_receipt_status",
                side_effect=mutate_between_captures,
            ):
                result = service.status("manual")

            self.assertEqual("stale", result["state"])
            self.assertNotEqual("current", result["outputs"][0]["state"])

    def test_changeset_preview_is_deterministic_escaped_and_isolated(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            changesets = ChangesetStore(project, "alpha-editor")
            service = RenderService(project, changesets)
            canonical_before = self.project_content_hash(root)
            canonical_render = service.render("manual")
            committed_output = root / ".docforge/rendered/manual.html"
            committed_before = committed_output.read_bytes()

            created = changesets.create("user-preview")
            proposed = changesets.propose_update(
                changeset_id="user-preview",
                expected_changeset_hash=created["changeset_hash"],
                node_id="guide.workflow",
                expected_content_hash=self.node_hash(project, "guide.workflow"),
                metadata={"summary": "A summary visible only in the preview."},
                content="<script>alert('unsafe')</script>\n\n**Rendered safely.**",
                relationship_changes=[],
                rationale="Show the proposed content through the declared view.",
            )
            first = service.preview("user-preview", "manual")
            preview_path = root / ".docforge/previews/user-preview/manual.html"
            preview_bytes = preview_path.read_bytes()
            second = service.preview("user-preview", "manual")

            self.assertEqual(proposed["changeset_hash"], first["changeset_hash"])
            self.assertEqual(first["preview_identity"], second["preview_identity"])
            self.assertEqual(preview_bytes, preview_path.read_bytes())
            self.assertNotEqual(
                canonical_render["output"]["render_identity"], first["preview_identity"]
            )
            html = preview_bytes.decode("utf-8")
            self.assertIn("&lt;script&gt;", html)
            self.assertNotIn("<script>", html)
            self.assertIn("<strong>Rendered safely.</strong>", html)
            self.assertIn("A summary visible only in the preview.", html)
            self.assertEqual(
                ".docforge/previews/user-preview/manual.html", first["preview"]["path"]
            )
            self.assertEqual(canonical_before, self.project_content_hash(root))
            self.assertEqual(committed_before, committed_output.read_bytes())
            self.assertEqual("current", service.status("manual")["state"])

    def test_manual_and_preview_publication_fsync_their_directories(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            changesets = ChangesetStore(project, "alpha-editor")
            service = RenderService(project, changesets)
            proposal = changesets.create("durable-preview")
            proposal = changesets.propose_update(
                changeset_id="durable-preview",
                expected_changeset_hash=str(proposal["changeset_hash"]),
                node_id="guide.workflow",
                expected_content_hash=self.node_hash(project, "guide.workflow"),
                metadata={"summary": "Durable preview output."},
                content=None,
                relationship_changes=[],
                rationale="Exercise durable preview publication.",
            )
            del proposal
            real_fsync = os.fsync
            fsynced_directories: set[Path] = set()

            def record_fsync(descriptor: int) -> None:
                try:
                    path = Path(os.readlink(f"/proc/self/fd/{descriptor}"))
                    if path.is_dir():
                        fsynced_directories.add(path)
                except OSError:
                    pass
                real_fsync(descriptor)

            with mock.patch(
                "docforge._fs_safety.os.fsync",
                side_effect=record_fsync,
            ):
                service.render("manual")
                service.preview("durable-preview", "manual")

            self.assertIn(root / ".docforge/rendered", fsynced_directories)
            self.assertIn(
                root / ".docforge/previews/durable-preview",
                fsynced_directories,
            )

    def test_failed_and_mid_input_renders_preserve_previous_outputs(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            service = RenderService(project)
            service.render("manual")
            output = root / ".docforge/rendered/manual.html"
            before = output.read_bytes()
            workflow = root / "docs/content/workflow.md"
            original_verify = service._verify_canonical

            def mutate_before_replace(snapshot, view, template_bytes) -> None:
                workflow.write_text(
                    workflow.read_text(encoding="utf-8") + "\nChanged during render.\n",
                    encoding="utf-8",
                )
                original_verify(snapshot, view, template_bytes)

            with (
                mock.patch.object(service, "_verify_canonical", side_effect=mutate_before_replace),
                self.assertRaisesRegex(DocForgeError, "changed during rendering") as changed,
            ):
                service.render("manual")
            self.assertEqual("render_input_changed", changed.exception.code)
            self.assertEqual(before, output.read_bytes())
            self.assertFalse(tuple(output.parent.glob(".docforge-render-*")))

            fresh_root = self.copy_fixture("alpha", Path(directory) / "invalid")
            fresh_project = Project.open(fresh_root)
            fresh_service = RenderService(fresh_project)
            fresh_service.render("manual")
            fresh_output = fresh_root / ".docforge/rendered/manual.html"
            fresh_before = fresh_output.read_bytes()
            template = fresh_root / "docs/templates/manual.html"
            template.write_text("<html>{{ unsupported }}</html>", encoding="utf-8")
            with self.assertRaisesRegex(DocForgeError, "unsupported tokens"):
                fresh_service.render("manual")
            self.assertEqual(fresh_before, fresh_output.read_bytes())

    def test_render_configuration_paths_commands_views_and_limits_fail_closed(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            command_root = self.copy_fixture("alpha", parent / "command")
            command_descriptor = command_root / ".docforge/project.toml"
            marker = parent / "command-ran"
            command_descriptor.write_text(
                command_descriptor.read_text(encoding="utf-8").replace(
                    'title = "Alpha Manual"\nfamilies = ["guide", "proof"]',
                    'title = "Alpha Manual"\n'
                    'families = ["guide", "proof"]\n'
                    f'command = "touch {marker.as_posix()}"',
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "unknown fields"):
                Project.open(command_root)
            self.assertFalse(marker.exists())

            renderer_root = self.copy_fixture("alpha", parent / "renderer")
            renderer_descriptor = renderer_root / ".docforge/project.toml"
            renderer_descriptor.write_text(
                renderer_descriptor.read_text(encoding="utf-8").replace(
                    'renderer = "generic_html"', 'renderer = "shell"'
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "unsupported built-in renderer"):
                Project.open(renderer_root)

            output_root = self.copy_fixture("alpha", parent / "output")
            output_descriptor = output_root / ".docforge/project.toml"
            output_descriptor.write_text(
                output_descriptor.read_text(encoding="utf-8").replace(
                    'output = ".docforge/rendered/manual.html"',
                    'output = "docs/content/manual.html"',
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "protected project root"):
                Project.open(output_root)

            template_root = self.copy_fixture("alpha", parent / "template")
            template_descriptor = template_root / ".docforge/project.toml"
            template_descriptor.write_text(
                template_descriptor.read_text(encoding="utf-8")
                .replace('template_root = "docs/templates"', 'template_root = "docs/content"')
                .replace('template = "manual.html"', 'template = "foundation.md"'),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "Template input must not overlap"):
                Project.open(template_root)

            limit_root = self.copy_fixture("alpha", parent / "limit")
            limit_descriptor = limit_root / ".docforge/project.toml"
            limit_descriptor.write_text(
                limit_descriptor.read_text(encoding="utf-8") + "\nmax_render_bytes = 100\n",
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "unknown fields"):
                Project.open(limit_root)
            limit_descriptor.write_text(
                limit_descriptor.read_text(encoding="utf-8")
                .replace("\nmax_render_bytes = 100\n", "")
                .replace(
                    "max_changeset_bytes = 100000",
                    "max_changeset_bytes = 100000\nmax_render_bytes = 100",
                ),
                encoding="utf-8",
            )
            limit_service = RenderService(Project.open(limit_root))
            with self.assertRaisesRegex(DocForgeError, "configured limit") as limit_error:
                limit_service.render("manual")
            self.assertEqual("render_too_large", limit_error.exception.code)

            excessive = self.copy_fixture("alpha", Path(directory) / "excessive")
            excessive_descriptor = excessive / ".docforge/project.toml"
            excessive_descriptor.write_text(
                excessive_descriptor.read_text(encoding="utf-8").replace(
                    "max_changeset_bytes = 100000",
                    "max_changeset_bytes = 100000\nmax_render_bytes = 20000001",
                ),
                encoding="utf-8",
            )
            permissive_limit = RenderService(Project.open(excessive)).render("manual")
            self.assertEqual("current", permissive_limit["receipt"]["state"])
            self.assertTrue((excessive / ".docforge/rendered/manual.html").is_file())
            self.assertFalse((limit_root / ".docforge/rendered/manual.html").exists())

            template_limit_root = self.copy_fixture("alpha", parent / "template-limit")
            template_limit_descriptor = template_limit_root / ".docforge/project.toml"
            template_limit_descriptor.write_text(
                template_limit_descriptor.read_text(encoding="utf-8").replace(
                    "max_changeset_bytes = 100000",
                    "max_changeset_bytes = 100000\nmax_template_bytes = 10",
                ),
                encoding="utf-8",
            )
            template_limit_service = RenderService(Project.open(template_limit_root))
            with self.assertRaisesRegex(DocForgeError, "template exceeds") as template_limit:
                template_limit_service.render("manual")
            self.assertEqual("template_too_large", template_limit.exception.code)
            self.assertFalse((template_limit_root / ".docforge/rendered/manual.html").exists())

            safe_root = self.copy_fixture("alpha", parent / "safe")
            safe_service = RenderService(Project.open(safe_root))
            with self.assertRaisesRegex(DocForgeError, "not declared") as unknown:
                safe_service.render("not-a-view")
            self.assertEqual("unknown_render_view", unknown.exception.code)

    def test_symlink_inputs_and_outputs_are_rejected_and_unconfigured_status_is_explicit(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            root = self.copy_fixture("alpha", parent)
            project = Project.open(root)
            service = RenderService(project)
            outside_template = parent / "outside-template.html"
            outside_template.write_text("{{ docforge_content }}", encoding="utf-8")
            template = root / "docs/templates/manual.html"
            template.unlink()
            template.symlink_to(outside_template)
            with self.assertRaisesRegex(DocForgeError, "missing or unsafe"):
                service.render("manual")

            template.unlink()
            shutil.copy2(FIXTURES / "alpha/docs/templates/manual.html", template)
            outside_output = parent / "outside-output.html"
            outside_output.write_text("do not replace", encoding="utf-8")
            output = root / ".docforge/rendered/manual.html"
            output.parent.mkdir(parents=True)
            output.symlink_to(outside_output)
            with self.assertRaisesRegex(DocForgeError, "output path is unsafe"):
                service.render("manual")
            self.assertEqual("do not replace", outside_output.read_text(encoding="utf-8"))

            beta = RenderService(Project.open(FIXTURES / "beta")).status()
            self.assertFalse(beta["configured"])
            self.assertEqual("not_configured", beta["state"])
            self.assertEqual([], beta["outputs"])

    def test_cli_exposes_declared_render_status_and_isolated_preview_only(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            changesets = ChangesetStore(project, "alpha-editor")
            created = changesets.create("cli-preview")
            changesets.propose_update(
                changeset_id="cli-preview",
                expected_changeset_hash=created["changeset_hash"],
                node_id="guide.workflow",
                expected_content_hash=self.node_hash(project, "guide.workflow"),
                metadata={"summary": "CLI preview summary."},
                content=None,
                relationship_changes=[],
                rationale="Exercise the explicit CLI preview path.",
            )

            commands = (
                ("render", "manual"),
                ("render-status", "manual"),
                ("preview", "cli-preview", "manual"),
                ("render-status", "manual", "--deep"),
            )
            results: list[dict] = []
            for command in commands:
                stream = io.StringIO()
                with contextlib.redirect_stdout(stream):
                    self.assertEqual(0, main(["--project-root", str(root), *command]))
                results.append(json.loads(stream.getvalue()))
            self.assertEqual("current", results[0]["state"])
            self.assertEqual("current", results[1]["state"])
            self.assertEqual("current", results[2]["state"])
            self.assertEqual("receipt", results[1]["verification"])
            self.assertEqual("deep", results[3]["verification"])
            self.assertTrue((root / ".docforge/rendered/manual.html").is_file())
            self.assertTrue((root / ".docforge/previews/cli-preview/manual.html").is_file())

            stream = io.StringIO()
            with contextlib.redirect_stdout(stream):
                self.assertEqual(
                    2,
                    main(["--project-root", str(root), "render", "undeclared"]),
                )
            self.assertEqual("unknown_render_view", json.loads(stream.getvalue())["error"]["code"])


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