Test/test_memory.Py

import asyncio
import pytest
import numpy as np
from cortex.memory.quantum_memory import QuantumMemory, QuantumMemoryCell
from cortex.memory.hierarchical import HierarchicalMemory
from cortex.memory.associative import AssociativeMemory

class TestQuantumMemory:
    """Tests pour la mémoire quantique"""
    
    @pytest.fixture
    async def quantum_memory(self):
        """Fixture pour initialiser la mémoire quantique"""
        memory = QuantumMemory(capacity=100)
        await memory.initialize()
        return memory
    
    @pytest.mark.asyncio
    async def test_memory_initialization(self, quantum_memory):
        """Test l'initialisation de la mémoire quantique"""
        assert quantum_memory is not None
        assert quantum_memory.capacity == 100
        assert len(quantum_memory.memory_cells) > 0
        assert quantum_memory.access_latency > 0
    
    @pytest.mark.asyncio
    async def test_data_integrity(self, quantum_memory):
        """Test l'intégrité des données stockées"""
        test_data = {
            "string": "Hello Quantum World",
            "number": 42,
            "list": [1, 2, 3, 4, 5],
            "dict": {"key": "value", "nested": {"a": 1, "b": 2}}
        }
        
        for key, value in test_data.items():
            address = await quantum_memory.store_quantum_data(value, f"test_{key}")
            retrieved = await quantum_memory.retrieve_quantum_data(address)
            assert retrieved == value, f"Data integrity failed for {key}"
    
    @pytest.mark.asyncio
    async def test_quantum_coherence(self, quantum_memory):
        """Test la préservation de la cohérence quantique"""
        address = await quantum_memory.store_quantum_data("coherence_test")
        cell = quantum_memory.memory_cells[address]
        
        # Vérifie la cohérence initiale
        assert cell.coherence > 0.9
        
        # Simule le temps qui passe
        await asyncio.sleep(0.1)
        await quantum_memory._maintain_coherence(address)
        
        # Vérifie que la cohérence a diminué mais pas trop
        assert 0.8 < cell.coherence < 1.0
    
    @pytest.mark.asyncio
    async def test_entanglement_persistence(self, quantum_memory):
        """Test la persistance de l'intrication"""
        addr1 = await quantum_memory.store_quantum_data("entangled_data_1")
        addr2 = await quantum_memory.store_quantum_data("entangled_data_2")
        
        # Crée l'intrication
        await quantum_memory.entangle_memory_cells(addr1, addr2)
        
        # Vérifie que l'intrication est persistée
        cell1 = quantum_memory.memory_cells[addr1]
        cell2 = quantum_memory.memory_cells[addr2]
        
        assert addr2 in cell1.entangled_with
        assert addr1 in cell2.entangled_with
    
    @pytest.mark.asyncio
    async def test_superposition_retrieval(self, quantum_memory):
        """Test la récupération depuis la superposition"""
        data_states = ["state_alpha", "state_beta", "state_gamma"]
        address = await quantum_memory.quantum_superposition_store(data_states)
        
        # Récupération sans mesure (devrait retourner toutes les données)
        retrieved = await quantum_memory.retrieve_quantum_data(address, measure=False)
        assert retrieved == data_states
        
        # Récupération avec mesure (devrait retourner un état spécifique)
        measured = await quantum_memory.retrieve_quantum_data(address, measure=True)
        assert measured in data_states
    
    @pytest.mark.asyncio
    async def test_pattern_recognition(self, quantum_memory):
        """Test la reconnaissance de pattern quantique"""
        # Stocke différents patterns
        patterns = [
            "quantum computing is amazing",
            "artificial intelligence revolution", 
            "quantum artificial intelligence",
            "machine learning with quantum"
        ]
        
        addresses = []
        for pattern in patterns:
            addr = await quantum_memory.store_quantum_data(pattern)
            addresses.append(addr)
        
        # Test la reconnaissance
        results = await quantum_memory.quantum_pattern_recognition("quantum")
        assert len(results) >= 2  # Au moins 2 patterns contiennent "quantum"
        
        results = await quantum_memory.quantum_pattern_recognition("artificial intelligence")
        assert len(results) >= 2
    
    @pytest.mark.asyncio
    async def test_memory_statistics(self, quantum_memory):
        """Test les statistiques de mémoire"""
        # Remplit partiellement la mémoire
        for i in range(25):
            await quantum_memory.store_quantum_data(f"test_data_{i}")
        
        stats = quantum_memory.get_memory_statistics()
        
        assert stats["total_capacity"] == 100
        assert stats["used_cells"] == 25
        assert stats["available_cells"] == 75
        assert 0 <= stats["memory_usage"] <= 1

class TestHierarchicalMemory:
    """Tests pour la mémoire hiérarchique"""
    
    @pytest.fixture
    async def hierarchical_memory(self):
        """Fixture pour initialiser la mémoire hiérarchique"""
        memory = HierarchicalMemory()
        await memory.initialize()
        return memory
    
    @pytest.mark.asyncio
    async def test_memory_hierarchy(self, hierarchical_memory):
        """Test la hiérarchie de mémoire"""
        # Test des différents niveaux
        data_immediate = "immediate data"
        data_short = "short term data" 
        data_long = "long term data"
        
        # Stockage à différents niveaux
        await hierarchical_memory.store(data_immediate, "immediate", priority=10)
        await hierarchical_memory.store(data_short, "short_term", priority=5)
        await hierarchical_memory.store(data_long, "long_term", priority=1)
        
        # Vérification de l'accès
        immediate = await hierarchical_memory.retrieve("immediate")
        assert immediate == data_immediate
        
        # Test de la promotion dans la hiérarchie
        await hierarchical_memory.promote("long_term")
        
        stats = hierarchical_memory.get_stats()
        assert stats["total_items"] == 3

class TestAssociativeMemory:
    """Tests pour la mémoire associative"""
    
    @pytest.fixture
    async def associative_memory(self):
        """Fixture pour initialiser la mémoire associative"""
        memory = AssociativeMemory()
        await memory.initialize()
        return memory
    
    @pytest.mark.asyncio
    async def test_associative_links(self, associative_memory):
        """Test les liens associatifs"""
        # Crée des associations
        await associative_memory.create_association("quantum", "computing", strength=0.9)
        await associative_memory.create_association("quantum", "physics", strength=0.8)
        await associative_memory.create_association("ai", "machine learning", strength=0.95)
        
        # Test la récupération associative
        quantum_associations = await associative_memory.get_associations("quantum")
        assert len(quantum_associations) == 2
        assert any("computing" in assoc for assoc in quantum_associations)
        
        ai_associations = await associative_memory.get_associations("ai")
        assert len(ai_associations) == 1
    
    @pytest.mark.asyncio
    async def test_pattern_completion(self, associative_memory):
        """Test la complétion de pattern"""
        # Entraîne la mémoire avec des patterns
        patterns = [
            ("cat", "animal", "feline", "pet"),
            ("dog", "animal", "canine", "pet"), 
            ("quantum", "physics", "computing", "mechanics")
        ]
        
        for pattern in patterns:
            for i, concept in enumerate(pattern):
                if i < len(pattern) - 1:
                    await associative_memory.create_association(
                        concept, pattern[i + 1], strength=0.8
                    )
        
        # Test la complétion
        completion = await associative_memory.pattern_completion("cat", "animal")
        assert "feline" in completion or "pet" in completion

@pytest.mark.asyncio
async def test_memory_system_integration():
    """Test d'intégration du système de mémoire complet"""
    # Initialisation de toutes les mémoires
    quantum_mem = QuantumMemory(50)
    hierarchical_mem = HierarchicalMemory()
    associative_mem = AssociativeMemory()
    
    await quantum_mem.initialize()
    await hierarchical_mem.initialize()
    await associative_mem.initialize()
    
    # Workflow de test intégré
    # 1. Stockage quantique
    quantum_data = "Quantum memory test data"
    q_address = await quantum_mem.store_quantum_data(quantum_data)
    
    # 2. Stockage hiérarchique
    await hierarchical_mem.store(quantum_data, "quantum_backup", priority=7)
    
    # 3. Création d'associations
    await associative_mem.create_association("quantum", "memory", strength=0.9)
    await associative_mem.create_association("memory", "storage", strength=0.8)
    
    # Vérifications
    assert q_address is not None
    hierarchical_data = await hierarchical_mem.retrieve("quantum_backup")
    assert hierarchical_data == quantum_data
    
    associations = await associative_mem.get_associations("quantum")
    assert len(associations) > 0
    
    print("✅ Tests d'intégration mémoire réussis!")

if __name__ == "__main__":
    # Exécution des tests
    asyncio.run(test_memory_system_integration())