test_mcp_functionality.py.disabledā¢7.04 kB
"""
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)