|
a/Allura/allura/tests/model/test_notification.py |
|
b/Allura/allura/tests/model/test_notification.py |
|
... |
|
... |
| 105 |
else:
|
105 |
else:
|
| 106 |
assert False, 'From address is wrong: %s' % task.kwargs['fromaddr']
|
106 |
assert False, 'From address is wrong: %s' % task.kwargs['fromaddr']
|
| 107 |
assert task.kwargs['text'].startswith('WikiPage Home modified by Test Admin')
|
107 |
assert task.kwargs['text'].startswith('WikiPage Home modified by Test Admin')
|
| 108 |
assert 'you indicated interest in ' in task.kwargs['text']
|
108 |
assert 'you indicated interest in ' in task.kwargs['text']
|
| 109 |
|
109 |
|
|
|
110 |
def test_permissions(self):
|
|
|
111 |
# Notification should only be delivered if user has read perms on the
|
|
|
112 |
# artifact. The perm check happens just before the mail task is
|
|
|
113 |
# posted.
|
|
|
114 |
u = M.User.query.get(username='test-admin')
|
|
|
115 |
self._subscribe(user=u)
|
|
|
116 |
# Simulate a permission check failure.
|
|
|
117 |
def patched_has_access(*args, **kw):
|
|
|
118 |
def predicate(*args, **kw):
|
|
|
119 |
return False
|
|
|
120 |
return predicate
|
|
|
121 |
from allura.model.notification import security
|
|
|
122 |
orig = security.has_access
|
|
|
123 |
security.has_access = patched_has_access
|
|
|
124 |
try:
|
|
|
125 |
# this will create a notification task
|
|
|
126 |
self._post_notification()
|
|
|
127 |
ThreadLocalORMSession.flush_all()
|
|
|
128 |
# running the notification task will create a mail task if the
|
|
|
129 |
# permission check passes...
|
|
|
130 |
M.MonQTask.run_ready()
|
|
|
131 |
ThreadLocalORMSession.flush_all()
|
|
|
132 |
# ...but in this case it doesn't create a mail task since we
|
|
|
133 |
# forced the perm check to fail
|
|
|
134 |
assert M.MonQTask.get() == None
|
|
|
135 |
finally:
|
|
|
136 |
security.has_access = orig
|
|
|
137 |
|
| 110 |
def _subscribe(self):
|
138 |
def _subscribe(self, **kw):
|
| 111 |
self.pg.subscribe(type='direct')
|
139 |
self.pg.subscribe(type='direct', **kw)
|
| 112 |
ThreadLocalORMSession.flush_all()
|
140 |
ThreadLocalORMSession.flush_all()
|
| 113 |
ThreadLocalORMSession.close_all()
|
141 |
ThreadLocalORMSession.close_all()
|
| 114 |
|
142 |
|
| 115 |
def _post_notification(self):
|
143 |
def _post_notification(self):
|
| 116 |
return M.Notification.post(self.pg, 'metadata')
|
144 |
return M.Notification.post(self.pg, 'metadata')
|