Notifications

NextCloud.py

@@ -90,6 +90,7 @@ def __init__(self, endpoint, user, password, js=False):
             "User": User(requester),
             "FederatedCloudShare": FederatedCloudShare(requester),
             "Activity": Activity(requester),
+            "Notifications": Notifications(requester),
         }
         for name, location in PUBLIC_API_NAME_CLASS_MAP.items():
             setattr(self, name, getattr(self.functionality[location], name))
@@ -746,6 +747,50 @@ def get_activities(self, since=None, limit=None, object_type=None, object_id=Non
         return self.requester.get(params=params)
 
 
+class Notifications(WithRequester):
+    API_URL = "/ocs/v2.php/apps/notifications/api/v2/notifications"
+
+    @nextcloud_method
+    def get_notifications(self):
+        """ Get list of notifications for a logged in user """
+        return self.requester.get()
+
+    @nextcloud_method
+    def get_notification(self, notification_id):
+        """
+        Get single notification by id for a user
+
+        Args:
+            notification_id (int): Notification id
+
+        Returns:
+
+        """
+        return self.requester.get(url=notification_id)
+
+    @nextcloud_method
+    def delete_notification(self, notification_id):
+        """
+        Delete single notification by id for a user
+
+        Args:
+            notification_id (int): Notification id
+
+        Returns:
+
+        """
+        return self.requester.delete(url=notification_id)
+
+    @nextcloud_method
+    def delete_all_notifications(self):
+        """ Delete all notification for a logged in user
+
+        Notes:
+            This endpoint was added for Nextcloud 14
+        """
+        return self.requester.delete()
+
+
 class OCSCode(enum.IntEnum):
     OK = 100
     SERVER_ERROR = 996

tests/test_notifications.py

@@ -0,0 +1,39 @@
+from .base import BaseTestCase
+
+
+class TestNotifications(BaseTestCase):
+
+    SUCCESS_CODE = 200
+
+    def test_get_delete_notifications(self):
+        # get all notifications
+        res = self.nxc.get_notifications()
+        all_data = res['ocs']['data']
+        assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE
+
+        if len(all_data):
+            notification = all_data[0]
+
+            # get single notification
+            res = self.nxc.get_notification(notification['notification_id'])
+            assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE
+            assert res['ocs']['data']['notification_id'] == notification['notification_id']
+
+            # delete single notification
+            res = self.nxc.delete_notification(notification['notification_id'])
+            assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE
+        else:
+            # assert get single notification will return 404 not found
+            res = self.nxc.get_notification(1)
+            assert res['ocs']['meta']['statuscode'] == self.NOT_FOUND_CODE
+
+        # delete all notifications success code check
+        res = self.nxc.delete_all_notifications()
+        assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE
+
+        # TODO: add more tests if WebDAV api will be implemented and there will be ability to make actions
+        #  using api which creates notifications (mentions in comments)
+        #  or when Federated file sharings api will be implemented (harder way)
+
+
+