[ckan-changes] commit/ckan: dread: [tests]: #1222 Mailer tests and fixes. #1221 Password reset tests and fixes.
Bitbucket
commits-noreply at bitbucket.org
Wed Jul 13 11:27:37 UTC 2011
1 new changeset in ckan:
http://bitbucket.org/okfn/ckan/changeset/4737c86d1d81/
changeset: 4737c86d1d81
branch: release-v1.4.2
user: dread
date: 2011-07-13 13:26:26
summary: [tests]: #1222 Mailer tests and fixes. #1221 Password reset tests and fixes.
affected #: 10 files (13.2 KB)
--- a/ckan/controllers/home.py Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/controllers/home.py Wed Jul 13 12:26:26 2011 +0100
@@ -43,7 +43,8 @@
query = query_for(model.Package)
query.run(query='*:*', facet_by=g.facets,
- limit=0, offset=0, username=c.user)
+ limit=0, offset=0, username=c.user,
+ order_by=None)
c.facets = query.facets
c.fields = []
c.package_count = query.count
--- a/ckan/controllers/user.py Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/controllers/user.py Wed Jul 13 12:26:26 2011 +0100
@@ -188,8 +188,17 @@
if request.method == 'POST':
id = request.params.get('user')
user = model.User.get(id)
+ if user is None and id and len(id)>2:
+ q = model.User.search(id)
+ if q.count() == 1:
+ user = q.one()
+ elif q.count() > 1:
+ users = ' '.join([user.name for user in q])
+ h.flash_error(_('"%s" matched several users') % (id))
+ return render("user/request_reset.html")
if user is None:
h.flash_error(_('No such user: %s') % id)
+ return render("user/request_reset.html")
try:
mailer.send_reset_link(user)
h.flash_success(_('Please check your inbox for a reset code.'))
@@ -204,8 +213,9 @@
abort(404)
c.reset_key = request.params.get('key')
if not mailer.verify_reset_link(user, c.reset_key):
- h.flash_error(_('Invalid reset key. Please try again.'))
- abort(403)
+ msg = _('Invalid reset key. Please try again.')
+ h.flash_error(msg)
+ abort(403, msg.encode('utf8'))
if request.method == 'POST':
try:
user.password = self._get_form_password()
--- a/ckan/lib/mailer.py Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/lib/mailer.py Wed Jul 13 12:26:26 2011 +0100
@@ -17,13 +17,16 @@
class MailerException(Exception):
pass
+def add_msg_niceties(recipient_name, body, sender_name, sender_url):
+ return _(u"Dear %s,") % recipient_name \
+ + u"\r\n\r\n%s\r\n\r\n" % body \
+ + u"--\r\n%s (%s)" % (sender_name, sender_url)
+
def _mail_recipient(recipient_name, recipient_email,
sender_name, sender_url, subject,
body, headers={}):
mail_from = config.get('ckan.mail_from')
- body = _(u"Dear %s,") % recipient_name \
- + u"\r\n\r\n%s\r\n\r\n" % body \
- + u"--\r\n%s (%s)" % (sender_name, sender_url)
+ body = add_msg_niceties(recipient_name, body, sender_name, sender_url)
msg = MIMEText(body.encode('utf-8'), 'plain', 'utf-8')
for k, v in headers.items(): msg[k] = v
subject = Header(subject.encode('utf-8'), 'utf-8')
@@ -34,7 +37,9 @@
msg['Date'] = Utils.formatdate(time())
msg['X-Mailer'] = "CKAN %s" % __version__
try:
- server = smtplib.SMTP(config.get('smtp_server', 'localhost'))
+ server = smtplib.SMTP(
+ config.get('test_smtp_server',
+ config.get('smtp_server', 'localhost')))
#server.set_debuglevel(1)
server.sendmail(mail_from, [recipient_email], msg.as_string())
server.quit()
@@ -48,15 +53,12 @@
g.site_title, g.site_url, subject, body, headers=headers)
def mail_user(recipient, subject, body, headers={}):
- if (recipient.email is None) and len(recipient.email):
+ if (recipient.email is None) or not len(recipient.email):
raise MailerException(_("No recipient email address available!"))
mail_recipient(recipient.display_name, recipient.email, subject,
body, headers=headers)
-def make_key():
- return uuid.uuid4().hex[:10]
-
RESET_LINK_MESSAGE = _(
'''You have requested your password on %(site_title)s to be reset.
@@ -65,16 +67,30 @@
%(reset_link)s
''')
-def send_reset_link(user):
- user.reset_key = make_key()
- model.Session.add(user)
- model.Session.commit()
+def make_key():
+ return uuid.uuid4().hex[:10]
+
+def create_reset_key(user):
+ user.reset_key = unicode(make_key())
+ model.repo.commit_and_remove()
+
+def get_reset_link(user):
+ return urljoin(g.site_url,
+ url_for(controller='user',
+ action='perform_reset',
+ id=user.id,
+ key=user.reset_key))
+
+def get_reset_link_body(user):
d = {
- 'reset_link': urljoin(g.site_url, url_for(controller='user',
- action='perform_reset', id=user.id, key=user.reset_key)),
+ 'reset_link': get_reset_link(user),
'site_title': g.site_title
}
- body = RESET_LINK_MESSAGE % d
+ return RESET_LINK_MESSAGE % d
+
+def send_reset_link(user):
+ create_reset_key(user)
+ body = get_reset_link_body(user)
mail_user(user, _('Reset your password'), body)
def verify_reset_link(user, key):
--- a/ckan/templates/user/request_reset.html Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/templates/user/request_reset.html Wed Jul 13 12:26:26 2011 +0100
@@ -14,7 +14,7 @@
Request a password reset
</h2>
- <form id="user-edit" action="" method="post" class="simple-form"
+ <form id="user-password-reset" action="" method="post" class="simple-form"
xmlns:py="http://genshi.edgewall.org/"
xmlns:xi="http://www.w3.org/2001/XInclude"
>
@@ -25,7 +25,7 @@
</fieldset><div>
- ${h.submit('save', _('Reset password'))}
+ ${h.submit('reset', _('Reset password'))}
</div></form></div>
--- a/ckan/tests/functional/test_home.py Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/tests/functional/test_home.py Wed Jul 13 12:26:26 2011 +0100
@@ -19,7 +19,6 @@
def teardown_class(self):
model.repo.rebuild_db()
- @search_related
def test_home_page(self):
offset = url_for('home')
res = self.app.get(offset)
@@ -48,7 +47,6 @@
res = self.app.get(offset)
res.click('Search', index=0)
- @search_related
def test_tags_link(self):
offset = url_for('home')
res = self.app.get(offset)
@@ -77,14 +75,12 @@
assert 'Search - ' in results_page, results_page
assert '>0<' in results_page, results_page
- @search_related
def test_template_footer_end(self):
offset = url_for('home')
res = self.app.get(offset)
assert '<strong>TEST TEMPLATE_FOOTER_END TEST</strong>'
# DISABLED because this is not on home page anymore
- @search_related
def _test_register_new_package(self):
offset = url_for('home')
res = self.app.get(offset)
@@ -94,7 +90,6 @@
assert 'Register a New Package' in results_page, results_page
assert '<input id="Package--title" name="Package--title" size="40" type="text" value="test title">' in results_page, results_page
- @search_related
def test_locale_change(self):
offset = url_for('home')
res = self.app.get(offset)
@@ -105,7 +100,6 @@
finally:
res = res.click('English')
- @search_related
def test_locale_change_with_false_hash(self):
offset = url_for('home')
res = self.app.get(offset)
--- a/ckan/tests/functional/test_user.py Mon Jul 11 11:20:41 2011 +0100
+++ b/ckan/tests/functional/test_user.py Wed Jul 13 12:26:26 2011 +0100
@@ -3,12 +3,17 @@
from ckan.tests import search_related, CreateTestData
from ckan.tests.html_check import HtmlCheckMethods
+from ckan.tests.pylons_controller import PylonsTestCase
+from ckan.tests.mock_mail_server import SmtpServerHarness
import ckan.model as model
from base import FunctionalTestCase
+from ckan.lib.mailer import get_reset_link, create_reset_key
-class TestUserController(FunctionalTestCase, HtmlCheckMethods):
+class TestUserController(FunctionalTestCase, HtmlCheckMethods, PylonsTestCase, SmtpServerHarness):
@classmethod
def setup_class(self):
+ PylonsTestCase.setup_class()
+ SmtpServerHarness.setup_class()
CreateTestData.create()
# make 3 changes, authored by annafan
@@ -26,6 +31,7 @@
@classmethod
def teardown_class(self):
+ SmtpServerHarness.teardown_class()
model.repo.rebuild_db()
def teardown(self):
@@ -444,3 +450,93 @@
# but for some reason this does not work ...
return res
+ def test_request_reset_user_password_link_user_incorrect(self):
+ offset = url_for(controller='user',
+ action='request_reset')
+ res = self.app.get(offset)
+ fv = res.forms['user-password-reset']
+ fv['user'] = 'unknown'
+ res = fv.submit()
+ main_res = self.main_div(res)
+ assert 'No such user: unknown' in main_res, main_res # error
+
+ def test_request_reset_user_password_using_search(self):
+ CreateTestData.create_user(name='larry1', email='kittens at john.com')
+ offset = url_for(controller='user',
+ action='request_reset')
+ res = self.app.get(offset)
+ fv = res.forms['user-password-reset']
+ fv['user'] = 'kittens'
+ res = fv.submit()
+ assert_equal(res.status, 302)
+ assert_equal(res.header_dict['Location'], 'http://localhost/')
+
+ CreateTestData.create_user(name='larry2', fullname='kittens')
+ res = self.app.get(offset)
+ fv = res.forms['user-password-reset']
+ fv['user'] = 'kittens'
+ res = fv.submit()
+ main_res = self.main_div(res)
+ assert '"kittens" matched several users' in main_res, main_res
+ assert 'larry1' not in main_res, main_res
+ assert 'larry2' not in main_res, main_res
+
+ res = self.app.get(offset)
+ fv = res.forms['user-password-reset']
+ fv['user'] = ''
+ res = fv.submit()
+ main_res = self.main_div(res)
+ assert 'No such user:' in main_res, main_res
+
+ res = self.app.get(offset)
+ fv = res.forms['user-password-reset']
+ fv['user'] = 'l'
+ res = fv.submit()
+ main_res = self.main_div(res)
+ assert 'No such user:' in main_res, main_res
+
+ def test_reset_user_password_link(self):
+ # Set password
+ CreateTestData.create_user(name='bob', email='bob at bob.net', password='test1')
+
+ # Set password to something new
+ model.User.by_name(u'bob').password = 'test2'
+ model.repo.commit_and_remove()
+ test2_encoded = model.User.by_name(u'bob').password
+ assert test2_encoded != 'test2'
+ assert model.User.by_name(u'bob').password == test2_encoded
+
+ # Click link from reset password email
+ create_reset_key(model.User.by_name(u'bob'))
+ reset_password_link = get_reset_link(model.User.by_name(u'bob'))
+ offset = reset_password_link.replace('http://test.ckan.net', '')
+ print offset
+ res = self.app.get(offset)
+
+ # Reset password form
+ fv = res.forms['user-reset']
+ fv['password1'] = 'test1'
+ fv['password2'] = 'test1'
+ res = fv.submit('save', status=302)
+
+ # Check a new password is stored
+ assert model.User.by_name(u'bob').password != test2_encoded
+
+ def test_perform_reset_user_password_link_key_incorrect(self):
+ CreateTestData.create_user(name='jack', password='test1')
+ # Make up a key - i.e. trying to hack this
+ user = model.User.by_name(u'jack')
+ offset = url_for(controller='user',
+ action='perform_reset',
+ id=user.id,
+ key='randomness') # i.e. incorrect
+ res = self.app.get(offset, status=403) # error
+
+ def test_perform_reset_user_password_link_user_incorrect(self):
+ # Make up a key - i.e. trying to hack this
+ user = model.User.by_name(u'jack')
+ offset = url_for(controller='user',
+ action='perform_reset',
+ id='randomness', # i.e. incorrect
+ key='randomness')
+ res = self.app.get(offset, status=404)
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/ckan/tests/lib/test_mailer.py Wed Jul 13 12:26:26 2011 +0100
@@ -0,0 +1,112 @@
+import time
+from nose.tools import assert_equal, assert_raises
+from pylons import config
+from email.mime.text import MIMEText
+
+from ckan import model
+from ckan.tests.pylons_controller import PylonsTestCase
+from ckan.tests.mock_mail_server import SmtpServerHarness
+from ckan.lib.mailer import mail_recipient, mail_user, send_reset_link, add_msg_niceties, MailerException, get_reset_link_body, get_reset_link
+from ckan.lib.create_test_data import CreateTestData
+from ckan.lib.base import g
+
+class TestMailer(SmtpServerHarness, PylonsTestCase):
+ @classmethod
+ def setup_class(cls):
+ CreateTestData.create_user(name='bob', email='bob at bob.net')
+ CreateTestData.create_user(name='mary') #NB No email addr provided
+ SmtpServerHarness.setup_class()
+ PylonsTestCase.setup_class()
+
+ @classmethod
+ def teardown_class(cls):
+ SmtpServerHarness.teardown_class()
+ model.repo.rebuild_db()
+
+ def setup(self):
+ self.clear_smtp_messages()
+
+ def mime_encode(self, msg, recipient_name):
+ sender_name = g.site_title
+ sender_url = g.site_url
+ body = add_msg_niceties(recipient_name, msg, sender_name, sender_url)
+ encoded_body = MIMEText(body.encode('utf-8'), 'plain', 'utf-8').get_payload().strip()
+ return encoded_body
+
+ def test_mail_recipient(self):
+ msgs = self.get_smtp_messages()
+ assert_equal(msgs, [])
+
+ # send email
+ test_email = {'recipient_name': 'Bob',
+ 'recipient_email':'bob at bob.net',
+ 'subject': 'Meeting',
+ 'body': 'The meeting is cancelled.',
+ 'headers': {'header1': 'value1'}}
+ mail_recipient(**test_email)
+ time.sleep(0.1)
+
+ # check it went to the mock smtp server
+ msgs = self.get_smtp_messages()
+ assert_equal(len(msgs), 1)
+ msg = msgs[0]
+ assert_equal(msg[1], config['ckan.mail_from'])
+ assert_equal(msg[2], [test_email['recipient_email']])
+ assert test_email['headers'].keys()[0] in msg[3], msg[3]
+ assert test_email['headers'].values()[0] in msg[3], msg[3]
+ assert test_email['subject'] in msg[3], msg[3]
+ expected_body = self.mime_encode(test_email['body'],
+ test_email['recipient_name'])
+ assert expected_body in msg[3], '%r not in %r' % (expected_body, msg[3])
+
+ def test_mail_user(self):
+ msgs = self.get_smtp_messages()
+ assert_equal(msgs, [])
+
+ # send email
+ test_email = {'recipient': model.User.by_name(u'bob'),
+ 'subject': 'Meeting',
+ 'body': 'The meeting is cancelled.',
+ 'headers': {'header1': 'value1'}}
+ mail_user(**test_email)
+ time.sleep(0.1)
+
+ # check it went to the mock smtp server
+ msgs = self.get_smtp_messages()
+ assert_equal(len(msgs), 1)
+ msg = msgs[0]
+ assert_equal(msg[1], config['ckan.mail_from'])
+ assert_equal(msg[2], [model.User.by_name(u'bob').email])
+ assert test_email['headers'].keys()[0] in msg[3], msg[3]
+ assert test_email['headers'].values()[0] in msg[3], msg[3]
+ assert test_email['subject'] in msg[3], msg[3]
+ expected_body = self.mime_encode(test_email['body'],
+ 'bob')
+ assert expected_body in msg[3], '%r not in %r' % (expected_body, msg[3])
+
+ def test_mail_user_without_email(self):
+ # send email
+ test_email = {'recipient': model.User.by_name(u'mary'),
+ 'subject': 'Meeting',
+ 'body': 'The meeting is cancelled.',
+ 'headers': {'header1': 'value1'}}
+ assert_raises(MailerException, mail_user, **test_email)
+
+ def test_send_reset_email(self):
+ # send email
+ send_reset_link(model.User.by_name(u'bob'))
+ time.sleep(0.1)
+
+ # check it went to the mock smtp server
+ msgs = self.get_smtp_messages()
+ assert_equal(len(msgs), 1)
+ msg = msgs[0]
+ assert_equal(msg[1], config['ckan.mail_from'])
+ assert_equal(msg[2], [model.User.by_name(u'bob').email])
+ assert 'Reset' in msg[3], msg[3]
+ test_msg = get_reset_link_body(model.User.by_name(u'bob'))
+ expected_body = self.mime_encode(test_msg,
+ u'bob')
+ assert expected_body in msg[3], '%r not in %r' % (expected_body, msg[3])
+
+ # reset link tested in user functional test
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/ckan/tests/misc/test_mock_mail_server.py Wed Jul 13 12:26:26 2011 +0100
@@ -0,0 +1,33 @@
+import time
+from nose.tools import assert_equal
+from pylons import config
+from email.mime.text import MIMEText
+
+from ckan.tests.pylons_controller import PylonsTestCase
+from ckan.tests.mock_mail_server import SmtpServerHarness
+from ckan.lib.mailer import mail_recipient
+
+class TestMockMailServer(SmtpServerHarness, PylonsTestCase):
+ @classmethod
+ def setup_class(cls):
+ SmtpServerHarness.setup_class()
+ PylonsTestCase.setup_class()
+
+ @classmethod
+ def teardown_class(cls):
+ SmtpServerHarness.teardown_class()
+
+ def test_basic(self):
+ msgs = self.get_smtp_messages()
+ assert_equal(msgs, [])
+
+ test_email = {'recipient_name': 'Bob',
+ 'recipient_email':'bob at bob.net',
+ 'subject': 'Meeting',
+ 'body': 'The meeting is cancelled.',
+ 'headers': {'header1': 'value1'}}
+ mail_recipient(**test_email)
+ time.sleep(0.1)
+
+ msgs = self.get_smtp_messages()
+ assert_equal(len(msgs), 1)
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/ckan/tests/mock_mail_server.py Wed Jul 13 12:26:26 2011 +0100
@@ -0,0 +1,82 @@
+import threading
+import asyncore
+import socket
+from smtpd import SMTPServer
+
+from pylons import config
+
+from ckan.lib.mailer import _mail_recipient
+
+class MockSmtpServer(SMTPServer):
+ '''A mock SMTP server that operates in an asyncore loop'''
+ def __init__(self, host, port):
+ self.msgs = []
+ SMTPServer.__init__(self, (host, port), None)
+
+ def process_message(self, peer, mailfrom, rcpttos, data):
+ self.msgs.append((peer, mailfrom, rcpttos, data))
+
+ def get_smtp_messages(self):
+ return self.msgs
+
+ def clear_smtp_messages(self):
+ self.msgs = []
+
+class MockSmtpServerThread(threading.Thread):
+ '''Runs the mock SMTP server in a thread'''
+ def __init__(self, host, port):
+ self.assert_port_free(host, port)
+ # init thread
+ self._stop_event = threading.Event()
+ self.thread_name = self.__class__
+ threading.Thread.__init__(self, name=self.thread_name)
+ # init smtp server
+ self.server = MockSmtpServer(host, port)
+
+ def assert_port_free(self, host, port):
+ test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ test_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
+ test_socket.getsockopt(socket.SOL_SOCKET,
+ socket.SO_REUSEADDR) | 1 )
+ test_socket.bind((host, port))
+ test_socket.close()
+
+ def run(self):
+ while not self._stop_event.isSet():
+ asyncore.loop(timeout=0.01, count=1)
+
+ def stop(self, timeout=None):
+ self._stop_event.set()
+ threading.Thread.join(self, timeout)
+ self.server.close()
+
+ def get_smtp_messages(self):
+ return self.server.get_smtp_messages()
+
+ def clear_smtp_messages(self):
+ return self.server.clear_smtp_messages()
+
+class SmtpServerHarness(object):
+ '''Derive from this class to run MockSMTP - a test harness that
+ records what email messages are requested to be sent by it.'''
+
+ @classmethod
+ def setup_class(cls):
+ smtp_server = config.get('test_smtp_server') or config['smtp_server']
+ if ':' in smtp_server:
+ host, port = smtp_server.split(':')
+ else:
+ host, port = smtp_server, 25
+ cls.smtp_thread = MockSmtpServerThread(host, int(port))
+ cls.smtp_thread.start()
+
+ @classmethod
+ def teardown_class(cls):
+ cls.smtp_thread.stop()
+
+ def get_smtp_messages(self):
+ return self.smtp_thread.get_smtp_messages()
+
+ def clear_smtp_messages(self):
+ return self.smtp_thread.clear_smtp_messages()
+
--- a/test-core.ini Mon Jul 11 11:20:41 2011 +0100
+++ b/test-core.ini Wed Jul 13 12:26:26 2011 +0100
@@ -47,3 +47,7 @@
# use <strong> so we can check that html is *not* escaped
ckan.template_footer_end = <strong>TEST TEMPLATE_FOOTER_END TEST</strong>
+
+# mailer
+test_smtp_server = localhost:6675
+ckan.mail_from = info at test.ckan.net
Repository URL: https://bitbucket.org/okfn/ckan/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
More information about the ckan-changes
mailing list