va/int-layer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Misc improvements

Browse Source
This commit is contained in:
Anibal Angulo
2026-02-20 06:59:31 +00:00
parent 734cade8d9
commit e9d80def08
33 changed files with 1844 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/conftest.py
View File

@@ -10,8 +10,7 @@ import pytest_asyncio
from fakeredis import aioredis as fakeredis
from capa_de_integracion.config import Settings
from capa_de_integracion.services.firestore_service import FirestoreService
from capa_de_integracion.services.redis_service import RedisService
from capa_de_integracion.services.storage import FirestoreService, RedisService
# Configure pytest-asyncio
pytest_plugins = ("pytest_asyncio",)

775
tests/services/test_conversation_service.py Normal file
View File

@@ -0,0 +1,775 @@
"""Unit tests for ConversationManagerService."""
from datetime import UTC, datetime, timedelta
from typing import Literal
from unittest.mock import AsyncMock, Mock, patch
from uuid import uuid4
import pytest
from capa_de_integracion.config import Settings
from capa_de_integracion.models import (
ConversationEntry,
ConversationRequest,
ConversationSession,
DetectIntentResponse,
QueryResult,
User,
)
from capa_de_integracion.services.conversation import ConversationManagerService
from capa_de_integracion.services.dlp import DLPService
from capa_de_integracion.services.rag import RAGServiceBase
from capa_de_integracion.services.storage.firestore import FirestoreService
from capa_de_integracion.services.storage.redis import RedisService
@pytest.fixture
def mock_settings() -> Settings:
"""Create mock settings."""
settings = Mock(spec=Settings)
settings.dlp_template_complete_flow = "test_template"
settings.conversation_context_message_limit = 60
settings.conversation_context_days_limit = 30
return settings
@pytest.fixture
def mock_dlp() -> DLPService:
"""Create mock DLP service."""
dlp = Mock(spec=DLPService)
dlp.get_obfuscated_string = AsyncMock(return_value="obfuscated message")
return dlp
@pytest.fixture
def mock_rag() -> RAGServiceBase:
"""Create mock RAG service."""
rag = Mock(spec=RAGServiceBase)
rag.query = AsyncMock(return_value="RAG response")
return rag
@pytest.fixture
def mock_redis() -> RedisService:
"""Create mock Redis service."""
redis = Mock(spec=RedisService)
redis.get_session = AsyncMock(return_value=None)
redis.save_session = AsyncMock()
redis.get_notification_session = AsyncMock(return_value=None)
redis.delete_notification_session = AsyncMock()
return redis
@pytest.fixture
def mock_firestore() -> FirestoreService:
"""Create mock Firestore service."""
firestore = Mock(spec=FirestoreService)
firestore.get_session_by_phone = AsyncMock(return_value=None)
firestore.create_session = AsyncMock()
firestore.save_session = AsyncMock()
firestore.save_entry = AsyncMock()
firestore.get_entries = AsyncMock(return_value=[])
firestore.update_notification_status = AsyncMock()
# Mock db.collection for notifications
mock_doc = Mock()
mock_doc.exists = False
mock_doc_ref = Mock()
mock_doc_ref.get = AsyncMock(return_value=mock_doc)
mock_collection = Mock()
mock_collection.document = Mock(return_value=mock_doc_ref)
firestore.db = Mock()
firestore.db.collection = Mock(return_value=mock_collection)
firestore.notifications_collection = "notifications"
return firestore
@pytest.fixture
def conversation_service(
mock_settings: Settings,
mock_rag: RAGServiceBase,
mock_redis: RedisService,
mock_firestore: FirestoreService,
mock_dlp: DLPService,
) -> ConversationManagerService:
"""Create conversation service with mocked dependencies."""
with patch(
"capa_de_integracion.services.conversation.QuickReplyContentService"
):
service = ConversationManagerService(
settings=mock_settings,
rag_service=mock_rag,
redis_service=mock_redis,
firestore_service=mock_firestore,
dlp_service=mock_dlp,
)
return service
@pytest.fixture
def sample_session() -> ConversationSession:
"""Create a sample conversation session."""
return ConversationSession(
session_id="test_session_123",
user_id="user_by_phone_1234567890",
telefono="1234567890",
last_modified=datetime.now(UTC),
last_message="Hello",
pantalla_contexto=None,
)
@pytest.fixture
def sample_request() -> ConversationRequest:
"""Create a sample conversation request."""
return ConversationRequest(
mensaje="Hello, I need help",
usuario=User(
telefono="1234567890",
nickname="TestUser",
),
canal="whatsapp",
)
# ============================================================================
# Test Session Management
# ============================================================================
class TestSessionManagement:
"""Tests for session management methods."""
@pytest.mark.asyncio
async def test_obtain_session_from_redis(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
sample_session: ConversationSession,
) -> None:
"""Test obtaining session from Redis."""
mock_redis.get_session = AsyncMock(return_value=sample_session)
result = await conversation_service._obtain_or_create_session("1234567890")
assert result == sample_session
mock_redis.get_session.assert_awaited_once_with("1234567890")
@pytest.mark.asyncio
async def test_obtain_session_from_firestore_when_redis_miss(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
mock_firestore: FirestoreService,
sample_session: ConversationSession,
) -> None:
"""Test obtaining session from Firestore when Redis misses."""
mock_redis.get_session = AsyncMock(return_value=None)
mock_firestore.get_session_by_phone = AsyncMock(return_value=sample_session)
result = await conversation_service._obtain_or_create_session("1234567890")
assert result == sample_session
mock_redis.get_session.assert_awaited_once()
mock_firestore.get_session_by_phone.assert_awaited_once_with("1234567890")
@pytest.mark.asyncio
async def test_create_new_session_when_both_miss(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
mock_firestore: FirestoreService,
sample_session: ConversationSession,
) -> None:
"""Test creating new session when both Redis and Firestore miss."""
mock_redis.get_session = AsyncMock(return_value=None)
mock_firestore.get_session_by_phone = AsyncMock(return_value=None)
mock_firestore.create_session = AsyncMock(return_value=sample_session)
result = await conversation_service._obtain_or_create_session("1234567890")
assert result == sample_session
mock_firestore.create_session.assert_awaited_once()
# Verify the session was auto-cached to Redis
mock_redis.save_session.assert_awaited_once_with(sample_session)
@pytest.mark.asyncio
async def test_session_auto_cached_to_redis(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
mock_firestore: FirestoreService,
sample_session: ConversationSession,
) -> None:
"""Test that newly created session is auto-cached to Redis."""
mock_redis.get_session = AsyncMock(return_value=None)
mock_firestore.get_session_by_phone = AsyncMock(return_value=None)
mock_firestore.create_session = AsyncMock(return_value=sample_session)
await conversation_service._obtain_or_create_session("1234567890")
mock_redis.save_session.assert_awaited_once_with(sample_session)
# ============================================================================
# Test Entry Persistence
# ============================================================================
class TestEntryPersistence:
"""Tests for conversation entry persistence methods."""
@pytest.mark.asyncio
async def test_save_conversation_turn_with_conversacion_type(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
) -> None:
"""Test saving conversation turn with CONVERSACION type."""
await conversation_service._save_conversation_turn(
session_id="test_session",
user_text="Hello",
assistant_text="Hi there",
entry_type="CONVERSACION",
canal="whatsapp",
)
assert mock_firestore.save_entry.await_count == 2
# Verify user entry
user_call = mock_firestore.save_entry.await_args_list[0]
assert user_call[0][0] == "test_session"
user_entry = user_call[0][1]
assert user_entry.entity == "user"
assert user_entry.text == "Hello"
assert user_entry.type == "CONVERSACION"
assert user_entry.canal == "whatsapp"
# Verify assistant entry
assistant_call = mock_firestore.save_entry.await_args_list[1]
assistant_entry = assistant_call[0][1]
assert assistant_entry.entity == "assistant"
assert assistant_entry.text == "Hi there"
assert assistant_entry.type == "CONVERSACION"
@pytest.mark.asyncio
async def test_save_conversation_turn_with_llm_type(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
) -> None:
"""Test saving conversation turn with LLM type."""
await conversation_service._save_conversation_turn(
session_id="test_session",
user_text="What's the weather?",
assistant_text="It's sunny",
entry_type="LLM",
canal="telegram",
)
assert mock_firestore.save_entry.await_count == 2
assistant_call = mock_firestore.save_entry.await_args_list[1]
assistant_entry = assistant_call[0][1]
assert assistant_entry.type == "LLM"
@pytest.mark.asyncio
async def test_save_conversation_turn_with_canal(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
) -> None:
"""Test saving conversation turn with canal parameter."""
await conversation_service._save_conversation_turn(
session_id="test_session",
user_text="Test",
assistant_text="Response",
entry_type="CONVERSACION",
canal="sms",
)
user_call = mock_firestore.save_entry.await_args_list[0]
user_entry = user_call[0][1]
assert user_entry.canal == "sms"
@pytest.mark.asyncio
async def test_save_conversation_turn_without_canal(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
) -> None:
"""Test saving conversation turn without canal parameter."""
await conversation_service._save_conversation_turn(
session_id="test_session",
user_text="Test",
assistant_text="Response",
entry_type="CONVERSACION",
)
user_call = mock_firestore.save_entry.await_args_list[0]
user_entry = user_call[0][1]
assert user_entry.canal is None
# ============================================================================
# Test Session Updates
# ============================================================================
class TestSessionUpdates:
"""Tests for session update methods."""
@pytest.mark.asyncio
async def test_update_session_sets_last_message(
self,
conversation_service: ConversationManagerService,
sample_session: ConversationSession,
) -> None:
"""Test that update_session sets last_message."""
await conversation_service._update_session_after_turn(
session=sample_session,
last_message="New message",
)
assert sample_session.last_message == "New message"
@pytest.mark.asyncio
async def test_update_session_sets_timestamp(
self,
conversation_service: ConversationManagerService,
sample_session: ConversationSession,
) -> None:
"""Test that update_session sets timestamp."""
old_timestamp = sample_session.last_modified
await conversation_service._update_session_after_turn(
session=sample_session,
last_message="New message",
)
assert sample_session.last_modified > old_timestamp
@pytest.mark.asyncio
async def test_update_session_saves_to_firestore(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_session: ConversationSession,
) -> None:
"""Test that update_session saves to Firestore."""
await conversation_service._update_session_after_turn(
session=sample_session,
last_message="New message",
)
mock_firestore.save_session.assert_awaited_once_with(sample_session)
@pytest.mark.asyncio
async def test_update_session_saves_to_redis(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
sample_session: ConversationSession,
) -> None:
"""Test that update_session saves to Redis."""
await conversation_service._update_session_after_turn(
session=sample_session,
last_message="New message",
)
mock_redis.save_session.assert_awaited_once_with(sample_session)
# ============================================================================
# Test Quick Reply Path
# ============================================================================
class TestQuickReplyPath:
"""Tests for quick reply path handling."""
@pytest.mark.asyncio
async def test_quick_reply_path_with_valid_context(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test quick reply path with valid pantalla_contexto."""
sample_session.pantalla_contexto = "screen_123"
sample_session.last_modified = datetime.now(UTC)
# Mock quick reply service
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Quick reply response"),
)
conversation_service._manage_quick_reply_conversation = AsyncMock(
return_value=mock_response
)
result = await conversation_service._handle_quick_reply_path(
request=sample_request,
session=sample_session,
)
assert result == mock_response
@pytest.mark.asyncio
async def test_quick_reply_path_with_stale_context_returns_none(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test quick reply path with stale pantalla_contexto returns None."""
sample_session.pantalla_contexto = "screen_123"
# Set timestamp to 11 minutes ago (stale)
sample_session.last_modified = datetime.now(UTC) - timedelta(minutes=11)
result = await conversation_service._handle_quick_reply_path(
request=sample_request,
session=sample_session,
)
assert result is None
@pytest.mark.asyncio
async def test_quick_reply_path_without_context_returns_none(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test quick reply path without pantalla_contexto returns None."""
sample_session.pantalla_contexto = None
result = await conversation_service._handle_quick_reply_path(
request=sample_request,
session=sample_session,
)
assert result is None
@pytest.mark.asyncio
async def test_quick_reply_path_saves_entries(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test quick reply path saves conversation entries."""
sample_session.pantalla_contexto = "screen_123"
sample_session.last_modified = datetime.now(UTC)
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Quick reply response"),
)
conversation_service._manage_quick_reply_conversation = AsyncMock(
return_value=mock_response
)
await conversation_service._handle_quick_reply_path(
request=sample_request,
session=sample_session,
)
assert mock_firestore.save_entry.await_count == 2
@pytest.mark.asyncio
async def test_quick_reply_path_updates_session(
self,
conversation_service: ConversationManagerService,
mock_redis: RedisService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test quick reply path updates session."""
sample_session.pantalla_contexto = "screen_123"
sample_session.last_modified = datetime.now(UTC)
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Quick reply response"),
)
conversation_service._manage_quick_reply_conversation = AsyncMock(
return_value=mock_response
)
await conversation_service._handle_quick_reply_path(
request=sample_request,
session=sample_session,
)
mock_firestore.save_session.assert_awaited_once()
mock_redis.save_session.assert_awaited_once()
# ============================================================================
# Test Standard Conversation Path
# ============================================================================
class TestStandardConversation:
"""Tests for standard conversation flow."""
@pytest.mark.asyncio
async def test_standard_conversation_loads_history(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation loads history."""
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
mock_firestore.get_entries.assert_awaited_once_with(
sample_session.session_id,
limit=60,
)
@pytest.mark.asyncio
async def test_standard_conversation_queries_rag(
self,
conversation_service: ConversationManagerService,
mock_rag: RAGServiceBase,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation queries RAG service."""
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
mock_rag.query.assert_awaited_once()
@pytest.mark.asyncio
async def test_standard_conversation_saves_entries(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation saves entries."""
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
assert mock_firestore.save_entry.await_count == 2
@pytest.mark.asyncio
async def test_standard_conversation_updates_session(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
mock_redis: RedisService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation updates session."""
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
# save_session is called in _update_session_after_turn
assert mock_firestore.save_session.await_count >= 1
assert mock_redis.save_session.await_count >= 1
@pytest.mark.asyncio
async def test_standard_conversation_marks_notifications_processed(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation marks notifications as processed."""
# Mock that there are active notifications
conversation_service._get_active_notifications = AsyncMock(
return_value=[Mock(texto="Test notification")]
)
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
mock_firestore.update_notification_status.assert_awaited_once()
@pytest.mark.asyncio
async def test_standard_conversation_without_notifications(
self,
conversation_service: ConversationManagerService,
mock_firestore: FirestoreService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test standard conversation without notifications."""
conversation_service._get_active_notifications = AsyncMock(return_value=[])
await conversation_service._handle_standard_conversation(
request=sample_request,
session=sample_session,
)
mock_firestore.update_notification_status.assert_not_awaited()
# ============================================================================
# Test Orchestration
# ============================================================================
class TestOrchestration:
"""Tests for main orchestration logic."""
@pytest.mark.asyncio
async def test_manage_conversation_applies_dlp(
self,
conversation_service: ConversationManagerService,
mock_dlp: DLPService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test manage_conversation applies DLP obfuscation."""
conversation_service._obtain_or_create_session = AsyncMock(
return_value=sample_session
)
conversation_service._handle_standard_conversation = AsyncMock(
return_value=DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Response"),
)
)
await conversation_service.manage_conversation(sample_request)
mock_dlp.get_obfuscated_string.assert_awaited_once()
assert sample_request.mensaje == "obfuscated message"
@pytest.mark.asyncio
async def test_manage_conversation_obtains_session(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test manage_conversation obtains session."""
conversation_service._obtain_or_create_session = AsyncMock(
return_value=sample_session
)
conversation_service._handle_standard_conversation = AsyncMock(
return_value=DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Response"),
)
)
await conversation_service.manage_conversation(sample_request)
conversation_service._obtain_or_create_session.assert_awaited_once_with(
"1234567890"
)
@pytest.mark.asyncio
async def test_manage_conversation_uses_quick_reply_path_when_valid(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test manage_conversation uses quick reply path when valid."""
sample_session.pantalla_contexto = "screen_123"
sample_session.last_modified = datetime.now(UTC)
conversation_service._obtain_or_create_session = AsyncMock(
return_value=sample_session
)
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Quick reply"),
)
conversation_service._handle_quick_reply_path = AsyncMock(
return_value=mock_response
)
conversation_service._handle_standard_conversation = AsyncMock()
result = await conversation_service.manage_conversation(sample_request)
assert result == mock_response
conversation_service._handle_quick_reply_path.assert_awaited_once()
conversation_service._handle_standard_conversation.assert_not_awaited()
@pytest.mark.asyncio
async def test_manage_conversation_uses_standard_path_when_no_context(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test manage_conversation uses standard path when no context."""
sample_session.pantalla_contexto = None
conversation_service._obtain_or_create_session = AsyncMock(
return_value=sample_session
)
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Standard response"),
)
conversation_service._handle_standard_conversation = AsyncMock(
return_value=mock_response
)
result = await conversation_service.manage_conversation(sample_request)
assert result == mock_response
conversation_service._handle_standard_conversation.assert_awaited_once()
@pytest.mark.asyncio
async def test_manage_conversation_uses_standard_path_when_stale_context(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
sample_session: ConversationSession,
) -> None:
"""Test manage_conversation uses standard path when context is stale."""
sample_session.pantalla_contexto = "screen_123"
sample_session.last_modified = datetime.now(UTC) - timedelta(minutes=11)
conversation_service._obtain_or_create_session = AsyncMock(
return_value=sample_session
)
mock_response = DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText="Standard response"),
)
conversation_service._handle_quick_reply_path = AsyncMock(return_value=None)
conversation_service._handle_standard_conversation = AsyncMock(
return_value=mock_response
)
result = await conversation_service.manage_conversation(sample_request)
assert result == mock_response
conversation_service._handle_standard_conversation.assert_awaited_once()
@pytest.mark.asyncio
async def test_manage_conversation_handles_exceptions(
self,
conversation_service: ConversationManagerService,
sample_request: ConversationRequest,
) -> None:
"""Test manage_conversation handles exceptions properly."""
conversation_service._obtain_or_create_session = AsyncMock(
side_effect=Exception("Test error")
)
with pytest.raises(Exception, match="Test error"):
await conversation_service.manage_conversation(sample_request)

4
tests/services/test_dlp_service.py
View File

@@ -6,7 +6,7 @@ import pytest
from google.cloud.dlp_v2 import types
from capa_de_integracion.config import Settings
from capa_de_integracion.services.dlp_service import DLPService
from capa_de_integracion.services import DLPService
@pytest.fixture
@@ -21,7 +21,7 @@ def mock_settings():
@pytest.fixture
def service(mock_settings):
"""Create DLPService instance with mocked client."""
with patch("capa_de_integracion.services.dlp_service.dlp_v2.DlpServiceAsyncClient"):
with patch("capa_de_integracion.services.dlp.dlp_v2.DlpServiceAsyncClient"):
return DLPService(mock_settings)

2
tests/services/test_firestore_service.py
View File

@@ -7,7 +7,7 @@ from inline_snapshot import snapshot
from capa_de_integracion.models import ConversationEntry, ConversationSession
from capa_de_integracion.models.notification import Notification
from capa_de_integracion.services.firestore_service import FirestoreService
from capa_de_integracion.services.storage import FirestoreService
@pytest.mark.vcr

36
tests/services/test_notification_manager.py
View File

@@ -6,10 +6,8 @@ import pytest
from capa_de_integracion.config import Settings
from capa_de_integracion.models.notification import ExternalNotificationRequest
from capa_de_integracion.services.dlp_service import DLPService
from capa_de_integracion.services.firestore_service import FirestoreService
from capa_de_integracion.services.notification_manager import NotificationManagerService
from capa_de_integracion.services.redis_service import RedisService
from capa_de_integracion.services import DLPService, NotificationManagerService
from capa_de_integracion.services.storage import FirestoreService, RedisService
@pytest.fixture
@@ -157,3 +155,33 @@ async def test_process_notification_generates_unique_id(service, mock_redis):
notification2 = mock_redis.save_or_append_notification.call_args[0][0]
assert notification1.id_notificacion != notification2.id_notificacion
@pytest.mark.asyncio
async def test_process_notification_firestore_exception_handling(
service, mock_redis, mock_firestore
):
"""Test that Firestore exceptions are handled gracefully in background task."""
import asyncio
# Make Firestore save fail
mock_firestore.save_or_append_notification = AsyncMock(
side_effect=Exception("Firestore connection error")
)
request = ExternalNotificationRequest(
telefono="555-1234",
texto="Test notification",
parametros_ocultos=None,
)
await service.process_notification(request)
# Redis should succeed
mock_redis.save_or_append_notification.assert_called_once()
# Give the background task time to execute
await asyncio.sleep(0.1)
# Firestore should have been attempted (and failed)
mock_firestore.save_or_append_notification.assert_called_once()

160
tests/services/test_quick_reply_content.py
View File

@@ -1,27 +1,29 @@
"""Tests for QuickReplyContentService."""
import json
from pathlib import Path
from unittest.mock import Mock, patch
from unittest.mock import Mock
import pytest
from capa_de_integracion.config import Settings
from capa_de_integracion.models.quick_replies import QuickReplyScreen
from capa_de_integracion.services.quick_reply_content import QuickReplyContentService
from capa_de_integracion.services import QuickReplyContentService
@pytest.fixture
def mock_settings():
def mock_settings(tmp_path):
"""Create mock settings for testing."""
settings = Mock(spec=Settings)
settings.base_path = Path("/tmp/test_resources")
# Create the quick_replies directory
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
settings.base_path = tmp_path
return settings
@pytest.fixture
def service(mock_settings):
"""Create QuickReplyContentService instance."""
"""Create QuickReplyContentService instance with empty cache."""
return QuickReplyContentService(mock_settings)
@@ -59,22 +61,19 @@ async def test_get_quick_replies_whitespace_screen_id(service):
@pytest.mark.asyncio
async def test_get_quick_replies_file_not_found(service, tmp_path):
"""Test get_quick_replies raises error when file not found."""
# Set service to use a temp directory where the file won't exist
service.quick_replies_path = tmp_path / "nonexistent_dir"
with pytest.raises(ValueError, match="Error loading quick replies"):
async def test_get_quick_replies_file_not_found(service):
"""Test get_quick_replies raises error when screen not in cache."""
# Cache is empty (no files loaded), so any screen_id should raise ValueError
with pytest.raises(ValueError, match="Quick reply not found for screen_id"):
await service.get_quick_replies("nonexistent")
@pytest.mark.asyncio
async def test_get_quick_replies_success(service, tmp_path):
"""Test get_quick_replies successfully loads file."""
# Create test JSON file
async def test_get_quick_replies_success(tmp_path):
"""Test get_quick_replies successfully retrieves from cache."""
# Create test JSON file BEFORE initializing service
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
service.quick_replies_path = quick_replies_dir
test_data = {
"header": "Test Header",
@@ -97,6 +96,11 @@ async def test_get_quick_replies_success(service, tmp_path):
test_file = quick_replies_dir / "test_screen.json"
test_file.write_text(json.dumps(test_data), encoding="utf-8")
# Initialize service - it will preload the file into cache
settings = Mock(spec=Settings)
settings.base_path = tmp_path
service = QuickReplyContentService(settings)
result = await service.get_quick_replies("test_screen")
assert isinstance(result, QuickReplyScreen)
@@ -114,25 +118,30 @@ async def test_get_quick_replies_success(service, tmp_path):
@pytest.mark.asyncio
async def test_get_quick_replies_invalid_json(service, tmp_path):
"""Test get_quick_replies raises error for invalid JSON."""
async def test_get_quick_replies_invalid_json(tmp_path):
"""Test that invalid JSON files are skipped during cache preload."""
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
service.quick_replies_path = quick_replies_dir
# Create invalid JSON file
test_file = quick_replies_dir / "invalid.json"
test_file.write_text("{ invalid json }", encoding="utf-8")
with pytest.raises(ValueError, match="Invalid JSON format"):
# Initialize service - invalid file should be logged but not crash
settings = Mock(spec=Settings)
settings.base_path = tmp_path
service = QuickReplyContentService(settings)
# Requesting the invalid screen should raise ValueError (not in cache)
with pytest.raises(ValueError, match="Quick reply not found for screen_id"):
await service.get_quick_replies("invalid")
@pytest.mark.asyncio
async def test_get_quick_replies_minimal_data(service, tmp_path):
"""Test get_quick_replies with minimal data."""
async def test_get_quick_replies_minimal_data(tmp_path):
"""Test get_quick_replies with minimal data from cache."""
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
service.quick_replies_path = quick_replies_dir
test_data = {
"preguntas": [],
@@ -141,6 +150,11 @@ async def test_get_quick_replies_minimal_data(service, tmp_path):
test_file = quick_replies_dir / "minimal.json"
test_file.write_text(json.dumps(test_data), encoding="utf-8")
# Initialize service - it will preload the file
settings = Mock(spec=Settings)
settings.base_path = tmp_path
service = QuickReplyContentService(settings)
result = await service.get_quick_replies("minimal")
assert isinstance(result, QuickReplyScreen)
@@ -168,3 +182,103 @@ async def test_validate_file_not_exists(service, tmp_path):
with pytest.raises(ValueError, match="Quick reply file not found"):
service._validate_file(test_file, "test")
@pytest.mark.asyncio
async def test_cache_preload_multiple_files(tmp_path):
"""Test that cache preloads multiple files correctly."""
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
# Create multiple test files
for screen_id in ["home", "pagos", "transferencia"]:
test_data = {
"header": f"{screen_id} header",
"preguntas": [
{"titulo": f"Q1 for {screen_id}", "respuesta": "Answer 1"},
],
}
test_file = quick_replies_dir / f"{screen_id}.json"
test_file.write_text(json.dumps(test_data), encoding="utf-8")
# Initialize service
settings = Mock(spec=Settings)
settings.base_path = tmp_path
service = QuickReplyContentService(settings)
# Verify all screens are in cache
home = await service.get_quick_replies("home")
assert home.header == "home header"
pagos = await service.get_quick_replies("pagos")
assert pagos.header == "pagos header"
transferencia = await service.get_quick_replies("transferencia")
assert transferencia.header == "transferencia header"
@pytest.mark.asyncio
async def test_cache_preload_with_mixed_valid_invalid(tmp_path):
"""Test cache preload handles mix of valid and invalid files."""
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
# Create valid file
valid_data = {"header": "valid", "preguntas": []}
(quick_replies_dir / "valid.json").write_text(
json.dumps(valid_data), encoding="utf-8",
)
# Create invalid file
(quick_replies_dir / "invalid.json").write_text(
"{ invalid }", encoding="utf-8",
)
# Initialize service - should not crash
settings = Mock(spec=Settings)
settings.base_path = tmp_path
service = QuickReplyContentService(settings)
# Valid file should be in cache
valid = await service.get_quick_replies("valid")
assert valid.header == "valid"
# Invalid file should not be in cache
with pytest.raises(ValueError, match="Quick reply not found"):
await service.get_quick_replies("invalid")
@pytest.mark.asyncio
async def test_cache_preload_handles_generic_exception(tmp_path, monkeypatch):
"""Test cache preload handles generic exceptions during file processing."""
from pathlib import Path
from unittest.mock import Mock
quick_replies_dir = tmp_path / "quick_replies"
quick_replies_dir.mkdir()
# Create valid JSON file
valid_data = {"header": "test", "preguntas": []}
(quick_replies_dir / "test.json").write_text(
json.dumps(valid_data), encoding="utf-8",
)
# Mock _parse_quick_reply_data to raise generic exception
def mock_parse_error(*args, **kwargs):
raise ValueError("Simulated parsing error")
settings = Mock(spec=Settings)
settings.base_path = tmp_path
# Initialize service and patch the parsing method
with monkeypatch.context() as m:
service = QuickReplyContentService(settings)
m.setattr(service, "_parse_quick_reply_data", mock_parse_error)
# Manually call _preload_cache to trigger the exception
service._cache.clear()
service._preload_cache()
# The file should not be in cache due to the exception
with pytest.raises(ValueError, match="Quick reply not found"):
await service.get_quick_replies("test")

166
tests/services/test_quick_reply_session.py Normal file
View File

@@ -0,0 +1,166 @@
"""Tests for QuickReplySessionService."""
from unittest.mock import AsyncMock, Mock
from uuid import uuid4
import pytest
from capa_de_integracion.models.conversation import ConversationSession
from capa_de_integracion.models.quick_replies import QuickReplyScreen
from capa_de_integracion.services import QuickReplySessionService
from capa_de_integracion.services.quick_reply.content import QuickReplyContentService
from capa_de_integracion.services.storage.firestore import FirestoreService
from capa_de_integracion.services.storage.redis import RedisService
@pytest.fixture
def mock_redis():
"""Create mock Redis service."""
redis = Mock(spec=RedisService)
redis.save_session = AsyncMock()
return redis
@pytest.fixture
def mock_firestore():
"""Create mock Firestore service."""
firestore = Mock(spec=FirestoreService)
firestore.get_session_by_phone = AsyncMock()
firestore.create_session = AsyncMock()
firestore.update_pantalla_contexto = AsyncMock()
return firestore
@pytest.fixture
def mock_content():
"""Create mock QuickReplyContentService."""
content = Mock(spec=QuickReplyContentService)
content.get_quick_replies = AsyncMock()
return content
@pytest.fixture
def service(mock_redis, mock_firestore, mock_content):
"""Create QuickReplySessionService instance."""
return QuickReplySessionService(
redis_service=mock_redis,
firestore_service=mock_firestore,
quick_reply_content_service=mock_content,
)
@pytest.mark.asyncio
async def test_validate_phone_empty_string(service):
"""Test phone validation with empty string."""
with pytest.raises(ValueError, match="Phone number is required"):
service._validate_phone("")
@pytest.mark.asyncio
async def test_validate_phone_whitespace(service):
"""Test phone validation with whitespace only."""
with pytest.raises(ValueError, match="Phone number is required"):
service._validate_phone(" ")
@pytest.mark.asyncio
async def test_start_session_new_user(service, mock_firestore, mock_redis, mock_content):
"""Test starting a quick reply session for a new user."""
# Setup mocks
mock_firestore.get_session_by_phone.return_value = None # No existing session
# Mock create_session to return a session with the ID that was passed in
def create_session_side_effect(session_id, user_id, telefono, pantalla_contexto):
return ConversationSession.create(
session_id=session_id,
user_id=user_id,
telefono=telefono,
pantalla_contexto=pantalla_contexto,
)
mock_firestore.create_session.side_effect = create_session_side_effect
test_quick_replies = QuickReplyScreen(
header="Home Screen",
body=None,
button=None,
header_section=None,
preguntas=[],
)
mock_content.get_quick_replies.return_value = test_quick_replies
# Execute
result = await service.start_quick_reply_session(
telefono="555-1234",
_nombre="John",
pantalla_contexto="home",
)
# Verify
assert result.session_id is not None # Session ID should be generated
assert result.quick_replies.header == "Home Screen"
mock_firestore.get_session_by_phone.assert_called_once_with("555-1234")
mock_firestore.create_session.assert_called_once()
# Verify create_session was called with correct parameters
call_args = mock_firestore.create_session.call_args
assert call_args[0][2] == "555-1234" # telefono
assert call_args[0][3] == "home" # pantalla_contexto
mock_redis.save_session.assert_called_once()
mock_content.get_quick_replies.assert_called_once_with("home")
@pytest.mark.asyncio
async def test_start_session_existing_user(service, mock_firestore, mock_redis, mock_content):
"""Test starting a quick reply session for an existing user."""
# Setup mocks - existing session
test_session_id = "existing-session-123"
test_session = ConversationSession.create(
session_id=test_session_id,
user_id="user_by_phone_5551234",
telefono="555-1234",
pantalla_contexto="old_screen",
)
mock_firestore.get_session_by_phone.return_value = test_session
test_quick_replies = QuickReplyScreen(
header="Payments Screen",
body=None,
button=None,
header_section=None,
preguntas=[],
)
mock_content.get_quick_replies.return_value = test_quick_replies
# Execute
result = await service.start_quick_reply_session(
telefono="555-1234",
_nombre="John",
pantalla_contexto="pagos",
)
# Verify
assert result.session_id == test_session_id
assert result.quick_replies.header == "Payments Screen"
mock_firestore.get_session_by_phone.assert_called_once_with("555-1234")
mock_firestore.update_pantalla_contexto.assert_called_once_with(
test_session_id,
"pagos",
)
mock_firestore.create_session.assert_not_called()
mock_redis.save_session.assert_called_once()
mock_content.get_quick_replies.assert_called_once_with("pagos")
@pytest.mark.asyncio
async def test_start_session_invalid_phone(service):
"""Test starting session with invalid phone number."""
with pytest.raises(ValueError, match="Phone number is required"):
await service.start_quick_reply_session(
telefono="",
_nombre="John",
pantalla_contexto="home",
)

19
tests/services/test_rag_services.py
View File

@@ -201,6 +201,25 @@ class TestHTTPRAGService:
mock_client.aclose.assert_called_once()
@pytest.mark.asyncio
async def test_http_generic_exception(self):
"""Test HTTP RAG service handles generic exceptions during processing."""
mock_response = Mock()
mock_response.raise_for_status = Mock()
# Make json() raise a generic exception
mock_response.json = Mock(side_effect=ValueError("Invalid response format"))
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_response)
mock_client_class.return_value = mock_client
service = HTTPRAGService(endpoint_url="http://test.example.com/rag")
messages = [{"role": "user", "content": "Hello"}]
with pytest.raises(ValueError, match="Invalid response format"):
await service.query(messages)
class TestRAGModels:
"""Tests for RAG data models."""

2
tests/services/test_redis_service.py
View File

@@ -9,7 +9,7 @@ from inline_snapshot import snapshot
from capa_de_integracion.config import Settings
from capa_de_integracion.models import ConversationEntry, ConversationSession
from capa_de_integracion.models.notification import Notification, NotificationSession
from capa_de_integracion.services.redis_service import RedisService
from capa_de_integracion.services.storage import RedisService
class TestConnectionManagement:

20
tests/test_dependencies.py
View File

@@ -11,6 +11,7 @@ from capa_de_integracion.dependencies import (
get_firestore_service,
get_notification_manager,
get_quick_reply_content_service,
get_quick_reply_session_service,
get_rag_service,
get_redis_service,
init_services,
@@ -22,10 +23,10 @@ from capa_de_integracion.services import (
DLPService,
NotificationManagerService,
QuickReplyContentService,
QuickReplySessionService,
)
from capa_de_integracion.services.firestore_service import FirestoreService
from capa_de_integracion.services.rag import EchoRAGService, HTTPRAGService
from capa_de_integracion.services.redis_service import RedisService
from capa_de_integracion.services.storage import FirestoreService, RedisService
def test_get_redis_service():
@@ -77,6 +78,21 @@ def test_get_quick_reply_content_service():
assert service is service2
def test_get_quick_reply_session_service():
"""Test get_quick_reply_session_service returns QuickReplySessionService."""
get_quick_reply_session_service.cache_clear()
get_redis_service.cache_clear()
get_firestore_service.cache_clear()
get_quick_reply_content_service.cache_clear()
service = get_quick_reply_session_service()
assert isinstance(service, QuickReplySessionService)
# Should return same instance (cached)
service2 = get_quick_reply_session_service()
assert service is service2
def test_get_notification_manager():
"""Test get_notification_manager returns NotificationManagerService."""
get_notification_manager.cache_clear()

81
tests/test_routers_simple.py
View File

@@ -8,6 +8,11 @@ from capa_de_integracion.models import ConversationRequest, DetectIntentResponse
from capa_de_integracion.models.notification import ExternalNotificationRequest
from capa_de_integracion.models.quick_replies import QuickReplyScreen
from capa_de_integracion.routers import conversation, notification, quick_replies
from capa_de_integracion.routers.quick_replies import (
QuickReplyScreenRequest,
QuickReplyUser,
)
from capa_de_integracion.services.quick_reply.session import QuickReplySessionResponse
@pytest.mark.asyncio
@@ -136,3 +141,79 @@ async def test_process_notification_general_error():
await notification.process_notification(request, mock_manager)
assert exc_info.value.status_code == 500
@pytest.mark.asyncio
async def test_start_quick_reply_session_success():
"""Test quick reply session endpoint with success."""
mock_service = Mock()
mock_result = QuickReplySessionResponse(
session_id="test-session-123",
quick_replies=QuickReplyScreen(
header="Test Header",
body=None,
button=None,
header_section=None,
preguntas=[],
),
)
mock_service.start_quick_reply_session = AsyncMock(return_value=mock_result)
request = QuickReplyScreenRequest(
usuario=QuickReplyUser(telefono="555-1234", nombre="John"),
pantallaContexto="home",
)
response = await quick_replies.start_quick_reply_session(request, mock_service)
assert response.response_id == "test-session-123"
assert response.quick_replies.header == "Test Header"
mock_service.start_quick_reply_session.assert_called_once_with(
telefono="555-1234",
_nombre="John",
pantalla_contexto="home",
)
@pytest.mark.asyncio
async def test_start_quick_reply_session_value_error():
"""Test quick reply session with ValueError."""
mock_service = Mock()
mock_service.start_quick_reply_session = AsyncMock(
side_effect=ValueError("Invalid screen"),
)
request = QuickReplyScreenRequest(
usuario=QuickReplyUser(telefono="555-1234", nombre="John"),
pantallaContexto="invalid",
)
from fastapi import HTTPException
with pytest.raises(HTTPException) as exc_info:
await quick_replies.start_quick_reply_session(request, mock_service)
assert exc_info.value.status_code == 400
assert "Invalid screen" in str(exc_info.value.detail)
@pytest.mark.asyncio
async def test_start_quick_reply_session_general_error():
"""Test quick reply session with general Exception."""
mock_service = Mock()
mock_service.start_quick_reply_session = AsyncMock(
side_effect=RuntimeError("Database error"),
)
request = QuickReplyScreenRequest(
usuario=QuickReplyUser(telefono="555-1234", nombre="John"),
pantallaContexto="home",
)
from fastapi import HTTPException
with pytest.raises(HTTPException) as exc_info:
await quick_replies.start_quick_reply_session(request, mock_service)
assert exc_info.value.status_code == 500
assert "Internal server error" in str(exc_info.value.detail)