79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
from django.conf import settings
|
|
from django.test import TestCase
|
|
|
|
from apps.notifications.services import store as services
|
|
from apps.notifications.services import RedisNotificationStore
|
|
from apps.notifications.tests.fakes import FakeRedis
|
|
|
|
|
|
class RedisNotificationStoreTests(TestCase):
|
|
def setUp(self):
|
|
self.fake_redis = FakeRedis()
|
|
self.original_redis_client = services.redis_client
|
|
services.redis_client = self.fake_redis
|
|
|
|
def tearDown(self):
|
|
services.redis_client = self.original_redis_client
|
|
|
|
def test_add_publishes_notification_and_unread_count(self):
|
|
with self.settings(NOTIFICATIONS_ENABLED=True):
|
|
notification = RedisNotificationStore.add(
|
|
"user-1",
|
|
{
|
|
"title": "Build finished",
|
|
"message": "Your deploy completed.",
|
|
"level": "success",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(notification["title"], "Build finished")
|
|
self.assertEqual(notification["message"], "Your deploy completed.")
|
|
self.assertEqual(notification["level"], "success")
|
|
self.assertEqual(len(self.fake_redis.published), 2)
|
|
|
|
channel, payload = self.fake_redis.published[0]
|
|
self.assertEqual(
|
|
channel,
|
|
f"{settings.NOTIFICATION_REDIS_CHANNEL_PREFIX}:user-1",
|
|
)
|
|
self.assertEqual(payload["event"], "notification")
|
|
self.assertEqual(payload["data"]["notification"]["id"], notification["id"])
|
|
self.assertEqual(payload["data"]["unread_count"], 1)
|
|
|
|
def test_mark_seen_and_mark_all_seen_publish_sync_events(self):
|
|
with self.settings(NOTIFICATIONS_ENABLED=True):
|
|
first = RedisNotificationStore.add("user-2", {"title": "First"})
|
|
RedisNotificationStore.add("user-2", {"title": "Second"})
|
|
self.fake_redis.published.clear()
|
|
|
|
payload = RedisNotificationStore.mark_seen("user-2", first["id"])
|
|
|
|
self.assertEqual(payload["notification_id"], first["id"])
|
|
self.assertFalse(payload["deleted"])
|
|
self.assertTrue(payload["notification"]["is_seen"])
|
|
self.assertEqual(self.fake_redis.published[0][1]["event"], "notification_seen")
|
|
|
|
self.fake_redis.published.clear()
|
|
updated = RedisNotificationStore.mark_all_seen("user-2")
|
|
|
|
self.assertEqual(updated, 2)
|
|
self.assertEqual(self.fake_redis.published[0][1]["event"], "notification_mark_all_read")
|
|
self.assertEqual(self.fake_redis.published[1][1]["event"], "unread_count")
|
|
self.assertEqual(self.fake_redis.published[1][1]["data"]["unread_count"], 0)
|
|
|
|
def test_list_returns_total_count_and_filtered_notifications(self):
|
|
RedisNotificationStore.add("user-3", {"title": "General", "type": "general"})
|
|
RedisNotificationStore.add("user-3", {"title": "Billing", "type": "billing"})
|
|
RedisNotificationStore.add("user-3", {"title": "General 2", "type": "general"})
|
|
|
|
notifications, total_count = RedisNotificationStore.list(
|
|
"user-3",
|
|
limit=1,
|
|
offset=0,
|
|
type_filter="general",
|
|
)
|
|
|
|
self.assertEqual(total_count, 2)
|
|
self.assertEqual(len(notifications), 1)
|
|
self.assertEqual(notifications[0]["type"], "general")
|