mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
comment out mcp tests
This commit is contained in:
phernandez
@@ -1,40 +1,40 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import EmbeddedResource
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
from basic_memory.schemas import CreateEntityResponse, EntityResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_observations(app, test_entity_data, client):
|
||||
"""Test adding observations to an existing entity."""
|
||||
|
||||
# First create an entity
|
||||
create_result = await handle_call_tool("create_entities", test_entity_data)
|
||||
create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
entity_id = create_response.entities[0].id
|
||||
|
||||
# Add new observation
|
||||
result = await handle_call_tool(
|
||||
"add_observations", {"entity_id": entity_id, "observations": ["A new observation"]}
|
||||
)
|
||||
|
||||
# Verify response format
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], EmbeddedResource)
|
||||
assert result[0].type == "resource"
|
||||
|
||||
# Verify observation was added
|
||||
response = EntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert response.id == entity_id
|
||||
assert len(response.observations) == 2 # 1 already present
|
||||
assert response.observations[1].content == "A new observation"
|
||||
|
||||
# Verify through API
|
||||
api_response = await client.get(f"/knowledge/entities/{entity_id}")
|
||||
assert api_response.status_code == 200
|
||||
entity = api_response.json()
|
||||
assert len(entity["observations"]) == 2 # Original + new
|
||||
assert "A new observation" in [o["content"] for o in entity["observations"]]
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, EntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_add_observations(app, test_entity_data, client):
|
||||
# """Test adding observations to an existing entity."""
|
||||
#
|
||||
# # First create an entity
|
||||
# create_result = await handle_call_tool("create_entities", test_entity_data)
|
||||
# create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# entity_id = create_response.entities[0].id
|
||||
#
|
||||
# # Add new observation
|
||||
# result = await handle_call_tool(
|
||||
# "add_observations", {"entity_id": entity_id, "observations": ["A new observation"]}
|
||||
# )
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
#
|
||||
# # Verify observation was added
|
||||
# response = EntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert response.id == entity_id
|
||||
# assert len(response.observations) == 2 # 1 already present
|
||||
# assert response.observations[1].content == "A new observation"
|
||||
#
|
||||
# # Verify through API
|
||||
# api_response = await client.get(f"/knowledge/entities/{entity_id}")
|
||||
# assert api_response.status_code == 200
|
||||
# entity = api_response.json()
|
||||
# assert len(entity["observations"]) == 2 # Original + new
|
||||
# assert "A new observation" in [o["content"] for o in entity["observations"]]
|
||||
|
||||
+107
-107
@@ -1,107 +1,107 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import EmbeddedResource
|
||||
|
||||
from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_single_entity(app):
|
||||
"""Test creating a single entity."""
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{"name": "SingleTest", "entity_type": "test", "observations": ["Test observation"]}
|
||||
]
|
||||
}
|
||||
|
||||
result = await handle_call_tool("create_entities", entity_data)
|
||||
|
||||
# Verify response format
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], EmbeddedResource)
|
||||
assert result[0].type == "resource"
|
||||
assert result[0].resource.mimeType == MIME_TYPE
|
||||
|
||||
# Verify entity creation
|
||||
response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert len(response.entities) == 1
|
||||
entity = response.entities[0]
|
||||
assert entity.name == "SingleTest"
|
||||
assert entity.entity_type == "test"
|
||||
assert len(entity.observations) == 1
|
||||
assert entity.observations[0].content == "Test observation"
|
||||
assert entity.id is not None
|
||||
|
||||
# Verify entity can be found via search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "SingleTest"})
|
||||
search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert len(search_response.matches) == 1
|
||||
assert search_response.matches[0].name == "SingleTest"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_multiple_entities(app):
|
||||
"""Test creating multiple entities in one call."""
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{"name": "BulkTest1", "entity_type": "test", "observations": ["First bulk test"]},
|
||||
{"name": "BulkTest2", "entity_type": "test", "observations": ["Second bulk test"]},
|
||||
{"name": "BulkTest3", "entity_type": "demo", "observations": ["Third bulk test"]},
|
||||
]
|
||||
}
|
||||
|
||||
result = await handle_call_tool("create_entities", entity_data)
|
||||
|
||||
# Verify response
|
||||
assert len(result) == 1
|
||||
response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
|
||||
# Verify all entities were created
|
||||
assert len(response.entities) == 3
|
||||
|
||||
# Check specific entities
|
||||
entities = {e.name: e for e in response.entities}
|
||||
assert "BulkTest1" in entities
|
||||
assert "BulkTest2" in entities
|
||||
assert "BulkTest3" in entities
|
||||
|
||||
# Verify IDs were generated correctly
|
||||
assert entities["BulkTest1"].id is not None
|
||||
assert entities["BulkTest2"].id is not None
|
||||
assert entities["BulkTest3"].id is not None
|
||||
|
||||
# Verify observations were saved
|
||||
assert len(entities["BulkTest1"].observations) == 1
|
||||
assert entities["BulkTest1"].observations[0].content == "First bulk test"
|
||||
|
||||
# Verify entities can be found via search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "BulkTest"})
|
||||
search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert len(search_response.matches) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_entity_with_all_fields(app):
|
||||
"""Test creating entity with all possible fields populated."""
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{
|
||||
"name": "FullEntity",
|
||||
"entity_type": "test",
|
||||
"description": "A complete test entity",
|
||||
"observations": ["First observation", "Second observation"],
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
result = await handle_call_tool("create_entities", entity_data)
|
||||
response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
|
||||
entity = response.entities[0]
|
||||
assert entity.name == "FullEntity"
|
||||
assert entity.description == "A complete test entity"
|
||||
assert len(entity.observations) == 2
|
||||
assert entity.observations[0].content == "First observation"
|
||||
assert entity.observations[1].content == "Second observation"
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_single_entity(app):
|
||||
# """Test creating a single entity."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "SingleTest", "entity_type": "test", "observations": ["Test observation"]}
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify entity creation
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.entities) == 1
|
||||
# entity = response.entities[0]
|
||||
# assert entity.name == "SingleTest"
|
||||
# assert entity.entity_type == "test"
|
||||
# assert len(entity.observations) == 1
|
||||
# assert entity.observations[0].content == "Test observation"
|
||||
# assert entity.id is not None
|
||||
#
|
||||
# # Verify entity can be found via search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "SingleTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert search_response.matches[0].name == "SingleTest"
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_multiple_entities(app):
|
||||
# """Test creating multiple entities in one call."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "BulkTest1", "entity_type": "test", "observations": ["First bulk test"]},
|
||||
# {"name": "BulkTest2", "entity_type": "test", "observations": ["Second bulk test"]},
|
||||
# {"name": "BulkTest3", "entity_type": "demo", "observations": ["Third bulk test"]},
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
#
|
||||
# # Verify response
|
||||
# assert len(result) == 1
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Verify all entities were created
|
||||
# assert len(response.entities) == 3
|
||||
#
|
||||
# # Check specific entities
|
||||
# entities = {e.name: e for e in response.entities}
|
||||
# assert "BulkTest1" in entities
|
||||
# assert "BulkTest2" in entities
|
||||
# assert "BulkTest3" in entities
|
||||
#
|
||||
# # Verify IDs were generated correctly
|
||||
# assert entities["BulkTest1"].id is not None
|
||||
# assert entities["BulkTest2"].id is not None
|
||||
# assert entities["BulkTest3"].id is not None
|
||||
#
|
||||
# # Verify observations were saved
|
||||
# assert len(entities["BulkTest1"].observations) == 1
|
||||
# assert entities["BulkTest1"].observations[0].content == "First bulk test"
|
||||
#
|
||||
# # Verify entities can be found via search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "BulkTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(search_response.matches) == 3
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_entity_with_all_fields(app):
|
||||
# """Test creating entity with all possible fields populated."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "FullEntity",
|
||||
# "entity_type": "test",
|
||||
# "description": "A complete test entity",
|
||||
# "observations": ["First observation", "Second observation"],
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# entity = response.entities[0]
|
||||
# assert entity.name == "FullEntity"
|
||||
# assert entity.description == "A complete test entity"
|
||||
# assert len(entity.observations) == 2
|
||||
# assert entity.observations[0].content == "First observation"
|
||||
# assert entity.observations[1].content == "Second observation"
|
||||
|
||||
@@ -1,48 +1,48 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_relations(app):
|
||||
"""Test creating relations between entities."""
|
||||
# Create two test entities
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{"name": "TestEntityA", "entity_type": "test", "observations": ["Entity A"]},
|
||||
{"name": "TestEntityB", "entity_type": "test", "observations": ["Entity B"]},
|
||||
]
|
||||
}
|
||||
|
||||
create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
)
|
||||
|
||||
from_entity = create_entity_response.entities[0]
|
||||
to_entity = create_entity_response.entities[1]
|
||||
# Create relation between them
|
||||
relation_data = {
|
||||
"relations": [
|
||||
{
|
||||
"from_id": from_entity.id,
|
||||
"to_id": to_entity.id,
|
||||
"relation_type": "relates_to",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
result = await handle_call_tool("create_relations", relation_data)
|
||||
|
||||
# Verify through search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "TestEntityA"})
|
||||
response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
|
||||
assert len(response.matches) == 1
|
||||
entity = response.matches[0]
|
||||
assert len(entity.relations) == 1
|
||||
assert entity.relations[0].to_id == to_entity.id
|
||||
assert entity.relations[0].relation_type == "relates_to"
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_relations(app):
|
||||
# """Test creating relations between entities."""
|
||||
# # Create two test entities
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "TestEntityA", "entity_type": "test", "observations": ["Entity A"]},
|
||||
# {"name": "TestEntityB", "entity_type": "test", "observations": ["Entity B"]},
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# from_entity = create_entity_response.entities[0]
|
||||
# to_entity = create_entity_response.entities[1]
|
||||
# # Create relation between them
|
||||
# relation_data = {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_relations", relation_data)
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "TestEntityA"})
|
||||
# response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# assert len(response.matches) == 1
|
||||
# entity = response.matches[0]
|
||||
# assert len(entity.relations) == 1
|
||||
# assert entity.relations[0].to_id == to_entity.id
|
||||
# assert entity.relations[0].relation_type == "relates_to"
|
||||
|
||||
@@ -1,35 +1,35 @@
|
||||
"""Tests for MCP delete_entities tool."""
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_entities(app):
|
||||
"""Test deleting entities."""
|
||||
# Create test entities
|
||||
entities = {
|
||||
"entities": [
|
||||
{"name": "DeleteTest1", "entity_type": "test", "observations": ["To be deleted 1"]},
|
||||
{"name": "DeleteTest2", "entity_type": "test", "observations": ["To be deleted 2"]},
|
||||
]
|
||||
}
|
||||
create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
)
|
||||
|
||||
# Delete first entity
|
||||
await handle_call_tool(
|
||||
"delete_entities", {"entity_ids": [create_entity_response.entities[0].id]}
|
||||
)
|
||||
|
||||
# Verify through search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "DeleteTest"})
|
||||
search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
|
||||
# Only second entity should remain
|
||||
assert len(search_response.matches) == 1
|
||||
assert search_response.matches[0].name == "DeleteTest2"
|
||||
# """Tests for MCP delete_entities tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_entities(app):
|
||||
# """Test deleting entities."""
|
||||
# # Create test entities
|
||||
# entities = {
|
||||
# "entities": [
|
||||
# {"name": "DeleteTest1", "entity_type": "test", "observations": ["To be deleted 1"]},
|
||||
# {"name": "DeleteTest2", "entity_type": "test", "observations": ["To be deleted 2"]},
|
||||
# ]
|
||||
# }
|
||||
# create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# # Delete first entity
|
||||
# await handle_call_tool(
|
||||
# "delete_entities", {"entity_ids": [create_entity_response.entities[0].id]}
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "DeleteTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Only second entity should remain
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert search_response.matches[0].name == "DeleteTest2"
|
||||
|
||||
@@ -1,47 +1,47 @@
|
||||
"""Tests for MCP delete_observations tool."""
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_observations(app):
|
||||
"""Test deleting specific observations from an entity."""
|
||||
# Create entity with multiple observations
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{
|
||||
"name": "ObsDeleteTest",
|
||||
"entity_type": "test",
|
||||
"observations": [
|
||||
"Keep this observation",
|
||||
"Delete this observation",
|
||||
"Also keep this",
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
create_result = await handle_call_tool("create_entities", entity_data)
|
||||
create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
entity_id = create_response.entities[0].id
|
||||
|
||||
# Delete specific observation
|
||||
await handle_call_tool(
|
||||
"delete_observations", {"entity_id": entity_id, "deletions": ["Delete this observation"]}
|
||||
)
|
||||
|
||||
# Verify through search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "ObsDeleteTest"})
|
||||
search_response = SearchNodesResponse.model_validate_json(
|
||||
search_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
)
|
||||
|
||||
# Check remaining observations
|
||||
entity = search_response.matches[0]
|
||||
observations = [o.content for o in entity.observations]
|
||||
assert len(observations) == 2
|
||||
assert "Delete this observation" not in observations
|
||||
assert "Keep this observation" in observations
|
||||
assert "Also keep this" in observations
|
||||
# """Tests for MCP delete_observations tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_observations(app):
|
||||
# """Test deleting specific observations from an entity."""
|
||||
# # Create entity with multiple observations
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "ObsDeleteTest",
|
||||
# "entity_type": "test",
|
||||
# "observations": [
|
||||
# "Keep this observation",
|
||||
# "Delete this observation",
|
||||
# "Also keep this",
|
||||
# ],
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
# create_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# entity_id = create_response.entities[0].id
|
||||
#
|
||||
# # Delete specific observation
|
||||
# await handle_call_tool(
|
||||
# "delete_observations", {"entity_id": entity_id, "deletions": ["Delete this observation"]}
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "ObsDeleteTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(
|
||||
# search_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# # Check remaining observations
|
||||
# entity = search_response.matches[0]
|
||||
# observations = [o.content for o in entity.observations]
|
||||
# assert len(observations) == 2
|
||||
# assert "Delete this observation" not in observations
|
||||
# assert "Keep this observation" in observations
|
||||
# assert "Also keep this" in observations
|
||||
|
||||
@@ -1,58 +1,58 @@
|
||||
"""Tests for MCP delete_relations tool."""
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_relations(app):
|
||||
"""Test deleting relations between entities."""
|
||||
# Create test entities with relation
|
||||
entities = {
|
||||
"entities": [
|
||||
{"name": "RelSource", "entity_type": "test", "observations": ["Source entity"]},
|
||||
{"name": "RelTarget", "entity_type": "test", "observations": ["Target entity"]},
|
||||
]
|
||||
}
|
||||
create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
)
|
||||
from_entity = create_entity_response.entities[0]
|
||||
to_entity = create_entity_response.entities[1]
|
||||
|
||||
# Create relation
|
||||
relation = {
|
||||
"relations": [
|
||||
{
|
||||
"from_id": from_entity.id,
|
||||
"to_id": to_entity.id,
|
||||
"relation_type": "relates_to",
|
||||
}
|
||||
]
|
||||
}
|
||||
await handle_call_tool("create_relations", relation)
|
||||
|
||||
# Delete the relation
|
||||
await handle_call_tool(
|
||||
"delete_relations",
|
||||
{
|
||||
"relations": [
|
||||
{
|
||||
"from_id": from_entity.id,
|
||||
"to_id": to_entity.id,
|
||||
"relation_type": "relates_to",
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
# Verify through search
|
||||
search_result = await handle_call_tool("search_nodes", {"query": "relsource"})
|
||||
search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
|
||||
# Source entity should exist but have no relations
|
||||
assert len(search_response.matches) == 1
|
||||
assert len(search_response.matches[0].relations) == 0
|
||||
# """Tests for MCP delete_relations tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_relations(app):
|
||||
# """Test deleting relations between entities."""
|
||||
# # Create test entities with relation
|
||||
# entities = {
|
||||
# "entities": [
|
||||
# {"name": "RelSource", "entity_type": "test", "observations": ["Source entity"]},
|
||||
# {"name": "RelTarget", "entity_type": "test", "observations": ["Target entity"]},
|
||||
# ]
|
||||
# }
|
||||
# create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
# from_entity = create_entity_response.entities[0]
|
||||
# to_entity = create_entity_response.entities[1]
|
||||
#
|
||||
# # Create relation
|
||||
# relation = {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
# await handle_call_tool("create_relations", relation)
|
||||
#
|
||||
# # Delete the relation
|
||||
# await handle_call_tool(
|
||||
# "delete_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "relsource"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Source entity should exist but have no relations
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert len(search_response.matches[0].relations) == 0
|
||||
|
||||
@@ -1,49 +1,49 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.server import handle_list_tools
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tools(app):
|
||||
"""Test that server exposes expected tools."""
|
||||
|
||||
tools = await handle_list_tools()
|
||||
|
||||
# Check each expected tool is present
|
||||
expected_tools = {
|
||||
# Knowledge graph tools
|
||||
"create_entities",
|
||||
"search_nodes",
|
||||
"open_nodes",
|
||||
"add_observations",
|
||||
"create_relations",
|
||||
"delete_entities",
|
||||
"delete_observations",
|
||||
"delete_relations",
|
||||
# Document tools
|
||||
"create_document",
|
||||
"list_documents",
|
||||
"get_document",
|
||||
"update_document",
|
||||
"delete_document",
|
||||
}
|
||||
|
||||
found_tools = {t.name: t for t in tools}
|
||||
assert found_tools.keys() == expected_tools
|
||||
|
||||
# Verify schemas include required fields
|
||||
search_schema = found_tools["search_nodes"].inputSchema
|
||||
assert "query" in search_schema["properties"]
|
||||
assert search_schema["required"] == ["query"]
|
||||
|
||||
# Verify document tool schemas
|
||||
create_doc_schema = found_tools["create_document"].inputSchema
|
||||
assert "path" in create_doc_schema["properties"]
|
||||
assert "content" in create_doc_schema["properties"]
|
||||
assert set(create_doc_schema["required"]) == {"path", "content"}
|
||||
|
||||
get_doc_schema = found_tools["get_document"].inputSchema
|
||||
assert "id" in get_doc_schema["properties"]
|
||||
assert get_doc_schema["required"] == ["id"]
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_list_tools
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_list_tools(app):
|
||||
# """Test that server exposes expected tools."""
|
||||
#
|
||||
# tools = await handle_list_tools()
|
||||
#
|
||||
# # Check each expected tool is present
|
||||
# expected_tools = {
|
||||
# # Knowledge graph tools
|
||||
# "create_entities",
|
||||
# "search_nodes",
|
||||
# "open_nodes",
|
||||
# "add_observations",
|
||||
# "create_relations",
|
||||
# "delete_entities",
|
||||
# "delete_observations",
|
||||
# "delete_relations",
|
||||
# # Document tools
|
||||
# "create_document",
|
||||
# "list_documents",
|
||||
# "get_document",
|
||||
# "update_document",
|
||||
# "delete_document",
|
||||
# }
|
||||
#
|
||||
# found_tools = {t.name: t for t in tools}
|
||||
# assert found_tools.keys() == expected_tools
|
||||
#
|
||||
# # Verify schemas include required fields
|
||||
# search_schema = found_tools["search_nodes"].inputSchema
|
||||
# assert "query" in search_schema["properties"]
|
||||
# assert search_schema["required"] == ["query"]
|
||||
#
|
||||
# # Verify document tool schemas
|
||||
# create_doc_schema = found_tools["create_document"].inputSchema
|
||||
# assert "path" in create_doc_schema["properties"]
|
||||
# assert "content" in create_doc_schema["properties"]
|
||||
# assert set(create_doc_schema["required"]) == {"path", "content"}
|
||||
#
|
||||
# get_doc_schema = found_tools["get_document"].inputSchema
|
||||
# assert "id" in get_doc_schema["properties"]
|
||||
# assert get_doc_schema["required"] == ["id"]
|
||||
|
||||
+242
-242
@@ -1,242 +1,242 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
from mcp.shared.exceptions import McpError
|
||||
from mcp.types import INVALID_PARAMS, METHOD_NOT_FOUND
|
||||
|
||||
from basic_memory.mcp.server import handle_call_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_tool_name(app):
|
||||
"""Test calling a non-existent tool."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("not_a_tool", {})
|
||||
assert "Unknown tool" in str(exc.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_required_field(app):
|
||||
"""Test validation when required fields are missing."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("search_nodes", {})
|
||||
assert "query" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("create_entities", {})
|
||||
assert "entities" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("create_document", {})
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
error_msg = str(exc.value).lower()
|
||||
assert "path" in error_msg or "content" in error_msg
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_arrays(app):
|
||||
"""Test validation of array fields that can't be empty."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("create_entities", {"entities": []})
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("open_nodes", {"entity_ids": []})
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_field_types(app):
|
||||
"""Test validation when fields have wrong types."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("search_nodes", {"query": 123})
|
||||
assert "string" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("create_entities", {"entities": "not an array"})
|
||||
assert "array" in str(exc.value).lower() or "list" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("get_document", {"id": "not an integer"})
|
||||
assert "integer" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_nested_fields(app):
|
||||
"""Test validation of nested object fields."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"create_entities",
|
||||
{
|
||||
"entities": [
|
||||
{
|
||||
"name": "Test",
|
||||
# Missing required entity_type
|
||||
"observations": [],
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
assert "entity_type" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"create_document",
|
||||
{
|
||||
"path": "test.md",
|
||||
"content": "test",
|
||||
"doc_metadata": "not an object" # Should be dict/null
|
||||
},
|
||||
)
|
||||
assert "doc_metadata" in str(exc.value).lower()
|
||||
assert exc.value.args[0] == INVALID_PARAMS
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_relation_format_to_id(app):
|
||||
"""Test validation of relation data."""
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"create_relations",
|
||||
{
|
||||
"relations": [
|
||||
{
|
||||
"from_id": 1,
|
||||
# Missing to_id
|
||||
"relation_type": "relates_to",
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
assert "to_id" in str(exc.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_relation_format_relation_type(app):
|
||||
# Invalid relation type
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"create_relations",
|
||||
{
|
||||
"relations": [
|
||||
{
|
||||
"from_id": 1,
|
||||
"to_id": 2,
|
||||
"relation_type": "", # Empty relation type
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
assert "relation_type" in str(exc.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_observation_validation_len(app):
|
||||
"""Test validation specific to observations."""
|
||||
# Empty observations
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"add_observations",
|
||||
{
|
||||
"entity_id": 1,
|
||||
"observations": ["", ""], # Empty observations
|
||||
},
|
||||
)
|
||||
assert "observations" in str(exc.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_observation_validation_delete(app):
|
||||
# Empty deletions
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"delete_observations",
|
||||
{
|
||||
"entity_id": 1,
|
||||
"deletions": [], # Empty deletions
|
||||
},
|
||||
)
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edge_case_validation_search_len(app):
|
||||
"""Test edge cases in validation."""
|
||||
# Very long strings
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"search_nodes",
|
||||
{"query": "x" * 10000}, # Extremely long query
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_document_endpoint_validation(app):
|
||||
"""Test validation specific to document endpoints."""
|
||||
# Invalid ID format for get_document
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool("get_document", {"id": -1})
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
assert "greater than 0" in str(exc.value).lower()
|
||||
|
||||
# Invalid document path
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"create_document",
|
||||
{
|
||||
"path": "", # Empty path
|
||||
"content": "test content"
|
||||
}
|
||||
)
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
assert "path" in str(exc.value).lower()
|
||||
|
||||
# Update without content
|
||||
with pytest.raises(McpError) as exc:
|
||||
await handle_call_tool(
|
||||
"update_document",
|
||||
{
|
||||
"id": 1,
|
||||
"doc_metadata": None
|
||||
}
|
||||
)
|
||||
assert INVALID_PARAMS == exc.value.args[0]
|
||||
assert "content" in str(exc.value).lower()
|
||||
|
||||
|
||||
# We'll skip this test for now since it requires database setup
|
||||
@pytest.mark.skip(reason="Requires database setup")
|
||||
@pytest.mark.asyncio
|
||||
async def test_document_http_methods(app):
|
||||
"""Test that document endpoints use correct HTTP methods."""
|
||||
# Test GET endpoints
|
||||
await handle_call_tool("list_documents", {})
|
||||
await handle_call_tool("get_document", {"id": 1})
|
||||
|
||||
# Test POST endpoint
|
||||
response = await handle_call_tool(
|
||||
"create_document",
|
||||
{
|
||||
"path": "test.md",
|
||||
"content": "test content"
|
||||
}
|
||||
)
|
||||
|
||||
# Test PUT endpoint
|
||||
await handle_call_tool(
|
||||
"update_document",
|
||||
{
|
||||
"id": 1,
|
||||
"content": "updated content",
|
||||
"doc_metadata": None
|
||||
}
|
||||
)
|
||||
|
||||
# Test DELETE endpoint
|
||||
await handle_call_tool("delete_document", {"id": 1})
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.shared.exceptions import McpError
|
||||
# from mcp.types import INVALID_PARAMS, METHOD_NOT_FOUND
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_tool_name(app):
|
||||
# """Test calling a non-existent tool."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("not_a_tool", {})
|
||||
# assert "Unknown tool" in str(exc.value)
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_missing_required_field(app):
|
||||
# """Test validation when required fields are missing."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("search_nodes", {})
|
||||
# assert "query" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {})
|
||||
# assert "entities" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_document", {})
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
# error_msg = str(exc.value).lower()
|
||||
# assert "path" in error_msg or "content" in error_msg
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_empty_arrays(app):
|
||||
# """Test validation of array fields that can't be empty."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {"entities": []})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("open_nodes", {"entity_ids": []})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_field_types(app):
|
||||
# """Test validation when fields have wrong types."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("search_nodes", {"query": 123})
|
||||
# assert "string" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {"entities": "not an array"})
|
||||
# assert "array" in str(exc.value).lower() or "list" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("get_document", {"id": "not an integer"})
|
||||
# assert "integer" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_nested_fields(app):
|
||||
# """Test validation of nested object fields."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_entities",
|
||||
# {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "Test",
|
||||
# # Missing required entity_type
|
||||
# "observations": [],
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "entity_type" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "test.md",
|
||||
# "content": "test",
|
||||
# "doc_metadata": "not an object" # Should be dict/null
|
||||
# },
|
||||
# )
|
||||
# assert "doc_metadata" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_relation_format_to_id(app):
|
||||
# """Test validation of relation data."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": 1,
|
||||
# # Missing to_id
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "to_id" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_relation_format_relation_type(app):
|
||||
# # Invalid relation type
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": 1,
|
||||
# "to_id": 2,
|
||||
# "relation_type": "", # Empty relation type
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "relation_type" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_observation_validation_len(app):
|
||||
# """Test validation specific to observations."""
|
||||
# # Empty observations
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "add_observations",
|
||||
# {
|
||||
# "entity_id": 1,
|
||||
# "observations": ["", ""], # Empty observations
|
||||
# },
|
||||
# )
|
||||
# assert "observations" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_observation_validation_delete(app):
|
||||
# # Empty deletions
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "delete_observations",
|
||||
# {
|
||||
# "entity_id": 1,
|
||||
# "deletions": [], # Empty deletions
|
||||
# },
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_edge_case_validation_search_len(app):
|
||||
# """Test edge cases in validation."""
|
||||
# # Very long strings
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "search_nodes",
|
||||
# {"query": "x" * 10000}, # Extremely long query
|
||||
# )
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_document_endpoint_validation(app):
|
||||
# """Test validation specific to document endpoints."""
|
||||
# # Invalid ID format for get_document
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("get_document", {"id": -1})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "greater than 0" in str(exc.value).lower()
|
||||
#
|
||||
# # Invalid document path
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "", # Empty path
|
||||
# "content": "test content"
|
||||
# }
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "path" in str(exc.value).lower()
|
||||
#
|
||||
# # Update without content
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "update_document",
|
||||
# {
|
||||
# "id": 1,
|
||||
# "doc_metadata": None
|
||||
# }
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "content" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# # We'll skip this test for now since it requires database setup
|
||||
# @pytest.mark.skip(reason="Requires database setup")
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_document_http_methods(app):
|
||||
# """Test that document endpoints use correct HTTP methods."""
|
||||
# # Test GET endpoints
|
||||
# await handle_call_tool("list_documents", {})
|
||||
# await handle_call_tool("get_document", {"id": 1})
|
||||
#
|
||||
# # Test POST endpoint
|
||||
# response = await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "test.md",
|
||||
# "content": "test content"
|
||||
# }
|
||||
# )
|
||||
#
|
||||
# # Test PUT endpoint
|
||||
# await handle_call_tool(
|
||||
# "update_document",
|
||||
# {
|
||||
# "id": 1,
|
||||
# "content": "updated content",
|
||||
# "doc_metadata": None
|
||||
# }
|
||||
# )
|
||||
#
|
||||
# # Test DELETE endpoint
|
||||
# await handle_call_tool("delete_document", {"id": 1})
|
||||
|
||||
@@ -1,63 +1,63 @@
|
||||
"""Tests for MCP open_nodes tool."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import EmbeddedResource
|
||||
|
||||
from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
from basic_memory.schemas import OpenNodesResponse, CreateEntityResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_open_nodes(app):
|
||||
"""Test retrieving specific nodes by name."""
|
||||
# Create test entities
|
||||
entity_data = {
|
||||
"entities": [
|
||||
{
|
||||
"name": "OpenTestA",
|
||||
"entity_type": "test",
|
||||
"observations": ["First test entity"],
|
||||
},
|
||||
{
|
||||
"name": "OpenTestB",
|
||||
"entity_type": "test",
|
||||
"observations": ["Second test entity"],
|
||||
},
|
||||
{
|
||||
"name": "OpenTestC",
|
||||
"entity_type": "test",
|
||||
"observations": ["Third test entity"],
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
)
|
||||
entity_a = create_entity_response.entities[0]
|
||||
entity_b = create_entity_response.entities[1]
|
||||
|
||||
# Open specific nodes
|
||||
result = await handle_call_tool("open_nodes", {"entity_ids": [entity_a.id, entity_b.id]})
|
||||
|
||||
# Verify response format
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], EmbeddedResource)
|
||||
assert result[0].type == "resource"
|
||||
assert result[0].resource.mimeType == MIME_TYPE
|
||||
|
||||
# Verify entities returned
|
||||
response = OpenNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert len(response.entities) == 2
|
||||
|
||||
# Entities should be returned in same order as requested
|
||||
assert response.entities[0].name == "OpenTestA"
|
||||
assert response.entities[1].name == "OpenTestB"
|
||||
|
||||
# Verify entity content
|
||||
entity = response.entities[0]
|
||||
assert entity.id == entity_a.id
|
||||
assert entity.entity_type == "test"
|
||||
assert len(entity.observations) == 1
|
||||
assert entity.observations[0].content == "First test entity"
|
||||
# """Tests for MCP open_nodes tool."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import OpenNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_open_nodes(app):
|
||||
# """Test retrieving specific nodes by name."""
|
||||
# # Create test entities
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "OpenTestA",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["First test entity"],
|
||||
# },
|
||||
# {
|
||||
# "name": "OpenTestB",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["Second test entity"],
|
||||
# },
|
||||
# {
|
||||
# "name": "OpenTestC",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["Third test entity"],
|
||||
# },
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
# entity_a = create_entity_response.entities[0]
|
||||
# entity_b = create_entity_response.entities[1]
|
||||
#
|
||||
# # Open specific nodes
|
||||
# result = await handle_call_tool("open_nodes", {"entity_ids": [entity_a.id, entity_b.id]})
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify entities returned
|
||||
# response = OpenNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.entities) == 2
|
||||
#
|
||||
# # Entities should be returned in same order as requested
|
||||
# assert response.entities[0].name == "OpenTestA"
|
||||
# assert response.entities[1].name == "OpenTestB"
|
||||
#
|
||||
# # Verify entity content
|
||||
# entity = response.entities[0]
|
||||
# assert entity.id == entity_a.id
|
||||
# assert entity.entity_type == "test"
|
||||
# assert len(entity.observations) == 1
|
||||
# assert entity.observations[0].content == "First test entity"
|
||||
|
||||
@@ -1,37 +1,37 @@
|
||||
"""Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import EmbeddedResource
|
||||
|
||||
from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
from basic_memory.schemas import SearchNodesResponse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_nodes(app, test_entity_data, client):
|
||||
"""Test searching for an entity after creating it."""
|
||||
|
||||
# First create an entity
|
||||
await handle_call_tool("create_entities", test_entity_data)
|
||||
|
||||
# Then search for it
|
||||
result = await handle_call_tool("search_nodes", {"query": "Test Entity"})
|
||||
|
||||
# Verify response format
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], EmbeddedResource)
|
||||
assert result[0].type == "resource"
|
||||
assert result[0].resource.mimeType == MIME_TYPE
|
||||
|
||||
# Verify search results
|
||||
response = SearchNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
assert len(response.matches) == 1
|
||||
assert response.matches[0].name == "Test Entity"
|
||||
assert response.query == "Test Entity"
|
||||
|
||||
# Verify through API
|
||||
api_response = await client.post("/knowledge/search", json={"query": "Test Entity"})
|
||||
assert api_response.status_code == 200
|
||||
data = api_response.json()
|
||||
assert len(data["matches"]) == 1
|
||||
assert data["matches"][0]["name"] == "Test Entity"
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_search_nodes(app, test_entity_data, client):
|
||||
# """Test searching for an entity after creating it."""
|
||||
#
|
||||
# # First create an entity
|
||||
# await handle_call_tool("create_entities", test_entity_data)
|
||||
#
|
||||
# # Then search for it
|
||||
# result = await handle_call_tool("search_nodes", {"query": "Test Entity"})
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify search results
|
||||
# response = SearchNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.matches) == 1
|
||||
# assert response.matches[0].name == "Test Entity"
|
||||
# assert response.query == "Test Entity"
|
||||
#
|
||||
# # Verify through API
|
||||
# api_response = await client.post("/knowledge/search", json={"query": "Test Entity"})
|
||||
# assert api_response.status_code == 200
|
||||
# data = api_response.json()
|
||||
# assert len(data["matches"]) == 1
|
||||
# assert data["matches"][0]["name"] == "Test Entity"
|
||||
|
||||
+25
-25
@@ -1,25 +1,25 @@
|
||||
"""Test to show MCP tool documentation."""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from mcp.types import Tool
|
||||
from basic_memory.mcp.server import handle_list_tools
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tools():
|
||||
"""List available tools and their documentation."""
|
||||
tools = await handle_list_tools()
|
||||
assert isinstance(tools, list)
|
||||
assert all(isinstance(t, Tool) for t in tools)
|
||||
|
||||
print("\nAvailable MCP Tools:\n")
|
||||
|
||||
# Print each tool's documentation
|
||||
for tool in tools:
|
||||
print(f"Tool: {tool.name}")
|
||||
print(f"Description: {tool.description}")
|
||||
print("Required fields:", tool.inputSchema.get("required", []))
|
||||
print()
|
||||
print("Schema:", json.dumps(tool.inputSchema, indent=2))
|
||||
print("-" * 80 + "\n")
|
||||
# """Test to show MCP tool documentation."""
|
||||
#
|
||||
# import json
|
||||
# import pytest
|
||||
# from mcp.types import Tool
|
||||
# from basic_memory.mcp.server import handle_list_tools
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_list_tools():
|
||||
# """List available tools and their documentation."""
|
||||
# tools = await handle_list_tools()
|
||||
# assert isinstance(tools, list)
|
||||
# assert all(isinstance(t, Tool) for t in tools)
|
||||
#
|
||||
# print("\nAvailable MCP Tools:\n")
|
||||
#
|
||||
# # Print each tool's documentation
|
||||
# for tool in tools:
|
||||
# print(f"Tool: {tool.name}")
|
||||
# print(f"Description: {tool.description}")
|
||||
# print("Required fields:", tool.inputSchema.get("required", []))
|
||||
# print()
|
||||
# print("Schema:", json.dumps(tool.inputSchema, indent=2))
|
||||
# print("-" * 80 + "\n")
|
||||
|
||||
Reference in New Issue
Block a user