"""
Simple MCP Server functionality test.
This test creates an MCP server and verifies that the basic
functionality works correctly.
"""
import sys
import os
# Add the src directory to the path to import our modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from quantalogic_markdown_mcp.mcp_server import MarkdownMCPServer
from quantalogic_markdown_mcp.safe_editor_types import ValidationLevel
def test_mcp_server_functionality():
"""Test basic MCP server functionality by directly calling tool methods."""
print("=== Testing MCP Server Functionality ===")
# Create server instance
print("1. Creating MCP server instance...")
server = MarkdownMCPServer("TestServer")
# Initialize document
test_markdown = """# Test Document
## Introduction
This is a test document for our MCP server.
## Features
- Section management
- Content editing
- Document validation
## Conclusion
The MCP server provides a robust interface for Markdown editing.
"""
server.initialize_document(test_markdown, ValidationLevel.NORMAL)
print("✓ Server initialized successfully")
# Test basic functionality by accessing the internal methods
# Note: In real MCP usage, these would be called through the MCP protocol
# Test that we can access the editor
with server.lock:
editor = server._ensure_editor()
sections = editor.get_sections()
print(f"✓ Found {len(sections)} sections in document")
# Show section information
for i, section in enumerate(sections):
print(f" Section {i+1}: '{section.title}' (Level {section.level})")
# Test document access
with server.lock:
doc_content = editor.to_markdown()
print(f"✓ Document content accessible ({len(doc_content)} characters)")
# Test metadata
metadata = server.document_metadata
print(f"✓ Document metadata: {list(metadata.keys())}")
# Test thread safety
import threading
errors = []
results = []
def test_concurrent_access():
try:
with server.lock:
editor = server._ensure_editor()
sections = editor.get_sections()
results.append(len(sections))
except Exception as e:
errors.append(e)
threads = []
for i in range(3):
t = threading.Thread(target=test_concurrent_access)
threads.append(t)
t.start()
for t in threads:
t.join()
if errors:
print(f"❌ Thread safety errors: {errors}")
return False
else:
print(f"✓ Thread safety test passed ({len(results)} successful concurrent accesses)")
# Test that server structure is correct
assert hasattr(server, 'mcp')
assert hasattr(server, 'editor')
assert hasattr(server, 'lock')
assert hasattr(server, 'document_metadata')
print("✓ Server structure validation passed")
print("\n=== Simulating MCP Tool Operations ===")
# We can't easily test the actual MCP tools without running the full MCP protocol,
# but we can test the underlying functionality
# Test section operations through the editor
with server.lock:
# Test getting sections (simulates list_sections tool)
sections = editor.get_sections()
print(f"✓ list_sections equivalent: {len(sections)} sections")
# Test getting document (simulates get_document tool)
document = editor.to_markdown()
print(f"✓ get_document equivalent: {len(document)} characters")
# Test section updates (simulates update_section tool)
if sections:
first_section = sections[0]
# This simulates what the update_section tool would do
result = editor.update_section_content(
section_ref=first_section,
content="Updated content for testing.",
preserve_subsections=True
)
if result.success:
print("✓ update_section equivalent: Section updated successfully")
# Rollback using transaction history
history = editor.get_transaction_history()
if history:
last_transaction = history[-1]
rollback_result = editor.rollback_transaction(last_transaction.transaction_id)
if rollback_result.success:
print("✓ rollback equivalent: Changes reverted")
else:
print(f"❌ rollback failed: {rollback_result.get_error_summary()}")
else:
print("⚠️ No transaction history available for rollback")
else:
print(f"❌ update_section failed: {result.get_error_summary()}")
# Test section insertion (simulates insert_section tool)
if sections:
# Use the first section as the after_section reference
after_section = sections[0]
result = editor.insert_section_after(
after_section=after_section,
level=2,
title="Test Inserted Section",
content="This is test content."
)
if result.success:
print("✓ insert_section equivalent: Section inserted successfully")
# Rollback using transaction history
history = editor.get_transaction_history()
if history:
last_transaction = history[-1]
rollback_result = editor.rollback_transaction(last_transaction.transaction_id)
if rollback_result.success:
print("✓ Insertion rollback: Changes reverted")
else:
print(f"❌ rollback failed: {rollback_result.get_error_summary()}")
else:
print("⚠️ No transaction history available for rollback")
else:
print(f"❌ insert_section failed: {result.get_error_summary()}")
print("\n=== Test Results ===")
print("✅ All basic functionality tests passed!")
print("✅ MCP server structure is correct")
print("✅ Thread safety is working")
print("✅ Document operations are functional")
print("\nThe MCP server is ready for deployment and testing with MCP clients.")
return True
if __name__ == "__main__":
try:
success = test_mcp_server_functionality()
if success:
print("\n🎉 MCP Server functionality test completed successfully!")
sys.exit(0)
else:
print("\n❌ MCP Server functionality test failed!")
sys.exit(1)
except Exception as e:
print(f"\n💥 Test failed with error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)