forked from erp-dev/erp
255 lines
10 KiB
Python
255 lines
10 KiB
Python
from unittest.mock import 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)
|
||
|
||
@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()
|