fix(api): only create deduplication notifications for specific user

api/passport_admin/notification_generators/expired_stamp.py

@@ -1,11 +1,11 @@
 import hashlib
 
 import dag_cbor
+from django.utils import timezone
+
 from account.models import Community
 from ceramic_cache.api.v1 import handle_get_scorer_weights
 from ceramic_cache.models import CeramicCache
-from django.utils import timezone
-
 from passport_admin.models import Notification
 
 
@@ -46,4 +46,5 @@ def generate_stamp_expired_notifications(address, community: Community):
                     is_active=True,
                     content=f"Your {cc.provider} stamp has expired. Please reverify to keep your Passport up to date.",
                     link=cc.provider,
+                    eth_address=address,
                 )

api/passport_admin/notification_generators/on_chain_expired.py

@@ -1,9 +1,10 @@
-import dag_cbor
 import hashlib
-
 from typing import List
-from passport_admin.schema import ChainSchema
+
+import dag_cbor
+
 from passport_admin.models import Notification
+from passport_admin.schema import ChainSchema
 
 
 def generate_on_chain_expired_notifications(address, expired_chains: List[ChainSchema]):
@@ -30,4 +31,5 @@ def generate_on_chain_expired_notifications(address, expired_chains: List[ChainS
                 type="on_chain_expiry",
                 is_active=True,
                 content=f"Your on-chain Passport on {chain.name} has expired. Update now to maintain your active status.",
+                eth_address=address,
             )

api/passport_admin/tests/test_notifications.py

@@ -7,12 +7,13 @@
 
 import dag_cbor
 import pytest
-from account.models import Community
-from ceramic_cache.api.v1 import DbCacheToken
-from ceramic_cache.models import CeramicCache
 from django.conf import settings
 from django.test import Client
 from django.utils import timezone
+
+from account.models import Community
+from ceramic_cache.api.v1 import DbCacheToken
+from ceramic_cache.models import CeramicCache
 from passport_admin.models import Notification, NotificationStatus
 from registry.models import Event
 from scorer_weighted.models import BinaryWeightedScorer, Scorer
@@ -872,3 +873,47 @@ def test_max_20_active_notifications(self, sample_token, sample_address, communi
         )  # Oldest notification
 
         assert res["items"][-1]["created_at"] == oldest_created_at
+
+    def test_ndeduplication_events_are_returned_for_signed_in_user(
+        self, sample_token, sample_address, community
+    ):
+        """
+        This tests that only a users notifications are returned
+        """
+        notification = Notification.objects.create(
+            notification_id=f"notification_1",
+            type="custom",
+            is_active=True,
+            content=f"Hello! This is a custom notification 1",
+            eth_address=sample_address,
+            community=community,
+        )
+        notification.created_at = timezone.now() - timedelta(days=1)
+        notification.save()
+
+        notification = Notification.objects.create(
+            notification_id=f"notification_2",
+            type="custom",
+            is_active=True,
+            content=f"Hello! This is a custom notification 2",
+            eth_address="sample_address",
+            community=community,
+        )
+        notification.created_at = timezone.now() - timedelta(days=2)
+        notification.save()
+
+        assert Notification.objects.all().count() == 2
+
+        notifications = Notification.objects.filter(eth_address=sample_address)
+        assert notifications.count() == 1
+
+        # Call the same endpoint again to check if the same notifications are returned
+        response = client.post(
+            "/passport-admin/notifications",
+            {"scorer_id": community.id},
+            HTTP_AUTHORIZATION=f"Bearer {sample_token}",
+            content_type="application/json",
+        )
+
+        res = response.json()
+        assert len(res["items"]) == 1