139 lines
3.8 KiB
Python
139 lines
3.8 KiB
Python
"""Simplified router tests using direct function calls."""
|
|
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from capa_de_integracion.models import ConversationRequest, DetectIntentResponse, User
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_intent_success():
|
|
"""Test detect intent endpoint with success."""
|
|
mock_manager = Mock()
|
|
mock_manager.manage_conversation = AsyncMock(
|
|
return_value=DetectIntentResponse(
|
|
response_id="test-123",
|
|
query_result=None,
|
|
),
|
|
)
|
|
|
|
request = ConversationRequest(
|
|
mensaje="Hello",
|
|
usuario=User(telefono="555-1234"),
|
|
canal="web",
|
|
)
|
|
|
|
response = await conversation.detect_intent(request, mock_manager)
|
|
|
|
assert response.response_id == "test-123"
|
|
mock_manager.manage_conversation.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_intent_value_error():
|
|
"""Test detect intent with ValueError."""
|
|
mock_manager = Mock()
|
|
mock_manager.manage_conversation = AsyncMock(
|
|
side_effect=ValueError("Invalid input"),
|
|
)
|
|
|
|
request = ConversationRequest(
|
|
mensaje="Test",
|
|
usuario=User(telefono="555-1234"),
|
|
canal="web",
|
|
)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await conversation.detect_intent(request, mock_manager)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "Invalid input" in str(exc_info.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_intent_general_error():
|
|
"""Test detect intent with general Exception."""
|
|
mock_manager = Mock()
|
|
mock_manager.manage_conversation = AsyncMock(
|
|
side_effect=RuntimeError("Server error"),
|
|
)
|
|
|
|
request = ConversationRequest(
|
|
mensaje="Test",
|
|
usuario=User(telefono="555-1234"),
|
|
canal="web",
|
|
)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await conversation.detect_intent(request, mock_manager)
|
|
|
|
assert exc_info.value.status_code == 500
|
|
assert "Internal server error" in str(exc_info.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_notification_success():
|
|
"""Test notification processing success."""
|
|
mock_manager = Mock()
|
|
mock_manager.process_notification = AsyncMock()
|
|
|
|
request = ExternalNotificationRequest(
|
|
telefono="555-1234",
|
|
texto="Your card was blocked",
|
|
)
|
|
|
|
result = await notification.process_notification(request, mock_manager)
|
|
|
|
assert result is None
|
|
mock_manager.process_notification.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_notification_value_error():
|
|
"""Test notification with ValueError."""
|
|
mock_manager = Mock()
|
|
mock_manager.process_notification = AsyncMock(
|
|
side_effect=ValueError("Invalid phone"),
|
|
)
|
|
|
|
request = ExternalNotificationRequest(
|
|
telefono="",
|
|
texto="Test",
|
|
)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await notification.process_notification(request, mock_manager)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_notification_general_error():
|
|
"""Test notification with general error."""
|
|
mock_manager = Mock()
|
|
mock_manager.process_notification = AsyncMock(
|
|
side_effect=RuntimeError("Server error"),
|
|
)
|
|
|
|
request = ExternalNotificationRequest(
|
|
telefono="555-1234",
|
|
texto="Test",
|
|
)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await notification.process_notification(request, mock_manager)
|
|
|
|
assert exc_info.value.status_code == 500
|