from django.test import TestCase, Client from django.urls import reverse from django.contrib.auth import get_user_model from liuyao.models import liuyao from accounts.models import UserProfile from unittest.mock import patch, MagicMock from rest_framework import status from django.utils import timezone from datetime import date, time, datetime, timedelta import json User = get_user_model() class LiuYaoAnalysisTests(TestCase): """Test suite for the LiuYao analysis endpoints.""" def setUp(self): """Set up test data.""" # Create test users self.regular_user = User.objects.create_user( phone='1234567890', email='regular@example.com', password='testpassword123' ) self.regen_user = User.objects.create_user( phone='2345678901', email='regen@example.com', password='testpassword123' ) self.staff_user = User.objects.create_user( phone='3456789012', email='staff@example.com', password='testpassword123', is_staff=True ) # Update user profiles (automatically created by signals) self.regular_user.profile.can_regenerate_ai = False self.regular_user.profile.save() self.regen_user.profile.can_regenerate_ai = True self.regen_user.profile.save() # Create test LiuYao records self.liuyao_regular = liuyao.objects.create( question='测试问题 - 普通用户', qdate=timezone.now(), y1='111', # 静爻 y2='0', # 动爻 y3='111', # 静爻 y4='1', # 动爻 y5='111', # 静爻 y6='111', # 静爻 user=self.regular_user, reading='用户自己的解读' ) self.liuyao_regen = liuyao.objects.create( question='测试问题 - 可重新生成用户', qdate=timezone.now(), y1='000', # 静爻 y2='111', # 静爻 y3='1', # 动爻 y4='000', # 静爻 y5='0', # 动爻 y6='111', # 静爻 user=self.regen_user, reading='' ) self.liuyao_staff = liuyao.objects.create( question='测试问题 - 管理员', qdate=timezone.now(), y1='1', # 动爻 y2='1', # 动爻 y3='0', # 动爻 y4='0', # 动爻 y5='111', # 静爻 y6='000', # 静爻 user=self.staff_user, reading='管理员的解读' ) self.liuyao_public = liuyao.objects.create( question='公开测试问题', qdate=timezone.now(), y1='111', y2='111', y3='111', y4='111', y5='111', y6='111', user=None, # Public record reading='公开记录' ) # Set up the client self.client = Client() # Mock analysis data self.mock_analysis_data = { "liuyao_analysis": "这是一个测试分析结果。根据您的六爻卦象，您询问的事情将会有积极的发展。", "think": "思考过程：分析卦象的变化和六神的作用...", "provider": "test_provider", "model": "test_model", "prompt": "测试提示词内容", "analysis": "详细的六爻分析内容" } # Add analysis to one LiuYao record self.liuyao_regular.ai_analysis = self.mock_analysis_data self.liuyao_regular.analysis_status = 'completed' self.liuyao_regular.analysis_timestamp = timezone.now() self.liuyao_regular.save() # URL templates for testing self.analysis_url_template = '/api/liuyao/{}/analysis/' # Updated to use LiuyaoViewSet endpoints with the correct URL structure: # GET /api/liuyao/liuyao/{id}/ -> LiuyaoViewSet.retrieve() self.detail_url_template = '/api/liuyao/liuyao/{}/' # DELETE /api/liuyao/liuyao/{id}/ -> LiuyaoViewSet.destroy() self.delete_url_template = '/api/liuyao/liuyao/{}/' # ----- Normal Parameter Tests ----- def test_get_existing_analysis(self): """Test GET request to retrieve existing LiuYao analysis.""" url = self.analysis_url_template.format(self.liuyao_regular.id) self.client.login(phone='1234567890', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], 'completed') self.assertEqual(response.data['analysis'], self.mock_analysis_data) self.assertIn('timestamp', response.data) def test_get_nonexistent_analysis(self): """Test GET request for LiuYao without analysis.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], 'not_found') self.assertIn('message', response.data) @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_post_generate_analysis(self, mock_analyze): """Test POST request to generate a new LiuYao analysis.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], 'completed') self.assertEqual(response.data['analysis'], self.mock_analysis_data) mock_analyze.assert_called_once() @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_post_with_provider_model_params(self, mock_analyze): """Test POST request with provider and model parameters.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data # Use JSON content type response = self.client.post( url, data={'provider': 'groq', 'model': 'llama-3.1-70b-versatile'}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) # Check that analyze_liuyao was called with the correct parameters mock_analyze.assert_called_once() args, kwargs = mock_analyze.call_args self.assertEqual(args[0], self.liuyao_regen) # First arg should be liuyao object # Check if model_key and provider are passed correctly self.assertIn('model_key', kwargs) self.assertIn('provider', kwargs) self.assertEqual(kwargs['model_key'], 'llama-3.1-70b-versatile') self.assertEqual(kwargs['provider'], 'groq') def test_get_liuyao_detail(self): """Test GET request to retrieve LiuYao details.""" url = self.detail_url_template.format(self.liuyao_regular.id) self.client.login(phone='1234567890', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['question'], '测试问题 - 普通用户') self.assertEqual(response.data['y1'], '111') self.assertEqual(response.data['y2'], '0') self.assertIn('qdate', response.data) self.assertIn('reading', response.data) def test_delete_liuyao_record(self): """Test DELETE request to remove LiuYao record.""" url = self.delete_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) # Verify the record is actually deleted with self.assertRaises(liuyao.DoesNotExist): liuyao.objects.get(id=self.liuyao_regen.id) # ----- Error Parameter Tests ----- def test_nonexistent_liuyao(self): """Test request for a non-existent LiuYao record.""" url = self.analysis_url_template.format(9999) # Assuming ID 9999 doesn't exist self.client.login(phone='1234567890', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) self.assertEqual(response.data['error'], 'LiuYao divination not found') @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_analysis_failure(self, mock_analyze): """Test handling of analysis generation failure.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to raise an exception mock_analyze.side_effect = Exception("LiuYao analysis generation failed") response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('error', response.data) self.assertIn('detail', response.data) @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_analysis_returns_none(self, mock_analyze): """Test handling when analysis function returns None.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to return None mock_analyze.return_value = None response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) # Check for the actual response format - it returns 'status' and 'message' instead of 'error' self.assertIn('status', response.data) self.assertEqual(response.data['status'], 'error') self.assertIn('message', response.data) # ----- Permission and Authentication Tests ----- @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_regenerate_without_authentication(self, mock_analyze): """Test regeneration attempt without authentication.""" url = self.analysis_url_template.format(self.liuyao_regular.id) # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data response = self.client.post( url, data={'regenerate': True}, content_type='application/json' ) # Note: The endpoint has AllowAny permission, so unauthenticated users can access it # But they shouldn't be able to regenerate existing analysis # Check the actual response based on the implementation self.assertIn(response.status_code, [status.HTTP_403_FORBIDDEN, status.HTTP_200_OK]) mock_analyze.assert_not_called() @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_regenerate_without_permission(self, mock_analyze): """Test regeneration attempt by user without regeneration permission.""" url = self.analysis_url_template.format(self.liuyao_regular.id) self.client.login(phone='1234567890', password='testpassword123') # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data response = self.client.post( url, data={'regenerate': True}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('error', response.data) mock_analyze.assert_not_called() @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_regenerate_with_permission(self, mock_analyze): """Test regeneration by user with permission.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data response = self.client.post( url, data={'regenerate': True}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_regenerate_by_staff(self, mock_analyze): """Test regeneration by staff user (should override permission check).""" url = self.analysis_url_template.format(self.liuyao_staff.id) self.client.login(phone='3456789012', password='testpassword123') # Configure the mock to return our mock analysis data mock_analyze.return_value = self.mock_analysis_data response = self.client.post( url, data={'regenerate': True}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() def test_user_accessing_other_users_liuyao(self): """Test user attempting to access another user's LiuYao record.""" url = self.detail_url_template.format(self.liuyao_staff.id) self.client.login(phone='1234567890', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_user_deleting_other_users_liuyao(self): """Test user attempting to delete another user's LiuYao record.""" url = self.delete_url_template.format(self.liuyao_staff.id) self.client.login(phone='1234567890', password='testpassword123') response = self.client.delete(url) # Should return PermissionDenied as 403 or 404 depending on implementation self.assertIn(response.status_code, [status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND]) def test_anonymous_access_to_public_liuyao(self): """Test anonymous access to a public LiuYao record.""" url = self.analysis_url_template.format(self.liuyao_public.id) response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) def test_staff_access_to_any_liuyao(self): """Test that staff can access any LiuYao record.""" # Staff accessing regular user's LiuYao url = self.analysis_url_template.format(self.liuyao_regular.id) self.client.login(phone='3456789012', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # Staff accessing regen user's LiuYao url = self.analysis_url_template.format(self.liuyao_regen.id) response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # ----- Edge Cases ----- def test_boolean_regenerate_param(self): """Test handling of boolean regenerate parameter.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') with patch('ai.utils.liuyao_analysis.analyze_liuyao') as mock_analyze: mock_analyze.return_value = self.mock_analysis_data # Test with boolean true response = self.client.post( url, data={'regenerate': True}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() def test_various_regenerate_string_values(self): """Test handling of different string values for regenerate parameter.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') with patch('ai.utils.liuyao_analysis.analyze_liuyao') as mock_analyze: mock_analyze.return_value = self.mock_analysis_data # Test with string 'true' response = self.client.post( url, data={'regenerate': 'true'}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() mock_analyze.reset_mock() # Test with 'yes' response = self.client.post( url, data={'regenerate': 'yes'}, content_type='application/json' ) # The implementation only checks for 'true', so 'yes' should be treated as False self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() mock_analyze.reset_mock() # Test with '1' response = self.client.post( url, data={'regenerate': '1'}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() mock_analyze.reset_mock() # Test with 't' response = self.client.post( url, data={'regenerate': 't'}, content_type='application/json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_empty_results_handling(self, mock_analyze): """Test handling of empty analysis results.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Configure the mock to return empty results (which is treated as None/falsy) mock_analyze.return_value = {} response = self.client.post(url) # Empty dict is falsy, so it should be treated as an error self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertIn('status', response.data) self.assertEqual(response.data['status'], 'error') def test_liuyao_without_user(self): """Test access to a LiuYao record without a user field.""" url = self.analysis_url_template.format(self.liuyao_public.id) # Try access with different users self.client.login(phone='1234567890', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.client.login(phone='2345678901', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.client.login(phone='3456789012', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # Try with anonymous user self.client.logout() response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_analysis_with_pending_status(self, mock_analyze): """Test handling when a LiuYao record has pending analysis status.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Set the LiuYao to pending status self.liuyao_regen.analysis_status = 'pending' self.liuyao_regen.save() response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], 'pending') @patch('ai.utils.liuyao_analysis.analyze_liuyao') def test_analysis_with_error_status(self, mock_analyze): """Test handling when a LiuYao record has error analysis status.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') # Set the LiuYao to error status self.liuyao_regen.analysis_status = 'error' self.liuyao_regen.save() response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], 'error') def test_malformed_liuyao_data(self): """Test handling of LiuYao records with malformed yao data.""" # Create a LiuYao with invalid yao values malformed_liuyao = liuyao.objects.create( question='测试问题 - 格式错误', qdate=timezone.now(), y1='invalid', # Invalid value y2='999', # Invalid value y3='', # Empty value y4='111', # Valid value y5='0', # Valid value y6='1', # Valid value user=self.regen_user ) url = self.analysis_url_template.format(malformed_liuyao.id) self.client.login(phone='2345678901', password='testpassword123') # Should still return the record but analysis might handle the invalid data response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) def test_liuyao_with_complex_data_field(self): """Test LiuYao records with complex data in the data field.""" # Create a LiuYao with complex data complex_data = { "ly": { "ogua": {"gua": 0, "palace": {"palace": 0, "shi": 3, "yin": 6}}, "cgua": {"gua": 1} }, "bz": { "year": {"g": 0, "e": 0}, "month": {"g": 1, "e": 1}, "day": {"g": 2, "e": 2}, "hour": {"g": 3, "e": 3} } } complex_liuyao = liuyao.objects.create( question='测试问题 - 复杂数据', qdate=timezone.now(), y1='111', y2='0', y3='111', y4='1', y5='111', y6='111', user=self.regen_user, data=complex_data ) url = self.analysis_url_template.format(complex_liuyao.id) self.client.login(phone='2345678901', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # Test analysis generation with complex data with patch('ai.utils.liuyao_analysis.analyze_liuyao') as mock_analyze: mock_analyze.return_value = self.mock_analysis_data response = self.client.post(url) self.assertEqual(response.status_code, status.HTTP_200_OK) mock_analyze.assert_called_once() def test_unicode_question_handling(self): """Test handling of Unicode characters in LiuYao questions.""" unicode_questions = [ "测试中文问题？🔮", "Test with émojis ✨💫", "Mixed 中英文 question", "Special chars: @#$%^&*()", "古代漢字問題" ] for question in unicode_questions: unicode_liuyao = liuyao.objects.create( question=question, qdate=timezone.now(), y1='111', y2='0', y3='111', y4='1', y5='111', y6='111', user=self.regen_user ) url = self.analysis_url_template.format(unicode_liuyao.id) self.client.login(phone='2345678901', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # ----- Performance and Stress Tests ----- def test_concurrent_analysis_requests(self): """Test handling of multiple concurrent analysis requests.""" url = self.analysis_url_template.format(self.liuyao_regen.id) self.client.login(phone='2345678901', password='testpassword123') with patch('ai.utils.liuyao_analysis.analyze_liuyao') as mock_analyze: mock_analyze.return_value = self.mock_analysis_data # Simulate multiple requests (in a real test, you might use threading) responses = [] for i in range(5): response = self.client.post(url) responses.append(response) # All should succeed for response in responses: self.assertEqual(response.status_code, status.HTTP_200_OK) def test_large_reading_text(self): """Test handling of LiuYao records with very large reading text.""" large_text = "这是一个非常长的解读内容。" * 1000 # 30,000 characters large_reading_liuyao = liuyao.objects.create( question='测试问题 - 大量文本', qdate=timezone.now(), y1='111', y2='0', y3='111', y4='1', y5='111', y6='111', user=self.regen_user, reading=large_text ) url = self.analysis_url_template.format(large_reading_liuyao.id) self.client.login(phone='2345678901', password='testpassword123') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK)