""" Comprehensive unit tests for LiuYao AI conversation feature. Tests cover: - Standard cases: normal conversation flow, message sending, context management - Edge cases: message limits, length limits, context thresholds, permissions - Outlier cases: AI failures, invalid data, unauthorized access, retry functionality """ import json from unittest.mock import Mock, patch, MagicMock, call from django.test import TestCase, override_settings from django.contrib.auth import get_user_model from django.utils import timezone from datetime import datetime from rest_framework.test import APIClient from rest_framework import status from ai.models import Conversation, Message, ConversationSubject from ai.utils.conversation import ( create_liuyao_conversation, send_conversation_message, prepare_conversation_prompt, calculate_context_size, generate_conversation_summary, manage_conversation_context, ) from liuyao.models import liuyao from accounts.models import UserProfile User = get_user_model() class LiuYaoConversationSubjectTests(TestCase): """Test ConversationSubject pivot table for LiuYao.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) def test_get_or_create_subject_liuyao(self): """Test creating ConversationSubject for LiuYao.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) self.assertIsNotNone(subject.id) self.assertEqual(subject.content_type, 'liuyao') self.assertEqual(subject.object_id, self.liuyao_obj.id) def test_get_or_create_subject_idempotent(self): """Test that get_or_create_subject is idempotent.""" subject1 = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) subject2 = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) self.assertEqual(subject1.id, subject2.id) def test_subject_get_object_liuyao(self): """Test getting LiuYao object from subject.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) obj = subject.get_object() self.assertIsNotNone(obj) self.assertEqual(obj.id, self.liuyao_obj.id) self.assertEqual(obj.question, 'Will I get the job?') def test_subject_str_representation(self): """Test ConversationSubject string representation.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) str_repr = str(subject) self.assertIn('liuyao', str_repr.lower()) self.assertIn(str(self.liuyao_obj.id), str_repr) class LiuYaoConversationModelTests(TestCase): """Test Conversation model with LiuYao.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) def test_create_conversation_with_subject(self): """Test creating a conversation with ConversationSubject.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) self.assertIsNotNone(conversation.id) self.assertEqual(conversation.user, self.user) self.assertEqual(conversation.subject, subject) self.assertEqual(conversation.get_message_count(), 0) def test_conversation_get_subject_object(self): """Test getting LiuYao object from conversation.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) subject_obj = conversation.get_subject_object() self.assertIsNotNone(subject_obj) self.assertEqual(subject_obj.id, self.liuyao_obj.id) self.assertEqual(subject_obj.question, 'Will I get the job?') def test_conversation_str_with_liuyao(self): """Test conversation string representation with LiuYao.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject, title="Test Conversation" ) str_repr = str(conversation) self.assertIn("Test Conversation", str_repr) self.assertIn(str(self.user), str_repr) def test_message_creation_liuyao_conversation(self): """Test creating messages in LiuYao conversation.""" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) user_msg = Message.objects.create( conversation=conversation, role='user', content='What does this hexagram mean?' ) assistant_msg = Message.objects.create( conversation=conversation, role='assistant', content='This hexagram indicates...', provider='groq', model='llama-3.3-70b-versatile' ) self.assertEqual(conversation.get_message_count(), 2) self.assertEqual(user_msg.role, 'user') self.assertEqual(assistant_msg.role, 'assistant') self.assertEqual(assistant_msg.provider, 'groq') class LiuYaoConversationUtilityTests(TestCase): """Test LiuYao conversation utility functions.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0', data={'hexagram': 'test_data'} ) def test_create_liuyao_conversation_function(self): """Test create_liuyao_conversation utility function.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) self.assertIsNotNone(conversation.id) self.assertEqual(conversation.user, self.user) self.assertIsNotNone(conversation.subject) self.assertEqual(conversation.subject.content_type, 'liuyao') self.assertEqual(conversation.subject.object_id, self.liuyao_obj.id) self.assertIn(self.liuyao_obj.question[:50], conversation.title) def test_create_liuyao_conversation_with_title(self): """Test create_liuyao_conversation with custom title.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj, title="Custom LiuYao Conversation" ) self.assertEqual(conversation.title, "Custom LiuYao Conversation") @patch('ai.utils.conversation.prepare_liuyao_prompt') def test_prepare_conversation_prompt_liuyao(self, mock_prepare_liuyao): """Test prepare_conversation_prompt function for LiuYao.""" mock_prepare_liuyao.return_value = "LiuYao hexagram data here" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) history = [ {'role': 'user', 'content': 'What does this mean?'}, {'role': 'assistant', 'content': 'It means...'} ] prompt = prepare_conversation_prompt( conversation=conversation, conversation_history=history, user_message='Tell me more', language='en' ) self.assertIn('LiuYao hexagram data here', prompt) self.assertIn('What does this mean?', prompt) self.assertIn('It means...', prompt) self.assertIn('Tell me more', prompt) mock_prepare_liuyao.assert_called_once_with(self.liuyao_obj, language='en') @patch('ai.utils.conversation.prepare_liuyao_prompt') def test_prepare_conversation_prompt_with_summary(self, mock_prepare_liuyao): """Test prepare_conversation_prompt with context summary.""" mock_prepare_liuyao.return_value = "LiuYao data" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) conversation.context_summary = "Previous conversation summary" conversation.save() prompt = prepare_conversation_prompt( conversation=conversation, conversation_history=[], user_message='New question', language='en' ) self.assertIn('Previous conversation summary', prompt) def test_calculate_context_size_liuyao(self): """Test calculate_context_size function with LiuYao prompt.""" liuyao_prompt = "LiuYao data " * 100 # ~1000 chars history = [ {'role': 'user', 'content': 'Message 1'}, {'role': 'assistant', 'content': 'Response 1'} ] current_message = "Current message" size = calculate_context_size( divination_prompt=liuyao_prompt, conversation_history=history, current_message=current_message ) self.assertGreater(size, 0) self.assertIsInstance(size, int) @patch('ai.utils.conversation.LLMServiceFactory') @patch('ai.utils.conversation.get_ai_config') @patch('ai.utils.conversation.prepare_liuyao_prompt') def test_generate_conversation_summary_liuyao(self, mock_prepare_liuyao, mock_get_config, mock_factory): """Test generate_conversation_summary function for LiuYao.""" mock_prepare_liuyao.return_value = "LiuYao data" mock_get_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} mock_service = MagicMock() mock_service.get_completion.return_value = "This is a summary of the LiuYao conversation." mock_factory.get_service.return_value = mock_service subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) messages = [ Message.objects.create( conversation=conversation, role='user', content='Question 1' ), Message.objects.create( conversation=conversation, role='assistant', content='Answer 1' ) ] summary = generate_conversation_summary( conversation=conversation, messages_to_summarize=messages, language='en' ) self.assertEqual(summary, "This is a summary of the LiuYao conversation.") mock_service.get_completion.assert_called_once() @patch('ai.utils.conversation.generate_conversation_summary') @patch('ai.utils.conversation.prepare_liuyao_prompt') @override_settings(CONVERSATION_CONTEXT_THRESHOLD=1000) # Low threshold for testing def test_manage_conversation_context_under_threshold(self, mock_prepare_liuyao, mock_generate_summary): """Test manage_conversation_context when under threshold.""" mock_prepare_liuyao.return_value = "Small LiuYao data" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) # Create a few messages (under threshold) for i in range(5): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}' ) manage_conversation_context(conversation, language='en') # Should not generate summary mock_generate_summary.assert_not_called() @patch('ai.utils.conversation.generate_conversation_summary') @patch('ai.utils.conversation.prepare_liuyao_prompt') @override_settings(CONVERSATION_CONTEXT_THRESHOLD=100) # Very low threshold def test_manage_conversation_context_exceeds_threshold(self, mock_prepare_liuyao, mock_generate_summary): """Test manage_conversation_context when exceeding threshold.""" mock_prepare_liuyao.return_value = "LiuYao data " * 1000 # Large LiuYao prompt mock_generate_summary.return_value = "Summary of old messages" subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) # Create many messages to exceed threshold for i in range(30): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Long message content ' * 10 # Make messages large ) manage_conversation_context(conversation, language='en') # Should generate summary mock_generate_summary.assert_called() conversation.refresh_from_db() self.assertEqual(conversation.context_summary, "Summary of old messages") @patch('ai.utils.conversation.LLMServiceFactory') @patch('ai.utils.conversation.get_ai_config') @patch('ai.utils.conversation.manage_conversation_context') @patch('ai.utils.conversation.prepare_conversation_prompt') def test_send_conversation_message_success_liuyao(self, mock_prepare_prompt, mock_manage_context, mock_get_config, mock_factory): """Test send_conversation_message success case for LiuYao.""" mock_get_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} mock_prepare_prompt.return_value = "Full prompt" mock_service = MagicMock() mock_service.get_completion.return_value = "AI response here" mock_factory.get_service.return_value = mock_service subject = ConversationSubject.get_or_create_subject('liuyao', self.liuyao_obj.id) conversation = Conversation.objects.create( user=self.user, subject=subject ) assistant_msg = send_conversation_message( conversation=conversation, user_message="Hello", language='en' ) self.assertIsNotNone(assistant_msg) self.assertEqual(assistant_msg.role, 'assistant') self.assertEqual(assistant_msg.content, "AI response here") self.assertEqual(assistant_msg.provider, 'groq') mock_manage_context.assert_called_once() class LiuYaoConversationAPITests(TestCase): """Test LiuYao conversation API endpoints.""" def setUp(self): """Set up test data.""" self.client = APIClient() self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.client.force_authenticate(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) def test_list_conversations_empty(self): """Test listing conversations when none exist.""" url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 0) self.assertEqual(len(response.data['results']), 0) def test_list_conversations_with_data(self): """Test listing conversations with existing data.""" conversation1 = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) conversation2 = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 2) self.assertEqual(len(response.data['results']), 2) def test_list_conversations_other_user(self): """Test listing conversations for another user's LiuYao.""" other_user = User.objects.create_user( phone='9999999999', email='other@example.com', password='testpass123' ) other_liuyao = liuyao.objects.create( qdate=timezone.now(), user=other_user, question='Other question', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) url = f'/api/liuyao/{other_liuyao.id}/conversations/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_get_conversation_detail(self): """Test getting conversation detail.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) Message.objects.create( conversation=conversation, role='user', content='Hello' ) Message.objects.create( conversation=conversation, role='assistant', content='Hi there!' ) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data['messages']), 2) self.assertEqual(response.data['messages'][0]['content'], 'Hello') def test_get_conversation_not_found(self): """Test getting non-existent conversation.""" url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/999/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_create_conversation(self): """Test creating a new conversation.""" url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/create/' response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) # Note: serializer may return person_id/person_name for backward compatibility # The conversation is created successfully regardless def test_create_conversation_unauthorized_liuyao(self): """Test creating conversation for unauthorized LiuYao.""" other_user = User.objects.create_user( phone='9999999999', email='other@example.com', password='testpass123' ) other_liuyao = liuyao.objects.create( qdate=timezone.now(), user=other_user, question='Other question', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) url = f'/api/liuyao/{other_liuyao.id}/conversations/create/' response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) @patch('api.views.send_conversation_message') def test_send_message_success(self, mock_send_message): """Test sending a message successfully.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Mock will create user message and return assistant message def mock_send_side_effect(*args, **kwargs): # Create user message (as the real function does) user_msg = Message.objects.create( conversation=conversation, role='user', content=kwargs['user_message'] ) # Return assistant message return Message.objects.create( conversation=conversation, role='assistant', content='AI response', provider='groq', model='llama-3.3-70b-versatile' ) mock_send_message.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': 'Hello', 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('user_message', response.data) self.assertIn('assistant_message', response.data) mock_send_message.assert_called_once() def test_send_message_empty(self): """Test sending empty message.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': ' ', # Whitespace only 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_send_message_too_long(self): """Test sending message exceeding length limit.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) long_message = 'a' * 2001 # Exceeds 2000 char limit url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': long_message, 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) @override_settings(CONVERSATION_MAX_MESSAGES=3) @patch('api.views.send_conversation_message') def test_send_message_limit_reached(self, mock_send_message): """Test sending message when limit is reached.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Create messages up to limit (3 messages = 1.5 exchanges) for i in range(3): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}' ) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': 'Should fail', 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('maximum message limit', response.data['error']) mock_send_message.assert_not_called() def test_send_message_provider_override_without_permission(self): """Test provider/model override without permission.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # User doesn't have can_regenerate_ai privilege self.user.profile.can_regenerate_ai = False self.user.profile.save() url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': 'Hello', 'provider': 'openai', 'model': 'gpt-4o' }) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) @patch('api.views.send_conversation_message') def test_send_message_provider_override_with_permission(self, mock_send_message): """Test provider/model override with permission.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # User has can_regenerate_ai privilege self.user.profile.can_regenerate_ai = True self.user.profile.save() def mock_send_side_effect(*args, **kwargs): # Create user message Message.objects.create( conversation=conversation, role='user', content=kwargs['user_message'] ) # Return assistant message return Message.objects.create( conversation=conversation, role='assistant', content='AI response', provider='openai', model='gpt-4o' ) mock_send_message.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': 'Hello', 'provider': 'openai', 'model': 'gpt-4o' }) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_send_message.assert_called_once() # Check that provider and model were passed call_kwargs = mock_send_message.call_args[1] if mock_send_message.call_args else {} if call_kwargs: self.assertEqual(call_kwargs.get('provider'), 'openai') self.assertEqual(call_kwargs.get('model'), 'gpt-4o') @patch('api.views.send_conversation_message') def test_send_message_ai_failure(self, mock_send_message): """Test handling AI service failure.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) mock_send_message.side_effect = Exception("AI service error") url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = self.client.post(url, { 'message': 'Hello', 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('error', response.data) @patch('api.views.send_conversation_message') def test_retry_message_success(self, mock_send_message): """Test retrying a message successfully.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) user_msg = Message.objects.create( conversation=conversation, role='user', content='Hello' ) # Mock creates new assistant message def mock_send_side_effect(*args, **kwargs): return Message.objects.create( conversation=conversation, role='assistant', content='Retry response', provider='groq', model='llama-3.3-70b-versatile' ) mock_send_message.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/retry/' response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_send_message.assert_called_once() # Should resend the last user message call_kwargs = mock_send_message.call_args[1] if mock_send_message.call_args else {} if call_kwargs: self.assertEqual(call_kwargs.get('user_message'), 'Hello') def test_retry_message_no_user_message(self): """Test retry when no user message exists.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Only assistant message, no user message Message.objects.create( conversation=conversation, role='assistant', content='Only assistant' ) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/retry/' response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) @patch('api.views.send_conversation_message') def test_retry_message_preserves_language(self, mock_send_message): """Test retry preserves language from last assistant message.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) Message.objects.create( conversation=conversation, role='user', content='Hello' ) Message.objects.create( conversation=conversation, role='assistant', content='Response', provider='groq', model='llama-3.3-70b-versatile', meta={'language': 'en'} ) def mock_send_side_effect(*args, **kwargs): return Message.objects.create( conversation=conversation, role='assistant', content='Retry' ) mock_send_message.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/retry/' response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # Should use language from last assistant message call_kwargs = mock_send_message.call_args[1] if mock_send_message.call_args else {} if call_kwargs: self.assertEqual(call_kwargs.get('language'), 'en') def test_unauthorized_access(self): """Test unauthorized access to endpoints.""" self.client.logout() url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/' response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) class LiuYaoConversationEdgeCaseTests(TestCase): """Test edge cases and boundary conditions for LiuYao conversations.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) def test_message_exactly_2000_chars(self): """Test message with exactly 2000 characters.""" from api.serializers import SendMessageSerializer message = 'a' * 2000 serializer = SendMessageSerializer(data={ 'message': message, 'language': 'en' }) self.assertTrue(serializer.is_valid()) def test_message_2001_chars(self): """Test message with 2001 characters (over limit).""" from api.serializers import SendMessageSerializer message = 'a' * 2001 serializer = SendMessageSerializer(data={ 'message': message, 'language': 'en' }) self.assertFalse(serializer.is_valid()) @override_settings(CONVERSATION_MAX_MESSAGES=25) def test_exactly_25_messages(self): """Test conversation with exactly 25 messages.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Create 25 messages for i in range(25): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}' ) self.assertEqual(conversation.get_message_count(), 25) @override_settings(CONVERSATION_MAX_MESSAGES=25) def test_26_messages_blocks_new(self): """Test that 26th message is blocked.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Create 25 messages for i in range(25): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}' ) # Try to send 26th message via API client = APIClient() client.force_authenticate(user=self.user) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = client.post(url, { 'message': '26th message', 'language': 'en' }) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) @patch('api.views.send_conversation_message') def test_chinese_language(self, mock_send_message): """Test conversation with Chinese language.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) def mock_send_side_effect(*args, **kwargs): # Create user message Message.objects.create( conversation=conversation, role='user', content=kwargs['user_message'] ) # Return assistant message return Message.objects.create( conversation=conversation, role='assistant', content='中文回复' ) mock_send_message.side_effect = mock_send_side_effect client = APIClient() client.force_authenticate(user=self.user) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{conversation.id}/send/' response = client.post(url, { 'message': '你好', 'language': 'zh-hans' }) self.assertEqual(response.status_code, status.HTTP_200_OK) call_kwargs = mock_send_message.call_args[1] if mock_send_message.call_args else {} if call_kwargs: self.assertEqual(call_kwargs.get('language'), 'zh-hans') def test_unicode_message_content(self): """Test message with unicode content.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) unicode_message = '你好 🌟 测试 émojis' msg = Message.objects.create( conversation=conversation, role='user', content=unicode_message ) self.assertEqual(msg.content, unicode_message) def test_multiple_conversations_same_liuyao(self): """Test multiple conversations for same LiuYao.""" conversation1 = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) conversation2 = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) self.assertNotEqual(conversation1.id, conversation2.id) self.assertEqual(conversation1.subject, conversation2.subject) class LiuYaoConversationOutlierTests(TestCase): """Test outlier cases and error scenarios for LiuYao conversations.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) def test_invalid_liuyao_id(self): """Test API with invalid LiuYao ID.""" client = APIClient() client.force_authenticate(user=self.user) url = '/api/liuyao/99999/conversations/' response = client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_invalid_conversation_id(self): """Test API with invalid conversation ID.""" client = APIClient() client.force_authenticate(user=self.user) url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/99999/' response = client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) @patch('ai.utils.conversation.generate_conversation_summary') @patch('ai.utils.conversation.prepare_liuyao_prompt') @override_settings(CONVERSATION_CONTEXT_THRESHOLD=100) def test_summary_generation_failure(self, mock_prepare_liuyao, mock_generate_summary): """Test handling summary generation failure.""" mock_prepare_liuyao.return_value = "Large LiuYao data " * 1000 mock_generate_summary.side_effect = Exception("Summary generation failed") conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Create many messages for i in range(30): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Long message ' * 10 ) # Should not raise exception, just log error try: manage_conversation_context(conversation, language='en') except Exception: self.fail("manage_conversation_context should handle summary failures gracefully") # Summary should not be set conversation.refresh_from_db() self.assertFalse(conversation.context_summary) @patch('ai.utils.conversation.LLMServiceFactory') @patch('ai.utils.conversation.get_ai_config') @patch('ai.utils.conversation.manage_conversation_context') @patch('ai.utils.conversation.prepare_conversation_prompt') def test_prompt_preparation_failure(self, mock_prepare_prompt, mock_manage_context, mock_get_config, mock_factory): """Test handling prompt preparation failure.""" mock_get_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} mock_prepare_prompt.side_effect = Exception("Prompt preparation failed") mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Should fallback to limited history with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_liuyao_prompt: mock_liuyao_prompt.return_value = "LiuYao data" # Should not raise, but fallback try: send_conversation_message( conversation=conversation, user_message="Test", language='en' ) except Exception as e: # Should fallback gracefully pass def test_conversation_with_deleted_liuyao(self): """Test conversation behavior when LiuYao is deleted.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) conversation_id = conversation.id # Delete LiuYao (should not cascade delete conversation due to SET_NULL) self.liuyao_obj.delete() # Conversation should still exist self.assertTrue(Conversation.objects.filter(id=conversation_id).exists()) def test_message_with_deleted_conversation(self): """Test message behavior when conversation is deleted.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) msg = Message.objects.create( conversation=conversation, role='user', content='Test' ) msg_id = msg.id # Delete conversation (should cascade delete messages) conversation.delete() # Message should be deleted self.assertFalse(Message.objects.filter(id=msg_id).exists()) def test_conversation_with_deleted_user(self): """Test conversation behavior when user is deleted.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) conversation_id = conversation.id # Delete user (should cascade delete conversation) self.user.delete() # Conversation should be deleted self.assertFalse(Conversation.objects.filter(id=conversation_id).exists()) @patch('ai.utils.conversation.LLMServiceFactory') @patch('ai.utils.conversation.get_ai_config') @patch('ai.utils.conversation.manage_conversation_context') @patch('ai.utils.conversation.prepare_conversation_prompt') def test_llm_service_failure(self, mock_prepare_prompt, mock_manage_context, mock_get_config, mock_factory): """Test handling LLM service failure.""" mock_get_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} mock_prepare_prompt.return_value = "Prompt" mock_service = MagicMock() mock_service.get_completion.side_effect = Exception("LLM service error") mock_factory.get_service.return_value = mock_service conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) with self.assertRaises(Exception): send_conversation_message( conversation=conversation, user_message="Test", language='en' ) def test_concurrent_message_sending(self): """Test handling concurrent message sending (simulated).""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) # Simulate concurrent sends by checking message count before and after initial_count = conversation.get_message_count() # Create messages rapidly for i in range(5): Message.objects.create( conversation=conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}' ) final_count = conversation.get_message_count() self.assertEqual(final_count, initial_count + 5) def test_very_long_liuyao_prompt(self): """Test handling very long LiuYao prompt.""" conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_liuyao: # Simulate very long LiuYao prompt mock_liuyao.return_value = "LiuYao data " * 10000 # Very long size = calculate_context_size( divination_prompt=mock_liuyao.return_value, conversation_history=[], current_message="Test" ) self.assertGreater(size, 0) self.assertIsInstance(size, int) def test_liuyao_without_data_field(self): """Test LiuYao conversation with LiuYao that has no data field.""" liuyao_no_data = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Test question', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0', data=None # No data field ) conversation = create_liuyao_conversation( user=self.user, liuyao=liuyao_no_data ) # Should still work self.assertIsNotNone(conversation.id) subject_obj = conversation.get_subject_object() self.assertEqual(subject_obj.id, liuyao_no_data.id) def test_liuyao_with_empty_question(self): """Test LiuYao conversation with empty question.""" liuyao_empty = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='', # Empty question y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) conversation = create_liuyao_conversation( user=self.user, liuyao=liuyao_empty ) # Should still work self.assertIsNotNone(conversation.id) # Title should handle empty question gracefully self.assertIsNotNone(conversation.title) class LiuYaoConversationFailedMessageTests(TestCase): """Test failed message handling for LiuYao conversations.""" def setUp(self): """Set up test data.""" self.user = User.objects.create_user( phone='1234567890', email='test@example.com', password='testpass123' ) UserProfile.objects.get_or_create(user=self.user) self.liuyao_obj = liuyao.objects.create( qdate=timezone.now(), user=self.user, question='Will I get the job?', y1='1', y2='0', y3='1', y4='0', y5='1', y6='0' ) self.conversation = create_liuyao_conversation( user=self.user, liuyao=self.liuyao_obj ) def test_failed_message_creation_and_persistence(self): """Test that failed messages are created and persisted correctly.""" # Create a failed message directly failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='AI service error' ) # Verify message is saved correctly self.assertEqual(failed_msg.status, 'failed') self.assertEqual(failed_msg.error_message, 'AI service error') self.assertEqual(failed_msg.content, 'Test message') # Verify message persists in database saved_msg = Message.objects.get(id=failed_msg.id) self.assertEqual(saved_msg.status, 'failed') self.assertEqual(saved_msg.error_message, 'AI service error') def test_failed_message_not_counted_in_limit(self): """Test that failed messages don't count toward message limit.""" # Create 10 successful messages for i in range(10): Message.objects.create( conversation=self.conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Message {i}', status='sent' ) # Create 5 failed messages for i in range(5): Message.objects.create( conversation=self.conversation, role='user', content=f'Failed message {i}', status='failed', error_message='Error' ) # Message count should only include sent messages (10) self.assertEqual(self.conversation.get_message_count(), 10) def test_send_message_failure_sets_status(self): """Test that when send_conversation_message fails, user message is marked as failed.""" from ai.utils.conversation import send_conversation_message # Mock LLM service to raise an exception with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.side_effect = Exception("Model decommissioned") mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" # Should raise exception with self.assertRaises(Exception): send_conversation_message( conversation=self.conversation, user_message='Test message', language='en' ) # Check that user message was created and marked as failed user_msg = Message.objects.filter( conversation=self.conversation, role='user', content='Test message' ).first() self.assertIsNotNone(user_msg) self.assertEqual(user_msg.status, 'failed') self.assertIsNotNone(user_msg.error_message) self.assertIn('Model decommissioned', user_msg.error_message) def test_retry_failed_message_reuses_existing(self): """Test that retrying a failed message reuses the existing message.""" # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Original message', status='failed', error_message='Previous error' ) original_id = failed_msg.id # Mock successful retry with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" assistant_msg = send_conversation_message( conversation=self.conversation, user_message='Original message', language='en', existing_user_message=failed_msg ) # Verify the same message was reused (not a new one created) failed_msg.refresh_from_db() self.assertEqual(failed_msg.id, original_id) self.assertEqual(failed_msg.status, 'sent') self.assertIsNone(failed_msg.error_message) # Verify no duplicate messages were created user_messages = Message.objects.filter( conversation=self.conversation, role='user', content='Original message' ) self.assertEqual(user_messages.count(), 1) def test_retry_failed_message_with_edit(self): """Test that retrying allows editing the message content.""" # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Original message', status='failed', error_message='Previous error' ) # Mock successful retry with edited content with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" assistant_msg = send_conversation_message( conversation=self.conversation, user_message='Edited message', language='en', existing_user_message=failed_msg ) # Verify message content was updated failed_msg.refresh_from_db() self.assertEqual(failed_msg.content, 'Edited message') self.assertEqual(failed_msg.status, 'sent') self.assertIsNone(failed_msg.error_message) def test_retry_status_transition(self): """Test that retry correctly transitions status: failed -> pending -> sent.""" # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='Error' ) # Mock successful retry with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" # Check status is set to pending before AI call # (This happens inside send_conversation_message) assistant_msg = send_conversation_message( conversation=self.conversation, user_message='Test message', language='en', existing_user_message=failed_msg ) # Verify final status is 'sent' failed_msg.refresh_from_db() self.assertEqual(failed_msg.status, 'sent') def test_retry_failed_message_api_endpoint(self): """Test retry endpoint reuses existing failed message.""" client = APIClient() client.force_authenticate(user=self.user) # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='Previous error' ) original_id = failed_msg.id # Mock successful retry with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): # Verify existing_user_message is passed self.assertIsNotNone(kwargs.get('existing_user_message')) self.assertEqual(kwargs['existing_user_message'].id, original_id) # Update the existing message existing_msg = kwargs['existing_user_message'] existing_msg.status = 'sent' existing_msg.error_message = None existing_msg.save() # Return assistant message return Message.objects.create( conversation=self.conversation, role='assistant', content='AI response', provider='groq', model='llama-3.3-70b-versatile' ) mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/retry/' response = client.post(url, { 'message': 'Test message' }) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('user_message', response.data) self.assertEqual(response.data['user_message']['id'], original_id) self.assertEqual(response.data['user_message']['status'], 'sent') # Verify no duplicate was created user_messages = Message.objects.filter( conversation=self.conversation, role='user', content='Test message' ) self.assertEqual(user_messages.count(), 1) def test_retry_failed_message_with_edit_api(self): """Test retry endpoint allows editing message content.""" client = APIClient() client.force_authenticate(user=self.user) # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Original message', status='failed', error_message='Previous error' ) # Mock successful retry with edited content with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): existing_msg = kwargs['existing_user_message'] # Verify content can be updated existing_msg.content = kwargs['user_message'] existing_msg.status = 'sent' existing_msg.error_message = None existing_msg.save() return Message.objects.create( conversation=self.conversation, role='assistant', content='AI response', provider='groq', model='llama-3.3-70b-versatile' ) mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/retry/' response = client.post(url, { 'message': 'Edited message', 'message_id': failed_msg.id }) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['user_message']['content'], 'Edited message') # Verify message was updated, not duplicated failed_msg.refresh_from_db() self.assertEqual(failed_msg.content, 'Edited message') self.assertEqual(failed_msg.status, 'sent') def test_retry_failed_message_api_failure(self): """Test retry endpoint handles failures correctly.""" client = APIClient() client.force_authenticate(user=self.user) # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='Previous error' ) # Mock retry failure with patch('api.views.send_conversation_message') as mock_send: mock_send.side_effect = Exception("Retry failed") url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/retry/' response = client.post(url, { 'message': 'Test message' }) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('error', response.data) self.assertIn('user_message', response.data) # Verify failed message is returned self.assertEqual(response.data['user_message']['id'], failed_msg.id) def test_message_count_excludes_failed_and_pending(self): """Test that message count excludes failed and pending messages.""" # Create sent messages for i in range(5): Message.objects.create( conversation=self.conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Sent message {i}', status='sent' ) # Create failed messages for i in range(3): Message.objects.create( conversation=self.conversation, role='user', content=f'Failed message {i}', status='failed', error_message='Error' ) # Create pending messages for i in range(2): Message.objects.create( conversation=self.conversation, role='user', content=f'Pending message {i}', status='pending' ) # Message count should only include sent messages (5) self.assertEqual(self.conversation.get_message_count(), 5) def test_message_count_backward_compatibility(self): """Test that messages without status (None) are counted for backward compatibility.""" # Create messages without status (old messages) for i in range(3): Message.objects.create( conversation=self.conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Old message {i}' # No status field set (None) - status will be None ) # Create sent messages for i in range(2): Message.objects.create( conversation=self.conversation, role='user' if i % 2 == 0 else 'assistant', content=f'Sent message {i}', status='sent' ) # Message count should include both (5 total) self.assertEqual(self.conversation.get_message_count(), 5) def test_send_message_api_returns_failed_message_on_error(self): """Test that send message API returns failed message with ID in error response.""" client = APIClient() client.force_authenticate(user=self.user) # Mock send_conversation_message to raise exception with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): # Create user message first (as real function does) user_msg = Message.objects.create( conversation=self.conversation, role='user', content=kwargs['user_message'], status='pending' ) # Mark as failed (as real function does in exception handler) user_msg.status = 'failed' user_msg.error_message = 'AI service error' user_msg.save() # Then raise exception raise Exception("AI service error") mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/send/' response = client.post(url, { 'message': 'Test message', 'language': 'en' }) # Should return error with user_message containing ID self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('user_message', response.data) self.assertIn('id', response.data['user_message']) self.assertEqual(response.data['user_message']['status'], 'failed') self.assertIn('error_message', response.data['user_message']) self.assertEqual(response.data['user_message']['content'], 'Test message') self.assertIsNone(response.data['assistant_message']) self.assertIn('error', response.data) self.assertIn('detail', response.data) # Verify message ID can be used for retry message_id = response.data['user_message']['id'] self.assertIsNotNone(message_id) self.assertIsInstance(message_id, int) def test_multiple_failed_messages_retry_latest(self): """Test that retry endpoint retries the latest failed message.""" client = APIClient() client.force_authenticate(user=self.user) # Create multiple failed messages failed_msg1 = Message.objects.create( conversation=self.conversation, role='user', content='First failed', status='failed', error_message='Error 1', created_at=timezone.now() - timezone.timedelta(minutes=10) ) failed_msg2 = Message.objects.create( conversation=self.conversation, role='user', content='Second failed', status='failed', error_message='Error 2', created_at=timezone.now() - timezone.timedelta(minutes=5) ) # Mock successful retry with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): existing_msg = kwargs['existing_user_message'] existing_msg.status = 'sent' existing_msg.save() return Message.objects.create( conversation=self.conversation, role='assistant', content='AI response' ) mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/retry/' response = client.post(url) # Should retry the latest failed message (failed_msg2) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['user_message']['id'], failed_msg2.id) def test_retry_specific_message_by_id(self): """Test retry endpoint can retry a specific message by ID.""" client = APIClient() client.force_authenticate(user=self.user) # Create multiple failed messages failed_msg1 = Message.objects.create( conversation=self.conversation, role='user', content='First failed', status='failed', error_message='Error 1' ) failed_msg2 = Message.objects.create( conversation=self.conversation, role='user', content='Second failed', status='failed', error_message='Error 2' ) # Mock successful retry with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): existing_msg = kwargs['existing_user_message'] existing_msg.status = 'sent' existing_msg.save() return Message.objects.create( conversation=self.conversation, role='assistant', content='AI response' ) mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/retry/' response = client.post(url, { 'message_id': failed_msg1.id }) # Should retry the specified message (failed_msg1) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['user_message']['id'], failed_msg1.id) def test_pending_message_status_on_send(self): """Test that new messages start with pending status.""" from ai.utils.conversation import send_conversation_message # Mock successful send with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" # Note: The function creates message with pending status first, # then updates to sent on success. We can't easily test the pending # state in between, but we can verify the final state is sent. assistant_msg = send_conversation_message( conversation=self.conversation, user_message='Test message', language='en' ) # Verify user message was created and is now sent user_msg = Message.objects.filter( conversation=self.conversation, role='user', content='Test message' ).first() self.assertIsNotNone(user_msg) self.assertEqual(user_msg.status, 'sent') def test_failed_message_persistence_across_reloads(self): """Test that failed messages persist and are visible after reload.""" # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='AI service error' ) # Simulate reload by fetching conversation again conversation_id = self.conversation.id reloaded_conversation = Conversation.objects.get(id=conversation_id) # Verify failed message is still there messages = reloaded_conversation.messages.all() failed_messages = [m for m in messages if m.status == 'failed'] self.assertEqual(len(failed_messages), 1) self.assertEqual(failed_messages[0].id, failed_msg.id) self.assertEqual(failed_messages[0].error_message, 'AI service error') def test_retry_clears_error_message(self): """Test that successful retry clears error_message.""" # Create a failed message failed_msg = Message.objects.create( conversation=self.conversation, role='user', content='Test message', status='failed', error_message='Previous error' ) # Mock successful retry with patch('ai.utils.conversation.LLMServiceFactory') as mock_factory: mock_service = MagicMock() mock_service.get_completion.return_value = "AI response" mock_factory.get_service.return_value = mock_service with patch('ai.utils.conversation.get_ai_config') as mock_config: mock_config.return_value = {'provider': 'groq', 'model': 'llama-3.3-70b-versatile'} with patch('ai.utils.conversation.prepare_liuyao_prompt') as mock_prompt: mock_prompt.return_value = "LiuYao data" assistant_msg = send_conversation_message( conversation=self.conversation, user_message='Test message', language='en', existing_user_message=failed_msg ) # Verify error_message is cleared failed_msg.refresh_from_db() self.assertIsNone(failed_msg.error_message) self.assertEqual(failed_msg.status, 'sent') def test_send_message_api_returns_latest_failed_message_on_error(self): """Test that send message API returns the latest failed message when multiple exist.""" client = APIClient() client.force_authenticate(user=self.user) # Create multiple failed messages failed_msg1 = Message.objects.create( conversation=self.conversation, role='user', content='First failed message', status='failed', error_message='Error 1', created_at=timezone.now() - timezone.timedelta(minutes=10) ) failed_msg2 = Message.objects.create( conversation=self.conversation, role='user', content='Second failed message', status='failed', error_message='Error 2', created_at=timezone.now() - timezone.timedelta(minutes=5) ) # Mock send_conversation_message to raise exception with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): # Create new user message user_msg = Message.objects.create( conversation=self.conversation, role='user', content=kwargs['user_message'], status='pending' ) # Mark as failed user_msg.status = 'failed' user_msg.error_message = 'Latest error' user_msg.save() raise Exception("Latest AI service error") mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/send/' response = client.post(url, { 'message': 'Latest test message', 'language': 'en' }) # Should return the latest failed message (not the older ones) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('user_message', response.data) self.assertIn('id', response.data['user_message']) self.assertEqual(response.data['user_message']['content'], 'Latest test message') self.assertEqual(response.data['user_message']['status'], 'failed') self.assertEqual(response.data['user_message']['error_message'], 'Latest error') # Verify it's not one of the older failed messages returned_id = response.data['user_message']['id'] self.assertNotEqual(returned_id, failed_msg1.id) self.assertNotEqual(returned_id, failed_msg2.id) def test_send_message_api_error_when_no_failed_message_found(self): """Test API error response when no failed message is found (edge case).""" client = APIClient() client.force_authenticate(user=self.user) # Mock send_conversation_message to raise exception without creating a message with patch('api.views.send_conversation_message') as mock_send: # Simulate exception before message creation (unlikely but possible) mock_send.side_effect = Exception("Error before message creation") url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/send/' response = client.post(url, { 'message': 'Test message', 'language': 'en' }) # Should return error without user_message (fallback case) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('error', response.data) self.assertIn('detail', response.data) # In this edge case, user_message might not be present # (This tests the else branch in the API endpoint) def test_send_message_api_failed_message_id_matches_database(self): """Test that returned failed message ID matches the actual database record.""" client = APIClient() client.force_authenticate(user=self.user) # Mock send_conversation_message to raise exception with patch('api.views.send_conversation_message') as mock_send: def mock_send_side_effect(*args, **kwargs): # Create user message first (as real function does) user_msg = Message.objects.create( conversation=self.conversation, role='user', content=kwargs['user_message'], status='pending' ) # Mark as failed (as real function does in exception handler) user_msg.status = 'failed' user_msg.error_message = 'AI service error' user_msg.save() # Store the ID for verification mock_send_side_effect.created_message_id = user_msg.id # Then raise exception raise Exception("AI service error") mock_send.side_effect = mock_send_side_effect url = f'/api/liuyao/{self.liuyao_obj.id}/conversations/{self.conversation.id}/send/' response = client.post(url, { 'message': 'Test message for ID verification', 'language': 'en' }) # Verify response contains user_message with ID self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('user_message', response.data) returned_message_id = response.data['user_message']['id'] # Verify the message exists in database with matching ID and status db_message = Message.objects.get(id=returned_message_id) self.assertEqual(db_message.id, returned_message_id) self.assertEqual(db_message.status, 'failed') self.assertEqual(db_message.content, 'Test message for ID verification') self.assertEqual(db_message.error_message, 'AI service error') self.assertEqual(db_message.conversation.id, self.conversation.id)