diff --git a/NextCloud.py b/NextCloud.py index 07ca9e7..479476a 100644 --- a/NextCloud.py +++ b/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 diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 index 0000000..b8dc2cd --- /dev/null +++ b/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) + + +