flux1-edit/debug_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FLUX.1 Edit MCP Server Debug Test

This script helps diagnose issues with the MCP server
"""

# Force UTF-8 encoding setup
import os
import sys
import locale

# Force UTF-8 environment variables
os.environ['PYTHONIOENCODING'] = 'utf-8'
os.environ['PYTHONUTF8'] = '1'
os.environ['LC_ALL'] = 'C.UTF-8'

# Windows-specific UTF-8 setup
if sys.platform.startswith('win'):
    try:
        os.system('chcp 65001 >nul 2>&1')
    except Exception:
        pass
    
    try:
        locale.setlocale(locale.LC_ALL, 'C.UTF-8')
    except locale.Error:
        try:
            locale.setlocale(locale.LC_ALL, '')
        except locale.Error:
            pass

import traceback

# Add paths
current_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(current_dir, 'src'))
sys.path.insert(0, current_dir)

def test_imports():
    """Test all required imports"""
    print("=== Import Tests ===")
    
    # Test MCP imports
    try:
        import mcp.types as types
        from mcp.server import Server
        from mcp.server.stdio import stdio_server
        print("[SUCCESS] MCP imports successful")
    except ImportError as e:
        print(f"[ERROR] MCP import failed: {e}")
        return False
    
    # Test local imports
    try:
        from src.connector.config import Config
        print("[SUCCESS] Config import successful")
    except ImportError as e:
        print(f"[ERROR] Config import failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False
    
    try:
        from src.server.models import TOOL_DEFINITIONS, ToolName
        print("[SUCCESS] Models import successful")
    except ImportError as e:
        print(f"[ERROR] Models import failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False
    
    try:
        from src.server.handlers import ToolHandlers
        print("[SUCCESS] Handlers import successful")
    except ImportError as e:
        print(f"[ERROR] Handlers import failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False
    
    # Test aiohttp
    try:
        import aiohttp
        print(f"[SUCCESS] aiohttp version: {aiohttp.__version__}")
    except ImportError as e:
        print(f"[ERROR] aiohttp import failed: {e}")
        return False
    
    return True

def test_config():
    """Test configuration loading"""
    print("\n=== Configuration Tests ===")
    
    try:
        from src.connector.config import Config
        config = Config()
        print("[SUCCESS] Config created successfully")
        
        # Check API key
        if config.api_key:
            print(f"[SUCCESS] API key configured: ***{config.api_key[-4:]}")
        else:
            print("[WARNING] API key not configured")
        
        # Check paths
        print(f"[SUCCESS] Input path: {config.input_path}")
        print(f"[SUCCESS] Output path: {config.generated_images_path}")
        
        # Test validation
        if config.validate():
            print("[SUCCESS] Configuration validation passed")
        else:
            print("[ERROR] Configuration validation failed")
            
        return True
        
    except Exception as e:
        print(f"[ERROR] Configuration test failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False

def test_handlers():
    """Test handler creation"""
    print("\n=== Handler Tests ===")
    
    try:
        from src.connector.config import Config
        from src.server.handlers import ToolHandlers
        
        config = Config()
        handlers = ToolHandlers(config)
        print("[SUCCESS] Handlers created successfully")
        
        return True
        
    except Exception as e:
        print(f"[ERROR] Handler test failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False

def test_server_creation():
    """Test MCP server creation"""
    print("\n=== Server Creation Tests ===")
    
    try:
        from mcp.server import Server
        
        server = Server("flux1-edit-test")
        print("[SUCCESS] MCP server created successfully")
        
        return True
        
    except Exception as e:
        print(f"[ERROR] Server creation failed: {e}")
        print(f"Error details: {traceback.format_exc()}")
        return False

def main():
    """Run all tests"""
    print("FLUX.1 Edit MCP Server Debug Test")
    print("=" * 50)
    
    tests = [
        test_imports,
        test_config,
        test_handlers,
        test_server_creation
    ]
    
    passed = 0
    for test in tests:
        try:
            if test():
                passed += 1
        except Exception as e:
            print(f"[ERROR] Test crashed: {e}")
    
    print(f"\n=== Results ===")
    print(f"Passed: {passed}/{len(tests)}")
    
    if passed == len(tests):
        print("[SUCCESS] All tests passed! The server should work.")
        return 0
    else:
        print("[ERROR] Some tests failed. Check the errors above.")
        return 1

if __name__ == "__main__":
    sys.exit(main())