Files
markitect-tool/tests/test_memory_graph_contracts.py

221 lines
8.0 KiB
Python

import json
from pathlib import Path
import yaml
from click.testing import CliRunner
from markitect_tool.cli import main
from markitect_tool.memory import (
compile_memory_graph_selection_to_context_package,
load_memory_graph_file,
load_memory_graph_selection_file,
load_memory_profile_file,
plan_memory_profile,
validate_memory_graph,
validate_memory_profile,
)
def test_memory_profile_validation_and_plan(tmp_path: Path):
profile_path = _write_profile(tmp_path)
profile = load_memory_profile_file(profile_path)
result = validate_memory_profile(profile, path=profile_path)
plan = plan_memory_profile(profile)
assert result.valid
assert result.metadata["memory_kinds"] == ["reasoning", "knowledge", "package"]
assert plan["runtime_boundary"]["services_launched_by_markitect_tool"] is False
assert plan["activation"]["max_items"] == 4
def test_memory_graph_validation_and_context_package_compile(tmp_path: Path):
profile_path = _write_profile(tmp_path)
graph_path = _write_graph(tmp_path)
selection_path = _write_selection(tmp_path)
graph = load_memory_graph_file(graph_path)
profile = load_memory_profile_file(profile_path)
selection = load_memory_graph_selection_file(selection_path)
result = validate_memory_graph(graph, path=graph_path)
package = compile_memory_graph_selection_to_context_package(graph, selection, profile=profile)
assert result.valid
assert package.title == "Decision package"
assert len(package.items) == 1
assert package.items[0].source.path == "decision.md"
assert package.metadata["memory_graph"]["selected_edges"] == ["edge.supports"]
assert package.retrieval_recipes[0].kind == "memory-graph-selection"
def test_memory_graph_example_fixtures_cover_reasoning_conversation_and_knowledge():
examples = Path("examples/memory")
profile = load_memory_profile_file(examples / "memory-profile.local.yaml")
for graph_name, selection_name, package_title, expected_node_count in [
(
"decision-graph.yaml",
"decision-graph-selection.yaml",
"Memory Contract Boundary Package",
3,
),
(
"conversation-path.yaml",
"conversation-path-selection.yaml",
"Conversation Episode Package",
4,
),
(
"knowledge-neighborhood.yaml",
"knowledge-neighborhood-selection.yaml",
"Knowledge Neighborhood Package",
6,
),
]:
graph = load_memory_graph_file(examples / graph_name)
selection = load_memory_graph_selection_file(examples / selection_name)
validation = validate_memory_graph(graph, path=examples / graph_name)
package = compile_memory_graph_selection_to_context_package(graph, selection, profile=profile)
assert validation.valid, validation.to_dict()
assert package.title == package_title
assert package.metadata["memory_graph"]["selected_nodes"] == selection.node_ids
assert len(selection.node_ids) == expected_node_count
assert package.metadata["memory_profile"]["id"] == "local-agent-memory"
def test_memory_graph_invalid_example_fixtures_emit_errors():
examples = Path("examples/memory")
graph = load_memory_graph_file(examples / "invalid-memory-graph.yaml")
profile = load_memory_profile_file(examples / "invalid-memory-profile.yaml")
graph_result = validate_memory_graph(graph, path=examples / "invalid-memory-graph.yaml")
profile_result = validate_memory_profile(profile, path=examples / "invalid-memory-profile.yaml")
graph_codes = {diagnostic.code for diagnostic in graph_result.diagnostics}
profile_codes = {diagnostic.code for diagnostic in profile_result.diagnostics}
assert not graph_result.valid
assert "memory.graph.node.unknown_kind" in graph_codes
assert "memory.graph.edge.unknown_target" in graph_codes
assert "memory.graph.event.unknown_kind" in graph_codes
assert not profile_result.valid
assert "memory.profile.unknown_kind" in profile_codes
assert "memory.profile.store_missing" in profile_codes
def test_mkt_memory_blueprint_and_graph_cli(tmp_path: Path):
_write_profile(tmp_path)
_write_graph(tmp_path)
selection_path = _write_selection(tmp_path)
runner = CliRunner()
blueprint = runner.invoke(
main,
["memory", "blueprint", "validate", str(tmp_path / "profile.yaml"), "--format", "json"],
)
plan = runner.invoke(
main,
["memory", "blueprint", "plan", str(tmp_path / "profile.yaml"), "--format", "json"],
)
graph = runner.invoke(
main,
["memory", "graph", "validate", str(tmp_path / "graph.yaml"), "--format", "json"],
)
packed = runner.invoke(main, ["memory", "graph", "pack", str(selection_path), "--format", "json"])
assert blueprint.exit_code == 0, blueprint.output
assert json.loads(blueprint.output)["valid"] is True
assert plan.exit_code == 0, plan.output
assert json.loads(plan.output)["profile"]["id"] == "test-profile"
assert graph.exit_code == 0, graph.output
assert json.loads(graph.output)["metadata"]["nodes"] == 1
assert packed.exit_code == 0, packed.output
package = json.loads(packed.output)
assert package["title"] == "Decision package"
assert package["metadata"]["memory_profile"]["id"] == "test-profile"
def _write_profile(root: Path) -> Path:
path = root / "profile.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.profile.v1",
"id": "test-profile",
"title": "Test profile",
"intent": "Exercise local memory profile contracts.",
"memory_kinds": ["reasoning", "knowledge", "package"],
"stores": {
"reasoning": "event-log",
"knowledge": "graph-store",
"package": "context-registry",
},
"activation": {"max_items": 4, "max_tokens": 1000},
},
sort_keys=False,
),
encoding="utf-8",
)
return path
def _write_graph(root: Path) -> Path:
path = root / "graph.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.graph.v1",
"id": "test-graph",
"title": "Test graph",
"intent": "Compile one decision into context.",
"nodes": [
{
"id": "decision.keep-contract-local",
"kind": "decision",
"text": "Keep memory graph contracts in markitect-tool.",
"source_spans": [
{
"path": "decision.md",
"unit_kind": "section",
"selector": "sections[heading=Decision]",
"engine": "selector",
}
],
}
],
"edges": [
{
"id": "edge.supports",
"kind": "supports",
"source": "decision.keep-contract-local",
"target": "decision.keep-contract-local",
}
],
},
sort_keys=False,
),
encoding="utf-8",
)
return path
def _write_selection(root: Path) -> Path:
path = root / "selection.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.selection.v1",
"graph": "graph.yaml",
"profile": "profile.yaml",
"title": "Decision package",
"intent": "Activate graph decisions.",
"node_ids": ["decision.keep-contract-local"],
"budget": {"max_items": 2},
},
sort_keys=False,
),
encoding="utf-8",
)
return path