""" Integration tests for ChatGPT-compatible MCP tools. Tests the complete flow of search and fetch tools designed for ChatGPT integration, ensuring they properly wrap Basic Memory's MCP tools and return OpenAI-compatible MCP content array format. """ import json import pytest from fastmcp import Client def extract_mcp_json_content(mcp_result): """ Helper to extract JSON content from MCP CallToolResult. FastMCP auto-serializes our List[Dict[str, Any]] return values, so we need to: 1. Get the content list from the CallToolResult 2. Parse the JSON string in the text field (which is our serialized list) 3. Extract the actual JSON from the MCP content array structure """ content_list = mcp_result.content mcp_content_list = json.loads(content_list[0].text) return json.loads(mcp_content_list[0]["text"]) @pytest.mark.asyncio async def test_chatgpt_search_basic(mcp_server, app, test_project): """Test basic ChatGPT search functionality with MCP content array format.""" async with Client(mcp_server) as client: # Create test notes for searching await client.call_tool( "write_note", { "project": test_project.name, "title": "Machine Learning Fundamentals", "directory": "ai", "content": ( "# Machine Learning Fundamentals\n\nIntroduction to ML concepts and algorithms." ), "tags": "ml,ai,fundamentals", }, ) await client.call_tool( "write_note", { "project": test_project.name, "title": "Deep Learning with PyTorch", "directory": "ai", "content": ( "# Deep Learning with PyTorch\n\n" "Building neural networks using PyTorch framework." ), "tags": "pytorch,deep-learning,ai", }, ) await client.call_tool( "write_note", { "project": test_project.name, "title": "Data Visualization Guide", "directory": "data", "content": ( "# Data Visualization Guide\n\nCreating charts and graphs for data analysis." ), "tags": "visualization,data,charts", }, ) # Test ChatGPT search tool search_result = await client.call_tool( "search", { "query": "Machine Learning", }, ) # Extract JSON content from MCP result results_json = extract_mcp_json_content(search_result) assert "results" in results_json assert len(results_json["results"]) > 0 # Check result structure first_result = results_json["results"][0] assert "id" in first_result assert "title" in first_result assert "url" in first_result # Verify correct content found — target note must be present titles = [r["title"] for r in results_json["results"]] assert "Machine Learning Fundamentals" in titles @pytest.mark.asyncio async def test_chatgpt_search_empty_results(mcp_server, app, test_project): """Test ChatGPT search with no matching results.""" async with Client(mcp_server) as client: # Search for non-existent content search_result = await client.call_tool( "search", { "query": "NonExistentTopic12345", }, ) # Extract JSON content from MCP result results_json = extract_mcp_json_content(search_result) assert "results" in results_json assert len(results_json["results"]) == 0 assert results_json["query"] == "NonExistentTopic12345" @pytest.mark.asyncio async def test_chatgpt_search_with_boolean_operators(mcp_server, app, test_project): """Test ChatGPT search with boolean operators.""" async with Client(mcp_server) as client: # Create test notes await client.call_tool( "write_note", { "project": test_project.name, "title": "Python Web Frameworks", "directory": "dev", "content": ( "# Python Web Frameworks\n\nComparing Django and Flask for web development." ), "tags": "python,web,frameworks", }, ) await client.call_tool( "write_note", { "project": test_project.name, "title": "JavaScript Frameworks", "directory": "dev", "content": "# JavaScript Frameworks\n\nReact, Vue, and Angular comparison.", "tags": "javascript,web,frameworks", }, ) # Test with AND operator search_result = await client.call_tool( "search", { "query": "Python AND frameworks", }, ) results_json = extract_mcp_json_content(search_result) titles = [r["title"] for r in results_json["results"]] # Python note must appear; JS note may also appear since FTS # tokenizes broadly on shared terms like "frameworks" assert "Python Web Frameworks" in titles @pytest.mark.asyncio async def test_chatgpt_fetch_document(mcp_server, app, test_project): """Test ChatGPT fetch tool for retrieving full document content.""" async with Client(mcp_server) as client: # Create a test note note_content = """# Advanced Python Techniques ## Overview This document covers advanced Python programming techniques. ## Topics Covered - Decorators - Context Managers - Metaclasses - Async/Await patterns ## Code Examples ```python def my_decorator(func): def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper ``` """ await client.call_tool( "write_note", { "project": test_project.name, "title": "Advanced Python Techniques", "directory": "programming", "content": note_content, "tags": "python,advanced,programming", }, ) # Fetch the document using its title fetch_result = await client.call_tool( "fetch", { "id": "Advanced Python Techniques", }, ) # Extract JSON content from MCP result document_json = extract_mcp_json_content(fetch_result) assert "id" in document_json assert "title" in document_json assert "text" in document_json assert "url" in document_json assert "metadata" in document_json # Verify content assert document_json["title"] == "Advanced Python Techniques" assert "Decorators" in document_json["text"] assert "Context Managers" in document_json["text"] assert "def my_decorator" in document_json["text"] @pytest.mark.asyncio async def test_chatgpt_fetch_by_permalink(mcp_server, app, test_project): """Test ChatGPT fetch using permalink identifier.""" async with Client(mcp_server) as client: # Create a note with known content await client.call_tool( "write_note", { "project": test_project.name, "title": "Test Document", "directory": "test", "content": "# Test Document\n\nThis is test content for permalink fetching.", "tags": "test", }, ) # First search to get the permalink search_result = await client.call_tool( "search", { "query": "Test Document", }, ) results_json = extract_mcp_json_content(search_result) assert len(results_json["results"]) > 0 permalink = results_json["results"][0]["id"] # Fetch using the permalink fetch_result = await client.call_tool( "fetch", { "id": permalink, }, ) # Verify the fetched document document_json = extract_mcp_json_content(fetch_result) assert document_json["id"] == permalink assert "Test Document" in document_json["title"] assert "test content for permalink fetching" in document_json["text"] @pytest.mark.asyncio async def test_chatgpt_fetch_nonexistent_document(mcp_server, app, test_project): """Test ChatGPT fetch with non-existent document ID.""" async with Client(mcp_server) as client: # Try to fetch a non-existent document fetch_result = await client.call_tool( "fetch", { "id": "NonExistentDocument12345", }, ) # Extract JSON content from MCP result document_json = extract_mcp_json_content(fetch_result) # Should have document structure even for errors assert "id" in document_json assert "title" in document_json assert "text" in document_json # Check for error indication assert document_json["id"] == "NonExistentDocument12345" assert "Not Found" in document_json["text"] or "not found" in document_json["text"] @pytest.mark.asyncio async def test_chatgpt_fetch_with_empty_title(mcp_server, app, test_project): """Test ChatGPT fetch handles documents with empty or missing titles.""" async with Client(mcp_server) as client: # Create a note without a title in the content await client.call_tool( "write_note", { "project": test_project.name, "title": "untitled-note", "directory": "misc", "content": "This is content without a markdown header.\n\nJust plain text.", "tags": "misc", }, ) # Fetch the document fetch_result = await client.call_tool( "fetch", { "id": "untitled-note", }, ) # Parse JSON response document_json = extract_mcp_json_content(fetch_result) # Should have a title even if content doesn't have one assert "title" in document_json assert document_json["title"] != "" assert document_json["title"] is not None assert "content without a markdown header" in document_json["text"] @pytest.mark.asyncio async def test_chatgpt_search_pagination_default(mcp_server, app, test_project): """Test that ChatGPT search uses reasonable pagination defaults.""" async with Client(mcp_server) as client: # Create more than 10 notes to test pagination for i in range(15): await client.call_tool( "write_note", { "project": test_project.name, "title": f"Test Note {i}", "directory": "bulk", "content": f"# Test Note {i}\n\nThis is test content number {i}.", "tags": "test,bulk", }, ) # Search should return max 10 results by default search_result = await client.call_tool( "search", { "query": "Test Note", }, ) results_json = extract_mcp_json_content(search_result) # Should have at most 10 results (the default page_size) assert len(results_json["results"]) <= 10 assert results_json["total_count"] <= 10 @pytest.mark.asyncio async def test_chatgpt_tools_error_handling(mcp_server, app, test_project): """Test error handling in ChatGPT tools returns proper MCP format.""" async with Client(mcp_server) as client: # Test search with invalid query (if validation exists) # Using empty query to potentially trigger an error search_result = await client.call_tool( "search", { "query": "", # Empty query might cause an error }, ) # Should still return MCP content array format assert hasattr(search_result, "content") content_list = search_result.content assert isinstance(content_list, list) assert len(content_list) == 1 assert content_list[0].type == "text" # Should be valid JSON even on error results_json = extract_mcp_json_content(search_result) assert "results" in results_json # Should have results key even if empty @pytest.mark.asyncio async def test_chatgpt_integration_workflow(mcp_server, app, test_project): """Test complete workflow: search then fetch, as ChatGPT would use it.""" async with Client(mcp_server) as client: # Step 1: Create multiple documents docs = [ { "title": "API Design Best Practices", "content": ( "# API Design Best Practices\n\nRESTful API design principles and patterns." ), "tags": "api,rest,design", }, { "title": "GraphQL vs REST", "content": "# GraphQL vs REST\n\nComparing GraphQL and REST API architectures.", "tags": "api,graphql,rest", }, { "title": "Database Design Patterns", "content": ( "# Database Design Patterns\n\n" "Common database design patterns and anti-patterns." ), "tags": "database,design,patterns", }, ] for doc in docs: await client.call_tool( "write_note", { "project": test_project.name, "title": doc["title"], "directory": "architecture", "content": doc["content"], "tags": doc["tags"], }, ) # Step 2: Search for API-related content (as ChatGPT would) search_result = await client.call_tool( "search", { "query": "API", }, ) results_json = extract_mcp_json_content(search_result) assert len(results_json["results"]) >= 2 # Step 3: Fetch one of the search results (as ChatGPT would) first_result_id = results_json["results"][0]["id"] fetch_result = await client.call_tool( "fetch", { "id": first_result_id, }, ) document_json = extract_mcp_json_content(fetch_result) # Verify the fetched document matches search result assert document_json["id"] == first_result_id assert "API" in document_json["text"] or "api" in document_json["text"].lower() # Verify document has expected structure assert document_json["metadata"]["format"] == "markdown"