# GNU MediaGoblin -- federated, autonomous media hosting # Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # 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 Affero 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 . import pytest import six.moves.urllib.parse as urlparse from mediagoblin.tools import template, mail from mediagoblin.db.models import Notification, CommentSubscription from mediagoblin.db.base import Session from mediagoblin.notifications import mark_comment_notification_seen from mediagoblin.tests.tools import fixture_add_comment, \ fixture_media_entry, fixture_add_user, \ fixture_comment_subscription class TestNotifications: @pytest.fixture(autouse=True) def setup(self, test_app): self.test_app = test_app # TODO: Possibly abstract into a decorator like: # @as_authenticated_user('chris') self.test_user = fixture_add_user(privileges=[u'active',u'commenter']) self.current_user = None self.login() def login(self, username=u'chris', password=u'toast'): response = self.test_app.post( '/auth/login/', { 'username': username, 'password': password}) response.follow() assert urlparse.urlsplit(response.location)[2] == '/' assert 'mediagoblin/root.html' in template.TEMPLATE_TEST_CONTEXT ctx = template.TEMPLATE_TEST_CONTEXT['mediagoblin/root.html'] assert Session.merge(ctx['request'].user).username == username self.current_user = ctx['request'].user def logout(self): self.test_app.get('/auth/logout/') self.current_user = None @pytest.mark.parametrize('wants_email', [True, False]) def test_comment_notification(self, wants_email): ''' Test - if a notification is created when posting a comment on another users media entry. - that the comment data is consistent and exists. ''' user = fixture_add_user('otherperson', password='nosreprehto', wants_comment_notification=wants_email, privileges=[u'active',u'commenter']) assert user.wants_comment_notification == wants_email user_id = user.id media_entry = fixture_media_entry(uploader=user.id, state=u'processed') media_entry_id = media_entry.id subscription = fixture_comment_subscription(media_entry) subscription_id = subscription.id media_uri_id = '/u/{0}/m/{1}/'.format(user.username, media_entry.id) media_uri_slug = '/u/{0}/m/{1}/'.format(user.username, media_entry.slug) self.test_app.post( media_uri_id + 'comment/add/', { 'comment_content': u'Test comment #42' } ) notifications = Notification.query.filter_by( user_id=user.id).all() assert len(notifications) == 1 notification = notifications[0] assert notification.seen == False assert notification.user_id == user.id assert notification.obj().comment().get_actor.id == self.test_user.id assert notification.obj().comment().content == u'Test comment #42' if wants_email == True: # Why the `or' here? In Werkzeug 0.11.0 and above # werkzeug stopped showing the port for localhost when # rendering something like this. As long as we're # supporting pre-0.11.0 we'll keep this `or', but maybe # in the future we can kill it. assert ( mail.EMAIL_TEST_MBOX_INBOX == [ {'from': 'notice@mediagoblin.example.org', 'message': 'Content-Type: text/plain; \ charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: \ base64\nSubject: GNU MediaGoblin - chris commented on your \ post\nFrom: notice@mediagoblin.example.org\nTo: \ otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyIHBvc3QgKGh0dHA6Ly9sb2Nh\nbGhvc3Q6ODAvdS9vdGhlcnBlcnNvbi9tL3NvbWUtdGl0bGUvYy8xLyNjb21tZW50KSBhdCBHTlUg\nTWVkaWFHb2JsaW4KClRlc3QgY29tbWVudCAjNDIKCkdOVSBNZWRpYUdvYmxpbg==\n', 'to': [u'otherperson@example.com']}] or mail.EMAIL_TEST_MBOX_INBOX == [ {'from': 'notice@mediagoblin.example.org', 'message': 'Content-Type: text/plain; \ charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: \ base64\nSubject: GNU MediaGoblin - chris commented on your \ post\nFrom: notice@mediagoblin.example.org\nTo: \ otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyIHBvc3QgKGh0dHA6Ly9sb2Nh\nbGhvc3QvdS9vdGhlcnBlcnNvbi9tL3NvbWUtdGl0bGUvYy8xLyNjb21tZW50KSBhdCBHTlUgTWVk\naWFHb2JsaW4KClRlc3QgY29tbWVudCAjNDIKCkdOVSBNZWRpYUdvYmxpbg==\n', 'to': [u'otherperson@example.com']}]) else: assert mail.EMAIL_TEST_MBOX_INBOX == [] # Save the ids temporarily because of DetachedInstanceError notification_id = notification.id comment_id = notification.obj().id self.logout() self.login('otherperson', 'nosreprehto') self.test_app.get(media_uri_slug + 'c/{0}/'.format(comment_id)) notification = Notification.query.filter_by(id=notification_id).first() assert notification.seen == True self.test_app.get(media_uri_slug + 'notifications/silence/') subscription = CommentSubscription.query.filter_by(id=subscription_id)\ .first() assert subscription.notify == False notifications = Notification.query.filter_by( user_id=user_id).all() # User should not have been notified assert len(notifications) == 1 def test_mark_all_comment_notifications_seen(self): """ Test that mark_all_comments_seen works""" user = fixture_add_user('otherperson', password='nosreprehto', privileges=[u'active']) media_entry = fixture_media_entry(uploader=user.id, state=u'processed') fixture_comment_subscription(media_entry) media_uri_id = '/u/{0}/m/{1}/'.format(user.username, media_entry.id) # add 2 comments self.test_app.post( media_uri_id + 'comment/add/', { 'comment_content': u'Test comment #43' } ) self.test_app.post( media_uri_id + 'comment/add/', { 'comment_content': u'Test comment #44' } ) notifications = Notification.query.filter_by( user_id=user.id).all() assert len(notifications) == 2 # both comments should not be marked seen assert notifications[0].seen == False assert notifications[1].seen == False # login with other user to mark notifications seen self.logout() self.login('otherperson', 'nosreprehto') # mark all comment notifications seen res = self.test_app.get('/notifications/comments/mark_all_seen/') res.follow() assert urlparse.urlsplit(res.location)[2] == '/' notifications = Notification.query.filter_by( user_id=user.id).all() # both notifications should be marked seen assert notifications[0].seen == True assert notifications[1].seen == True