1536 lines
66 KiB
Python
1536 lines
66 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
import tomllib
|
|
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 import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
|
|
from docforge._version import __version__
|
|
from docforge.changeset_contract import document_hash
|
|
from docforge.cli import _parser, _run, main
|
|
from docforge.client_config import (
|
|
_read_existing,
|
|
_validate_configuration_result,
|
|
generate_client_configuration,
|
|
)
|
|
from docforge.doctor import run_doctor
|
|
from docforge.errors import DocForgeError
|
|
from docforge.index import ProjectIndex
|
|
from docforge.mcp_server import READ_TOOLS
|
|
from docforge.project import MAX_PROJECT_DESCRIPTOR_BYTES, Project
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIXTURES = ROOT / "tests" / "fixtures"
|
|
SCHEMAS = ROOT / "schemas"
|
|
CONFIGURATION_SCHEMA = json.loads(
|
|
(SCHEMAS / "client-configuration.schema.json").read_text(encoding="utf-8")
|
|
)
|
|
DOCTOR_SCHEMA = json.loads((SCHEMAS / "doctor-result.schema.json").read_text(encoding="utf-8"))
|
|
POLICY_SCHEMA = json.loads((SCHEMAS / "policy.schema.json").read_text(encoding="utf-8"))
|
|
|
|
GRAPH_RENDER_CONFIG = """
|
|
|
|
[graph_render]
|
|
output_root = ".docforge/portable-graph"
|
|
|
|
[[graph_render.views]]
|
|
id = "architecture"
|
|
renderer = "portable_graph_html"
|
|
output = "architecture.html"
|
|
title = "Alpha architecture"
|
|
root = "guide.workflow"
|
|
"""
|
|
|
|
|
|
class ClientIntegrationTests(unittest.TestCase):
|
|
def copy_fixture(self, destination: Path) -> Path:
|
|
root = destination / "alpha"
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
|
return root
|
|
|
|
@staticmethod
|
|
def tree_snapshot(root: Path) -> dict[str, tuple[int, int, str]]:
|
|
return {
|
|
path.relative_to(root).as_posix(): (
|
|
path.stat().st_mode,
|
|
path.stat().st_size,
|
|
path.read_bytes().hex(),
|
|
)
|
|
for path in sorted(root.rglob("*"))
|
|
if path.is_file()
|
|
}
|
|
|
|
def test_exact_outline_cli_forms_and_legacy_project_root_forms_parse(self) -> None:
|
|
parser = _parser()
|
|
for client in ("codex", "claude", "openclaw"):
|
|
configured = parser.parse_args(["configure", client, "--project", "/tmp/project"])
|
|
self.assertEqual("configure", configured.command)
|
|
self.assertEqual(client, configured.client)
|
|
self.assertEqual(Path("/tmp/project"), configured.project)
|
|
doctor = parser.parse_args(["doctor", "--client", "codex"])
|
|
self.assertEqual("doctor", doctor.command)
|
|
self.assertIsNone(doctor.project)
|
|
legacy = parser.parse_args(["--project-root", "/tmp/project", "info"])
|
|
self.assertEqual(Path("/tmp/project"), legacy.project_root)
|
|
with self.assertRaises(DocForgeError) as missing:
|
|
_run(parser.parse_args(["info"]))
|
|
self.assertEqual("missing_project_root", missing.exception.code)
|
|
|
|
def test_configuration_preview_is_deterministic_parseable_and_secret_free(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
before = self.tree_snapshot(root)
|
|
previous = os.environ.get("DOCFORGE_TEST_SECRET")
|
|
os.environ["DOCFORGE_TEST_SECRET"] = "must-not-appear"
|
|
try:
|
|
results: dict[str, dict[str, object]] = {}
|
|
for client in ("codex", "claude", "openclaw"):
|
|
first = generate_client_configuration(project, client, no_ast=True)
|
|
second = generate_client_configuration(project, client, no_ast=True)
|
|
self.assertEqual(first, second)
|
|
Draft202012Validator(CONFIGURATION_SCHEMA).validate(first)
|
|
self.assertEqual(__version__, first["docforge_version"])
|
|
Draft202012Validator(POLICY_SCHEMA).validate(first["effective_policy"])
|
|
self.assertEqual("read", first["binding"]["capability_mode"])
|
|
self.assertEqual(
|
|
"preserve-no-ast",
|
|
first["binding"]["adapter_policy"]["mode"],
|
|
)
|
|
self.assertEqual({}, first["binding"]["environment"])
|
|
self.assertNotIn(
|
|
"must-not-appear",
|
|
json.dumps(first, sort_keys=True),
|
|
)
|
|
results[client] = first
|
|
finally:
|
|
if previous is None:
|
|
os.environ.pop("DOCFORGE_TEST_SECRET", None)
|
|
else:
|
|
os.environ["DOCFORGE_TEST_SECRET"] = previous
|
|
|
|
codex_content = results["codex"]["artifact"]["content"]
|
|
self.assertIn("mcp_servers", tomllib.loads(codex_content))
|
|
openclaw_content = results["openclaw"]["artifact"]["content"]
|
|
self.assertIn("mcp", json.loads(openclaw_content))
|
|
claude_content = results["claude"]["artifact"]["content"]
|
|
self.assertIn("mcpServers", json.loads(claude_content))
|
|
self.assertEqual(before, self.tree_snapshot(root))
|
|
|
|
def test_capability_bindings_fail_closed_and_render_policy_is_derived(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
project = Project.open(self.copy_fixture(Path(directory)))
|
|
with self.assertRaises(DocForgeError) as missing_writer:
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
capability_mode="proposal",
|
|
)
|
|
self.assertEqual("capability_unavailable", missing_writer.exception.code)
|
|
proposal = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
capability_mode="proposal",
|
|
proposal_writer="alpha-editor",
|
|
)
|
|
self.assertEqual("explicit", proposal["binding"]["render_policy"]["manual"])
|
|
application = generate_client_configuration(
|
|
project,
|
|
"openclaw",
|
|
capability_mode="application",
|
|
proposal_writer="alpha-editor",
|
|
canonical_applier="alpha-editor",
|
|
)
|
|
self.assertEqual("auto", application["binding"]["render_policy"]["manual"])
|
|
with self.assertRaises(DocForgeError) as mismatch:
|
|
generate_client_configuration(
|
|
project,
|
|
"openclaw",
|
|
capability_mode="application",
|
|
proposal_writer="alpha-editor",
|
|
canonical_applier="other",
|
|
)
|
|
self.assertEqual("capability_unavailable", mismatch.exception.code)
|
|
with self.assertRaises(DocForgeError) as escalated:
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
proposal_writer="alpha-editor",
|
|
)
|
|
self.assertEqual("invalid_capability_binding", escalated.exception.code)
|
|
|
|
def test_nondefault_projection_policy_selectors_serialize_and_validate(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + GRAPH_RENDER_CONFIG,
|
|
encoding="utf-8",
|
|
)
|
|
project = Project.open(root)
|
|
expected_arguments = [
|
|
"-I",
|
|
"-m",
|
|
"docforge.mcp_server",
|
|
"--project-root",
|
|
str(root),
|
|
"--capability-mode",
|
|
"read",
|
|
"--manual-render-policy",
|
|
"disabled",
|
|
"--portable-graph-policy",
|
|
"disabled",
|
|
"--live-viewer-policy",
|
|
"disabled",
|
|
]
|
|
for client in ("codex", "claude", "openclaw"):
|
|
with self.subTest(client=client):
|
|
result = generate_client_configuration(
|
|
project,
|
|
client,
|
|
manual_render_policy="disabled",
|
|
portable_graph_policy="disabled",
|
|
live_viewer_policy="disabled",
|
|
)
|
|
Draft202012Validator(CONFIGURATION_SCHEMA).validate(result)
|
|
_validate_configuration_result(result)
|
|
self.assertEqual(
|
|
{
|
|
"schema_version": 2,
|
|
"manual": "disabled",
|
|
"portable_graph": "disabled",
|
|
"live_viewer": "disabled",
|
|
},
|
|
result["projection_policy"],
|
|
)
|
|
self.assertEqual(expected_arguments, result["binding"]["args"])
|
|
content = result["artifact"]["content"]
|
|
if client == "codex":
|
|
document = tomllib.loads(content)
|
|
serialized = document["mcp_servers"][result["server_name"]]["args"]
|
|
elif client == "claude":
|
|
document = json.loads(content)
|
|
serialized = document["mcpServers"][result["server_name"]]["args"]
|
|
else:
|
|
document = json.loads(content)
|
|
serialized = document["mcp"]["servers"][result["server_name"]]["args"]
|
|
self.assertEqual(expected_arguments, serialized)
|
|
canonical = json.dumps(
|
|
result["projection_policy"],
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
self.assertEqual(
|
|
hashlib.sha256(canonical).hexdigest(),
|
|
result["projection_policy_hash"],
|
|
)
|
|
|
|
def test_projection_policy_schema_and_configuration_validation_reject_drift(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
project = Project.open(self.copy_fixture(Path(directory)))
|
|
result = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
manual_render_policy="disabled",
|
|
live_viewer_policy="disabled",
|
|
)
|
|
validator = Draft202012Validator(CONFIGURATION_SCHEMA)
|
|
validator.validate(result)
|
|
_validate_configuration_result(result)
|
|
|
|
version_drift = json.loads(json.dumps(result))
|
|
version_drift["docforge_version"] = "0.0.0"
|
|
with self.assertRaisesRegex(AssertionError, "product version"):
|
|
_validate_configuration_result(version_drift)
|
|
|
|
for field, value in (
|
|
("schema_version", 1),
|
|
("manual", "on-demand"),
|
|
("portable_graph", "auto"),
|
|
("live_viewer", "explicit"),
|
|
):
|
|
with self.subTest(field=field):
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["projection_policy"][field] = value
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
|
|
missing = json.loads(json.dumps(result))
|
|
missing.pop("projection_policy")
|
|
self.assertTrue(list(validator.iter_errors(missing)))
|
|
extra = json.loads(json.dumps(result))
|
|
extra["projection_policy"]["project_path"] = "/private/project"
|
|
self.assertTrue(list(validator.iter_errors(extra)))
|
|
|
|
mismatched = json.loads(json.dumps(result))
|
|
mismatched["projection_policy"]["manual"] = "explicit"
|
|
canonical = json.dumps(
|
|
mismatched["projection_policy"],
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
mismatched["projection_policy_hash"] = hashlib.sha256(canonical).hexdigest()
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(mismatched)
|
|
|
|
bad_hash = json.loads(json.dumps(result))
|
|
bad_hash["projection_policy_hash"] = "0" * 64
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(bad_hash)
|
|
|
|
defaulted = generate_client_configuration(project, "codex")
|
|
self.assertNotIn(
|
|
"--manual-render-policy",
|
|
defaulted["binding"]["args"],
|
|
)
|
|
defaulted["projection_policy"]["manual"] = "disabled"
|
|
canonical = json.dumps(
|
|
defaulted["projection_policy"],
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
defaulted["projection_policy_hash"] = hashlib.sha256(canonical).hexdigest()
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(defaulted)
|
|
|
|
unavailable = json.loads(json.dumps(result))
|
|
unavailable["projection_availability"]["manual_configured"] = False
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(unavailable)
|
|
|
|
descriptor = project.descriptor.descriptor_path
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + GRAPH_RENDER_CONFIG,
|
|
encoding="utf-8",
|
|
)
|
|
graph_project = Project.open(project.descriptor.root)
|
|
graph_defaulted = generate_client_configuration(graph_project, "codex")
|
|
self.assertNotIn(
|
|
"--portable-graph-policy",
|
|
graph_defaulted["binding"]["args"],
|
|
)
|
|
coordinated_graph_drift = json.loads(json.dumps(graph_defaulted))
|
|
coordinated_graph_drift["projection_policy"]["portable_graph"] = "disabled"
|
|
coordinated_graph_drift["projection_availability"]["portable_graph_configured"] = False
|
|
canonical = json.dumps(
|
|
coordinated_graph_drift["projection_policy"],
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
coordinated_graph_drift["projection_policy_hash"] = hashlib.sha256(
|
|
canonical
|
|
).hexdigest()
|
|
coordinated_graph_drift["configuration_hash"] = document_hash(
|
|
{
|
|
"schema_version": 1,
|
|
"client": coordinated_graph_drift["client"],
|
|
"server_name": coordinated_graph_drift["server_name"],
|
|
"project": coordinated_graph_drift["project"],
|
|
"binding": coordinated_graph_drift["binding"],
|
|
"effective_policy": coordinated_graph_drift["effective_policy"],
|
|
"projection_policy": coordinated_graph_drift["projection_policy"],
|
|
"projection_policy_hash": coordinated_graph_drift["projection_policy_hash"],
|
|
"projection_availability": coordinated_graph_drift["projection_availability"],
|
|
"artifact_format": coordinated_graph_drift["artifact"]["format"],
|
|
"artifact_content_sha256": coordinated_graph_drift["artifact"][
|
|
"content_sha256"
|
|
],
|
|
}
|
|
)
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(coordinated_graph_drift)
|
|
|
|
invalid_selections = (
|
|
{"manual_render_policy": "sometimes"},
|
|
{"portable_graph_policy": "auto"},
|
|
{"live_viewer_policy": "always"},
|
|
)
|
|
for selection in invalid_selections:
|
|
with self.subTest(selection=selection):
|
|
with self.assertRaises(DocForgeError) as raised:
|
|
generate_client_configuration(graph_project, "codex", **selection)
|
|
self.assertEqual("invalid_projection_policy", raised.exception.code)
|
|
|
|
def test_doctor_round_trips_projection_selectors_and_rejects_invalid_modes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
ProjectIndex(project).build()
|
|
config = parent / "openclaw.json"
|
|
generated = generate_client_configuration(
|
|
project,
|
|
"openclaw",
|
|
manual_render_policy="disabled",
|
|
live_viewer_policy="disabled",
|
|
output=config,
|
|
)
|
|
healthy = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(healthy)
|
|
self.assertEqual("healthy", healthy["doctor_state"])
|
|
self.assertIn(
|
|
"effective_policy_valid",
|
|
[check["code"] for check in healthy["checks"]],
|
|
)
|
|
|
|
document = json.loads(config.read_text(encoding="utf-8"))
|
|
entry = document["mcp"]["servers"][generated["server_name"]]
|
|
arguments = entry["args"]
|
|
manual_position = arguments.index("--manual-render-policy") + 1
|
|
arguments[manual_position] = "sometimes"
|
|
config.write_text(json.dumps(document, sort_keys=True), encoding="utf-8")
|
|
invalid = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(invalid)
|
|
self.assertEqual("unhealthy", invalid["doctor_state"])
|
|
policy_check = next(
|
|
check for check in invalid["checks"] if check["check_id"] == "policy.effective"
|
|
)
|
|
self.assertEqual("invalid_projection_policy", policy_check["code"])
|
|
|
|
arguments[manual_position] = "auto"
|
|
config.write_text(json.dumps(document, sort_keys=True), encoding="utf-8")
|
|
unavailable = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(unavailable)
|
|
self.assertEqual("unhealthy", unavailable["doctor_state"])
|
|
policy_check = next(
|
|
check for check in unavailable["checks"] if check["check_id"] == "policy.effective"
|
|
)
|
|
self.assertEqual("projection_policy_unavailable", policy_check["code"])
|
|
|
|
def test_explicit_fragment_write_is_atomic_conflict_aware_and_private(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
output = parent / "client" / "docforge.toml"
|
|
output.parent.mkdir()
|
|
created = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("created", created["artifact"]["write_state"])
|
|
self.assertEqual(
|
|
stat.S_IMODE(output.stat().st_mode),
|
|
0o600,
|
|
)
|
|
unchanged = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("unchanged", unchanged["artifact"]["write_state"])
|
|
|
|
output.write_text("different\n", encoding="utf-8")
|
|
with self.assertRaises(DocForgeError) as conflict:
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("output_conflict", conflict.exception.code)
|
|
self.assertEqual("different\n", output.read_text(encoding="utf-8"))
|
|
output.unlink()
|
|
output.symlink_to(parent / "outside")
|
|
with self.assertRaises(DocForgeError) as unsafe:
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("unsafe_output", unsafe.exception.code)
|
|
self.assertEqual([], list(output.parent.glob(".docforge-client-*")))
|
|
|
|
linked_parent = parent / "linked-client"
|
|
outside = parent / "outside-client"
|
|
outside.mkdir()
|
|
linked_parent.symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaises(DocForgeError) as escaped:
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=linked_parent / "fragment.toml",
|
|
)
|
|
self.assertEqual("unsafe_output", escaped.exception.code)
|
|
self.assertFalse((outside / "fragment.toml").exists())
|
|
|
|
def test_committed_fragment_reports_unconfirmed_directory_durability(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
output = parent / "client" / "docforge.toml"
|
|
output.parent.mkdir()
|
|
with mock.patch(
|
|
"docforge.client_config.os.fsync",
|
|
side_effect=[None, OSError("directory fsync unavailable")],
|
|
):
|
|
created = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
Draft202012Validator(CONFIGURATION_SCHEMA).validate(created)
|
|
self.assertEqual("created", created["artifact"]["write_state"])
|
|
self.assertEqual("unconfirmed", created["artifact"]["durability"])
|
|
self.assertEqual(
|
|
[{"code": "publication_durability_unconfirmed"}],
|
|
created["warnings"],
|
|
)
|
|
self.assertTrue(output.is_file())
|
|
self.assertEqual(
|
|
created["artifact"]["content"],
|
|
output.read_text(encoding="utf-8"),
|
|
)
|
|
unchanged = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("unchanged", unchanged["artifact"]["write_state"])
|
|
|
|
def test_configuration_fails_closed_on_stale_or_unimportable_runtime(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaises(DocForgeError) as stale:
|
|
generate_client_configuration(project, "codex")
|
|
self.assertEqual("source_changed", stale.exception.code)
|
|
|
|
current = Project.open(root)
|
|
failed_probe = mock.Mock(returncode=1)
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.subprocess.run",
|
|
return_value=failed_probe,
|
|
),
|
|
self.assertRaises(DocForgeError) as unavailable,
|
|
):
|
|
generate_client_configuration(current, "codex")
|
|
self.assertEqual(
|
|
"client_configuration_unavailable",
|
|
unavailable.exception.code,
|
|
)
|
|
|
|
def test_fragment_publication_rejects_parent_replacement_and_os_errors(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
output_parent = parent / "client"
|
|
output_parent.mkdir()
|
|
output = output_parent / "docforge.toml"
|
|
moved = parent / "moved-client"
|
|
original_link = os.link
|
|
swapped = False
|
|
|
|
def swap_parent(*args: object, **kwargs: object) -> None:
|
|
nonlocal swapped
|
|
if not swapped:
|
|
output_parent.rename(moved)
|
|
output_parent.mkdir()
|
|
swapped = True
|
|
original_link(*args, **kwargs)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.os.link",
|
|
side_effect=swap_parent,
|
|
),
|
|
self.assertRaises(DocForgeError) as changed,
|
|
):
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("output_changed", changed.exception.code)
|
|
self.assertFalse(output.exists())
|
|
self.assertEqual([], list(moved.glob("docforge.toml")))
|
|
self.assertEqual([], list(moved.glob(".docforge-client-*")))
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.os.link",
|
|
side_effect=PermissionError("denied"),
|
|
),
|
|
self.assertRaises(DocForgeError) as denied,
|
|
):
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("output_publication_failed", denied.exception.code)
|
|
self.assertEqual([], list(output_parent.glob(".docforge-client-*")))
|
|
|
|
def test_identical_existing_fragment_must_already_be_private_and_singly_linked(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
output = parent / "docforge.toml"
|
|
preview = generate_client_configuration(project, "codex")
|
|
output.write_text(preview["artifact"]["content"], encoding="utf-8")
|
|
output.chmod(0o644)
|
|
with self.assertRaises(DocForgeError) as public:
|
|
generate_client_configuration(project, "codex", output=output)
|
|
self.assertEqual("unsafe_output", public.exception.code)
|
|
|
|
def test_unchanged_fragment_revalidates_binding_after_read(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
output = parent / "docforge.toml"
|
|
generate_client_configuration(project, "codex", output=output)
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
|
|
def mutate_after_read(*args: object, **kwargs: object) -> bytes:
|
|
content = _read_existing(*args, **kwargs) # type: ignore[arg-type]
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return content
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config._read_existing",
|
|
side_effect=mutate_after_read,
|
|
),
|
|
self.assertRaises(DocForgeError) as stale,
|
|
):
|
|
generate_client_configuration(project, "codex", output=output)
|
|
self.assertEqual("source_changed", stale.exception.code)
|
|
|
|
def test_fragment_publication_revalidates_descriptor_after_link(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
output_parent = parent / "client"
|
|
output_parent.mkdir()
|
|
output = output_parent / "docforge.toml"
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
original_link = os.link
|
|
|
|
def mutate_descriptor(*args: object, **kwargs: object) -> None:
|
|
original_link(*args, **kwargs)
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.os.link",
|
|
side_effect=mutate_descriptor,
|
|
),
|
|
self.assertRaises(DocForgeError) as stale,
|
|
):
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertEqual("source_changed", stale.exception.code)
|
|
self.assertFalse(output.exists())
|
|
self.assertEqual([], list(output_parent.glob(".docforge-client-*")))
|
|
|
|
def test_unprovable_postlink_binding_returns_schema_valid_degraded_evidence(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
output_parent = parent / "client"
|
|
output_parent.mkdir()
|
|
output = output_parent / "docforge.toml"
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
original_link = os.link
|
|
|
|
def mutate_descriptor(*args: object, **kwargs: object) -> None:
|
|
original_link(*args, **kwargs)
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.os.link",
|
|
side_effect=mutate_descriptor,
|
|
),
|
|
mock.patch(
|
|
"docforge.client_config._rollback_link",
|
|
return_value=False,
|
|
),
|
|
):
|
|
result = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
Draft202012Validator(CONFIGURATION_SCHEMA).validate(result)
|
|
self.assertEqual("created", result["artifact"]["write_state"])
|
|
self.assertEqual("unconfirmed", result["artifact"]["durability"])
|
|
self.assertIsNone(result["artifact"]["output_path"])
|
|
self.assertIn(
|
|
{"code": "publication_binding_unconfirmed"},
|
|
result["warnings"],
|
|
)
|
|
|
|
def test_fragment_publication_revalidates_after_directory_fsync(self) -> None:
|
|
for scenario in ("descriptor", "target", "parent"):
|
|
with self.subTest(scenario=scenario), tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
output_parent = parent / "client"
|
|
output_parent.mkdir()
|
|
output = output_parent / "docforge.toml"
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
moved = parent / "moved-client"
|
|
original_fsync = os.fsync
|
|
calls = 0
|
|
|
|
def mutate_at_directory_fsync(
|
|
file_descriptor: int,
|
|
selected_scenario: str = scenario,
|
|
selected_descriptor: Path = descriptor,
|
|
selected_output: Path = output,
|
|
selected_parent: Path = output_parent,
|
|
selected_moved: Path = moved,
|
|
selected_fsync: Callable[[int], None] = original_fsync,
|
|
) -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
selected_fsync(file_descriptor)
|
|
if calls != 2:
|
|
return
|
|
if selected_scenario == "descriptor":
|
|
selected_descriptor.write_text(
|
|
selected_descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
elif selected_scenario == "target":
|
|
selected_output.write_text("tampered\n", encoding="utf-8")
|
|
else:
|
|
selected_parent.rename(selected_moved)
|
|
selected_parent.mkdir()
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.client_config.os.fsync",
|
|
side_effect=mutate_at_directory_fsync,
|
|
),
|
|
self.assertRaises(DocForgeError) as changed,
|
|
):
|
|
generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
output=output,
|
|
)
|
|
self.assertIn(changed.exception.code, {"source_changed", "output_changed"})
|
|
self.assertFalse(output.exists())
|
|
self.assertFalse((moved / "docforge.toml").exists())
|
|
|
|
def test_doctor_round_trip_is_schema_valid_and_performs_no_hidden_work(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
config = parent / "codex.toml"
|
|
generated = generate_client_configuration(
|
|
project,
|
|
"codex",
|
|
no_ast=True,
|
|
output=config,
|
|
)
|
|
before_project = self.tree_snapshot(root)
|
|
before_config = config.read_bytes()
|
|
with (
|
|
mock.patch.object(
|
|
Project,
|
|
"load",
|
|
side_effect=AssertionError("doctor must not load project sources"),
|
|
),
|
|
mock.patch(
|
|
"docforge.index.ProjectIndex.check",
|
|
side_effect=AssertionError("doctor must not check SQLite"),
|
|
),
|
|
mock.patch(
|
|
"docforge.index.ProjectIndex.synchronize",
|
|
side_effect=AssertionError("doctor must not synchronize"),
|
|
),
|
|
mock.patch(
|
|
"docforge.index.ProjectIndex.build",
|
|
side_effect=AssertionError("doctor must not build"),
|
|
),
|
|
mock.patch(
|
|
"subprocess.run",
|
|
side_effect=AssertionError("doctor must not execute configured commands"),
|
|
),
|
|
):
|
|
result = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(result)
|
|
self.assertEqual("healthy", result["doctor_state"])
|
|
self.assertEqual(0, result["summary"]["warning"])
|
|
self.assertEqual(0, result["summary"]["failed"])
|
|
self.assertTrue(result["guarantees"]["read_only"])
|
|
no_ast = next(
|
|
check for check in result["checks"] if check["check_id"] == "policy.no_ast"
|
|
)
|
|
self.assertEqual("no_ast_policy_valid", no_ast["code"])
|
|
self.assertEqual("unverifiable", no_ast["details"]["adapter_internals"])
|
|
self.assertEqual(before_project, self.tree_snapshot(root))
|
|
self.assertEqual(before_config, config.read_bytes())
|
|
|
|
def test_doctor_reports_missing_malformed_and_secret_environment_without_echoing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
missing = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=parent / "missing.toml",
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(missing)
|
|
self.assertEqual("unhealthy", missing["doctor_state"])
|
|
malformed_path = parent / "malformed.toml"
|
|
malformed_path.write_text("[invalid", encoding="utf-8")
|
|
malformed = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=malformed_path,
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(malformed)
|
|
self.assertEqual("unhealthy", malformed["doctor_state"])
|
|
|
|
generated = generate_client_configuration(project, "openclaw")
|
|
document = json.loads(generated["artifact"]["content"])
|
|
entry = next(iter(document["mcp"]["servers"].values()))
|
|
entry["env"] = {"TOKEN": "top-secret-value"}
|
|
secret_path = parent / "openclaw.json"
|
|
secret_path.write_text(
|
|
json.dumps(document, sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|
|
secret = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=secret_path,
|
|
server_name=generated["server_name"],
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(secret)
|
|
self.assertEqual("degraded", secret["doctor_state"])
|
|
encoded = json.dumps(secret, sort_keys=True)
|
|
self.assertIn("TOKEN", encoded)
|
|
self.assertNotIn("top-secret-value", encoded)
|
|
|
|
def test_cli_doctor_exit_codes_are_stable(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
config = parent / "codex.toml"
|
|
generated = generate_client_configuration(project, "codex", output=config)
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
healthy = main(
|
|
[
|
|
"doctor",
|
|
"--client",
|
|
"codex",
|
|
"--project",
|
|
str(root),
|
|
"--config",
|
|
str(config),
|
|
"--server-name",
|
|
str(generated["server_name"]),
|
|
]
|
|
)
|
|
self.assertEqual(0, healthy)
|
|
self.assertEqual("healthy", json.loads(output.getvalue())["doctor_state"])
|
|
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
degraded = main(
|
|
[
|
|
"doctor",
|
|
"--client",
|
|
"codex",
|
|
"--project",
|
|
str(root),
|
|
"--config",
|
|
str(parent / "missing.toml"),
|
|
]
|
|
)
|
|
self.assertEqual(2, degraded)
|
|
self.assertEqual("unhealthy", json.loads(output.getvalue())["doctor_state"])
|
|
|
|
malformed = parent / "malformed.toml"
|
|
malformed.write_text("[invalid", encoding="utf-8")
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
unhealthy = main(
|
|
[
|
|
"doctor",
|
|
"--client",
|
|
"codex",
|
|
"--project",
|
|
str(root),
|
|
"--config",
|
|
str(malformed),
|
|
]
|
|
)
|
|
self.assertEqual(2, unhealthy)
|
|
self.assertEqual("unhealthy", json.loads(output.getvalue())["doctor_state"])
|
|
|
|
def test_project_descriptor_read_is_bounded_stable_and_symlink_safe(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
original = descriptor.read_bytes()
|
|
descriptor.unlink()
|
|
external = parent / "external.toml"
|
|
external.write_bytes(original)
|
|
descriptor.symlink_to(external)
|
|
with self.assertRaises(DocForgeError) as symlinked:
|
|
Project.open(root)
|
|
self.assertEqual("project_descriptor_unsafe", symlinked.exception.code)
|
|
|
|
descriptor.unlink()
|
|
descriptor.write_bytes(b"x" * (MAX_PROJECT_DESCRIPTOR_BYTES + 1))
|
|
with self.assertRaises(DocForgeError) as oversized:
|
|
Project.open(root)
|
|
self.assertEqual("project_descriptor_oversized", oversized.exception.code)
|
|
|
|
shutil.rmtree(root / ".docforge")
|
|
external_directory = parent / "external-docforge"
|
|
external_directory.mkdir()
|
|
(external_directory / "project.toml").write_bytes(original)
|
|
(root / ".docforge").symlink_to(external_directory, target_is_directory=True)
|
|
with self.assertRaises(DocForgeError) as escaped:
|
|
Project.open(root)
|
|
self.assertEqual("project_descriptor_unsafe", escaped.exception.code)
|
|
|
|
def test_project_load_revalidates_the_bounded_descriptor(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_bytes(b"x" * (MAX_PROJECT_DESCRIPTOR_BYTES + 1))
|
|
with self.assertRaises(DocForgeError) as oversized:
|
|
project.load()
|
|
self.assertEqual("project_descriptor_oversized", oversized.exception.code)
|
|
|
|
def test_doctor_ignores_unrelated_entries_and_honors_codex_home(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
codex_home = parent / "codex-home"
|
|
codex_home.mkdir()
|
|
config = codex_home / "config.toml"
|
|
generated = generate_client_configuration(project, "codex", output=config)
|
|
config.write_text(
|
|
config.read_text(encoding="utf-8")
|
|
+ '\n[mcp_servers."remote"]\nurl = "https://example.invalid/mcp"\n',
|
|
encoding="utf-8",
|
|
)
|
|
with mock.patch.dict(os.environ, {"CODEX_HOME": str(codex_home)}):
|
|
result = run_doctor(project, "codex")
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(result)
|
|
self.assertEqual("healthy", result["doctor_state"])
|
|
self.assertEqual(generated["server_name"], result["config"]["server_name"])
|
|
|
|
openclaw_config = parent / "openclaw.json"
|
|
openclaw = generate_client_configuration(
|
|
project,
|
|
"openclaw",
|
|
output=openclaw_config,
|
|
)
|
|
document = json.loads(openclaw_config.read_text(encoding="utf-8"))
|
|
document["mcp"]["servers"]["remote"] = {
|
|
"type": "http",
|
|
"url": "https://example.invalid/mcp",
|
|
"enabled": True,
|
|
}
|
|
openclaw_config.write_text(
|
|
json.dumps(document, sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|
|
openclaw_result = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=openclaw_config,
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(openclaw_result)
|
|
self.assertEqual("healthy", openclaw_result["doctor_state"])
|
|
self.assertEqual(
|
|
openclaw["server_name"],
|
|
openclaw_result["config"]["server_name"],
|
|
)
|
|
|
|
selected = next(iter(document["mcp"]["servers"].values()))
|
|
selected["cwd"] = str(root)
|
|
selected["toolFilter"] = {"include": ["docforge_bootstrap"]}
|
|
openclaw_config.write_text(
|
|
json.dumps(document, sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|
|
filtered = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=openclaw_config,
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(filtered)
|
|
self.assertEqual("degraded", filtered["doctor_state"])
|
|
self.assertIn(
|
|
"client_tool_filter_unverified",
|
|
[check["code"] for check in filtered["checks"]],
|
|
)
|
|
|
|
def test_doctor_rejects_nonlaunching_or_malformed_matching_arguments(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
generated = generate_client_configuration(project, "openclaw")
|
|
document = json.loads(generated["artifact"]["content"])
|
|
entry = next(iter(document["mcp"]["servers"].values()))
|
|
entry["args"] = entry["args"][3:]
|
|
config = parent / "missing-prefix.json"
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
missing_prefix = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
)
|
|
self.assertEqual("unhealthy", missing_prefix["doctor_state"])
|
|
self.assertIn(
|
|
"server_arguments_invalid",
|
|
[check["code"] for check in missing_prefix["checks"]],
|
|
)
|
|
|
|
entry["args"] = [
|
|
"-I",
|
|
"-m",
|
|
"docforge.mcp_server",
|
|
"--project-root",
|
|
str(root),
|
|
"--bogus",
|
|
]
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
malformed = run_doctor(project, "openclaw", config_path=config)
|
|
self.assertEqual("unhealthy", malformed["doctor_state"])
|
|
self.assertIn(
|
|
"server_arguments_invalid",
|
|
[check["code"] for check in malformed["checks"]],
|
|
)
|
|
|
|
secret = "super-secret-argument"
|
|
entry["args"][-1] = secret
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
redacted = run_doctor(project, "openclaw", config_path=config)
|
|
self.assertEqual("unhealthy", redacted["doctor_state"])
|
|
self.assertNotIn(secret, json.dumps(redacted, sort_keys=True))
|
|
|
|
def test_doctor_rejects_ambiguous_duplicate_roots_and_malformed_controls(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
generated = generate_client_configuration(project, "openclaw")
|
|
document = json.loads(generated["artifact"]["content"])
|
|
name, first = next(iter(document["mcp"]["servers"].items()))
|
|
duplicate = json.loads(json.dumps(first))
|
|
root_value = duplicate["args"].index("--project-root") + 1
|
|
duplicate["args"][root_value] = "/first/does/not/match"
|
|
for index in range(64):
|
|
duplicate["args"].extend(("--project-root", f"/duplicate/does/not/match/{index}"))
|
|
duplicate["args"].extend(("--project-root", str(root)))
|
|
document["mcp"]["servers"][f"{name}-duplicate"] = duplicate
|
|
config = parent / "openclaw.json"
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
ambiguous = run_doctor(project, "openclaw", config_path=config)
|
|
self.assertEqual("unhealthy", ambiguous["doctor_state"])
|
|
self.assertIn(
|
|
"client_entry_ambiguous",
|
|
[check["code"] for check in ambiguous["checks"]],
|
|
)
|
|
|
|
document["mcp"]["servers"].pop(f"{name}-duplicate")
|
|
first["supportsParallelToolCalls"] = "not-a-bool"
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
malformed_parallel = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
)
|
|
self.assertEqual("unhealthy", malformed_parallel["doctor_state"])
|
|
|
|
first["supportsParallelToolCalls"] = False
|
|
first["cwd"] = str(parent / "missing-directory")
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
missing_cwd = run_doctor(project, "openclaw", config_path=config)
|
|
self.assertEqual("unhealthy", missing_cwd["doctor_state"])
|
|
self.assertIn(
|
|
"client_cwd_invalid",
|
|
[check["code"] for check in missing_cwd["checks"]],
|
|
)
|
|
|
|
for malformed_filter in (
|
|
["not-an-object"],
|
|
{"include": "not-a-list"},
|
|
{"include": [1, 2]},
|
|
):
|
|
first.pop("cwd", None)
|
|
first["toolFilter"] = malformed_filter
|
|
config.write_text(json.dumps(document), encoding="utf-8")
|
|
malformed = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=config,
|
|
)
|
|
self.assertEqual("unhealthy", malformed["doctor_state"])
|
|
self.assertIn(
|
|
"client_entry_invalid",
|
|
[check["code"] for check in malformed["checks"]],
|
|
)
|
|
|
|
def test_doctor_bounds_parser_failures_inputs_and_check_inventory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
deeply_nested = parent / "deep.json"
|
|
deeply_nested.write_text("[" * 2_000 + "]" * 2_000, encoding="utf-8")
|
|
deep = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=deeply_nested,
|
|
)
|
|
self.assertEqual("unhealthy", deep["doctor_state"])
|
|
|
|
huge_integer = parent / "integer.json"
|
|
huge_integer.write_text(
|
|
'{"mcp":{"servers":{}},"value":' + "9" * 100_000 + "}",
|
|
encoding="utf-8",
|
|
)
|
|
integer = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=huge_integer,
|
|
)
|
|
self.assertEqual("unhealthy", integer["doctor_state"])
|
|
|
|
malformed_openclaw = parent / "malformed-openclaw.json"
|
|
malformed_openclaw.write_text('{"mcp":"wrong"}', encoding="utf-8")
|
|
malformed_driver = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=malformed_openclaw,
|
|
)
|
|
self.assertEqual("unhealthy", malformed_driver["doctor_state"])
|
|
self.assertIn(
|
|
"client_config_invalid",
|
|
[check["code"] for check in malformed_driver["checks"]],
|
|
)
|
|
|
|
oversized_name = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=parent / "missing.json",
|
|
server_name="x" * 5_000,
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(oversized_name)
|
|
self.assertEqual("unhealthy", oversized_name["doctor_state"])
|
|
self.assertEqual(
|
|
14,
|
|
len(oversized_name["checks"]),
|
|
)
|
|
self.assertEqual(
|
|
len(oversized_name["checks"]),
|
|
len({check["check_id"] for check in oversized_name["checks"]}),
|
|
)
|
|
oversized_path = run_doctor(
|
|
project,
|
|
"openclaw",
|
|
config_path=Path("/" + "x" * 40_000),
|
|
)
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(oversized_path)
|
|
self.assertEqual("unhealthy", oversized_path["doctor_state"])
|
|
self.assertLessEqual(
|
|
len(json.dumps(oversized_path, sort_keys=True).encode("utf-8")),
|
|
32_768,
|
|
)
|
|
|
|
def test_doctor_rejects_descriptor_config_and_index_parent_swaps(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
descriptor_parent = root / ".docforge"
|
|
moved_descriptor = root / ".docforge-old"
|
|
original_project_open = os.open
|
|
descriptor_swapped = False
|
|
|
|
def swap_descriptor_parent(
|
|
path: object,
|
|
flags: int,
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> int:
|
|
nonlocal descriptor_swapped
|
|
descriptor = original_project_open(path, flags, *args, **kwargs)
|
|
if (
|
|
not descriptor_swapped
|
|
and Path(path) == descriptor_parent
|
|
and flags & os.O_DIRECTORY
|
|
):
|
|
descriptor_parent.rename(moved_descriptor)
|
|
descriptor_parent.mkdir()
|
|
shutil.copy2(
|
|
moved_descriptor / "project.toml",
|
|
descriptor_parent / "project.toml",
|
|
)
|
|
descriptor_swapped = True
|
|
return descriptor
|
|
|
|
with (
|
|
mock.patch(
|
|
"docforge.project.os.open",
|
|
side_effect=swap_descriptor_parent,
|
|
),
|
|
self.assertRaises(DocForgeError) as changed,
|
|
):
|
|
run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=parent / "missing.toml",
|
|
)
|
|
self.assertEqual("project_descriptor_changed", changed.exception.code)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
config_parent = parent / "client"
|
|
config_parent.mkdir()
|
|
config = config_parent / "config.toml"
|
|
generated = generate_client_configuration(project, "codex", output=config)
|
|
moved_config = parent / "client-old"
|
|
original_doctor_open = os.open
|
|
config_swapped = False
|
|
|
|
def swap_config_parent(
|
|
path: object,
|
|
flags: int,
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> int:
|
|
nonlocal config_swapped
|
|
descriptor = original_doctor_open(path, flags, *args, **kwargs)
|
|
if not config_swapped and Path(path) == config_parent and flags & os.O_DIRECTORY:
|
|
config_parent.rename(moved_config)
|
|
config_parent.mkdir()
|
|
(config_parent / "config.toml").write_text("", encoding="utf-8")
|
|
config_swapped = True
|
|
return descriptor
|
|
|
|
with mock.patch(
|
|
"docforge.doctor.os.open",
|
|
side_effect=swap_config_parent,
|
|
):
|
|
changed_config = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
self.assertEqual("unhealthy", changed_config["doctor_state"])
|
|
self.assertIn(
|
|
"client_config_changed",
|
|
[check["code"] for check in changed_config["checks"]],
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
config = parent / "config.toml"
|
|
generated = generate_client_configuration(project, "codex", output=config)
|
|
index_parent = project.descriptor.index_path.parent
|
|
moved_index_parent = parent / "cache-old"
|
|
original_doctor_open = os.open
|
|
index_swapped = False
|
|
|
|
def swap_index_parent(
|
|
path: object,
|
|
flags: int,
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> int:
|
|
nonlocal index_swapped
|
|
descriptor = original_doctor_open(path, flags, *args, **kwargs)
|
|
if not index_swapped and Path(path) == index_parent and flags & os.O_DIRECTORY:
|
|
index_parent.rename(moved_index_parent)
|
|
index_parent.mkdir()
|
|
index_swapped = True
|
|
return descriptor
|
|
|
|
with mock.patch(
|
|
"docforge.doctor.os.open",
|
|
side_effect=swap_index_parent,
|
|
):
|
|
unsafe_index = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
self.assertEqual("degraded", unsafe_index["doctor_state"])
|
|
self.assertIn(
|
|
"index_unsafe",
|
|
[check["code"] for check in unsafe_index["checks"]],
|
|
)
|
|
|
|
def test_doctor_rechecks_the_exact_client_file_before_return(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
project = Project.open(root)
|
|
ProjectIndex(project).build()
|
|
config = parent / "config.toml"
|
|
generated = generate_client_configuration(project, "codex", output=config)
|
|
raw = config.read_bytes()
|
|
with mock.patch(
|
|
"docforge.doctor._read_stable_regular",
|
|
side_effect=[
|
|
(raw, (1, 1, len(raw), 1, 1)),
|
|
(raw, (1, 2, len(raw), 1, 1)),
|
|
],
|
|
):
|
|
result = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=config,
|
|
server_name=generated["server_name"],
|
|
)
|
|
self.assertEqual("unhealthy", result["doctor_state"])
|
|
self.assertIn(
|
|
"client_config_changed",
|
|
[check["code"] for check in result["checks"]],
|
|
)
|
|
|
|
def test_descriptor_currency_and_custom_adapter_generation_fail_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
project = Project.open(root)
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_text(
|
|
descriptor.read_text(encoding="utf-8") + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaises(DocForgeError) as changed:
|
|
run_doctor(project, "codex", config_path=root / "missing.toml")
|
|
self.assertEqual("source_changed", changed.exception.code)
|
|
|
|
current = Project.open(root)
|
|
current.descriptor = replace(current.descriptor, adapter="custom-adapter")
|
|
with self.assertRaises(DocForgeError) as custom:
|
|
generate_client_configuration(current, "codex")
|
|
self.assertEqual(
|
|
"client_configuration_unavailable",
|
|
custom.exception.code,
|
|
)
|
|
|
|
def test_diagnostic_cli_results_validate_their_dedicated_schemas(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
root = self.copy_fixture(parent)
|
|
preview_output = io.StringIO()
|
|
with contextlib.redirect_stdout(preview_output):
|
|
preview_code = main(
|
|
[
|
|
"--diagnostics",
|
|
"configure",
|
|
"codex",
|
|
"--project",
|
|
str(root),
|
|
]
|
|
)
|
|
self.assertEqual(0, preview_code)
|
|
preview = json.loads(preview_output.getvalue())
|
|
Draft202012Validator(CONFIGURATION_SCHEMA).validate(preview)
|
|
self.assertEqual("cli.configure", preview["diagnostics"]["operation"])
|
|
|
|
unhealthy_output = io.StringIO()
|
|
with contextlib.redirect_stdout(unhealthy_output):
|
|
unhealthy_code = main(
|
|
[
|
|
"--diagnostics",
|
|
"doctor",
|
|
"--client",
|
|
"codex",
|
|
"--project",
|
|
str(root),
|
|
"--config",
|
|
str(parent / "missing.toml"),
|
|
]
|
|
)
|
|
self.assertEqual(2, unhealthy_code)
|
|
unhealthy = json.loads(unhealthy_output.getvalue())
|
|
Draft202012Validator(DOCTOR_SCHEMA).validate(unhealthy)
|
|
self.assertEqual("cli.doctor", unhealthy["diagnostics"]["operation"])
|
|
|
|
def test_configuration_schema_rejects_cross_field_drift(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
project = Project.open(self.copy_fixture(Path(directory)))
|
|
result = generate_client_configuration(project, "codex")
|
|
validator = Draft202012Validator(CONFIGURATION_SCHEMA)
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["artifact"]["format"] = "openclaw-json-fragment-v1"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["effective_policy"] = {}
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["action"] = "write"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["artifact"]["durability"] = "unconfirmed"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["binding"]["capability_mode"] = "proposal"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["binding"]["adapter_policy"] = {
|
|
"schema_version": 1,
|
|
"mode": "preserve-no-ast",
|
|
"adapter_evolution": "preserve",
|
|
"ast_forbidden": True,
|
|
"logic_indexing": "off",
|
|
"blocked_tools": ["docforge_get_logic"],
|
|
"instruction": "Preserve adapter behavior without AST or Logic publication.",
|
|
}
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["binding"]["render_policy"]["manual"] = "disabled"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["binding"]["args"][6] = "proposal"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
no_ast = generate_client_configuration(project, "codex", no_ast=True)
|
|
drifted = json.loads(json.dumps(no_ast))
|
|
drifted["binding"]["args"].remove("--no-ast")
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
drifted = json.loads(json.dumps(result))
|
|
drifted["effective_policy"]["prohibitions"][2] = "made_up_permission"
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(drifted)
|
|
drifted = json.loads(json.dumps(no_ast))
|
|
drifted["binding"]["adapter_policy"]["instruction"] = "AST use is allowed."
|
|
self.assertTrue(list(validator.iter_errors(drifted)))
|
|
with self.assertRaises(AssertionError):
|
|
_validate_configuration_result(drifted)
|
|
|
|
def test_doctor_schema_rejects_duplicate_inventory_and_inflated_summary(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent = Path(directory)
|
|
project = Project.open(self.copy_fixture(parent))
|
|
result = run_doctor(
|
|
project,
|
|
"codex",
|
|
config_path=parent / "missing.toml",
|
|
)
|
|
validator = Draft202012Validator(DOCTOR_SCHEMA)
|
|
duplicate = json.loads(json.dumps(result))
|
|
duplicate["checks"][1]["check_id"] = duplicate["checks"][0]["check_id"]
|
|
self.assertTrue(list(validator.iter_errors(duplicate)))
|
|
inflated = json.loads(json.dumps(result))
|
|
inflated["summary"]["failed"] = 99
|
|
self.assertTrue(list(validator.iter_errors(inflated)))
|
|
|
|
|
|
class GeneratedClientLaunchTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_generated_command_starts_real_project_bound_stdio_server(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory) / "alpha"
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
|
generated = generate_client_configuration(Project.open(root), "codex")
|
|
binding = generated["binding"]
|
|
parameters = StdioServerParameters(
|
|
command=binding["command"],
|
|
args=binding["args"],
|
|
)
|
|
async with (
|
|
stdio_client(parameters) as (read_stream, write_stream),
|
|
ClientSession(read_stream, write_stream) as session,
|
|
):
|
|
await session.initialize()
|
|
tools = await session.list_tools()
|
|
bootstrap = await session.call_tool("docforge_bootstrap", {})
|
|
self.assertEqual(list(READ_TOOLS), [tool.name for tool in tools.tools])
|
|
self.assertEqual("ok", bootstrap.structuredContent["status"])
|
|
self.assertEqual(
|
|
"read", bootstrap.structuredContent["effective_policy"]["capability_mode"]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|