core-api/tests/test_memory_simple.py

#!/usr/bin/env python3
"""
Simple integration tests for Phase 2 Memory System
No external dependencies beyond the memory system itself
"""
import asyncio
import sys
from datetime import datetime

# Add src to path
sys.path.insert(0, '/app')

from src.memory import (
    ConversationBufferMemory,
    QdrantConversationMemory,
    ConversationTurn,
    MessageRole,
    TokenUsage,
    get_buffer_memory,
    get_qdrant_memory
)
from src.models.embeddings import get_embedding_client


def test_embedding_client():
    """Test 1: Embedding Client"""
    print("\n" + "="*60)
    print("Test 1: Embedding Client")
    print("="*60)

    try:
        # Initialize
        client = get_embedding_client()
        assert client is not None
        assert client.dimension == 384
        print(f"✓ Embedding client initialized: {client.model_name}")
        print(f"✓ Embedding dimension: {client.dimension}")

        # Single embedding
        text = "Hello, this is a test message for embedding generation"
        embedding = client.embed_text(text)
        assert isinstance(embedding, list)
        assert len(embedding) == 384
        assert all(isinstance(x, float) for x in embedding)
        print(f"✓ Single embedding generated: {len(embedding)} dimensions")
        print(f"  Sample values: [{embedding[0]:.4f}, {embedding[1]:.4f}, {embedding[2]:.4f}, ...]")

        # Batch embedding
        texts = [
            "First message about Python programming",
            "Second message about machine learning",
            "Third message about data science"
        ]
        embeddings = client.embed_batch(texts)
        assert len(embeddings) == 3
        assert all(len(emb) == 384 for emb in embeddings)
        print(f"✓ Batch embeddings generated: {len(embeddings)} texts")

        print("\n✅ Embedding Client Tests: PASSED")
        return True

    except Exception as e:
        print(f"\n❌ Embedding Client Tests: FAILED")
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
        return False


async def test_qdrant_memory():
    """Test 2: Qdrant Memory Storage"""
    print("\n" + "="*60)
    print("Test 2: Qdrant Memory Storage")
    print("="*60)

    test_conv_id = f"test_conv_{int(datetime.utcnow().timestamp())}"

    try:
        # Initialize
        qdrant_memory = get_qdrant_memory()
        assert qdrant_memory.client is not None
        assert qdrant_memory.collection_name == "core_api_conversations"
        print(f"✓ Connected to Qdrant: {qdrant_memory.host}:{qdrant_memory.port}")
        print(f"✓ Collection: {qdrant_memory.collection_name}")

        # Add single turn
        turn1 = ConversationTurn(
            role=MessageRole.USER,
            content="What is Python?",
            turn_number=1,
            tokens=TokenUsage(prompt=10, completion=0, total=10)
        )
        await qdrant_memory.add_turn(test_conv_id, turn1)
        print(f"✓ Turn 1 stored in Qdrant")

        # Verify it exists
        exists = await qdrant_memory.conversation_exists(test_conv_id)
        assert exists is True
        print(f"✓ Conversation exists: {test_conv_id}")

        # Add more turns for chronological test
        turn2 = ConversationTurn(
            role=MessageRole.ASSISTANT,
            content="Python is a high-level programming language known for simplicity and readability",
            turn_number=2
        )
        turn3 = ConversationTurn(
            role=MessageRole.USER,
            content="How do I learn Python programming?",
            turn_number=3
        )

        await qdrant_memory.add_turn(test_conv_id, turn2)
        await qdrant_memory.add_turn(test_conv_id, turn3)
        print(f"✓ Turns 2-3 stored in Qdrant")

        # Test chronological retrieval (Tier 2 mode)
        retrieved = await qdrant_memory.get_turns(test_conv_id)
        assert len(retrieved) == 3
        assert retrieved[0].turn_number == 1
        assert retrieved[1].turn_number == 2
        assert retrieved[2].turn_number == 3
        assert retrieved[0].content == "What is Python?"
        print(f"✓ Chronological retrieval works: {len(retrieved)} turns")
        for i, turn in enumerate(retrieved, 1):
            print(f"  Turn {turn.turn_number}: {turn.role.value} - {turn.content[:50]}...")

        # Add turns with distinct topics for semantic search
        turn10 = ConversationTurn(
            role=MessageRole.USER,
            content="I love machine learning and neural networks and artificial intelligence",
            turn_number=10
        )
        turn11 = ConversationTurn(
            role=MessageRole.USER,
            content="Pizza is my favorite food and I enjoy eating pasta",
            turn_number=11
        )
        turn12 = ConversationTurn(
            role=MessageRole.USER,
            content="Deep learning models and transformers are fascinating AI technologies",
            turn_number=12
        )

        await qdrant_memory.add_turn(test_conv_id, turn10)
        await qdrant_memory.add_turn(test_conv_id, turn11)
        await qdrant_memory.add_turn(test_conv_id, turn12)
        print(f"✓ Added 3 more turns for semantic search test")

        # Test semantic search (Tier 3 mode)
        search_results = await qdrant_memory.similarity_search(
            query="artificial intelligence and deep learning",
            conversation_id=test_conv_id,
            limit=3
        )
        assert len(search_results) > 0
        print(f"✓ Semantic search works: {len(search_results)} matches")

        # Top result should be about AI/ML, not food
        top_result = search_results[0]
        print(f"  Top match (score: {top_result['score']:.3f}): {top_result['content'][:60]}...")
        assert top_result["score"] > 0.5, "Semantic similarity score too low"

        # Verify top matches are AI-related
        ai_keywords = ["machine learning", "neural networks", "Deep learning", "AI", "artificial intelligence"]
        top_content = search_results[0]["content"]
        assert any(keyword in top_content for keyword in ai_keywords), "Top result not AI-related"
        print(f"✓ Semantic relevance verified (AI-related content ranked higher)")

        # Test conversation stats
        stats = await qdrant_memory.get_conversation_stats(test_conv_id)
        assert stats["conversation_id"] == test_conv_id
        assert stats["total_turns"] == 6
        print(f"✓ Stats retrieved: {stats['total_turns']} turns, {stats['total_tokens']} tokens")

        # Cleanup
        await qdrant_memory.clear_conversation(test_conv_id)
        exists_after = await qdrant_memory.conversation_exists(test_conv_id)
        assert exists_after is False
        print(f"✓ Conversation cleared successfully")

        print("\n✅ Qdrant Memory Tests: PASSED")
        return True

    except Exception as e:
        print(f"\n❌ Qdrant Memory Tests: FAILED")
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()

        # Cleanup on error
        try:
            await qdrant_memory.clear_conversation(test_conv_id)
        except:
            pass

        return False


async def test_full_integration():
    """Test 3: Full Integration (Tier 1 + Tier 2/3)"""
    print("\n" + "="*60)
    print("Test 3: Full Integration (Tier 1 + Tier 2/3)")
    print("="*60)

    test_conv_id = f"integration_test_{int(datetime.utcnow().timestamp())}"

    try:
        # Initialize both tiers
        buffer_memory = get_buffer_memory()
        qdrant_memory = get_qdrant_memory()
        print(f"✓ Initialized Tier 1 (Buffer) and Tier 2/3 (Qdrant)")

        # 1. Add turns to buffer (Tier 1)
        turns = [
            ConversationTurn(role=MessageRole.USER, content="Hello!", turn_number=1),
            ConversationTurn(role=MessageRole.ASSISTANT, content="Hi there! How can I help?", turn_number=2),
            ConversationTurn(role=MessageRole.USER, content="How are you?", turn_number=3),
            ConversationTurn(role=MessageRole.ASSISTANT, content="I'm doing great, thanks!", turn_number=4),
        ]

        for turn in turns:
            await buffer_memory.add_turn(test_conv_id, turn)

        # Verify buffer has them
        buffer = await buffer_memory.get_buffer(test_conv_id)
        assert len(buffer.turns) == 4
        print(f"✓ Tier 1 buffer: {len(buffer.turns)} turns stored")

        # 2. Move to Qdrant (Tier 2/3) - simulating consolidation
        for turn in buffer.turns:
            await qdrant_memory.add_turn(test_conv_id, turn)

        # Verify Qdrant has them
        qdrant_turns = await qdrant_memory.get_turns(test_conv_id)
        assert len(qdrant_turns) == 4
        print(f"✓ Tier 2/3 Qdrant: {len(qdrant_turns)} turns stored")

        # 3. Test semantic search across consolidated data
        search_results = await qdrant_memory.similarity_search(
            query="greeting hello",
            conversation_id=test_conv_id,
            limit=2
        )
        assert len(search_results) > 0
        print(f"✓ Semantic search: {len(search_results)} matches found")
        print(f"  Best match: '{search_results[0]['content']}' (score: {search_results[0]['score']:.3f})")

        # 4. Test data consistency
        buffer_content = [t.content for t in buffer.turns]
        qdrant_content = [t.content for t in qdrant_turns]
        assert buffer_content == qdrant_content
        print(f"✓ Data consistency verified (Buffer ↔ Qdrant)")

        # Cleanup
        await qdrant_memory.clear_conversation(test_conv_id)
        await buffer_memory.clear_conversation(test_conv_id)
        print(f"✓ Cleanup complete")

        print("\n✅ Full Integration Tests: PASSED")
        return True

    except Exception as e:
        print(f"\n❌ Full Integration Tests: FAILED")
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()

        # Cleanup on error
        try:
            await qdrant_memory.clear_conversation(test_conv_id)
            await buffer_memory.clear_conversation(test_conv_id)
        except:
            pass

        return False


def main():
    """Run all tests"""
    print("\n" + "="*60)
    print("PHASE 2 MEMORY SYSTEM - INTEGRATION TESTS")
    print("="*60)
    print(f"Start time: {datetime.utcnow().isoformat()}")

    results = []

    # Test 1: Embedding Client
    results.append(("Embedding Client", test_embedding_client()))

    # Test 2: Qdrant Memory
    results.append(("Qdrant Memory", asyncio.run(test_qdrant_memory())))

    # Test 3: Full Integration
    results.append(("Full Integration", asyncio.run(test_full_integration())))

    # Summary
    print("\n" + "="*60)
    print("TEST SUMMARY")
    print("="*60)

    for test_name, passed in results:
        status = "✅ PASSED" if passed else "❌ FAILED"
        print(f"{test_name:.<40} {status}")

    total = len(results)
    passed = sum(1 for _, p in results if p)
    failed = total - passed

    print(f"\nTotal: {total} | Passed: {passed} | Failed: {failed}")
    print(f"Success rate: {(passed/total)*100:.1f}%")

    if all(p for _, p in results):
        print("\n" + "="*60)
        print("🎉 ALL TESTS PASSED!")
        print("="*60)
        print("\nPhase 2 Memory System Status: ✅ FUNCTIONAL")
        print("- Embedding client working (384d vectors)")
        print("- Qdrant storage working (chronological + semantic)")
        print("- Full integration working (Tier 1 ↔ Tier 2/3)")
        return 0
    else:
        print("\n" + "="*60)
        print("❌ SOME TESTS FAILED")
        print("="*60)
        return 1


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