79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
|
|
def load_module(root: Path, docs: Path, runtime: Path, state: Path):
|
|
os.environ.update({
|
|
"ATHENA_REPO_ROOT": str(root),
|
|
"ATHENA_DOCS_ROOT": str(docs),
|
|
"ATHENA_RUNTIME_FILE": str(runtime),
|
|
"ATHENA_CONTEXT_STATE": str(state),
|
|
"ATHENA_DOC_WRITE_MODE": "enabled",
|
|
})
|
|
source = Path(__file__).parents[1] / "platform/mcp/platform_context_mcp.py"
|
|
spec = importlib.util.spec_from_file_location("platform_context_mcp_test", source)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
base = Path(tmp)
|
|
repo = base / "repo"
|
|
docs = base / "docs"
|
|
state = base / "state"
|
|
runtime = base / "runtime.json"
|
|
(repo / "docs").mkdir(parents=True)
|
|
(repo / "config").mkdir(parents=True)
|
|
docs.mkdir()
|
|
(repo / "docs/PLATFORM_OVERVIEW.md").write_text("# Athena\nRouter and recovery.\n")
|
|
(repo / "docs/OPERATIONS.md").write_text("# Operations\nUse bounded tools.\n")
|
|
(docs / "PLATFORM_OVERVIEW.md").write_text("# Athena\nRouter and recovery.\n")
|
|
(repo / "config/service-catalog.json").write_text(json.dumps({
|
|
"version": 1,
|
|
"services": [{"id": "test", "name": "Test", "address": "127.0.0.1", "port": 9, "protocol": "tcp", "probe": "tcp"}],
|
|
}))
|
|
runtime.write_text(json.dumps({"generated_unix": 4102444800, "generated_at": "2100-01-01T00:00:00Z", "source_commit": "abc", "containers": []}))
|
|
m = load_module(repo, docs, runtime, state)
|
|
|
|
assert len(m.TOOLS) == 10
|
|
assert m.overview()["source"] == "docs/PLATFORM_OVERVIEW.md"
|
|
assert m.current_state()["available"] is True
|
|
assert m.external_services()["services"][0]["id"] == "test"
|
|
assert m.search_knowledge({"query": "recovery", "max_results": 3})["count"] >= 1
|
|
assert "Operations" in m.read_source({"path": "docs/OPERATIONS.md"})["content"]
|
|
try:
|
|
m.safe_repo_path("config/secret.env")
|
|
raise AssertionError("secret path was accepted")
|
|
except ValueError:
|
|
pass
|
|
|
|
proposal = m.prepare_update({
|
|
"summary": "Update overview test",
|
|
"evidence": "Verified synthetic test state",
|
|
"updates": [{"path": "docs/PLATFORM_OVERVIEW.md", "content": "# Athena\nUpdated safely.\n"}],
|
|
})
|
|
assert (docs / "PLATFORM_OVERVIEW.md").read_text().endswith("recovery.\n")
|
|
result = m.apply_update({"proposal_id": proposal["proposal_id"], "confirmation": proposal["required_confirmation"]})
|
|
assert result["documentation_applied"] is True
|
|
assert result["git_commit_complete"] is False
|
|
assert (docs / "PLATFORM_OVERVIEW.md").read_text().endswith("safely.\n")
|
|
assert m.maintenance_status()["latest_applied_documentation_change"]
|
|
|
|
for tool in m.TOOLS:
|
|
for prop in tool["inputSchema"].get("properties", {}).values():
|
|
pattern = prop.get("pattern")
|
|
if pattern:
|
|
assert pattern.startswith("^") and pattern.endswith("$")
|
|
print("PLATFORM_CONTEXT_MCP_TEST_OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|