test_new_features.py•2.6 kB
#!/usr/bin/env python3
"""
Test script for new PostgreSQL configuration and log features
"""
import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from services.mcp_resources_service import MCPResourcesService
from services.mcp_tools_service import MCPToolsService
async def test_pgconfig_resource():
"""Test the new PostgreSQL configuration resource"""
print("Testing PostgreSQL Configuration Resource...")
try:
resources_service = MCPResourcesService({})
result = await resources_service.read_resource("database://pgconfig")
if "error" in result.get("text", ""):
print("ERROR: PostgreSQL configuration resource failed:", result["text"][:200])
else:
print("SUCCESS: PostgreSQL configuration resource working!")
# Parse the JSON to check structure
import json
data = json.loads(result["text"])
print(f" Found {data['postgresql_configuration'].get('total_settings', 0)} configuration settings")
print(f" Server version: {data['postgresql_configuration']['server_info']['version'][:50]}...")
except Exception as e:
print("ERROR testing pgconfig resource:", str(e))
async def test_postgresql_logs_tool():
"""Test the new PostgreSQL logs tool"""
print("\nTesting PostgreSQL Logs Tool...")
try:
tools_service = MCPToolsService({})
# Test recent logs
result = await tools_service.get_postgresql_logs(log_type="recent", lines=3)
if result.get("success"):
print("SUCCESS: PostgreSQL logs tool working!")
print(f" Retrieved {len(result['log_data']['log_entries'])} log entries")
print(f" Active connections: {result['log_data']['summary']['active_connections']}")
else:
print("ERROR: PostgreSQL logs tool failed:", result.get("error", "Unknown error"))
# Test connections log type
result2 = await tools_service.get_postgresql_logs(log_type="connections", lines=1)
if result2.get("success"):
print("SUCCESS: PostgreSQL connections logs working!")
else:
print("ERROR: PostgreSQL connections logs failed:", result2.get("error", "Unknown error"))
except Exception as e:
print("ERROR testing logs tool:", str(e))
async def main():
print("Testing New PostgreSQL Features\n")
await test_pgconfig_resource()
await test_postgresql_logs_tool()
print("\nFeature testing complete!")
if __name__ == "__main__":
asyncio.run(main())