1 # GNU MediaGoblin -- federated, autonomous media hosting
2 # Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS.
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published by
6 # the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19 import six
.moves
.urllib
.parse
as urlparse
21 from mediagoblin
.tools
import template
, mail
23 from mediagoblin
.db
.models
import Notification
, CommentSubscription
24 from mediagoblin
.db
.base
import Session
26 from mediagoblin
.notifications
import mark_comment_notification_seen
28 from mediagoblin
.tests
.tools
import fixture_add_comment
, \
29 fixture_media_entry
, fixture_add_user
, \
30 fixture_comment_subscription
33 class TestNotifications
:
34 @pytest.fixture(autouse
=True)
35 def setup(self
, test_app
):
36 self
.test_app
= test_app
38 # TODO: Possibly abstract into a decorator like:
39 # @as_authenticated_user('chris')
40 self
.test_user
= fixture_add_user(privileges
=[u
'active',u
'commenter'])
42 self
.current_user
= None
46 def login(self
, username
=u
'chris', password
=u
'toast'):
47 response
= self
.test_app
.post(
50 'password': password
})
54 assert urlparse
.urlsplit(response
.location
)[2] == '/'
55 assert 'mediagoblin/root.html' in template
.TEMPLATE_TEST_CONTEXT
57 ctx
= template
.TEMPLATE_TEST_CONTEXT
['mediagoblin/root.html']
59 assert Session
.merge(ctx
['request'].user
).username
== username
61 self
.current_user
= ctx
['request'].user
64 self
.test_app
.get('/auth/logout/')
65 self
.current_user
= None
67 @pytest.mark
.parametrize('wants_email', [True, False])
68 def test_comment_notification(self
, wants_email
):
71 - if a notification is created when posting a comment on
72 another users media entry.
73 - that the comment data is consistent and exists.
76 user
= fixture_add_user('otherperson', password
='nosreprehto',
77 wants_comment_notification
=wants_email
,
78 privileges
=[u
'active',u
'commenter'])
80 assert user
.wants_comment_notification
== wants_email
84 media_entry
= fixture_media_entry(uploader
=user
.id, state
=u
'processed')
86 media_entry_id
= media_entry
.id
88 subscription
= fixture_comment_subscription(media_entry
)
90 subscription_id
= subscription
.id
92 media_uri_id
= '/u/{0}/m/{1}/'.format(user
.username
,
94 media_uri_slug
= '/u/{0}/m/{1}/'.format(user
.username
,
98 media_uri_id
+ 'comment/add/',
100 'comment_content': u
'Test comment #42'
104 notifications
= Notification
.query
.filter_by(
105 user_id
=user
.id).all()
107 assert len(notifications
) == 1
109 notification
= notifications
[0]
111 assert notification
.seen
== False
112 assert notification
.user_id
== user
.id
113 assert notification
.obj().get_actor
.id == self
.test_user
.id
114 assert notification
.obj().content
== u
'Test comment #42'
116 if wants_email
== True:
117 # Why the `or' here? In Werkzeug 0.11.0 and above
118 # werkzeug stopped showing the port for localhost when
119 # rendering something like this. As long as we're
120 # supporting pre-0.11.0 we'll keep this `or', but maybe
121 # in the future we can kill it.
123 mail
.EMAIL_TEST_MBOX_INBOX
== [
124 {'from': 'notice@mediagoblin.example.org',
125 'message': 'Content-Type: text/plain; \
126 charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: \
127 base64\nSubject: GNU MediaGoblin - chris commented on your \
128 post\nFrom: notice@mediagoblin.example.org\nTo: \
129 otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyIHBvc3QgKGh0dHA6Ly9sb2Nh\nbGhvc3Q6ODAvdS9vdGhlcnBlcnNvbi9tL3NvbWUtdGl0bGUvYy8xLyNjb21tZW50KSBhdCBHTlUg\nTWVkaWFHb2JsaW4KClRlc3QgY29tbWVudCAjNDIKCkdOVSBNZWRpYUdvYmxpbg==\n',
130 'to': [u
'otherperson@example.com']}]
131 or mail
.EMAIL_TEST_MBOX_INBOX
== [
132 {'from': 'notice@mediagoblin.example.org',
133 'message': 'Content-Type: text/plain; \
134 charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: \
135 base64\nSubject: GNU MediaGoblin - chris commented on your \
136 post\nFrom: notice@mediagoblin.example.org\nTo: \
137 otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyIHBvc3QgKGh0dHA6Ly9sb2Nh\nbGhvc3QvdS9vdGhlcnBlcnNvbi9tL3NvbWUtdGl0bGUvYy8xLyNjb21tZW50KSBhdCBHTlUgTWVk\naWFHb2JsaW4KClRlc3QgY29tbWVudCAjNDIKCkdOVSBNZWRpYUdvYmxpbg==\n',
138 'to': [u
'otherperson@example.com']}])
140 assert mail
.EMAIL_TEST_MBOX_INBOX
== []
143 # Save the ids temporarily because of DetachedInstanceError
144 notification_id
= notification
.id
145 comment_id
= notification
.obj().get_comment_link().id
148 self
.login('otherperson', 'nosreprehto')
150 self
.test_app
.get(media_uri_slug
+ 'c/{0}/'.format(comment_id
))
152 notification
= Notification
.query
.filter_by(id=notification_id
).first()
154 assert notification
.seen
== True
156 self
.test_app
.get(media_uri_slug
+ 'notifications/silence/')
158 subscription
= CommentSubscription
.query
.filter_by(id=subscription_id
)\
161 assert subscription
.notify
== False
163 notifications
= Notification
.query
.filter_by(
164 user_id
=user_id
).all()
166 # User should not have been notified
167 assert len(notifications
) == 1
169 def test_mark_all_comment_notifications_seen(self
):
170 """ Test that mark_all_comments_seen works"""
172 user
= fixture_add_user('otherperson', password
='nosreprehto',
173 privileges
=[u
'active'])
175 media_entry
= fixture_media_entry(uploader
=user
.id, state
=u
'processed')
177 fixture_comment_subscription(media_entry
)
179 media_uri_id
= '/u/{0}/m/{1}/'.format(user
.username
,
184 media_uri_id
+ 'comment/add/',
186 'comment_content': u
'Test comment #43'
191 media_uri_id
+ 'comment/add/',
193 'comment_content': u
'Test comment #44'
197 notifications
= Notification
.query
.filter_by(
198 user_id
=user
.id).all()
200 assert len(notifications
) == 2
202 # both comments should not be marked seen
203 assert notifications
[0].seen
== False
204 assert notifications
[1].seen
== False
206 # login with other user to mark notifications seen
208 self
.login('otherperson', 'nosreprehto')
210 # mark all comment notifications seen
211 res
= self
.test_app
.get('/notifications/comments/mark_all_seen/')
214 assert urlparse
.urlsplit(res
.location
)[2] == '/'
216 notifications
= Notification
.query
.filter_by(
217 user_id
=user
.id).all()
219 # both notifications should be marked seen
220 assert notifications
[0].seen
== True
221 assert notifications
[1].seen
== True