diff --git a/cognee-mcp/src/test_client.py b/cognee-mcp/src/test_client.py index 23160d8b2..bce7f807f 100755 --- a/cognee-mcp/src/test_client.py +++ b/cognee-mcp/src/test_client.py @@ -3,7 +3,7 @@ Test client for Cognee MCP Server functionality. This script tests all the tools and functions available in the Cognee MCP server, -including cognify, codify, search, prune, status checks, and utility functions. +including cognify, search, prune, status checks, and utility functions. Usage: # Set your OpenAI API key first @@ -23,6 +23,7 @@ import tempfile import time from contextlib import asynccontextmanager from cognee.shared.logging_utils import setup_logging +from logging import ERROR, INFO from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client @@ -35,7 +36,7 @@ from src.server import ( load_class, ) -# Set timeout for cognify/codify to complete in +# Set timeout for cognify to complete in TIMEOUT = 5 * 60 # 5 min in seconds @@ -151,12 +152,9 @@ DEBUG = True expected_tools = { "cognify", - "codify", "search", "prune", "cognify_status", - "codify_status", - "cognee_add_developer_rules", "list_data", "delete", } @@ -247,106 +245,6 @@ DEBUG = True } print(f"❌ {test_name} test failed: {e}") - async def test_codify(self): - """Test the codify functionality using MCP client.""" - print("\n🧪 Testing codify functionality...") - try: - async with self.mcp_server_session() as session: - codify_result = await session.call_tool( - "codify", arguments={"repo_path": self.test_repo_dir} - ) - - start = time.time() # mark the start - while True: - try: - # Wait a moment - await asyncio.sleep(5) - - # Check if codify processing is finished - status_result = await session.call_tool("codify_status", arguments={}) - if hasattr(status_result, "content") and status_result.content: - status_text = ( - status_result.content[0].text - if status_result.content - else str(status_result) - ) - else: - status_text = str(status_result) - - if str(PipelineRunStatus.DATASET_PROCESSING_COMPLETED) in status_text: - break - elif time.time() - start > TIMEOUT: - raise TimeoutError("Codify did not complete in 5min") - except DatabaseNotCreatedError: - if time.time() - start > TIMEOUT: - raise TimeoutError("Database was not created in 5min") - - self.test_results["codify"] = { - "status": "PASS", - "result": codify_result, - "message": "Codify executed successfully", - } - print("✅ Codify test passed") - - except Exception as e: - self.test_results["codify"] = { - "status": "FAIL", - "error": str(e), - "message": "Codify test failed", - } - print(f"❌ Codify test failed: {e}") - - async def test_cognee_add_developer_rules(self): - """Test the cognee_add_developer_rules functionality using MCP client.""" - print("\n🧪 Testing cognee_add_developer_rules functionality...") - try: - async with self.mcp_server_session() as session: - result = await session.call_tool( - "cognee_add_developer_rules", arguments={"base_path": self.test_data_dir} - ) - - start = time.time() # mark the start - while True: - try: - # Wait a moment - await asyncio.sleep(5) - - # Check if developer rule cognify processing is finished - status_result = await session.call_tool("cognify_status", arguments={}) - if hasattr(status_result, "content") and status_result.content: - status_text = ( - status_result.content[0].text - if status_result.content - else str(status_result) - ) - else: - status_text = str(status_result) - - if str(PipelineRunStatus.DATASET_PROCESSING_COMPLETED) in status_text: - break - elif time.time() - start > TIMEOUT: - raise TimeoutError( - "Cognify of developer rules did not complete in 5min" - ) - except DatabaseNotCreatedError: - if time.time() - start > TIMEOUT: - raise TimeoutError("Database was not created in 5min") - - self.test_results["cognee_add_developer_rules"] = { - "status": "PASS", - "result": result, - "message": "Developer rules addition executed successfully", - } - print("✅ Developer rules test passed") - - except Exception as e: - self.test_results["cognee_add_developer_rules"] = { - "status": "FAIL", - "error": str(e), - "message": "Developer rules test failed", - } - print(f"❌ Developer rules test failed: {e}") - async def test_search_functionality(self): """Test the search functionality with different search types using MCP client.""" print("\n🧪 Testing search functionality...") @@ -359,7 +257,11 @@ DEBUG = True # Go through all Cognee search types for search_type in SearchType: # Don't test these search types - if search_type in [SearchType.NATURAL_LANGUAGE, SearchType.CYPHER]: + if search_type in [ + SearchType.NATURAL_LANGUAGE, + SearchType.CYPHER, + SearchType.TRIPLET_COMPLETION, + ]: break try: async with self.mcp_server_session() as session: @@ -681,9 +583,6 @@ class TestModel: test_name="Cognify2", ) - await self.test_codify() - await self.test_cognee_add_developer_rules() - # Test list_data and delete functionality await self.test_list_data() await self.test_delete() @@ -739,7 +638,5 @@ async def main(): if __name__ == "__main__": - from logging import ERROR - logger = setup_logging(log_level=ERROR) asyncio.run(main())