minions-ai-agents/tests/test_episodic_memory.py

"""
Tests for Flywheel Module - Episodic Memory.

Tests lesson storage and retrieval.
"""

import pytest
from unittest.mock import Mock, patch, AsyncMock

from src.flywheel.episodic_memory import (
    EpisodicMemory,
    MemoryEntry,
    MemorySearchResult,
    MemoryType,
    get_episodic_memory
)


class TestMemoryEntry:
    """Tests for MemoryEntry dataclass."""
    
    def test_create_entry(self):
        """Test creating a memory entry."""
        from datetime import datetime, timezone
        
        entry = MemoryEntry(
            id="mem-001",
            memory_type=MemoryType.LESSON,
            tenant_id="tenant-001",
            technology="linux",
            problem_summary="Servidor sem memória",
            problem_symptoms=["alta utilização", "OOM killer"],
            resolution_summary="Reiniciou serviço Apache",
            resolution_steps=["Identificar processo", "Reiniciar serviço"],
            ticket_id="TKT-001",
            resolved_at=datetime.now(timezone.utc),
            resolution_time_minutes=30,
            confidence_score=0.9
        )
        
        assert entry.id == "mem-001"
        assert entry.memory_type == MemoryType.LESSON
        assert len(entry.resolution_steps) == 2


class TestEpisodicMemory:
    """Tests for EpisodicMemory class."""
    
    @pytest.fixture
    def memory(self):
        """Create memory with mocked Qdrant and Ollama."""
        with patch('src.flywheel.episodic_memory.get_qdrant_client') as mock_qdrant, \
             patch('src.flywheel.episodic_memory.get_ollama_client') as mock_ollama:
            
            mock_qdrant.return_value = Mock()
            mock_qdrant.return_value.upsert_document = Mock(return_value=True)
            mock_qdrant.return_value.search = Mock(return_value=[])
            
            mock_ollama.return_value = Mock()
            mock_ollama.return_value.get_embeddings = AsyncMock(return_value=[0.1] * 384)
            
            return EpisodicMemory()
    
    def test_generate_id(self, memory):
        """Test ID generation is unique."""
        id1 = memory._generate_id("ticket-1", "tenant-a")
        id2 = memory._generate_id("ticket-2", "tenant-a")
        
        assert id1.startswith("mem-")
        assert id1 != id2
    
    def test_create_search_content(self, memory):
        """Test search content creation."""
        from datetime import datetime, timezone
        
        entry = MemoryEntry(
            id="mem-001",
            memory_type=MemoryType.LESSON,
            tenant_id="tenant-001",
            technology="nginx",
            problem_summary="Erro 502",
            problem_symptoms=["gateway timeout"],
            resolution_summary="Aumentar buffer",
            resolution_steps=["Editar nginx.conf"],
            ticket_id="TKT-001",
            resolved_at=datetime.now(timezone.utc),
            resolution_time_minutes=15,
            confidence_score=0.85,
            tags=["proxy", "buffer"]
        )
        
        content = memory._create_search_content(entry)
        
        assert "Erro 502" in content
        assert "nginx" in content
        assert "buffer" in content.lower()
    
    @pytest.mark.asyncio
    async def test_generate_embedding(self, memory):
        """Test embedding generation."""
        emb = await memory._generate_embedding("test text")
        
        # Check against list since mock returns list, not numpy array
        assert isinstance(emb, list)
        assert len(emb) == 384
    
    @pytest.mark.asyncio
    async def test_store_lesson_success(self, memory):
        """Test successful lesson storage."""
        result = await memory.store_lesson(
            ticket_id="TKT-001",
            tenant_id="tenant-001",
            technology="linux",
            problem_summary="High CPU usage",
            problem_symptoms=["100% CPU", "slow response"],
            resolution_summary="Killed runaway process",
            resolution_steps=["top -c", "kill -9 PID"],
            resolution_time_minutes=20,
            confidence_score=0.9
        )
        
        assert result is not None
        assert result.startswith("mem-")
    
    @pytest.mark.asyncio
    async def test_store_antipattern(self, memory):
        """Test antipattern storage."""
        result = await memory.store_antipattern(
            ticket_id="TKT-002",
            tenant_id="tenant-001",
            technology="postgresql",
            problem_summary="Database slow",
            failed_approach="Reiniciar PostgreSQL",
            why_failed="Perda de cache causou lentidão maior",
            better_approach="Analisar queries lentas primeiro"
        )
        
        assert result is not None
        assert result.startswith("mem-")
    
    def test_explain_relevance_technology(self, memory):
        """Test relevance explanation for technology match."""
        from datetime import datetime, timezone
        
        entry = MemoryEntry(
            id="mem-001",
            memory_type=MemoryType.LESSON,
            tenant_id="tenant-001",
            technology="nginx",
            problem_summary="Erro 502",
            problem_symptoms=[],
            resolution_summary="Fix",
            resolution_steps=[],
            ticket_id="TKT-001",
            resolved_at=datetime.now(timezone.utc),
            resolution_time_minutes=10,
            confidence_score=0.8
        )
        
        explanation = memory._explain_relevance(entry, "nginx retornando erro")
        
        assert "nginx" in explanation.lower()
    
    def test_explain_relevance_antipattern(self, memory):
        """Test relevance explanation for antipattern."""
        from datetime import datetime, timezone
        
        entry = MemoryEntry(
            id="mem-001",
            memory_type=MemoryType.ANTIPATTERN,
            tenant_id="tenant-001",
            technology="linux",
            problem_summary="Memory issue",
            problem_symptoms=[],
            resolution_summary="Don't do this",
            resolution_steps=[],
            ticket_id="TKT-001",
            resolved_at=datetime.now(timezone.utc),
            resolution_time_minutes=0,
            confidence_score=1.0
        )
        
        explanation = memory._explain_relevance(entry, "memory problem")
        
        assert "ANTIPADRÃO" in explanation


class TestEpisodicMemorySingleton:
    """Tests for singleton."""
    
    def test_singleton(self):
        """Test singleton returns same instance."""
        import src.flywheel.episodic_memory as module
        module._memory = None
        
        with patch('src.flywheel.episodic_memory.get_qdrant_client'):
            m1 = get_episodic_memory()
            m2 = get_episodic_memory()
            
            assert m1 is m2