discord-fishbowl/test_system.py

#!/usr/bin/env python3
"""
Fixed test script for the Discord Fishbowl system
"""

import asyncio
import sys
import os
from pathlib import Path

# Set up proper Python path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))

# Set environment variables for testing
os.environ['DATABASE_URL'] = 'sqlite+aiosqlite:///fishbowl_test.db'
os.environ['ENVIRONMENT'] = 'development'
os.environ['LOG_LEVEL'] = 'INFO'

def test_imports():
    """Test all module imports"""
    print("🔧 Testing imports...")
    
    try:
        # Test basic imports
        from src.database import models
        from src.rag import vector_store
        from src.collaboration import creative_projects
        from src.utils import config
        print("✅ Basic imports successful")
        
        # Test specific classes
        from src.collaboration.creative_projects import CollaborativeCreativeManager
        from src.rag.vector_store import VectorStoreManager
        from src.rag.memory_sharing import MemorySharingManager
        print("✅ Class imports successful")
        
        return True
        
    except Exception as e:
        print(f"❌ Import failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_database():
    """Test database initialization"""
    print("\n🗄️  Testing database...")
    
    try:
        from src.database.connection import init_database, create_tables
        
        await init_database()
        await create_tables()
        print("✅ Database initialization successful")
        return True
        
    except Exception as e:
        print(f"❌ Database test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_vector_store():
    """Test vector store initialization"""
    print("\n🧠 Testing vector store...")
    
    try:
        from src.rag.vector_store import VectorStoreManager
        
        vector_store = VectorStoreManager("./data/vector_stores")
        character_names = ["Alex", "Sage", "Luna", "Echo"]
        await vector_store.initialize(character_names)
        print("✅ Vector store initialization successful")
        return True
        
    except Exception as e:
        print(f"❌ Vector store test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_memory_sharing():
    """Test memory sharing system"""
    print("\n🤝 Testing memory sharing...")
    
    try:
        from src.rag.vector_store import VectorStoreManager
        from src.rag.memory_sharing import MemorySharingManager
        
        vector_store = VectorStoreManager("./data/vector_stores")
        character_names = ["Alex", "Sage", "Luna", "Echo"]
        await vector_store.initialize(character_names)
        
        memory_sharing = MemorySharingManager(vector_store)
        await memory_sharing.initialize(character_names)
        print("✅ Memory sharing initialization successful")
        return True
        
    except Exception as e:
        print(f"❌ Memory sharing test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_creative_collaboration():
    """Test creative collaboration system"""
    print("\n🎨 Testing creative collaboration...")
    
    try:
        from src.rag.vector_store import VectorStoreManager
        from src.rag.memory_sharing import MemorySharingManager
        from src.collaboration.creative_projects import CollaborativeCreativeManager
        
        # Initialize components
        vector_store = VectorStoreManager("./data/vector_stores")
        character_names = ["Alex", "Sage", "Luna", "Echo"]
        await vector_store.initialize(character_names)
        
        memory_sharing = MemorySharingManager(vector_store)
        await memory_sharing.initialize(character_names)
        
        creative_manager = CollaborativeCreativeManager(vector_store, memory_sharing)
        await creative_manager.initialize(character_names)
        print("✅ Creative collaboration initialization successful")
        
        # Test project creation
        project_data = {
            "title": "Test Creative Project",
            "description": "A test project to verify the creative collaboration system works",
            "project_type": "story",
            "target_collaborators": ["Sage", "Luna"],
            "goals": ["Test system functionality", "Verify data persistence"],
            "estimated_duration": "test"
        }
        
        success, message = await creative_manager.propose_project("Alex", project_data)
        if success:
            print("✅ Project creation successful")
            
            # Test project suggestions
            suggestions = await creative_manager.get_project_suggestions("Luna")
            print(f"✅ Generated {len(suggestions)} project suggestions")
            
            return True
        else:
            print(f"❌ Project creation failed: {message}")
            return False
        
    except Exception as e:
        print(f"❌ Creative collaboration test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def test_mcp_integration():
    """Test MCP server integration"""
    print("\n🔧 Testing MCP integration...")
    
    try:
        from src.rag.vector_store import VectorStoreManager
        from src.rag.memory_sharing import MemorySharingManager
        from src.collaboration.creative_projects import CollaborativeCreativeManager
        from src.mcp.creative_projects_server import CreativeProjectsMCPServer
        from src.mcp.memory_sharing_server import MemorySharingMCPServer
        
        # Initialize components
        vector_store = VectorStoreManager("./data/vector_stores")
        character_names = ["Alex", "Sage", "Luna", "Echo"]
        await vector_store.initialize(character_names)
        
        memory_sharing = MemorySharingManager(vector_store)
        await memory_sharing.initialize(character_names)
        
        creative_manager = CollaborativeCreativeManager(vector_store, memory_sharing)
        await creative_manager.initialize(character_names)
        
        # Test MCP servers
        creative_mcp = CreativeProjectsMCPServer(creative_manager)
        memory_mcp = MemorySharingMCPServer(memory_sharing)
        
        await creative_mcp.set_character_context("Alex")
        await memory_mcp.set_character_context("Alex")
        
        print("✅ MCP server initialization successful")
        return True
        
    except Exception as e:
        print(f"❌ MCP integration test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

async def run_all_tests():
    """Run all system tests"""
    print("🐠 Discord Fishbowl System Tests")
    print("=" * 60)
    
    tests = [
        ("Module Imports", test_imports),
        ("Database System", test_database),
        ("Vector Store", test_vector_store),
        ("Memory Sharing", test_memory_sharing),
        ("Creative Collaboration", test_creative_collaboration),
        ("MCP Integration", test_mcp_integration),
    ]
    
    passed = 0
    total = len(tests)
    
    for test_name, test_func in tests:
        print(f"\n{'=' * 60}")
        print(f"🧪 {test_name}")
        print("=" * 60)
        
        try:
            if asyncio.iscoroutinefunction(test_func):
                result = await test_func()
            else:
                result = test_func()
            
            if result:
                passed += 1
                print(f"✅ {test_name} PASSED")
            else:
                print(f"❌ {test_name} FAILED")
                
        except Exception as e:
            print(f"❌ {test_name} FAILED with exception: {e}")
    
    print(f"\n{'=' * 60}")
    print(f"🎯 TEST RESULTS: {passed}/{total} tests passed")
    print("=" * 60)
    
    if passed == total:
        print("🎉 ALL TESTS PASSED! System is working correctly!")
        print("\nNext steps:")
        print("1. Install Ollama: https://ollama.ai/")
        print("2. Pull a model: ollama pull llama2")
        print("3. Get Discord tokens and update .env")
        print("4. Run the full system: cd src && python main.py")
        return True
    else:
        print(f"💥 {total - passed} tests failed. System needs attention.")
        return False

def main():
    """Main test function"""
    return asyncio.run(run_all_tests())

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)