359 lines
15 KiB
Python
359 lines
15 KiB
Python
import asyncio
|
|
import json
|
|
import pytest
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from django.test import TestCase, Client
|
|
from app.feeds.notification_publisher import publish_notification
|
|
|
|
|
|
def collect_stream(response):
|
|
"""Consume an async streaming response and return the decoded content."""
|
|
async def _collect():
|
|
chunks = []
|
|
async for chunk in response.streaming_content:
|
|
if isinstance(chunk, bytes):
|
|
chunks.append(chunk)
|
|
else:
|
|
chunks.append(chunk.encode("utf-8"))
|
|
return b"".join(chunks).decode("utf-8")
|
|
return asyncio.run(_collect())
|
|
|
|
|
|
def make_async_pubsub(messages=None, error=None):
|
|
"""Build an async pubsub mock that returns messages then stops."""
|
|
mock_pubsub = AsyncMock()
|
|
|
|
if error:
|
|
mock_pubsub.get_message.side_effect = error
|
|
elif messages is not None:
|
|
# Yield each message, then raise RedisError to break the loop
|
|
import redis.asyncio as aioredis
|
|
mock_pubsub.get_message.side_effect = messages + [aioredis.RedisError("done")]
|
|
else:
|
|
import redis.asyncio as aioredis
|
|
mock_pubsub.get_message.side_effect = aioredis.RedisError("done")
|
|
|
|
return mock_pubsub
|
|
|
|
|
|
class TestSSENotificationsEndpoint(TestCase):
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.client = Client()
|
|
self.feed_url = "https://example.com/social.org"
|
|
|
|
def _patch_redis(self, pubsub):
|
|
mock_redis = AsyncMock()
|
|
mock_redis.pubsub = MagicMock(return_value=pubsub) # pubsub() is a sync call
|
|
return patch("app.sse_notifications.views.aioredis.Redis", return_value=mock_redis)
|
|
|
|
def test_sse_endpoint_without_feed_returns_global_stream(self):
|
|
"""Test that SSE endpoint without feed parameter returns global stream."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The SSE endpoint is requested without a feed parameter
|
|
response = self.client.get("/sse/notifications/")
|
|
|
|
# Then: An event stream response is returned
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "text/event-stream")
|
|
|
|
def test_sse_endpoint_accepts_valid_feed(self):
|
|
"""Test that SSE endpoint accepts valid feed parameter."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The SSE endpoint is requested for a specific feed
|
|
response = self.client.get("/sse/notifications/", {"feed": self.feed_url})
|
|
|
|
# Then: An event stream response is returned with anti-buffering headers
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "text/event-stream")
|
|
self.assertEqual(response["Cache-Control"], "no-cache")
|
|
self.assertEqual(response["X-Accel-Buffering"], "no")
|
|
self.assertEqual(response["Access-Control-Allow-Origin"], "*")
|
|
|
|
def test_sse_sends_connection_event_with_feed(self):
|
|
"""Test that per-feed SSE sends initial connection event with feed field."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The per-feed stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/", {"feed": self.feed_url})
|
|
content = collect_stream(response)
|
|
|
|
# Then: The stream starts with a connection event naming the feed
|
|
self.assertIn("event: connected", content)
|
|
self.assertIn(f'"feed": "{self.feed_url}"', content)
|
|
self.assertIn('"status": "connected"', content)
|
|
|
|
def test_sse_sends_connection_event_global(self):
|
|
"""Test that global SSE sends initial connection event without feed field."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The global stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/")
|
|
content = collect_stream(response)
|
|
|
|
# Then: The stream starts with a connection event without a feed field
|
|
self.assertIn("event: connected", content)
|
|
self.assertIn('"status": "connected"', content)
|
|
self.assertNotIn('"feed":', content)
|
|
|
|
def test_sse_receives_notification_from_redis(self):
|
|
"""Test that per-feed SSE receives and forwards notifications from Redis."""
|
|
# Given: A Redis pubsub holding one mention notification
|
|
notification_data = {
|
|
"type": "mention",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
}
|
|
pubsub = make_async_pubsub(
|
|
messages=[{"type": "message", "data": json.dumps(notification_data)}]
|
|
)
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The per-feed stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/", {"feed": self.feed_url})
|
|
content = collect_stream(response)
|
|
|
|
# Then: The notification is forwarded as an SSE event
|
|
self.assertIn("event: notification", content)
|
|
self.assertIn('"type": "mention"', content)
|
|
self.assertIn("https://alice.com/social.org#2024-01-01T10:00:00+0000", content)
|
|
|
|
def test_global_sse_adds_target_feed_to_notifications(self):
|
|
"""Test that global SSE adds target_feed field extracted from channel name."""
|
|
# Given: A Redis pubsub holding one pattern message for a feed channel
|
|
notification_data = {
|
|
"type": "mention",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
}
|
|
target_feed = "https://example.com/social.org"
|
|
pubsub = make_async_pubsub(
|
|
messages=[
|
|
{
|
|
"type": "pmessage",
|
|
"pattern": "notifications:*",
|
|
"channel": f"notifications:{target_feed}",
|
|
"data": json.dumps(notification_data),
|
|
}
|
|
]
|
|
)
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The global stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/")
|
|
content = collect_stream(response)
|
|
|
|
# Then: The forwarded event includes the target_feed from the channel
|
|
self.assertIn("event: notification", content)
|
|
self.assertIn(f'"target_feed": "{target_feed}"', content)
|
|
self.assertIn('"type": "mention"', content)
|
|
|
|
def test_global_sse_uses_psubscribe(self):
|
|
"""Test that global SSE subscribes with psubscribe to all notification channels."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The global stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/")
|
|
collect_stream(response)
|
|
|
|
# Then: The view subscribed by pattern to every notification channel
|
|
pubsub.psubscribe.assert_awaited_once_with("notifications:*")
|
|
|
|
def test_per_feed_sse_uses_subscribe(self):
|
|
"""Test that per-feed SSE subscribes with subscribe to the specific channel."""
|
|
# Given: A Redis pubsub with no pending messages
|
|
pubsub = make_async_pubsub()
|
|
|
|
with self._patch_redis(pubsub):
|
|
# When: The per-feed stream is requested and consumed
|
|
response = self.client.get("/sse/notifications/", {"feed": self.feed_url})
|
|
collect_stream(response)
|
|
|
|
# Then: The view subscribed only to that feed's channel
|
|
pubsub.subscribe.assert_awaited_once_with(f"notifications:{self.feed_url}")
|
|
|
|
|
|
class TestNotificationPublisher(TestCase):
|
|
"""Test the notification publisher module"""
|
|
|
|
@patch("app.feeds.notification_publisher.redis.Redis")
|
|
def test_publish_mention_notification(self, mock_redis):
|
|
"""Test publishing a mention notification"""
|
|
# Given: A working Redis connection and a mention to notify
|
|
mock_redis_instance = MagicMock()
|
|
mock_redis.return_value = mock_redis_instance
|
|
|
|
target_feed = "https://bob.com/social.org"
|
|
post_url = "https://alice.com/social.org#2024-01-01T10:00:00+0000"
|
|
|
|
# When: The mention notification is published
|
|
result = publish_notification(
|
|
target_feed_url=target_feed, notification_type="mention", post_url=post_url
|
|
)
|
|
|
|
# Then: One message is published on the target feed's channel
|
|
self.assertTrue(result)
|
|
mock_redis_instance.publish.assert_called_once()
|
|
|
|
call_args = mock_redis_instance.publish.call_args
|
|
channel, data = call_args[0]
|
|
self.assertEqual(channel, f"notifications:{target_feed}")
|
|
notification = json.loads(data)
|
|
self.assertEqual(notification["type"], "mention")
|
|
self.assertEqual(notification["post"], post_url)
|
|
|
|
@patch("app.feeds.notification_publisher.redis.Redis")
|
|
def test_publish_reaction_notification_with_emoji(self, mock_redis):
|
|
"""Test publishing a reaction notification with emoji"""
|
|
# Given: A working Redis connection and a reaction to notify
|
|
mock_redis_instance = MagicMock()
|
|
mock_redis.return_value = mock_redis_instance
|
|
|
|
# When: The reaction notification is published
|
|
result = publish_notification(
|
|
target_feed_url="https://bob.com/social.org",
|
|
notification_type="reaction",
|
|
post_url="https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
emoji="❤",
|
|
parent="https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
)
|
|
|
|
# Then: The published message carries the reaction type and emoji
|
|
self.assertTrue(result)
|
|
call_args = mock_redis_instance.publish.call_args
|
|
notification = json.loads(call_args[0][1])
|
|
self.assertEqual(notification["type"], "reaction")
|
|
self.assertEqual(notification["emoji"], "❤")
|
|
|
|
@patch("app.feeds.notification_publisher.redis.Redis")
|
|
def test_publish_reply_notification(self, mock_redis):
|
|
"""Test publishing a reply notification"""
|
|
# Given: A working Redis connection and a reply to notify
|
|
mock_redis_instance = MagicMock()
|
|
mock_redis.return_value = mock_redis_instance
|
|
|
|
# When: The reply notification is published
|
|
result = publish_notification(
|
|
target_feed_url="https://bob.com/social.org",
|
|
notification_type="reply",
|
|
post_url="https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
parent="https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
)
|
|
|
|
# Then: The published message carries the reply type
|
|
self.assertTrue(result)
|
|
notification = json.loads(mock_redis_instance.publish.call_args[0][1])
|
|
self.assertEqual(notification["type"], "reply")
|
|
|
|
@patch("app.feeds.notification_publisher.redis.Redis")
|
|
def test_publish_boost_notification(self, mock_redis):
|
|
"""Test publishing a boost notification"""
|
|
# Given: A working Redis connection and a boost to notify
|
|
mock_redis_instance = MagicMock()
|
|
mock_redis.return_value = mock_redis_instance
|
|
|
|
# When: The boost notification is published
|
|
result = publish_notification(
|
|
target_feed_url="https://bob.com/social.org",
|
|
notification_type="boost",
|
|
post_url="https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
boosted="https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
)
|
|
|
|
# Then: The published message carries the boost type
|
|
self.assertTrue(result)
|
|
notification = json.loads(mock_redis_instance.publish.call_args[0][1])
|
|
self.assertEqual(notification["type"], "boost")
|
|
|
|
@patch("app.feeds.notification_publisher.redis.Redis")
|
|
def test_publish_notification_handles_redis_error(self, mock_redis):
|
|
"""Test that publish_notification handles Redis errors gracefully"""
|
|
# Given: A Redis connection that fails
|
|
mock_redis.side_effect = Exception("Redis connection failed")
|
|
|
|
# When: A notification is published
|
|
result = publish_notification(
|
|
target_feed_url="https://bob.com/social.org",
|
|
notification_type="mention",
|
|
post_url="https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
)
|
|
|
|
# Then: The failure is reported without raising
|
|
self.assertFalse(result)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class TestSSENotificationStructure:
|
|
"""Test that SSE notifications match the expected JSON structure"""
|
|
|
|
def test_mention_notification_structure(self):
|
|
# Given: A mention notification payload
|
|
notification = {
|
|
"type": "mention",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
}
|
|
|
|
# Then: It carries the mention type and a post URL with fragment
|
|
assert "type" in notification
|
|
assert notification["type"] == "mention"
|
|
assert "#" in notification["post"]
|
|
|
|
def test_reaction_notification_structure(self):
|
|
# Given: A reaction notification payload
|
|
notification = {
|
|
"type": "reaction",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
"emoji": "❤",
|
|
"parent": "https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
}
|
|
|
|
# Then: It carries the reaction type, emoji and parent
|
|
assert notification["type"] == "reaction"
|
|
assert "emoji" in notification
|
|
assert "parent" in notification
|
|
|
|
def test_reply_notification_structure(self):
|
|
# Given: A reply notification payload
|
|
notification = {
|
|
"type": "reply",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
"parent": "https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
}
|
|
|
|
# Then: It carries the reply type and parent
|
|
assert notification["type"] == "reply"
|
|
assert "parent" in notification
|
|
|
|
def test_boost_notification_structure(self):
|
|
# Given: A boost notification payload
|
|
notification = {
|
|
"type": "boost",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
"boosted": "https://bob.com/social.org#2024-01-01T09:00:00+0000",
|
|
}
|
|
|
|
# Then: It carries the boost type and the boosted post
|
|
assert notification["type"] == "boost"
|
|
assert "boosted" in notification
|
|
|
|
def test_global_notification_includes_target_feed(self):
|
|
# Given: A global stream notification payload
|
|
notification = {
|
|
"target_feed": "https://example.com/social.org",
|
|
"type": "mention",
|
|
"post": "https://alice.com/social.org#2024-01-01T10:00:00+0000",
|
|
}
|
|
|
|
# Then: It includes the target feed as an absolute URL
|
|
assert "target_feed" in notification
|
|
assert notification["target_feed"].startswith("http")
|