djangoldp_notification/tests/test_cache.py

+import uuid
+import json
+from rest_framework.test import APITestCase, APIClient
+from djangoldp_account.models import LDPUser
+from djangoldp_notification.models import Notification
+
+
+class TestSubscription(APITestCase):
+    def _get_random_user(self):
+        return LDPUser.objects.create(email='{}@test.co.uk'.format(str(uuid.uuid4())), first_name='Test',
+                                      last_name='Test', username=str(uuid.uuid4()))
+
+    def _get_random_notification(self, recipient, author):
+        return Notification.objects.create(user=recipient, author=author.urlid, object=author.urlid,
+                                           unread=True)
+
+    def setUpLoggedInUser(self):
+        self.user = self._get_random_user()
+        self.client.force_authenticate(user=self.user)
+
+    def setUp(self):
+        self.client = APIClient()
+
+    def test_indirect_cache(self):
+        self.setUpLoggedInUser()
+        author_user = self._get_random_user()
+        notification = self._get_random_notification(recipient=self.user, author=author_user)
+        self.assertEqual(notification.unread, True)
+
+        # GET the inbox - should set the cache
+        response = self.client.get("/users/{}/inbox/".format(self.user.username))
+        self.assertEqual(response.status_code, 200)
+        notif_serialized = response.data["ldp:contains"][0]
+        self.assertEqual(notif_serialized["unread"], True)
+
+        # PATCH the notification - should wipe the cache
+        patch = {
+            "unread": False,
+            "@context": {
+                "@vocab":"https://cdn.startinblox.com/owl#",
+                "unread": "https://cdn.startinblox.com/owl#unread"
+            }
+        }
+        response = self.client.patch("/notifications/{}/".format(notification.pk), data=json.dumps(patch),
+                                     content_type="application/ld+json")
+        notif_obj = Notification.objects.get(pk=notification.pk)
+        self.assertEqual(notif_obj.unread, False)
+
+        # GET the inbox - should now be read
+        response = self.client.get("/users/{}/inbox/".format(self.user.username))
+        self.assertEqual(response.status_code, 200)
+        notif_serialized = response.data["ldp:contains"][0]
+        self.assertEqual(notif_serialized["unread"], False)
+
+    # NOTE: this would be our ideal cache behaviour
+    # the functionality for optimising it was removed because of an issue with extensibility
+    #  https://git.startinblox.com/djangoldp-packages/djangoldp-notification/merge_requests/42#note_58559
+    '''def test_custom_cache_clear(self):
+        # going to create two notifications in two different inboxes
+        self.setUpLoggedInUser()
+        other_user = self._get_random_user()
+        notification = self._get_random_notification(recipient=self.user, author=other_user)
+        notification2 = self._get_random_notification(recipient=other_user, author=self.user)
+
+        # GET the inboxes and asser that the cache is set for both
+        self.client.get("/users/{}/inbox/".format(self.user.username))
+        self.client.get("/users/{}/inbox/".format(other_user.username))
+
+        # assert cache is set
+        my_container_urlid = '{}/users/{}/inbox/'.format(settings.SITE_URL, self.user.username)
+        their_container_urlid = '{}/users/{}/inbox/'.format(settings.SITE_URL, other_user.username)
+
+        self.assertTrue(GLOBAL_SERIALIZER_CACHE.has(getattr(Notification._meta, 'label', None), my_container_urlid))
+        self.assertTrue(GLOBAL_SERIALIZER_CACHE.has(getattr(Notification._meta, 'label', None), their_container_urlid))
+
+        # save my notification - should wipe the cache for my inbox...
+        notification.unread = False
+        notification.save()
+        self.assertFalse(GLOBAL_SERIALIZER_CACHE.has(getattr(Notification._meta, 'label', None), my_container_urlid))
+
+        # ...but not for theirs
+        self.assertTrue(GLOBAL_SERIALIZER_CACHE.has(getattr(Notification._meta, 'label', None), their_container_urlid))'''

djangoldp_notification/tests/test_models.py

+import uuid
+from rest_framework.test import APITestCase, APIClient
+
+from djangoldp_account.models import LDPUser
+from djangoldp_notification.models import Subscription
+
+
+class TestModel(APITestCase):
+    circle_user1_url = "http://localhost:8000/circles/1/"
+    circle_user2_url = "http://localhost:8000/circles/2/"
+
+    def _get_random_user(self):
+        return LDPUser.objects.create(email='{}@test.co.uk'.format(str(uuid.uuid4())), first_name='Test',
+                                      last_name='Test', username=str(uuid.uuid4()))
+
+    def _auth_as_user(self, user):
+        self.client.force_authenticate(user=user)
+
+    def setUpLoggedInUser(self):
+        self.user = self._get_random_user()
+        self._auth_as_user(self.user)
+
+    def setUp(self):
+        self.client = APIClient()
+
+        self.user1 = self._get_random_user()
+        Subscription.objects.create(object=self.circle_user1_url, inbox="http://testserver/users/karl_marx/inbox/")
+
+        self.user2 = self._get_random_user()
+        Subscription.objects.create(object=self.circle_user2_url,
+                                    inbox="http://testserver/users/piotr_kropotkine/inbox/")
+
+    def test_not_logged_fails(self):
+        response = self.client.get("/subscriptions/")
+        self.assertEqual(response.status_code, 403)
+
+    def test_logged_in_succeeds(self):
+        self._auth_as_user(self.user2)
+        response = self.client.get("/subscriptions/").data.get("ldp:contains")
+        self.assertEqual(len(response), 2)
+        response = response[1]
+        self.assertEqual(response["object"], self.circle_user2_url)
+        self.assertEqual(response["inbox"], "http://testserver/users/piotr_kropotkine/inbox/")
+        # No more permissions serialised on users
+        # self.assertIn({'mode': {'@type': 'view'}}, response["permissions"])
+        # self.assertIn({'mode': {'@type': 'delete'}}, response["permissions"])

djangoldp_notification/tests/test_subscriptions.py

+import uuid
+import json
+
+from django.conf import settings
+from rest_framework.test import APITestCase, APIClient
+
+from djangoldp_account.models import LDPUser
+from djangoldp_notification.models import Notification, Subscription
+
+
+class SubscriptionTestCase(APITestCase):
+    def setUp(self):
+        self.client = APIClient()
+
+    def _get_random_user(self):
+        return LDPUser.objects.create(email='{}@test.co.uk'.format(str(uuid.uuid4())), first_name='Test',
+                                      last_name='Test', username=str(uuid.uuid4()))
+
+    def _auth_as_user(self, user):
+        self.client.force_authenticate(user=user)
+
+    def setUpLoggedInUser(self):
+        self.user = self._get_random_user()
+        self._auth_as_user(self.user)
+
+    def test_can_subscribe_to_container(self):
+        self.setUpLoggedInUser()
+
+        users_container = "{}/users/".format(settings.SITE_URL)
+        Subscription.objects.create(object=users_container, inbox=self.user.urlid + "inbox/")
+        self.assertEqual(self.user.inbox.count(), 0)
+
+        self._get_random_user()
+        self.assertEqual(self.user.inbox.count(), 2)
+        notification = self.user.inbox.all()[0]
+        self.assertTrue(notification.object, users_container)

djangoldp_notification/views.py

+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.http import Http404
+from rest_framework import status
+from rest_framework.response import Response
+
+from djangoldp.models import Model
+from djangoldp.serializers import LDPSerializer
+from djangoldp.views import LDPViewSet
+from djangoldp.pagination import LDPPagination
+import logging
+
+
+logger = logging.getLogger('djangoldp')
+
+
+class LDPNotificationsPagination(LDPPagination):
+    default_limit = 80
+
+
+def filter_object_is_permitted(recipient, data):
+        '''
+        applies filter on passed object data
+        returns True if the object is permitted, False if not
+        '''
+        obj = data['object']
+
+        # if a str (urlid) is given then no filtering is to be applied
+        if isinstance(obj, str):
+            return True
+
+        # the type must be given for a filter to be resolved successfully
+        if not '@type' in obj or obj['@type'] is None:
+            logger.error('djangoldp_notification filter ERR in object serialization. received ' + str(obj) + ' without serialized type to identify model. Please include the @type in your serialization')
+            return False
+
+        object_model = Model.get_subclass_with_rdf_type(obj['@type'])
+
+        if object_model is None:
+            logger.error('djangoldp_notification filter ERR in object serialization. Cannot resolve type given ' + str(obj['type']))
+            return False
+        
+        if not hasattr(object_model, 'permit_notification'):
+            logger.error('djangoldp_notification filter ERR. Resolved type ' + str(obj['@type']) + ' but this Model did not have the required function permit_notification defined on it')
+            return False
+
+        return object_model.permit_notification(recipient, data)
+
+
+class LDPNotificationsViewSet(LDPViewSet):
+    '''overridden LDPViewSet to force pagination'''
+    pagination_class = LDPNotificationsPagination
+    depth = 0

setup.cfg

@@ -10,11 +10,13 @@ license = MIT
 [options]
 packages = find:
 install_requires =
-    djangoldp~=0.5
+    djangoldp~=3.1.0
+    djangoldp_account~=3.1.0
 
 [options.extras_require]
+include_package_data = True
 dev =
-    factory_boy>=2.11.0
+    factory_boy>=2.12.0
 
 [semantic_release]
 version_source = tag

setup.py

@@ -2,4 +2,6 @@
 
 from setuptools import setup
 
-setup()
+setup(
+    include_package_data=True
+)