210 lines
7.6 KiB
Python
210 lines
7.6 KiB
Python
# Copyright (C) 2010-2024 RhodeCode GmbH
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License, version 3
|
|
# (only), as published by the Free Software Foundation.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
# This program is dual-licensed. If you wish to learn more about the
|
|
# RhodeCode Enterprise Edition, including its added features, Support services,
|
|
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
|
|
|
import pytest
|
|
|
|
from rhodecode.tests.fixtures.rc_fixture import Fixture
|
|
|
|
from rhodecode.model.db import User, Notification, UserNotification
|
|
from rhodecode.model.meta import Session
|
|
from rhodecode.model.notification import NotificationModel
|
|
from rhodecode.model.user import UserModel
|
|
|
|
fixture = Fixture()
|
|
|
|
|
|
class TestNotifications(object):
|
|
destroy_users = set()
|
|
|
|
@classmethod
|
|
def teardown_class(cls):
|
|
fixture.destroy_users(cls.destroy_users)
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def create_users(self, request, app):
|
|
Session.remove()
|
|
self.u1 = UserModel().create_or_update(
|
|
username="u1", password="qweqwe", email="u1@rhodecode.org", firstname="u1", lastname="u1"
|
|
)
|
|
Session().commit()
|
|
self.u1 = self.u1.user_id
|
|
|
|
self.u2 = UserModel().create_or_update(
|
|
username="u2", password="qweqwe", email="u2@rhodecode.org", firstname="u2", lastname="u2"
|
|
)
|
|
Session().commit()
|
|
self.u2 = self.u2.user_id
|
|
|
|
self.u3 = UserModel().create_or_update(
|
|
username="u3", password="qweqwe", email="u3@rhodecode.org", firstname="u3", lastname="u3"
|
|
)
|
|
Session().commit()
|
|
self.u3 = self.u3.user_id
|
|
self.destroy_users.add("u1")
|
|
self.destroy_users.add("u2")
|
|
self.destroy_users.add("u3")
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_notifications(self, request, app):
|
|
for n in Notification.query().all():
|
|
Session().delete(n)
|
|
|
|
Session().commit()
|
|
assert [] == Notification.query().all()
|
|
assert [] == UserNotification.query().all()
|
|
|
|
def test_create_notification(self):
|
|
usrs = [self.u1, self.u2]
|
|
notification = NotificationModel().create(
|
|
created_by=self.u1, notification_subject="subj", notification_body="hi there", recipients=usrs
|
|
)
|
|
Session().commit()
|
|
u1 = User.get(self.u1)
|
|
u2 = User.get(self.u2)
|
|
notifications = Notification.query().all()
|
|
assert len(notifications) == 1
|
|
|
|
assert notifications[0].recipients, [u1 == u2]
|
|
assert notification.notification_id == notifications[0].notification_id
|
|
|
|
unotification = UserNotification.query().filter(UserNotification.notification == notification).all()
|
|
|
|
assert len(unotification) == len(usrs)
|
|
assert set([x.user.user_id for x in unotification]) == set(usrs)
|
|
|
|
def test_create_notification_fails_for_invalid_recipients(self):
|
|
with pytest.raises(Exception):
|
|
NotificationModel().create(
|
|
created_by=self.u1,
|
|
notification_subject="subj",
|
|
notification_body="hi there",
|
|
recipients=["bad_user_id"],
|
|
)
|
|
|
|
with pytest.raises(Exception):
|
|
NotificationModel().create(
|
|
created_by=self.u1, notification_subject="subj", notification_body="hi there", recipients=[]
|
|
)
|
|
|
|
def test_user_notifications(self):
|
|
notification1 = NotificationModel().create(
|
|
created_by=self.u1, notification_subject="subj", notification_body="hi there1", recipients=[self.u3]
|
|
)
|
|
Session().commit()
|
|
notification2 = NotificationModel().create(
|
|
created_by=self.u1, notification_subject="subj", notification_body="hi there2", recipients=[self.u3]
|
|
)
|
|
Session().commit()
|
|
u3 = Session().query(User).get(self.u3)
|
|
|
|
assert [x.notification for x in u3.notifications] == [notification2, notification1]
|
|
|
|
def test_delete_notifications(self):
|
|
notification = NotificationModel().create(
|
|
created_by=self.u1,
|
|
notification_subject="title",
|
|
notification_body="hi there3",
|
|
recipients=[self.u3, self.u1, self.u2],
|
|
)
|
|
Session().commit()
|
|
notifications = Notification.query().all()
|
|
assert notification in notifications
|
|
|
|
Notification.delete(notification.notification_id)
|
|
Session().commit()
|
|
|
|
notifications = Notification.query().all()
|
|
assert notification not in notifications
|
|
|
|
un = UserNotification.query().filter(UserNotification.notification == notification).all()
|
|
assert un == []
|
|
|
|
def test_delete_association(self):
|
|
notification = NotificationModel().create(
|
|
created_by=self.u1,
|
|
notification_subject="title",
|
|
notification_body="hi there3",
|
|
recipients=[self.u3, self.u1, self.u2],
|
|
)
|
|
Session().commit()
|
|
|
|
unotification = (
|
|
UserNotification.query()
|
|
.filter(UserNotification.notification == notification)
|
|
.filter(UserNotification.user_id == self.u3)
|
|
.scalar()
|
|
)
|
|
|
|
assert unotification.user_id == self.u3
|
|
|
|
NotificationModel().delete(self.u3, notification.notification_id)
|
|
Session().commit()
|
|
|
|
u3notification = (
|
|
UserNotification.query()
|
|
.filter(UserNotification.notification == notification)
|
|
.filter(UserNotification.user_id == self.u3)
|
|
.scalar()
|
|
)
|
|
|
|
assert u3notification is None
|
|
|
|
# notification object is still there
|
|
assert Notification.query().all() == [notification]
|
|
|
|
# u1 and u2 still have assignments
|
|
u1notification = (
|
|
UserNotification.query()
|
|
.filter(UserNotification.notification == notification)
|
|
.filter(UserNotification.user_id == self.u1)
|
|
.scalar()
|
|
)
|
|
assert u1notification is not None
|
|
u2notification = (
|
|
UserNotification.query()
|
|
.filter(UserNotification.notification == notification)
|
|
.filter(UserNotification.user_id == self.u2)
|
|
.scalar()
|
|
)
|
|
assert u2notification is not None
|
|
|
|
def test_notification_counter(self):
|
|
NotificationModel().create(
|
|
created_by=self.u1,
|
|
notification_subject="title",
|
|
notification_body="hi there_delete",
|
|
recipients=[self.u3, self.u1],
|
|
)
|
|
Session().commit()
|
|
|
|
# creator has it's own notification marked as read
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u2) == 0
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u3) == 1
|
|
|
|
NotificationModel().create(
|
|
created_by=self.u1,
|
|
notification_subject="title",
|
|
notification_body="hi there3",
|
|
recipients=[self.u3, self.u1, self.u2],
|
|
)
|
|
Session().commit()
|
|
# creator has it's own notification marked as read
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u2) == 1
|
|
assert NotificationModel().get_unread_cnt_for_user(self.u3) == 2
|