diff --git a/YuYuWechatV2_Client/client_app/tests.py b/YuYuWechatV2_Client/client_app/tests.py index 1d03c04..dc07caa 100755 --- a/YuYuWechatV2_Client/client_app/tests.py +++ b/YuYuWechatV2_Client/client_app/tests.py @@ -1,11 +1,12 @@ from django.test import TestCase from django.urls import reverse -from .models import WechatUser, Message, ServerConfig, ScheduledMessage +from .models import WechatUser, Message, ServerConfig, ScheduledMessage, Log from django.utils import timezone import json from unittest.mock import patch from django.test import Client from io import BytesIO +import subprocess class WechatUserModelTests(TestCase): def setUp(self): @@ -82,6 +83,28 @@ def test_home_view(self): self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'home.html') + def test_get_server_ip_view(self): + response = self.client.get(reverse('get_server_ip')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"server_ip": "127.0.0.1"}') + + def test_set_server_ip_view(self): + response = self.client.post(reverse('set_server_ip'), json.dumps({'server_ip': '192.168.0.1'}), + content_type='application/json') + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Server IP set to 192.168.0.1"}') + self.assertEqual(ServerConfig.objects.latest('id').server_ip, '192.168.0.1') + + def test_send_message_management_view(self): + response = self.client.get(reverse('send_message_management')) + self.assertEqual(response.status_code, 200) + self.assertTemplateUsed(response, 'send_message_management.html') + + def test_schedule_management_view(self): + response = self.client.get(reverse('schedule_management')) + self.assertEqual(response.status_code, 200) + self.assertTemplateUsed(response, 'schedule_management.html') + @patch('requests.post') def test_send_message_view(self, mock_post): mock_post.return_value.status_code = 200 @@ -92,7 +115,7 @@ def test_send_message_view(self, mock_post): 'text': 'Hello' }) self.assertEqual(response.status_code, 200) - self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Message sent to user1"}') + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Hello sent to user1"}') @patch('requests.post') def test_skip_execution_view(self, mock_post): @@ -108,7 +131,6 @@ def test_skip_execution_view(self, mock_post): scheduled_message.refresh_from_db() self.assertEqual(scheduled_message.execution_skip, 1) - def test_export_import_database_views(self): # 测试导出数据库 response = self.client.post(reverse('export_database')) @@ -127,12 +149,62 @@ def test_export_import_database_views(self): self.assertEqual(response.status_code, 200) self.assertEqual(response.content, b'Database imported successfully.') - def test_start_celery_view(self): + @patch('subprocess.Popen') + def test_start_celery_view(self, mock_popen): response = self.client.post(reverse('start_celery')) self.assertEqual(response.status_code, 200) self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Celery started"}') + self.assertTrue(mock_popen.called) - def test_stop_celery_view(self): + @patch('subprocess.call') + def test_stop_celery_view(self, mock_call): response = self.client.post(reverse('stop_celery')) self.assertEqual(response.status_code, 200) - self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Celery stopped"}') \ No newline at end of file + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Celery stopped"}') + self.assertTrue(mock_call.called) + + @patch('subprocess.run') + def test_check_celery_running_view(self, mock_run): + mock_run.return_value.stdout = b'celery process' + response = self.client.get(reverse('check_celery_running')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Celery is running"}') + + mock_run.return_value.stdout = b'' + response = self.client.get(reverse('check_celery_running')) + self.assertEqual(response.status_code, 404) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "Celery is not running"}') + + @patch('requests.post') + def test_check_wechat_status_view(self, mock_post): + mock_post.return_value.status_code = 200 + + response = self.client.post(reverse('check_wechat_status')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "success", "message": "WeChat status checked successfully"}') + + mock_post.return_value.status_code = 500 + + response = self.client.post(reverse('check_wechat_status')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "failure", "message": "微信不在线"}') + + def test_log_view(self): + response = self.client.get(reverse('log_view')) + self.assertEqual(response.status_code, 200) + self.assertTemplateUsed(response, 'log.html') + + def test_log_counts_view(self): + Log.objects.create(result=True, function_name='test_func', return_data='{}') + Log.objects.create(result=False, function_name='test_func', return_data='{}') + + response = self.client.get(reverse('log_counts')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"total": 2, "success": 1, "failure": 1}') + + def test_clear_logs_view(self): + Log.objects.create(result=True, function_name='test_func', return_data='{}') + response = self.client.post(reverse('clear_logs')) + self.assertEqual(response.status_code, 200) + self.assertJSONEqual(str(response.content, encoding='utf8'), '{"status": "success"}') + self.assertEqual(Log.objects.count(), 0) \ No newline at end of file