diff --git a/kallithea/tests/base.py b/kallithea/tests/base.py --- a/kallithea/tests/base.py +++ b/kallithea/tests/base.py @@ -24,7 +24,7 @@ from tg import config from webtest import TestApp from kallithea import model -from kallithea.model.db import Notification, User, UserNotification +from kallithea.model.db import User from kallithea.model.meta import Session from kallithea.lib.utils2 import safe_str @@ -149,15 +149,6 @@ class TestController(object): self.app = TestApp(testapp) return self.app - def remove_all_notifications(self): - # query().delete() does not (by default) trigger cascades - # ( http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes ) - # so delete the UserNotification first to ensure referential integrity. - UserNotification.query().delete() - - Notification.query().delete() - Session().commit() - def log_user(self, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS): self._logged_username = username diff --git a/kallithea/tests/functional/test_admin_notifications.py b/kallithea/tests/functional/test_admin_notifications.py deleted file mode 100644 --- a/kallithea/tests/functional/test_admin_notifications.py +++ /dev/null @@ -1,172 +0,0 @@ -from kallithea.tests.base import * -from kallithea.model.db import User - -from kallithea.model.user import UserModel -from kallithea.model.notification import NotificationModel -from kallithea.model.meta import Session -from kallithea.lib import helpers as h - -from tg.util.webtest import test_context - - -class TestNotificationsController(TestController): - def setup_method(self, method): - self.remove_all_notifications() - - def test_index(self, create_test_user): - self.log_user() - - u1 = create_test_user(dict(username='u1', password='qweqwe', - email='u1@example.com', - firstname=u'u1', lastname=u'u1', - active=True)) - u1 = u1.user_id - Session().commit() - - response = self.app.get(url('notifications')) - response.mustcontain('
No notifications here yet
') - - with test_context(self.app): - cur_user = self._get_logged_user() - notif = NotificationModel().create(created_by=u1, subject=u'test_notification_1', - body=u'notification_1', recipients=[cur_user]) - Session().commit() - - response = self.app.get(url('notifications')) - response.mustcontain('id="notification_%s"' % notif.notification_id) - - def test_delete(self, create_test_user): - self.log_user() - cur_user = self._get_logged_user() - - with test_context(self.app): - u1 = create_test_user(dict(username='u1', password='qweqwe', - email='u1@example.com', - firstname=u'u1', lastname=u'u1', - active=True)) - u2 = create_test_user(dict(username='u2', password='qweqwe', - email='u2@example.com', - firstname=u'u2', lastname=u'u2', - active=True)) - - # make notifications - notification = NotificationModel().create(created_by=cur_user, - subject=u'test', - body=u'hi there', - recipients=[cur_user, u1, u2]) - Session().commit() - u1 = User.get(u1.user_id) - u2 = User.get(u2.user_id) - - # check DB - get_notif = lambda un: [x.notification for x in un] - assert get_notif(cur_user.notifications) == [notification] - assert get_notif(u1.notifications) == [notification] - assert get_notif(u2.notifications) == [notification] - cur_usr_id = cur_user.user_id - - response = self.app.post( - url('notification_delete', notification_id=notification.notification_id), - params={'_authentication_token': self.authentication_token()}) - assert response.body == 'ok' - - cur_user = User.get(cur_usr_id) - assert cur_user.notifications == [] - - def test_show(self, create_test_user): - self.log_user() - with test_context(self.app): - cur_user = self._get_logged_user() - u1 = create_test_user(dict(username='u1', password='qweqwe', - email='u1@example.com', - firstname=u'u1', lastname=u'u1', - active=True)) - u2 = create_test_user(dict(username='u2', password='qweqwe', - email='u2@example.com', - firstname=u'u2', lastname=u'u2', - active=True)) - Session().commit() - - subject = u'test' - notif_body = u'hi there' - notification = NotificationModel().create(created_by=cur_user, - subject=subject, - body=notif_body, - recipients=[cur_user, u1, u2]) - - response = self.app.get(url('notification', - notification_id=notification.notification_id)) - - response.mustcontain(subject) - response.mustcontain(notif_body) - - def test_description_with_age(self): - self.log_user() - with test_context(self.app): - cur_user = self._get_logged_user() - subject = u'test' - notify_body = u'hi there' - - notification = NotificationModel().create(created_by = cur_user, - subject = subject, - body = notify_body) - - description = NotificationModel().make_description(notification) - assert description == "{0} sent message {1}".format( - cur_user.username, - h.age(notification.created_on) - ) - - def test_description_with_datetime(self): - self.log_user() - with test_context(self.app): - cur_user = self._get_logged_user() - subject = u'test' - notify_body = u'hi there' - notification = NotificationModel().create(created_by = cur_user, - subject = subject, - body = notify_body) - - description = NotificationModel().make_description(notification, False) - assert description == "{0} sent message at {1}".format( - cur_user.username, - h.fmt_date(notification.created_on) - ) - - def test_mark_all_read(self, create_test_user): - self.log_user() - with test_context(self.app): - u0 = self._get_logged_user() - u1 = create_test_user(dict(username='u1', password='qweqwe', - email='u1@example.com', - firstname=u'u1', lastname=u'u1', - active=True)) - u2 = create_test_user(dict(username='u2', password='qweqwe', - email='u2@example.com', - firstname=u'u2', lastname=u'u2', - active=True)) - notif = NotificationModel().create(created_by=u1, - subject=u'subject', - body=u'body', - recipients=[u0, u2]) - u0_id, u1_id, u2_id = u0.user_id, u1.user_id, u2.user_id - - assert [n.read for n in u0.notifications] == [False] - assert u1.notifications == [] - assert [n.read for n in u2.notifications] == [False] - - # Mark all read for current user. - - response = self.app.get(url('notifications_mark_all_read'), # TODO: should be POST - extra_environ=dict(HTTP_X_PARTIAL_XHR='1')) - - assert response.status_int == 200 - response.mustcontain('id="notification_%s"' % notif.notification_id) - - u0 = User.get(u0_id) - u1 = User.get(u1_id) - u2 = User.get(u2_id) - - assert [n.read for n in u0.notifications] == [True] - assert u1.notifications == [] - assert [n.read for n in u2.notifications] == [False] diff --git a/kallithea/tests/functional/test_changeset_pullrequests_comments.py b/kallithea/tests/functional/test_changeset_pullrequests_comments.py --- a/kallithea/tests/functional/test_changeset_pullrequests_comments.py +++ b/kallithea/tests/functional/test_changeset_pullrequests_comments.py @@ -2,8 +2,7 @@ import re from kallithea.tests.base import * from kallithea.model.changeset_status import ChangesetStatusModel -from kallithea.model.db import ChangesetComment, Notification, \ - UserNotification +from kallithea.model.db import ChangesetComment from kallithea.model.meta import Session @@ -14,8 +13,6 @@ class TestChangeSetCommentsController(Te Session().delete(x) Session().commit() - self.remove_all_notifications() - def test_create(self): self.log_user() rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc' @@ -38,17 +35,6 @@ class TestChangeSetCommentsController(Te # test DB assert ChangesetComment.query().count() == 1 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - - comment_id = ChangesetComment.query().first().comment_id - assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT - sbj = (u'/%s/changeset/' - '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' - % (HG_REPO, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject def test_create_inline(self): self.log_user() @@ -79,16 +65,6 @@ class TestChangeSetCommentsController(Te # test DB assert ChangesetComment.query().count() == 1 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - comment_id = ChangesetComment.query().first().comment_id - assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT - sbj = (u'/%s/changeset/' - '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' - % (HG_REPO, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject def test_create_with_mention(self): self.log_user() @@ -113,11 +89,6 @@ class TestChangeSetCommentsController(Te # test DB assert ChangesetComment.query().count() == 1 - assert Notification.query().count() == 2 - users = [x.user.username for x in UserNotification.query().all()] - - # test_regular gets notification by @mention - assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN] def test_create_status_change(self): self.log_user() @@ -142,17 +113,6 @@ class TestChangeSetCommentsController(Te # test DB assert ChangesetComment.query().count() == 1 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - - comment_id = ChangesetComment.query().first().comment_id - assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT - sbj = (u'/%s/changeset/' - '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' - % (HG_REPO, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject # check status status = ChangesetStatusModel().get_status(repo=HG_REPO, revision=rev) @@ -196,8 +156,6 @@ class TestPullrequestsCommentsController Session().delete(x) Session().commit() - self.remove_all_notifications() - def _create_pr(self): response = self.app.post(url(controller='pullrequests', action='create', repo_name=HG_REPO), @@ -238,15 +196,6 @@ class TestPullrequestsCommentsController # test DB assert ChangesetComment.query().count() == 2 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id - assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT - sbj = (u'/%s/pull-request/%s/_/stable#comment-%s' - % (HG_REPO, pr_id, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject def test_create_inline(self): self.log_user() @@ -277,15 +226,6 @@ class TestPullrequestsCommentsController # test DB assert ChangesetComment.query().count() == 2 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id - assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT - sbj = (u'/%s/pull-request/%s/_/stable#comment-%s' - % (HG_REPO, pr_id, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject def test_create_with_mention(self): self.log_user() @@ -309,11 +249,6 @@ class TestPullrequestsCommentsController # test DB assert ChangesetComment.query().count() == 2 - assert Notification.query().count() == 2 - users = [x.user.username for x in UserNotification.query().all()] - - # test_regular gets notification by @mention - assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN] def test_create_status_change(self): self.log_user() @@ -341,15 +276,6 @@ class TestPullrequestsCommentsController # test DB assert ChangesetComment.query().count() == 2 - assert Notification.query().count() == 1 - - notification = Notification.query().all()[0] - comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id - assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT - sbj = (u'/%s/pull-request/%s/_/stable#comment-%s' - % (HG_REPO, pr_id, comment_id)) - print "%s vs %s" % (sbj, notification.subject) - assert sbj in notification.subject # check status status = ChangesetStatusModel().get_status(repo=HG_REPO, pull_request=pr_id) diff --git a/kallithea/tests/functional/test_login.py b/kallithea/tests/functional/test_login.py --- a/kallithea/tests/functional/test_login.py +++ b/kallithea/tests/functional/test_login.py @@ -12,7 +12,7 @@ from kallithea.lib.auth import check_pas from kallithea.lib import helpers as h from kallithea.model.api_key import ApiKeyModel from kallithea.model import validators -from kallithea.model.db import User, Notification +from kallithea.model.db import User from kallithea.model.meta import Session from kallithea.model.user import UserModel @@ -22,9 +22,6 @@ fixture = Fixture() class TestLoginController(TestController): - def setup_method(self, method): - self.remove_all_notifications() - assert Notification.query().all() == [] def test_index(self): response = self.app.get(url(controller='login', action='index')) diff --git a/kallithea/tests/models/test_notifications.py b/kallithea/tests/models/test_notifications.py --- a/kallithea/tests/models/test_notifications.py +++ b/kallithea/tests/models/test_notifications.py @@ -6,7 +6,7 @@ import routes.util from kallithea.tests.base import * from kallithea.lib import helpers as h -from kallithea.model.db import User, Notification, UserNotification +from kallithea.model.db import User, Notification from kallithea.model.user import UserModel from kallithea.model.meta import Session from kallithea.model.notification import NotificationModel, EmailNotificationModel @@ -42,10 +42,6 @@ class TestNotifications(TestController): Session().commit() self.u3 = u3.user_id - self.remove_all_notifications() - assert [] == Notification.query().all() - assert [] == UserNotification.query().all() - def test_create_notification(self): with test_context(self.app): usrs = [self.u1, self.u2] @@ -60,118 +56,6 @@ class TestNotifications(TestController): notification = NotificationModel().create(created_by=self.u1, subject=u'subj', body=u'hi there', recipients=usrs) - Session().commit() - u1 = User.get(self.u1) - u2 = User.get(self.u2) - u3 = User.get(self.u3) - 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_id for x in unotification]) == set(usrs) - - def test_user_notifications(self): - with test_context(self.app): - notification1 = NotificationModel().create(created_by=self.u1, - subject=u'subj', body=u'hi there1', - recipients=[self.u3]) - Session().commit() - notification2 = NotificationModel().create(created_by=self.u1, - subject=u'subj', body=u'hi there2', - recipients=[self.u3]) - Session().commit() - u3 = Session().query(User).get(self.u3) - - assert sorted([x.notification for x in u3.notifications]) == sorted([notification2, notification1]) - - def test_delete_notifications(self): - with test_context(self.app): - notification = NotificationModel().create(created_by=self.u1, - subject=u'title', body=u'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): - with test_context(self.app): - notification = NotificationModel().create(created_by=self.u1, - subject=u'title', body=u'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 == 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 != None - u2notification = UserNotification.query() \ - .filter(UserNotification.notification == - notification) \ - .filter(UserNotification.user_id == self.u2) \ - .scalar() - assert u2notification != None - - def test_notification_counter(self): - with test_context(self.app): - NotificationModel().create(created_by=self.u1, - subject=u'title', body=u'hi there_delete', - recipients=[self.u3, self.u1]) - Session().commit() - - 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 - - notification = NotificationModel().create(created_by=self.u1, - subject=u'title', body=u'hi there3', - recipients=[self.u3, self.u1, self.u2]) - Session().commit() - - 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 @mock.patch.object(h, 'canonical_url', (lambda arg, **kwargs: 'http://%s/?%s' % (arg, '&'.join('%s=%s' % (k, v) for (k, v) in sorted(kwargs.items()))))) def test_dump_html_mails(self):