ed
3155518305
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.
80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
"""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)) |