""" Comprehensive unit tests for AI analysis reporting Django admin functionality. Tests cover: - PersonAdmin reporting features - LiuYaoAdmin reporting features - Admin actions for report resolution - Admin display methods - Bulk operations and edge cases """ from django.test import TestCase, RequestFactory from django.contrib.admin.sites import AdminSite from django.contrib.auth import get_user_model from django.contrib.admin import ModelAdmin from django.contrib.messages.storage.fallback import FallbackStorage from django.utils import timezone from datetime import date, time from unittest.mock import patch, MagicMock from bazi.models import Person from bazi.admin import PersonAdmin from liuyao.models import liuyao from liuyao.admin import LiuYaoAdmin User = get_user_model() class PersonAdminTests(TestCase): """Test PersonAdmin reporting functionality""" def setUp(self): """Set up test data""" self.factory = RequestFactory() self.admin_site = AdminSite() self.person_admin = PersonAdmin(Person, self.admin_site) self.admin_user = User.objects.create_user( phone='13800138001', email='admin@example.com', first_name='Admin', last_name='User', is_staff=True, is_superuser=True ) self.regular_user = User.objects.create_user( phone='13800138000', email='user@example.com', first_name='Test', last_name='User' ) self.person = Person.objects.create( name='测试人员', birth_date=date(1990, 1, 1), birth_time=time(12, 0), created_by=self.regular_user, analysis_status='completed', ai_analysis={'bazi_analysis': 'Test BaZi analysis'}, number_analysis_status='completed', number_ai_analysis={'number_analysis': 'Test Number analysis'} ) def test_bazi_report_status_display_no_report(self): """Test BaZi report status display when no report exists""" result = self.person_admin.bazi_report_status_display(self.person) self.assertIn('无举报', result) def test_bazi_report_status_display_pending(self): """Test BaZi report status display for pending reports""" self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'pending' result = self.person_admin.bazi_report_status_display(self.person) self.assertIn('待处理', result) self.assertIn('orange', result) def test_bazi_report_status_display_resolved(self): """Test BaZi report status display for resolved reports""" self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'resolved' result = self.person_admin.bazi_report_status_display(self.person) self.assertIn('已解决', result) self.assertIn('green', result) def test_bazi_report_status_display_dismissed(self): """Test BaZi report status display for dismissed reports""" self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'dismissed' result = self.person_admin.bazi_report_status_display(self.person) self.assertIn('已忽略', result) self.assertIn('gray', result) def test_number_report_status_display_all_statuses(self): """Test Number report status display for all statuses""" status_tests = [ ('pending', '待处理', 'orange'), ('reviewed', '已审核', 'blue'), ('resolved', '已解决', 'green'), ('dismissed', '已忽略', 'gray') ] for status, display_text, color in status_tests: with self.subTest(status=status): self.person.number_analysis_reported = True self.person.number_report_status = status result = self.person_admin.number_report_status_display(self.person) self.assertIn(display_text, result) self.assertIn(color, result) def test_resolve_bazi_reports_action_success(self): """Test successful BaZi report resolution action""" # Set up reported analysis self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'pending' self.person.bazi_report_category = 'inappropriate_content' self.person.bazi_report_message = 'Test report' self.person.save() # Create request with messages framework request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) messages = FallbackStorage(request) setattr(request, '_messages', messages) # Create queryset queryset = Person.objects.filter(id=self.person.id) # Mock email sending with patch('api.utils.send_user_resolution_notification') as mock_email: mock_email.return_value = True # Execute admin action self.person_admin.resolve_bazi_reports(request, queryset) # Verify database changes self.person.refresh_from_db() self.assertEqual(self.person.bazi_report_status, 'resolved') self.assertEqual(self.person.bazi_report_resolved_by, self.admin_user) self.assertIsNotNone(self.person.bazi_report_resolved_at) # Verify email was called mock_email.assert_called_once() def test_resolve_number_reports_action_success(self): """Test successful Number report resolution action""" # Set up reported analysis self.person.number_analysis_reported = True self.person.number_report_status = 'pending' self.person.number_report_category = 'inaccurate_analysis' self.person.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id=self.person.id) with patch('api.utils.send_user_resolution_notification') as mock_email: mock_email.return_value = True self.person_admin.resolve_number_reports(request, queryset) self.person.refresh_from_db() self.assertEqual(self.person.number_report_status, 'resolved') self.assertEqual(self.person.number_report_resolved_by, self.admin_user) def test_resolve_bazi_reports_no_pending_reports(self): """Test BaZi report resolution when no pending reports exist""" # Set up already resolved report self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'resolved' self.person.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id=self.person.id) # Should not change anything original_resolved_at = self.person.bazi_report_resolved_at self.person_admin.resolve_bazi_reports(request, queryset) self.person.refresh_from_db() self.assertEqual(self.person.bazi_report_resolved_at, original_resolved_at) def test_resolve_reports_email_failure_handling(self): """Test handling when email sending fails during resolution""" self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'pending' self.person.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id=self.person.id) # Mock email failure with patch('api.utils.send_user_resolution_notification') as mock_email: mock_email.side_effect = Exception("Email service down") # Should not crash self.person_admin.resolve_bazi_reports(request, queryset) # Report should still be marked as resolved self.person.refresh_from_db() self.assertEqual(self.person.bazi_report_status, 'resolved') def test_resolve_reports_no_user_email(self): """Test report resolution when user has no email""" # Remove user email self.regular_user.email = '' self.regular_user.save() self.person.bazi_analysis_reported = True self.person.bazi_report_status = 'pending' self.person.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id=self.person.id) # Should work without crashing self.person_admin.resolve_bazi_reports(request, queryset) self.person.refresh_from_db() self.assertEqual(self.person.bazi_report_status, 'resolved') def test_resolve_reports_bulk_operation(self): """Test bulk report resolution""" # Create multiple reported analyses persons = [] for i in range(3): person = Person.objects.create( name=f'Bulk Test {i}', birth_date=date(1990, 1, 1), created_by=self.regular_user, analysis_status='completed', ai_analysis={'test': f'data {i}'}, bazi_analysis_reported=True, bazi_report_status='pending' ) persons.append(person) request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id__in=[p.id for p in persons]) with patch('api.utils.send_user_resolution_notification'): self.person_admin.resolve_bazi_reports(request, queryset) # All should be resolved for person in persons: person.refresh_from_db() self.assertEqual(person.bazi_report_status, 'resolved') class LiuYaoAdminTests(TestCase): """Test LiuYaoAdmin reporting functionality""" def setUp(self): """Set up test data""" self.factory = RequestFactory() self.admin_site = AdminSite() self.liuyao_admin = LiuYaoAdmin(liuyao, self.admin_site) self.admin_user = User.objects.create_user( phone='13800138001', email='admin@example.com', first_name='Admin', last_name='User', is_staff=True, is_superuser=True ) self.regular_user = User.objects.create_user( phone='13800138000', email='user@example.com', first_name='Test', last_name='User' ) self.liuyao_entry = liuyao.objects.create( question='测试问题', user=self.regular_user, qdate=timezone.now(), y1='1', y2='0', y3='1', y4='0', y5='1', y6='0', analysis_status='completed', ai_analysis={'response': 'Test LiuYao analysis content'} ) def test_user_display(self): """Test user display method in LiuYao admin""" result = self.liuyao_admin.user_display(self.liuyao_entry) self.assertIn('Test User', result) self.assertIn(self.regular_user.phone, result) def test_user_display_no_user(self): """Test user display method when no user""" self.liuyao_entry.user = None result = self.liuyao_admin.user_display(self.liuyao_entry) self.assertIn('Guest User', result) def test_analysis_status_display(self): """Test analysis status display method""" status_tests = [ ('completed', '✓ Completed', 'green'), ('pending', '⟳ Pending', 'orange'), ('error', '✗ Error', 'red'), (None, 'Not Analyzed', None) ] for status, display_text, color in status_tests: with self.subTest(status=status): self.liuyao_entry.analysis_status = status result = self.liuyao_admin.analysis_status_display(self.liuyao_entry) self.assertIn(display_text, result) if color: self.assertIn(color, result) def test_report_status_display_all_statuses(self): """Test report status display for all statuses""" # Test no report result = self.liuyao_admin.report_status_display(self.liuyao_entry) self.assertIn('无举报', result) # Test various statuses status_tests = [ ('pending', '待处理', 'orange'), ('reviewed', '已审核', 'blue'), ('resolved', '已解决', 'green'), ('dismissed', '已忽略', 'gray') ] for status, display_text, color in status_tests: with self.subTest(status=status): self.liuyao_entry.analysis_reported = True self.liuyao_entry.report_status = status result = self.liuyao_admin.report_status_display(self.liuyao_entry) self.assertIn(display_text, result) self.assertIn(color, result) def test_ai_analysis_display(self): """Test AI analysis display method""" result = self.liuyao_admin.ai_analysis_display(self.liuyao_entry) self.assertIn('Test LiuYao analysis content', result) self.assertIn('Unknown', result) # Provider and model unknown def test_ai_analysis_display_no_analysis(self): """Test AI analysis display when no analysis exists""" self.liuyao_entry.ai_analysis = None result = self.liuyao_admin.ai_analysis_display(self.liuyao_entry) self.assertIn('No analysis available', result) def test_ai_analysis_display_truncation(self): """Test AI analysis display truncation for long content""" long_content = 'A' * 600 # Longer than 500 character limit self.liuyao_entry.ai_analysis = {'response': long_content} result = self.liuyao_admin.ai_analysis_display(self.liuyao_entry) self.assertIn('...', result) # Should be truncated # Check that the result contains the truncated content # The HTML markup will make it longer, but the content should be truncated self.assertIn('A' * 500, result) # Should contain first 500 characters self.assertNotIn('A' * 600, result) # Should not contain full content def test_resolve_reports_action_success(self): """Test successful LiuYao report resolution action""" # Set up reported analysis self.liuyao_entry.analysis_reported = True self.liuyao_entry.report_status = 'pending' self.liuyao_entry.report_category = 'inaccurate_analysis' self.liuyao_entry.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = liuyao.objects.filter(id=self.liuyao_entry.id) with patch('api.utils.send_user_resolution_notification') as mock_email: mock_email.return_value = True self.liuyao_admin.resolve_reports(request, queryset) # Verify database changes self.liuyao_entry.refresh_from_db() self.assertEqual(self.liuyao_entry.report_status, 'resolved') self.assertEqual(self.liuyao_entry.report_resolved_by, self.admin_user) self.assertIsNotNone(self.liuyao_entry.report_resolved_at) # Verify email was called mock_email.assert_called_once() def test_resolve_reports_already_resolved(self): """Test resolving already resolved LiuYao reports""" self.liuyao_entry.analysis_reported = True self.liuyao_entry.report_status = 'resolved' self.liuyao_entry.save() request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = liuyao.objects.filter(id=self.liuyao_entry.id) # Should not update already resolved reports original_resolved_at = self.liuyao_entry.report_resolved_at self.liuyao_admin.resolve_reports(request, queryset) self.liuyao_entry.refresh_from_db() self.assertEqual(self.liuyao_entry.report_resolved_at, original_resolved_at) class AdminDisplayMethodTests(TestCase): """Test admin display methods and edge cases""" def setUp(self): """Set up test data""" self.admin_site = AdminSite() self.person_admin = PersonAdmin(Person, self.admin_site) self.liuyao_admin = LiuYaoAdmin(liuyao, self.admin_site) self.user = User.objects.create_user( phone='13800138000', email='user@example.com', first_name='', # Empty first name last_name='' # Empty last name ) def test_created_by_display_no_name(self): """Test created_by display when user has no name""" person = Person.objects.create( name='No Name Test', birth_date=date(1990, 1, 1), created_by=self.user, analysis_status='completed', ai_analysis={'test': 'data'} ) result = self.person_admin.created_by_display(person) self.assertIn('No Name', result) self.assertIn(self.user.phone, result) def test_created_by_display_no_user(self): """Test created_by display when no user""" person = Person.objects.create( name='Orphan Test', birth_date=date(1990, 1, 1), created_by=None, analysis_status='completed', ai_analysis={'test': 'data'} ) result = self.person_admin.created_by_display(person) self.assertIn('Unknown User', result) def test_liuyao_user_display_no_name(self): """Test LiuYao user display when user has no name""" entry = liuyao.objects.create( question='No Name Test', user=self.user, qdate=timezone.now(), y1='1', y2='0', y3='1', y4='0', y5='1', y6='0', analysis_status='completed', ai_analysis={'response': 'Test'} ) result = self.liuyao_admin.user_display(entry) self.assertIn('No Name', result) self.assertIn(self.user.phone, result) def test_ai_analysis_display_malformed_data(self): """Test AI analysis display with malformed data""" person = Person.objects.create( name='Malformed Test', birth_date=date(1990, 1, 1), created_by=self.user, analysis_status='completed', ai_analysis={'invalid': 'structure'} # Missing expected fields ) # Should not crash result = self.person_admin.ai_analysis_display(person) self.assertIsNotNone(result) def test_ai_analysis_display_exception_handling(self): """Test AI analysis display exception handling""" # Create a person with analysis that might cause JSON errors person = Person.objects.create( name='Exception Test', birth_date=date(1990, 1, 1), created_by=self.user, analysis_status='completed', ai_analysis=None # Will cause issues when accessing ) # Mock to force an exception with patch.object(person, 'ai_analysis', side_effect=Exception("Test error")): result = self.person_admin.ai_analysis_display(person) # The display method should handle the exception gracefully self.assertIsNotNone(result) self.assertIn('No structured analysis content available', result) class AdminActionEdgeCaseTests(TestCase): """Test edge cases and error handling for admin actions""" def setUp(self): """Set up test data""" self.factory = RequestFactory() self.admin_site = AdminSite() self.person_admin = PersonAdmin(Person, self.admin_site) self.admin_user = User.objects.create_user( phone='13800138001', email='admin@example.com', is_staff=True, is_superuser=True ) def test_resolve_reports_empty_queryset(self): """Test resolving reports with empty queryset""" request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.none() # Should not crash self.person_admin.resolve_bazi_reports(request, queryset) # No assertions needed - just ensuring it doesn't crash def test_resolve_reports_mixed_statuses(self): """Test resolving reports with mixed statuses""" user = User.objects.create_user( phone='13800138000', email='user@example.com' ) # Create persons with different report statuses person1 = Person.objects.create( name='Pending Report', birth_date=date(1990, 1, 1), created_by=user, analysis_status='completed', ai_analysis={'test': 'data1'}, bazi_analysis_reported=True, bazi_report_status='pending' ) person2 = Person.objects.create( name='Already Resolved', birth_date=date(1990, 1, 1), created_by=user, analysis_status='completed', ai_analysis={'test': 'data2'}, bazi_analysis_reported=True, bazi_report_status='resolved' ) person3 = Person.objects.create( name='Not Reported', birth_date=date(1990, 1, 1), created_by=user, analysis_status='completed', ai_analysis={'test': 'data3'}, bazi_analysis_reported=False ) request = self.factory.post('/') request.user = self.admin_user setattr(request, 'session', {}) setattr(request, '_messages', FallbackStorage(request)) queryset = Person.objects.filter(id__in=[person1.id, person2.id, person3.id]) with patch('api.utils.send_user_resolution_notification'): self.person_admin.resolve_bazi_reports(request, queryset) # Only person1 should be updated person1.refresh_from_db() person2.refresh_from_db() person3.refresh_from_db() self.assertEqual(person1.bazi_report_status, 'resolved') self.assertEqual(person2.bazi_report_status, 'resolved') # Unchanged self.assertFalse(person3.bazi_analysis_reported) # Unchanged def test_admin_fieldsets_coverage(self): """Test that admin fieldsets include all reporting fields""" # Test Person admin fieldsets person_fieldsets = dict(self.person_admin.fieldsets) bazi_fields = person_fieldsets['BaZi Analysis Reports']['fields'] expected_bazi_fields = [ 'bazi_analysis_reported', 'bazi_report_category', 'bazi_report_message', 'bazi_report_timestamp', 'bazi_report_status', 'bazi_report_admin_notes', 'bazi_report_resolved_at' ] for field in expected_bazi_fields: self.assertIn(field, bazi_fields) number_fields = person_fieldsets['Number Analysis Reports']['fields'] expected_number_fields = [ 'number_analysis_reported', 'number_report_category', 'number_report_message', 'number_report_timestamp', 'number_report_status', 'number_report_admin_notes', 'number_report_resolved_at' ] for field in expected_number_fields: self.assertIn(field, number_fields) def test_admin_list_filters_include_reporting(self): """Test that admin list filters include reporting fields""" person_filters = self.person_admin.list_filter expected_filters = [ 'bazi_report_status', 'number_report_status', 'bazi_analysis_reported', 'number_analysis_reported' ] for filter_field in expected_filters: self.assertIn(filter_field, person_filters) def test_admin_search_fields_include_reporting(self): """Test that admin search fields include reporting fields""" person_search_fields = self.person_admin.search_fields expected_search_fields = ['bazi_report_message', 'number_report_message'] for search_field in expected_search_fields: self.assertIn(search_field, person_search_fields)