t6_mem0/final_functionality_test.py

#!/usr/bin/env python3
"""
Final comprehensive functionality test with fresh data
"""

import sys
from config import load_config, get_mem0_config
from mem0 import Memory
import time

def final_functionality_test():
    """Final comprehensive test with fresh data"""
    print("=" * 70)
    print("🎯 FINAL MEM0 FUNCTIONALITY TEST")
    print("=" * 70)
    
    try:
        # Initialize mem0
        system_config = load_config()
        config = get_mem0_config(system_config, "ollama")
        
        print("🚀 Initializing mem0...")
        memory = Memory.from_config(config)
        print("✅ mem0 initialized successfully")
        
        test_user = "final_test_user_2025"
        
        # Test 1: Add diverse memories
        print(f"\n📝 TEST 1: Adding diverse memories...")
        memories_to_add = [
            "I work as a software engineer specializing in Python and AI",
            "My current project involves building a RAG system with vector databases",
            "I prefer using local LLM models for privacy and cost reasons",
            "Supabase is my go-to choice for PostgreSQL with vector extensions",
            "I'm interested in learning more about graph databases like Neo4j"
        ]
        
        print(f"Adding {len(memories_to_add)} memories for user: {test_user}")
        for i, memory_text in enumerate(memories_to_add):
            result = memory.add(memory_text, user_id=test_user)
            status = "✅ Added" if result.get('results') else "📝 Processed"
            print(f"   {i+1}. {status}: {memory_text[:50]}...")
            if result.get('results'):
                for res in result['results']:
                    print(f"      → {res.get('event', 'UNKNOWN')}: {res.get('memory', 'N/A')[:40]}...")
        
        # Test 2: Comprehensive search
        print(f"\n🔍 TEST 2: Comprehensive search testing...")
        search_tests = [
            ("software engineer", "Job/Role search"),
            ("vector database", "Technology search"),
            ("privacy", "Concept search"),
            ("Python", "Programming language search"),
            ("graph database", "Database type search")
        ]
        
        for query, description in search_tests:
            print(f"   {description}: '{query}'")
            results = memory.search(query, user_id=test_user, limit=3)
            
            if results and 'results' in results:
                search_results = results['results']
                print(f"     Found {len(search_results)} results:")
                for j, result in enumerate(search_results):
                    score = result.get('score', 0)
                    memory_text = result.get('memory', 'N/A')
                    print(f"       {j+1}. Score: {score:.3f} | {memory_text[:45]}...")
            else:
                print("     No results found")
        
        # Test 3: Memory retrieval and count
        print(f"\n📊 TEST 3: Memory retrieval...")
        all_memories = memory.get_all(user_id=test_user)
        
        if all_memories and 'results' in all_memories:
            memories_list = all_memories['results']
            print(f"   Retrieved {len(memories_list)} memories for {test_user}:")
            for i, mem in enumerate(memories_list):
                created_at = mem.get('created_at', 'Unknown time')
                memory_text = mem.get('memory', 'N/A')
                print(f"     {i+1}. [{created_at[:19]}] {memory_text}")
        else:
            print(f"   No memories found or unexpected format: {all_memories}")
        
        # Test 4: User isolation test
        print(f"\n👥 TEST 4: User isolation...")
        other_user = "isolation_test_user"
        memory.add("This is a secret memory for testing user isolation", user_id=other_user)
        
        user1_memories = memory.get_all(user_id=test_user)
        user2_memories = memory.get_all(user_id=other_user)
        
        user1_count = len(user1_memories.get('results', []))
        user2_count = len(user2_memories.get('results', []))
        
        print(f"   User '{test_user}': {user1_count} memories")
        print(f"   User '{other_user}': {user2_count} memories")
        
        if user1_count > 0 and user2_count > 0:
            print("   ✅ User isolation working correctly")
        else:
            print("   ⚠️  User isolation test inconclusive")
        
        # Test 5: Memory updates/deduplication
        print(f"\n🔄 TEST 5: Memory update/deduplication...")
        
        # Add similar memory
        similar_result = memory.add("I work as a software engineer with expertise in Python and artificial intelligence", user_id=test_user)
        print(f"   Adding similar memory: {similar_result}")
        
        # Check if it was deduplicated or updated
        updated_memories = memory.get_all(user_id=test_user)
        updated_count = len(updated_memories.get('results', []))
        print(f"   Memory count after adding similar: {updated_count}")
        
        if updated_count == user1_count:
            print("   ✅ Deduplication working - no new memory added")
        elif updated_count > user1_count:
            print("   📝 New memory added - different enough to be separate")
        else:
            print("   ⚠️  Unexpected memory count change")
        
        # Test 6: Search relevance
        print(f"\n🎯 TEST 6: Search relevance testing...")
        specific_searches = [
            "What programming language do I use?",
            "What database technology do I prefer?",
            "What type of project am I working on?"
        ]
        
        for question in specific_searches:
            print(f"   Question: {question}")
            results = memory.search(question, user_id=test_user, limit=2)
            
            if results and 'results' in results:
                for result in results['results'][:1]:  # Show top result
                    score = result.get('score', 0)
                    memory_text = result.get('memory', 'N/A')
                    print(f"     Answer (score: {score:.3f}): {memory_text}")
            else:
                print("     No relevant memories found")
        
        print(f"\n🧹 CLEANUP: Removing test data...")
        # Clean up both test users
        try:
            for mem in user1_memories.get('results', []):
                memory.delete(mem['id'])
            for mem in user2_memories.get('results', []):
                memory.delete(mem['id'])
            print("   ✅ Test data cleaned up successfully")
        except Exception as e:
            print(f"   ⚠️  Cleanup note: {e}")
        
        print("\n" + "=" * 70)
        print("🎉 FINAL FUNCTIONALITY TEST COMPLETED!")
        print("✅ mem0 is fully functional with Supabase")
        print("✅ Memory storage, search, and retrieval working")
        print("✅ User isolation implemented correctly")
        print("✅ Vector embeddings and search operational")
        print("✅ Ollama LLM integration working")
        print("=" * 70)
        return True
        
    except Exception as e:
        print(f"\n❌ Test failed: {str(e)}")
        print(f"Error type: {type(e).__name__}")
        import traceback
        traceback.print_exc()
        return False

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