test(aggregate_directives): assert clean bodies + pollution-fix + max_chars cap + MCP impl

8 new importable-function tests in tests/test_aggregate_directives.py: - test_importable_function_returns_clean_body - test_importable_function_no_meta_md_pollution - test_importable_function_max_chars_truncates_with_suffix - test_importable_function_max_chars_zero_is_unlimited - test_importable_function_max_chars_above_size_is_unlimited - test_importable_function_missing_preset_raises - test_importable_function_missing_v1_in_preset_raises - test_importable_function_accepts_relative_preset_path 7 new MCP-server tests in tests/test_mcp_aggregate_directives.py: - test_mcp_server_exposes_aggregate_directives_spec - test_mcp_server_impl_default_preset - test_mcp_server_impl_explicit_preset_path - test_mcp_server_impl_max_chars_truncates - test_mcp_server_impl_missing_preset_returns_error_string - test_mcp_server_impl_zero_max_chars_is_unlimited - test_mcp_server_list_tools_includes_aggregate_directives Tests load modules via importlib.util.spec_from_file_location so the scripts/mcp_server.py main guard is not triggered during testing. sandbox-policy compliance: all write paths use tmp_path. 22 tests pass (15 aggregate_directives + 7 MCP); 7 pre-existing CLI tests still pass byte-identical.
2026-07-03 10:49:24 -04:00
parent a527f6224f
commit 3155518305
2 changed files with 160 additions and 1 deletions
@@ -5,6 +5,9 @@ These tests exercise the script via subprocess to verify the contract:
 - emits clean bodies separated by '====' header
 - exits 1 on missing files
 - emits to stdout by default; -o writes to file
+
+The importable aggregate_directives() function tests cover the API surface used
+by the MCP server (scripts/mcp_server.py:aggregate_directives tool).
 """
 from __future__ import annotations

@@ -12,6 +15,8 @@ import subprocess
 import sys
 from pathlib import Path

+import pytest
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SCRIPT = REPO_ROOT / "scripts" / "aggregate_directives.py"
 PRESET = REPO_ROOT / "conductor" / "directives" / "presets" / "current_baseline.md"
@@ -91,4 +96,78 @@ def test_every_v1_has_top_level_heading() -> None:
 def test_v1_heading_is_not_meta_provenance_format() -> None:
 for path in _iter_v1_files():
  first_line = path.read_text(encoding="utf-8").splitlines()[0]
-  assert not first_line.strip().endswith("- v1"), path.parent.name + " uses the meta.md provenance heading format in v1.md body"
+  assert not first_line.strip().endswith("- v1"), path.parent.name + " uses the meta.md provenance heading format in v1.md body"
+
+
+def _import_aggregate_directives_module():
+ import importlib.util
+ spec = importlib.util.spec_from_file_location("aggregate_directives_under_test", SCRIPT)
+ if spec is None or spec.loader is None:
+  raise RuntimeError("failed to build import spec for " + str(SCRIPT))
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+def test_importable_function_returns_clean_body() -> None:
+ mod = _import_aggregate_directives_module()
+ body = mod.aggregate_directives(str(PRESET))
+ assert "atomic_per_task_commits" in body
+ assert "ban_dict_any" in body
+ assert "knowledge_harvest_pattern" in body
+ assert body.count("========================================") >= 51 * 2
+
+
+def test_importable_function_no_meta_md_pollution() -> None:
+ mod = _import_aggregate_directives_module()
+ body = mod.aggregate_directives(str(PRESET))
+ assert "**Why this iteration:**" not in body
+ assert "**Source:**" not in body
+ assert "**Lifted:**" not in body
+ assert "## v1" not in body
+ assert "Provenance" not in body
+
+
+def test_importable_function_max_chars_truncates_with_suffix() -> None:
+ mod = _import_aggregate_directives_module()
+ body = mod.aggregate_directives(str(PRESET), max_chars=2000)
+ assert len(body) > 2000, "truncated body must include suffix text exceeding the cap"
+ assert "truncated" in body.lower()
+
+
+def test_importable_function_max_chars_zero_is_unlimited() -> None:
+ mod = _import_aggregate_directives_module()
+ full = mod.aggregate_directives(str(PRESET))
+ same = mod.aggregate_directives(str(PRESET), max_chars=0)
+ assert same == full
+
+
+def test_importable_function_max_chars_above_size_is_unlimited() -> None:
+ mod = _import_aggregate_directives_module()
+ full = mod.aggregate_directives(str(PRESET))
+ capped = mod.aggregate_directives(str(PRESET), max_chars=len(full) + 5000)
+ assert capped == full
+
+
+def test_importable_function_missing_preset_raises() -> None:
+ mod = _import_aggregate_directives_module()
+ bad = REPO_ROOT / "conductor" / "directives" / "presets" / "_nonexistent_preset_xyz.md"
+ with pytest.raises(FileNotFoundError) as exc_info:
+  mod.aggregate_directives(str(bad))
+ assert "preset file not found" in str(exc_info.value)
+
+
+def test_importable_function_missing_v1_in_preset_raises(tmp_path: Path) -> None:
+ mod = _import_aggregate_directives_module()
+ bad = tmp_path / "bad_preset.md"
+ bad.write_text("- missing_v1_xyz: conductor/directives/missing_v1_xyz/v1.md\n", encoding="utf-8")
+ with pytest.raises(FileNotFoundError) as exc_info:
+  mod.aggregate_directives(str(bad))
+ assert "missing_v1_xyz" in str(exc_info.value)
+
+
+def test_importable_function_accepts_relative_preset_path() -> None:
+ mod = _import_aggregate_directives_module()
+ rel = "conductor/directives/presets/current_baseline.md"
+ body = mod.aggregate_directives(rel, project_root=str(REPO_ROOT))
+ assert "ban_dict_any" in body
@@ -0,0 +1,80 @@
+"""Tests for the aggregate_directives MCP tool exposed by scripts/mcp_server.py.
+
+The MCP server module is loaded via importlib so its `if __name__ == "__main__"`
+guard does not run main(); we only exercise the tool-registration and dispatch
+machinery directly.
+"""
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+MCP_SERVER = REPO_ROOT / "scripts" / "mcp_server.py"
+PRESET = REPO_ROOT / "conductor" / "directives" / "presets" / "current_baseline.md"
+
+
+def _load_mcp_server_module():
+ spec = importlib.util.spec_from_file_location("mcp_server_under_test", MCP_SERVER)
+ if spec is None or spec.loader is None:
+  raise RuntimeError("failed to build import spec for " + str(MCP_SERVER))
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+def test_mcp_server_exposes_aggregate_directives_spec() -> None:
+ mod = _load_mcp_server_module()
+ assert hasattr(mod, "AGGREGATE_DIRECTIVES_SPEC")
+ spec = mod.AGGREGATE_DIRECTIVES_SPEC
+ assert spec["name"] == "aggregate_directives"
+ assert "preset_path" in spec["parameters"]["properties"]
+ assert "max_chars" in spec["parameters"]["properties"]
+ assert spec["parameters"]["properties"]["max_chars"]["type"] == "integer"
+
+
+def test_mcp_server_impl_default_preset() -> None:
+ mod = _load_mcp_server_module()
+ body = mod._aggregate_directives_impl("conductor/directives/presets/current_baseline.md")
+ assert "atomic_per_task_commits" in body
+ assert "ban_dict_any" in body
+ assert "**Why this iteration:**" not in body
+
+
+def test_mcp_server_impl_explicit_preset_path() -> None:
+ mod = _load_mcp_server_module()
+ body = mod._aggregate_directives_impl(str(PRESET))
+ assert "ban_dict_any" in body
+
+
+def test_mcp_server_impl_max_chars_truncates() -> None:
+ mod = _load_mcp_server_module()
+ body = mod._aggregate_directives_impl(str(PRESET), 1500)
+ assert len(body) > 1500, "truncated body must include suffix text exceeding the cap"
+ assert "truncated" in body.lower()
+
+
+def test_mcp_server_impl_missing_preset_returns_error_string() -> None:
+ mod = _load_mcp_server_module()
+ body = mod._aggregate_directives_impl("conductor/directives/presets/_nonexistent_xyz.md")
+ assert body.startswith("ERROR")
+ assert "preset file not found" in body or "not found" in body.lower()
+
+
+def test_mcp_server_impl_zero_max_chars_is_unlimited() -> None:
+ mod = _load_mcp_server_module()
+ full = mod._aggregate_directives_impl(str(PRESET), 0)
+ capped = mod._aggregate_directives_impl(str(PRESET), 0)
+ assert full == capped
+ assert "truncated" not in full.lower()
+
+
+def test_mcp_server_list_tools_includes_aggregate_directives() -> None:
+ import asyncio
+ mod = _load_mcp_server_module()
+ tools = asyncio.run(mod.list_tools())
+ names = [t.name for t in tools]
+ assert "aggregate_directives" in names
+ assert "run_powershell" in names
+ spec_count = len(mod.mcp_tool_specs.get_tool_schemas())
+ assert len(tools) == spec_count + 2, "expected " + str(spec_count + 2) + " tools (specs + run_powershell + aggregate_directives); got " + str(len(tools))