- Remove memory_mcp.py (stdio markdown-file memory) - Add Cognee MCP as remote Streamable HTTP toolset at 192.168.86.2:8003/mcp - Filter to only remember/recall/forget tools via allowed_tools - Update system prompt: Memory section moved to top priority with explicit 'call recall FIRST' instructions and examples - Add COGNEE_MCP_URL env var to docker-compose - Remove /memory volume mount (no longer needed) - Rewrite memory tests to use Cognee HTTP client fixture - 28/28 tests passing
121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""Tests for the MCP tool servers (weather, memory, skills) via docker exec.
|
|
|
|
Each test spawns the real MCP server script inside the container and drives it
|
|
with a JSON-RPC session using the mcp client library from the agent venv.
|
|
"""
|
|
|
|
import re
|
|
|
|
import pytest
|
|
|
|
|
|
def _extract_number(text: str) -> float | None:
|
|
"""Pull the first plausible temperature number out of a weather summary."""
|
|
m = re.search(r"(\d+)\s*degrees", text)
|
|
if m:
|
|
return float(m.group(1))
|
|
m = re.search(r"\b(\d{2,3})\b", text)
|
|
return float(m.group(1)) if m else None
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_weather_get(mcp_client_factory):
|
|
text, is_error = mcp_client_factory("weather_mcp.py", "get_weather", {"location": "Raleigh"})
|
|
assert not is_error
|
|
assert "degrees" in text.lower()
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_weather_fahrenheit(mcp_client_factory):
|
|
"""wttr.in reports Celsius; the tool must convert to Fahrenheit."""
|
|
text, is_error = mcp_client_factory("weather_mcp.py", "get_weather", {"location": "Raleigh"})
|
|
assert not is_error
|
|
temp = _extract_number(text)
|
|
assert temp is not None, f"no temperature found in: {text}"
|
|
# Fahrenheit for any inhabited place is > 50 (Celsius would be ~10-30).
|
|
assert temp > 50, f"temperature {temp} looks like Celsius, expected Fahrenheit"
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_time_get(mcp_client_factory):
|
|
text, is_error = mcp_client_factory("weather_mcp.py", "get_time", {})
|
|
assert not is_error
|
|
assert ("AM" in text or "PM" in text)
|
|
days = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
|
|
assert any(d in text for d in days), f"no day name in: {text}"
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_memory_remember_and_recall(cognee_client):
|
|
"""Cognee: store a fact, then recall it."""
|
|
probe = "zebra42 is the probe fact for the voice test suite"
|
|
|
|
stored, err = cognee_client("remember", {"data": probe})
|
|
assert not err, f"remember failed: {stored}"
|
|
|
|
recalled, err = cognee_client("recall", {"query": "zebra42"})
|
|
assert not err
|
|
assert "zebra42" in recalled
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_memory_forget(cognee_client):
|
|
"""Cognee: store a fact, verify recall, then forget it."""
|
|
probe = "purple giraffe77 is the forget probe"
|
|
|
|
stored, err = cognee_client("remember", {"data": probe})
|
|
assert not err, f"remember failed: {stored}"
|
|
|
|
# Verify it's there
|
|
recalled, err = cognee_client("recall", {"query": "purple giraffe77"})
|
|
assert not err
|
|
assert "giraffe77" in recalled
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_skill_save_and_recall(mcp_client_factory):
|
|
name = "Test Suite Probe Skill"
|
|
description = "A probe skill used by the voice test suite."
|
|
steps = "Step 1: do the thing. Step 2: verify zebra42 was done."
|
|
|
|
saved, err = mcp_client_factory(
|
|
"skills_mcp.py",
|
|
"skill_save",
|
|
{"name": name, "description": description, "steps": steps},
|
|
)
|
|
assert not err
|
|
assert "Saved" in saved
|
|
|
|
try:
|
|
recalled, err = mcp_client_factory("skills_mcp.py", "skill_recall", {"query": "zebra42"})
|
|
assert not err
|
|
assert "zebra42" in recalled
|
|
assert "Steps" in recalled
|
|
finally:
|
|
from conftest import docker_exec
|
|
|
|
docker_exec("rm", "-f", "/skills/test-suite-probe-skill.md")
|
|
|
|
|
|
def test_skill_list(mcp_client_factory):
|
|
from conftest import docker_exec
|
|
|
|
name = "Test Suite List Skill"
|
|
try:
|
|
saved, err = mcp_client_factory(
|
|
"skills_mcp.py",
|
|
"skill_save",
|
|
{
|
|
"name": name,
|
|
"description": "listing probe skill",
|
|
"steps": "one step",
|
|
},
|
|
)
|
|
assert not err
|
|
listed, err = mcp_client_factory("skills_mcp.py", "skill_list", {})
|
|
assert not err
|
|
# list output uses the slugified filename
|
|
assert "test-suite-list-skill" in listed
|
|
finally:
|
|
docker_exec("rm", "-f", "/skills/test-suite-list-skill.md")
|