From 315551830581fc0774bafc25464c4e4d2e403200 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Fri, 3 Jul 2026 10:49:24 -0400 Subject: [PATCH] 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. --- tests/test_aggregate_directives.py | 81 +++++++++++++++++++++++++- tests/test_mcp_aggregate_directives.py | 80 +++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 tests/test_mcp_aggregate_directives.py diff --git a/tests/test_aggregate_directives.py b/tests/test_aggregate_directives.py index 429e2bf2..43ea5465 100644 --- a/tests/test_aggregate_directives.py +++ b/tests/test_aggregate_directives.py @@ -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" \ No newline at end of file + 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 \ No newline at end of file diff --git a/tests/test_mcp_aggregate_directives.py b/tests/test_mcp_aggregate_directives.py new file mode 100644 index 00000000..1eeedaf4 --- /dev/null +++ b/tests/test_mcp_aggregate_directives.py @@ -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)) \ No newline at end of file