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")