MCP Memory Service

"""Unit tests for the exponential decay calculator."""

import pytest
from datetime import datetime, timedelta

from mcp_memory_service.consolidation.decay import ExponentialDecayCalculator, RelevanceScore
from mcp_memory_service.models.memory import Memory


@pytest.mark.unit
class TestExponentialDecayCalculator:
    """Test the exponential decay scoring system."""
    
    @pytest.fixture
    def decay_calculator(self, consolidation_config):
        return ExponentialDecayCalculator(consolidation_config)
    
    @pytest.mark.asyncio
    async def test_basic_decay_calculation(self, decay_calculator, sample_memories):
        """Test basic decay calculation functionality."""
        memories = sample_memories[:3]  # Use first 3 memories
        
        scores = await decay_calculator.process(memories)
        
        assert len(scores) == 3
        assert all(isinstance(score, RelevanceScore) for score in scores)
        assert all(score.total_score > 0 for score in scores)
        assert all(0 <= score.decay_factor <= 1 for score in scores)
    
    @pytest.mark.asyncio
    async def test_memory_age_affects_decay(self, decay_calculator):
        """Test that older memories have lower decay factors."""
        now = datetime.now()
        
        # Create memories of different ages
        recent_time = now - timedelta(days=1)
        old_time = now - timedelta(days=30)
        
        recent_memory = Memory(
            content="Recent memory",
            content_hash="recent",
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=recent_time.timestamp(),
            created_at_iso=recent_time.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
        )
        
        old_memory = Memory(
            content="Old memory",
            content_hash="old",
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=old_time.timestamp(),
            created_at_iso=old_time.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
        )
        
        scores = await decay_calculator.process([recent_memory, old_memory])
        
        recent_score = next(s for s in scores if s.memory_hash == "recent")
        old_score = next(s for s in scores if s.memory_hash == "old")
        
        # Recent memory should have higher decay factor
        assert recent_score.decay_factor > old_score.decay_factor
        assert recent_score.total_score > old_score.total_score
    
    @pytest.mark.asyncio
    async def test_memory_type_affects_retention(self, decay_calculator):
        """Test that different memory types have different retention periods."""
        now = datetime.now()
        age_days = 60  # 2 months old
        
        # Create memories of different types but same age
        critical_memory = Memory(
            content="Critical memory",
            content_hash="critical",
            tags=["critical"],
            memory_type="critical",
            embedding=[0.1] * 320,
            created_at=(now - timedelta(days=age_days)).timestamp(),
            created_at_iso=(now - timedelta(days=age_days)).isoformat() + 'Z'
        )
        
        temporary_memory = Memory(
            content="Temporary memory",
            content_hash="temporary",
            tags=["temp"],
            memory_type="temporary",
            embedding=[0.1] * 320,
            created_at=(now - timedelta(days=age_days)).timestamp(),
            created_at_iso=(now - timedelta(days=age_days)).isoformat() + 'Z'
        )
        
        scores = await decay_calculator.process([critical_memory, temporary_memory])
        
        critical_score = next(s for s in scores if s.memory_hash == "critical")
        temp_score = next(s for s in scores if s.memory_hash == "temporary")
        
        # Critical memory should decay slower (higher decay factor)
        assert critical_score.decay_factor > temp_score.decay_factor
        assert critical_score.metadata['retention_period'] > temp_score.metadata['retention_period']
    
    @pytest.mark.asyncio
    async def test_connections_boost_relevance(self, decay_calculator):
        """Test that memories with connections get relevance boost."""
        memory = Memory(
            content="Connected memory",
            content_hash="connected",
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        # Test with no connections
        scores_no_connections = await decay_calculator.process(
            [memory], 
            connections={}
        )
        
        # Test with connections
        scores_with_connections = await decay_calculator.process(
            [memory],
            connections={"connected": 3}
        )
        
        no_conn_score = scores_no_connections[0]
        with_conn_score = scores_with_connections[0]
        
        assert with_conn_score.connection_boost > no_conn_score.connection_boost
        assert with_conn_score.total_score > no_conn_score.total_score
        assert with_conn_score.metadata['connection_count'] == 3
    
    @pytest.mark.asyncio
    async def test_access_patterns_boost_relevance(self, decay_calculator):
        """Test that recent access boosts relevance."""
        memory = Memory(
            content="Accessed memory",
            content_hash="accessed",
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        # Test with no recent access
        scores_no_access = await decay_calculator.process([memory])
        
        # Test with recent access
        recent_access = {
            "accessed": datetime.now() - timedelta(hours=6)
        }
        scores_recent_access = await decay_calculator.process(
            [memory],
            access_patterns=recent_access
        )
        
        no_access_score = scores_no_access[0]
        recent_access_score = scores_recent_access[0]
        
        assert recent_access_score.access_boost > no_access_score.access_boost
        assert recent_access_score.total_score > no_access_score.total_score
    
    @pytest.mark.asyncio
    async def test_base_importance_from_metadata(self, decay_calculator):
        """Test that explicit importance scores are used."""
        high_importance_memory = Memory(
            content="Important memory",
            content_hash="important",
            tags=["test"],
            embedding=[0.1] * 320,
            metadata={"importance_score": 1.8},
            created_at=datetime.now().timestamp()
        )
        
        normal_memory = Memory(
            content="Normal memory",
            content_hash="normal", 
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        scores = await decay_calculator.process([high_importance_memory, normal_memory])
        
        important_score = next(s for s in scores if s.memory_hash == "important")
        normal_score = next(s for s in scores if s.memory_hash == "normal")
        
        assert important_score.base_importance > normal_score.base_importance
        assert important_score.total_score > normal_score.total_score
    
    @pytest.mark.asyncio
    async def test_base_importance_from_tags(self, decay_calculator):
        """Test that importance is derived from tags."""
        critical_memory = Memory(
            content="Critical memory",
            content_hash="critical_tag",
            tags=["critical", "system"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        temp_memory = Memory(
            content="Temporary memory",
            content_hash="temp_tag",
            tags=["temporary", "draft"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        scores = await decay_calculator.process([critical_memory, temp_memory])
        
        critical_score = next(s for s in scores if s.memory_hash == "critical_tag")
        temp_score = next(s for s in scores if s.memory_hash == "temp_tag")
        
        assert critical_score.base_importance > temp_score.base_importance
    
    @pytest.mark.asyncio
    async def test_protected_memory_minimum_relevance(self, decay_calculator):
        """Test that protected memories maintain minimum relevance."""
        # Create a very old memory that would normally have very low relevance
        old_critical_memory = Memory(
            content="Old critical memory",
            content_hash="old_critical",
            tags=["critical", "important"],
            memory_type="critical",
            embedding=[0.1] * 320,
            created_at=(datetime.now() - timedelta(days=500)).timestamp(),
            created_at_iso=(datetime.now() - timedelta(days=500)).isoformat() + 'Z'
        )
        
        scores = await decay_calculator.process([old_critical_memory])
        score = scores[0]
        
        # Even very old critical memory should maintain minimum relevance
        assert score.total_score >= 0.5  # Minimum for protected memories
        assert score.metadata['is_protected'] is True
    
    @pytest.mark.asyncio
    async def test_get_low_relevance_memories(self, decay_calculator, sample_memories):
        """Test filtering of low relevance memories."""
        scores = await decay_calculator.process(sample_memories)
        
        low_relevance = await decay_calculator.get_low_relevance_memories(scores, threshold=0.5)
        
        # Should find some low relevance memories
        assert len(low_relevance) > 0
        assert all(score.total_score < 0.5 for score in low_relevance)
    
    @pytest.mark.asyncio
    async def test_get_high_relevance_memories(self, decay_calculator, sample_memories):
        """Test filtering of high relevance memories."""
        scores = await decay_calculator.process(sample_memories)
        
        high_relevance = await decay_calculator.get_high_relevance_memories(scores, threshold=1.0)
        
        # Should find some high relevance memories  
        assert len(high_relevance) >= 0
        assert all(score.total_score >= 1.0 for score in high_relevance)
    
    @pytest.mark.asyncio
    async def test_update_memory_relevance_metadata(self, decay_calculator):
        """Test updating memory with relevance metadata."""
        memory = Memory(
            content="Test memory",
            content_hash="test",
            tags=["test"],
            embedding=[0.1] * 320,
            created_at=datetime.now().timestamp()
        )
        
        scores = await decay_calculator.process([memory])
        score = scores[0]
        
        updated_memory = await decay_calculator.update_memory_relevance_metadata(memory, score)
        
        assert 'relevance_score' in updated_memory.metadata
        assert 'relevance_calculated_at' in updated_memory.metadata
        assert 'decay_factor' in updated_memory.metadata
        assert 'connection_boost' in updated_memory.metadata
        assert 'access_boost' in updated_memory.metadata
        assert updated_memory.metadata['relevance_score'] == score.total_score
    
    @pytest.mark.asyncio
    async def test_empty_memories_list(self, decay_calculator):
        """Test handling of empty memories list."""
        scores = await decay_calculator.process([])
        assert scores == []
    
    @pytest.mark.asyncio
    async def test_memory_without_embedding(self, decay_calculator):
        """Test handling of memory without embedding."""
        memory = Memory(
            content="No embedding",
            content_hash="no_embedding",
            tags=["test"],
            embedding=None,  # No embedding
            created_at=datetime.now().timestamp()
        )
        
        scores = await decay_calculator.process([memory])
        
        # Should still work, just without embedding-based features
        assert len(scores) == 1
        assert scores[0].total_score > 0