import json from unittest.mock import MagicMock, patch from django.test import TestCase from django.utils import timezone from basic_info.models import Employee, Merchant, MerchantTypeEnum from mission.models import Mission, MissionCategory, MissionReply from notifier.models import NotificationEventKeyEnum, Notifier, NotifierChannelEnum, NotifierRoute from notifier.services import ( dispatch_notification_event, enqueue_notification_event, render_notification_content, send_notification_with_notifier, ) from notifier.tasks import notify_unreplied_missions_task class NotifierServiceTestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name="通知商户", type=MerchantTypeEnum.STORE) self.general_category = MissionCategory.objects.create(merchant=self.merchant, name="通用") self.after_sale_category = MissionCategory.objects.create(merchant=self.merchant, name="售后") self.notifier = Notifier.objects.create( merchant=self.merchant, name="任务创建通知", channel=NotifierChannelEnum.WECOM_WEBHOOK, template_key="mission_created", config={"key": "abc123", "msgtype": "markdown"}, ) self.global_route = NotifierRoute.objects.create( merchant=self.merchant, notifier=self.notifier, event_key=NotificationEventKeyEnum.MISSION_CREATED, ) self.creator = Employee.objects.create(merchant=self.merchant, name="任务创建者") def test_render_notification_content(self): content = render_notification_content( notifier=self.notifier, payload={ "mission_id": 12, "created_by_name": "张三", "creator_name": "张三", "category_name": "通用", "is_urgent": False, "participant_names_display": "李四、王五", "description": "检查打印质量", }, ) self.assertIn("任务已创建", content) self.assertIn("任务ID：12", content) self.assertIn("创建人：张三", content) def test_render_notification_content_with_structured_description_template(self): notifier = Notifier.objects.create( merchant=self.merchant, name="结构化描述通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_structured_description_text", config={"agent_id": 1000007}, ) content = render_notification_content( notifier=notifier, payload={ "mission_id": 12, "creator_name": "张三", "category_name": "通用", "description": "原始描述", "parsed_description_title": "标题一\n标题二", "parsed_description_body": "正文一\n正文二", "parsed_description_url": "https://example.com/detail", "parsed_description_image_url": "https://images.yuwen.cloud/abc.jpg", }, ) rendered_payload = json.loads(content) self.assertEqual(rendered_payload["msgtype"], "text") self.assertIn("标题：标题一\n标题二", rendered_payload["content"]) self.assertIn("正文：正文一\n正文二", rendered_payload["content"]) self.assertIn("链接：https://example.com/detail", rendered_payload["content"]) def test_render_notification_content_with_structured_description_news_template(self): notifier = Notifier.objects.create( merchant=self.merchant, name="结构化描述图文通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_structured_description_news", config={ "agent_ids": [1000007], "image_url": "https://cdn.example.com/covers/mission-news.png", "url": "https://erp.example.com/missions/fallback", }, ) content = render_notification_content( notifier=notifier, payload={ "mission_id": 12, "creator_name": "张三", "category_name": "通用", "description": "原始描述", "parsed_description_title": "标题一\n标题二", "parsed_description_body": "正文一\n正文二", "parsed_description_url": "https://example.com/detail", "parsed_description_image_url": "https://images.yuwen.cloud/abc.jpg", }, ) rendered_payload = json.loads(content) self.assertEqual(rendered_payload["msgtype"], "news") self.assertEqual(rendered_payload["title"], "标题一\n标题二") self.assertEqual(rendered_payload["description"], "正文一\n正文二") self.assertEqual(rendered_payload["url"], "https://example.com/detail") self.assertEqual(rendered_payload["image_url"], "https://images.yuwen.cloud/abc.jpg") def test_render_notification_content_with_structured_description_news_template_falls_back_to_default_image(self): notifier = Notifier.objects.create( merchant=self.merchant, name="结构化描述图文通知-默认图", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_structured_description_news", config={ "agent_ids": [1000007], "url": "https://erp.example.com/missions/fallback", }, ) with self.settings(MESSAGE_API_DEFAULT_NEWS_IMAGE_URL="https://via.placeholder.com/640x360.png?text=No+Image"): content = render_notification_content( notifier=notifier, payload={ "mission_id": 12, "creator_name": "张三", "category_name": "通用", "description": "原始描述", "parsed_description_title": "标题一\n标题二", "parsed_description_body": "正文一\n正文二", "parsed_description_url": "", "parsed_description_image_url": "", }, ) rendered_payload = json.loads(content) self.assertEqual(rendered_payload["url"], "https://erp.example.com/missions/fallback") self.assertEqual( rendered_payload["image_url"], "https://via.placeholder.com/640x360.png?text=No+Image", ) @patch("notifier.backends.send_wecom_webhook_message") def test_send_notification_with_notifier_uses_wecom_backend(self, mock_send): mock_send.return_value.ok = True mock_send.return_value.errcode = 0 mock_send.return_value.errmsg = "ok" result = send_notification_with_notifier( notifier=self.notifier, payload={ "mission_id": 12, "created_by_name": "张三", "creator_name": "张三", "category_name": "通用", "is_urgent": True, "participant_names_display": "无", "description": "检查打印质量", }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["channel"], NotifierChannelEnum.WECOM_WEBHOOK) mock_send.assert_called_once() @patch("notifier.services.send_notification_with_notifier") def test_dispatch_notification_event_filters_by_route_event_and_enabled(self, mock_send): mock_send.side_effect = lambda *, notifier, payload: { "notifier_id": notifier.id, "status": "sent", } replied_notifier = Notifier.objects.create( merchant=self.merchant, name="任务回应通知", channel=NotifierChannelEnum.WECOM_WEBHOOK, template_key="mission_replied", config={"key": "def456"}, ) NotifierRoute.objects.create( merchant=self.merchant, notifier=replied_notifier, event_key=NotificationEventKeyEnum.MISSION_REPLIED, ) disabled_notifier = Notifier.objects.create( merchant=self.merchant, name="停用通知器", channel=NotifierChannelEnum.WECOM_WEBHOOK, template_key="mission_created", is_enabled=False, config={"key": "ghi789"}, ) NotifierRoute.objects.create( merchant=self.merchant, notifier=disabled_notifier, event_key=NotificationEventKeyEnum.MISSION_CREATED, ) results = dispatch_notification_event( event_key=NotificationEventKeyEnum.MISSION_CREATED, merchant_id=self.merchant.id, payload={"mission_id": 99}, ) self.assertEqual( results, [ { "notifier_id": self.notifier.id, "status": "sent", "event_key": NotificationEventKeyEnum.MISSION_CREATED, "route_id": self.global_route.id, "route_event_key": NotificationEventKeyEnum.MISSION_CREATED, "route_mission_category_id": None, } ], ) self.assertEqual(mock_send.call_count, 1) self.assertEqual(mock_send.call_args.kwargs["notifier"].id, self.notifier.id) @patch("notifier.services.send_notification_with_notifier") def test_dispatch_notification_event_prefers_category_specific_route(self, mock_send): mock_send.side_effect = lambda *, notifier, payload: { "notifier_id": notifier.id, "status": "sent", } specific_route = NotifierRoute.objects.create( merchant=self.merchant, notifier=self.notifier, event_key=NotificationEventKeyEnum.MISSION_CREATED, mission_category=self.after_sale_category, ) results = dispatch_notification_event( event_key=NotificationEventKeyEnum.MISSION_CREATED, merchant_id=self.merchant.id, payload={"mission_id": 99, "category_id": self.after_sale_category.id}, ) self.assertEqual(mock_send.call_count, 1) self.assertEqual(results[0]["route_id"], specific_route.id) self.assertEqual(results[0]["route_mission_category_id"], self.after_sale_category.id) @patch("notifier.services.send_notification_with_notifier") def test_dispatch_notification_event_falls_back_to_global_route(self, mock_send): mock_send.side_effect = lambda *, notifier, payload: { "notifier_id": notifier.id, "status": "sent", } results = dispatch_notification_event( event_key=NotificationEventKeyEnum.MISSION_CREATED, merchant_id=self.merchant.id, payload={"mission_id": 99, "category_id": self.general_category.id}, ) self.assertEqual(mock_send.call_count, 1) self.assertEqual(results[0]["route_id"], self.global_route.id) self.assertIsNone(results[0]["route_mission_category_id"]) @patch("notifier.tasks.dispatch_notification_event_task.delay") def test_enqueue_notification_event_returns_task_id(self, mock_delay): mock_delay.return_value.id = "task-123" task_id = enqueue_notification_event( event_key=NotificationEventKeyEnum.MISSION_CREATED, merchant_id=self.merchant.id, payload={"mission_id": 1}, ) self.assertEqual(task_id, "task-123") mock_delay.assert_called_once() @patch("notifier.tasks.dispatch_notification_event_task.delay", side_effect=RuntimeError("broker down")) def test_enqueue_notification_event_fails_open(self, mock_delay): task_id = enqueue_notification_event( event_key=NotificationEventKeyEnum.MISSION_CREATED, merchant_id=self.merchant.id, payload={"mission_id": 1}, ) self.assertIsNone(task_id) mock_delay.assert_called_once() @patch("notifier.tasks.dispatch_notification_event") def test_notify_unreplied_missions_task_dispatches_due_mission_and_updates_state(self, mock_dispatch): mock_dispatch.return_value = [{"status": "sent"}] notifier = Notifier.objects.create( merchant=self.merchant, name="未回复提醒通知器", channel=NotifierChannelEnum.WECOM_WEBHOOK, template_key="mission_unreplied", config={"key": "unreplied-key"}, ) NotifierRoute.objects.create( merchant=self.merchant, notifier=notifier, event_key=NotificationEventKeyEnum.MISSION_UNREPLIED, ) mission = Mission.objects.create( merchant=self.merchant, category=self.general_category, creator=self.creator, description="超时未回复任务", notify_if_unreplied=True, unreplied_notify_interval_minutes=1, ) mission.created_at = timezone.now() - timezone.timedelta(minutes=3) mission.save(update_fields=["created_at", "updated_at"]) result = notify_unreplied_missions_task.run(limit=10) mission.refresh_from_db() self.assertEqual(result["sent_count"], 1) self.assertEqual(mission.unreplied_notify_sent_count, 1) self.assertIsNotNone(mission.unreplied_last_notified_at) mock_dispatch.assert_called_once() self.assertEqual(mock_dispatch.call_args.kwargs["event_key"], NotificationEventKeyEnum.MISSION_UNREPLIED) @patch("notifier.tasks.dispatch_notification_event") def test_notify_unreplied_missions_task_skips_mission_with_effective_reply(self, mock_dispatch): mission = Mission.objects.create( merchant=self.merchant, category=self.general_category, creator=self.creator, description="已有回复任务", notify_if_unreplied=True, unreplied_notify_interval_minutes=1, ) mission.created_at = timezone.now() - timezone.timedelta(minutes=3) mission.save(update_fields=["created_at", "updated_at"]) responder = Employee.objects.create(merchant=self.merchant, name="回应人") MissionReply.objects.create( merchant=self.merchant, mission=mission, responder=responder, content="收到", ) result = notify_unreplied_missions_task.run(limit=10) self.assertEqual(result["sent_count"], 0) mock_dispatch.assert_not_called() @patch("notifier.tasks.dispatch_notification_event") def test_notify_unreplied_missions_task_applies_category_payload_processor(self, mock_dispatch): mock_dispatch.return_value = [{"status": "sent"}] self.general_category.payload_processor = "structured_description_v1" self.general_category.save(update_fields=["payload_processor", "updated_at"]) notifier = Notifier.objects.create( merchant=self.merchant, name="未回复提醒通知器", channel=NotifierChannelEnum.WECOM_WEBHOOK, template_key="mission_unreplied", config={"key": "unreplied-key"}, ) NotifierRoute.objects.create( merchant=self.merchant, notifier=notifier, event_key=NotificationEventKeyEnum.MISSION_UNREPLIED, ) mission = Mission.objects.create( merchant=self.merchant, category=self.general_category, creator=self.creator, description=( "系统首行\n标题一\n款式图:https://images.yuwen.cloud/abc.jpg\n标题二\n\n正文一\n正文二\n" "手机端链接: https://example.com/detail" ), notify_if_unreplied=True, unreplied_notify_interval_minutes=1, ) mission.created_at = timezone.now() - timezone.timedelta(minutes=3) mission.save(update_fields=["created_at", "updated_at"]) result = notify_unreplied_missions_task.run(limit=10) self.assertEqual(result["sent_count"], 1) payload = mock_dispatch.call_args.kwargs["payload"] self.assertEqual(payload["payload_processor"], "structured_description_v1") self.assertEqual(payload["parsed_description_title"], "标题一\n标题二") self.assertEqual(payload["parsed_description_body"], "正文一\n正文二") self.assertEqual(payload["parsed_description_url"], "https://example.com/detail") self.assertEqual(payload["parsed_description_image_url"], "https://images.yuwen.cloud/abc.jpg") @patch("notifier.backends.send_message_api_text_message") def test_send_notification_with_notifier_uses_message_api_text_backend(self, mock_send): mock_send.return_value.errcode = 0 mock_send.return_value.errmsg = "ok" mock_send.return_value.raw = {"errcode": 0, "errmsg": "ok"} notifier = Notifier.objects.create( merchant=self.merchant, name="任务创建消息 API 通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_created", config={"agent_id": 1000007}, ) result = send_notification_with_notifier( notifier=notifier, payload={ "mission_id": 12, "created_by_name": "张三", "creator_name": "张三", "category_name": "通用", "is_urgent": True, "participant_names_display": "无", "description": "检查打印质量", }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["channel"], NotifierChannelEnum.MESSAGE_API) self.assertEqual(result["msgtype"], "text") self.assertEqual(mock_send.call_count, 1) self.assertEqual(mock_send.call_args.kwargs["agent_id"], 1000007) self.assertEqual(mock_send.call_args.kwargs["user_ids"], None) self.assertEqual(mock_send.call_args.kwargs["timeout_seconds"], 10.0) self.assertIn("任务已创建", mock_send.call_args.kwargs["content"]) @patch("notifier.backends.send_message_api_text_message") def test_send_notification_with_notifier_uses_structured_description_template(self, mock_send): mock_send.return_value.errcode = 0 mock_send.return_value.errmsg = "ok" mock_send.return_value.raw = {"errcode": 0, "errmsg": "ok"} notifier = Notifier.objects.create( merchant=self.merchant, name="结构化描述消息 API 通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_structured_description_text", config={"agent_id": 1000007}, ) result = send_notification_with_notifier( notifier=notifier, payload={ "mission_id": 12, "creator_name": "张三", "category_name": "通用", "description": "原始描述", "parsed_description_title": "标题一\n标题二", "parsed_description_body": "正文一\n正文二", "parsed_description_url": "https://example.com/detail", "parsed_description_image_url": "https://images.yuwen.cloud/abc.jpg", }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["channel"], NotifierChannelEnum.MESSAGE_API) self.assertEqual(result["msgtype"], "text") self.assertEqual(mock_send.call_count, 1) self.assertIn("标题：标题一\n标题二", mock_send.call_args.kwargs["content"]) self.assertIn("正文：正文一\n正文二", mock_send.call_args.kwargs["content"]) self.assertIn("链接：https://example.com/detail", mock_send.call_args.kwargs["content"]) @patch("notifier.backends.send_message_api_news_to_agents") def test_send_notification_with_notifier_uses_structured_description_news_template(self, mock_send): mock_send.return_value = [ { "agent_id": 1000007, "ok": True, "response": {"errcode": 0, "errmsg": "ok"}, "error": None, } ] notifier = Notifier.objects.create( merchant=self.merchant, name="结构化描述图文消息 API 通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="mission_structured_description_news", config={ "agent_ids": [1000007], "image_url": "https://cdn.example.com/covers/mission-news.png", }, ) result = send_notification_with_notifier( notifier=notifier, payload={ "mission_id": 12, "description": "原始描述", "parsed_description_title": "标题一\n标题二", "parsed_description_body": "正文一\n正文二", "parsed_description_url": "https://example.com/detail", "parsed_description_image_url": "https://images.yuwen.cloud/abc.jpg", }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["msgtype"], "news") mock_send.assert_called_once() self.assertEqual(mock_send.call_args.kwargs["agent_ids"], [1000007]) self.assertEqual(mock_send.call_args.kwargs["title"], "标题一\n标题二") self.assertEqual(mock_send.call_args.kwargs["description"], "正文一\n正文二") self.assertEqual(mock_send.call_args.kwargs["url"], "https://example.com/detail") self.assertEqual( mock_send.call_args.kwargs["image_url"], "https://images.yuwen.cloud/abc.jpg", ) @patch("notifier.backends.send_message_api_news_to_agents") def test_send_notification_with_notifier_uses_message_api_news_backend(self, mock_send): mock_send.return_value = [ { "agent_id": 1000007, "ok": True, "response": {"errcode": 0, "errmsg": "ok"}, "error": None, } ] notifier = Notifier.objects.create( merchant=self.merchant, name="任务创建图文通知", channel=NotifierChannelEnum.MESSAGE_API, template_key="test_message_news", config={"agent_ids": [1000007]}, ) result = send_notification_with_notifier( notifier=notifier, payload={ "mission_id": 12, "description": "检查打印质量", "category_name": "通用", }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["msgtype"], "news") self.assertEqual(result["ok_count"], 1) mock_send.assert_called_once() def test_send_notification_with_notifier_rejects_invalid_message_api_template(self): notifier = Notifier.objects.create( merchant=self.merchant, name="非法消息模板", channel=NotifierChannelEnum.MESSAGE_API, template_key="test_message_invalid", config={"agent_id": 1000007}, ) with self.assertRaisesMessage(Exception, "content"): send_notification_with_notifier( notifier=notifier, payload={"mission_id": 12}, ) @patch("notifier.backends.send_jpush_payload") def test_send_notification_with_notifier_uses_jpush_backend(self, mock_send): mock_send.return_value.raw = {"sendno": "123", "msg_id": "456"} notifier = Notifier.objects.create( merchant=self.merchant, name="任务创建极光推送", channel=NotifierChannelEnum.JPUSH, template_key="mission_created_jpush", config={ "app_key": "app-key-1", "master_secret": "master-secret-1", "alias_prefix": "emp_", "recipient_source": "participants", "platform": "all", "apns_production": True, "android_channel_id": "mission", }, ) result = send_notification_with_notifier( notifier=notifier, payload={ "mission_id": 12, "merchant_id": self.merchant.id, "description": "检查打印质量", "participant_ids": [21, 22, 21], }, ) self.assertEqual(result["status"], "sent") self.assertEqual(result["channel"], NotifierChannelEnum.JPUSH) self.assertEqual(result["aliases"], ["emp_21", "emp_22"]) mock_send.assert_called_once() self.assertEqual(mock_send.call_args.kwargs["app_key"], "app-key-1") self.assertEqual(mock_send.call_args.kwargs["master_secret"], "master-secret-1") payload = mock_send.call_args.kwargs["payload"] self.assertEqual(payload["audience"], {"alias": ["emp_21", "emp_22"]}) self.assertEqual(payload["notification"]["android"]["channel_id"], "mission") self.assertEqual(payload["notification"]["ios"]["sound"], "default") self.assertTrue(payload["options"]["apns_production"]) self.assertEqual(payload["notification"]["android"]["extras"]["mission_id"], "12") @patch("notifier.backends.send_jpush_payload") def test_send_notification_with_notifier_rejects_jpush_without_aliases(self, mock_send): notifier = Notifier.objects.create( merchant=self.merchant, name="无接收人极光推送", channel=NotifierChannelEnum.JPUSH, template_key="mission_created_jpush", config={"app_key": "app-key-1", "master_secret": "master-secret-1"}, ) with self.assertRaisesMessage(ValueError, "alias 不能为空"): send_notification_with_notifier( notifier=notifier, payload={"mission_id": 12, "description": "检查打印质量"}, ) mock_send.assert_not_called() class MessageAPIServiceTestCase(TestCase): @patch("notifier.message_api.urlopen") def test_send_message_api_text_message_calls_message_api_endpoint(self, mock_urlopen): from notifier.message_api import send_message_api_text_message response_obj = MagicMock() response_obj.read.return_value = b'{"errcode":0,"errmsg":"ok"}' response_obj.__enter__.return_value = response_obj response_obj.__exit__.return_value = False mock_urlopen.return_value = response_obj with self.settings( MESSAGE_API_BASE_URL="http://message-api.internal:8198", MESSAGE_API_AUTHORIZATION="secret-1", ): response = send_message_api_text_message(agent_id=1000007, content="hello") self.assertTrue(response.ok) self.assertEqual(response.errcode, 0) self.assertEqual(mock_urlopen.call_count, 1) request = mock_urlopen.call_args.args[0] self.assertEqual(request.full_url, "http://message-api.internal:8198/api/message/send") self.assertEqual(request.get_method(), "POST") self.assertEqual(request.get_header("Authorization"), "secret-1") @patch("notifier.message_api.urlopen") def test_send_message_api_news_to_agents_calls_message_api_endpoint(self, mock_urlopen): from notifier.message_api import send_message_api_news_to_agents response_obj = MagicMock() response_obj.read.return_value = ( b'{"results":[{"agent_id":1000007,"ok":true,"response":{"errcode":0,"errmsg":"ok"},"error":null}]}' ) response_obj.__enter__.return_value = response_obj response_obj.__exit__.return_value = False mock_urlopen.return_value = response_obj with self.settings( MESSAGE_API_BASE_URL="http://message-api.internal:8198", MESSAGE_API_AUTHORIZATION="secret-1", ): results = send_message_api_news_to_agents( agent_ids=[1000007], title="销售日报", description="点击查看今日各区域销售汇总", url="https://example.com/reports/daily-sales", image_url="https://example.com/static/daily-sales-cover.png", ) self.assertEqual(len(results), 1) request = mock_urlopen.call_args.args[0] self.assertEqual(request.full_url, "http://message-api.internal:8198/api/message/send/news") self.assertEqual(request.get_header("Authorization"), "secret-1")