a bit of code polish and an attempt to speed up the tests

This commit is contained in:
Matt Wright
2012-09-16 16:43:28 -04:00
parent 96f2be056d
commit 1f8fb48727
6 changed files with 58 additions and 43 deletions
+1 -13
View File
@@ -12,13 +12,10 @@
import base64
import hashlib
import hmac
import os
from contextlib import contextmanager
from datetime import datetime, timedelta
from functools import wraps
from flask import url_for, flash, current_app, request, session, redirect, \
render_template
from flask import url_for, flash, current_app, request, session, render_template
from flask.ext.login import login_user as _login_user, \
logout_user as _logout_user
from flask.ext.mail import Message
@@ -26,7 +23,6 @@ from flask.ext.principal import Identity, AnonymousIdentity, identity_changed
from itsdangerous import BadSignature, SignatureExpired
from werkzeug.local import LocalProxy
from .core import current_user
from .signals import user_registered, reset_password_instructions_sent, \
login_instructions_sent
@@ -38,14 +34,6 @@ _datastore = LocalProxy(lambda: _security.datastore)
_pwd_context = LocalProxy(lambda: _security.pwd_context)
def anonymous_user_required(f):
@wraps(f)
def wrapper(*args, **kwargs):
if current_user.is_authenticated():
return redirect(get_url(_security.post_login_view))
return f(*args, **kwargs)
return wrapper
def login_user(user, remember=True):
"""Performs the login and sends the appropriate signal."""
+1 -1
View File
@@ -174,7 +174,7 @@ def confirm_email(token):
"""View function which handles a email confirmation request."""
expired, invalid, user = confirm_email_token_status(token)
print expired, invalid, user
if invalid:
do_flash(*get_message('INVALID_CONFIRMATION_TOKEN'))
if expired:
+41 -18
View File
@@ -199,11 +199,22 @@ class DefaultSecurityTests(SecurityTest):
self.assertNotIn('BadSignature', r.data)
class ConfiguredSecurityTests(SecurityTest):
class ConfiguredPasswordHashSecurityTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORD_HASH': 'bcrypt',
'SECURITY_PASSWORD_SALT': 'so-salty',
'USER_COUNT': 1
}
def test_authenticate(self):
r = self.authenticate(endpoint="/login")
self.assertIn('Home Page', r.data)
class ConfiguredSecurityTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_REGISTERABLE': True,
'SECURITY_LOGOUT_URL': '/custom_logout',
'SECURITY_LOGIN_URL': '/custom_login',
@@ -265,6 +276,7 @@ class BadConfiguredSecurityTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORD_HASH': 'bcrypt',
'USER_COUNT': 1
}
def test_bad_configuration_raises_runtimer_error(self):
@@ -273,7 +285,8 @@ class BadConfiguredSecurityTests(SecurityTest):
class RegisterableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_REGISTERABLE': True
'SECURITY_REGISTERABLE': True,
'USER_COUNT': 1
}
def test_register_valid_user(self):
@@ -286,7 +299,8 @@ class RegisterableTests(SecurityTest):
class ConfirmableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True
'SECURITY_REGISTERABLE': True,
'USER_COUNT': 1
}
def test_login_before_confirmation(self):
@@ -345,7 +359,8 @@ class ExpiredConfirmationTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
'SECURITY_CONFIRM_EMAIL_WITHIN': '1 seconds'
'SECURITY_CONFIRM_EMAIL_WITHIN': '1 milliseconds',
'USER_COUNT': 1
}
def test_expired_confirmation_token_sends_email(self):
@@ -355,7 +370,7 @@ class ExpiredConfirmationTest(SecurityTest):
self.register(e)
token = registrations[0]['confirm_token']
time.sleep(3)
time.sleep(1.25)
with self.app.extensions['mail'].record_messages() as outbox:
r = self.client.get('/confirm/' + token, follow_redirects=True)
@@ -372,7 +387,8 @@ class LoginWithoutImmediateConfirmTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
'SECURITY_LOGIN_WITHOUT_CONFIRMATION': True
'SECURITY_LOGIN_WITHOUT_CONFIRMATION': True,
'USER_COUNT': 1
}
def test_register_valid_user_automatically_signs_in(self):
@@ -441,7 +457,7 @@ class ExpiredResetPasswordTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_RESET_PASSWORD_WITHIN': '1 seconds'
'SECURITY_RESET_PASSWORD_WITHIN': '1 milliseconds'
}
def test_reset_password_with_expired_token(self):
@@ -451,7 +467,7 @@ class ExpiredResetPasswordTest(SecurityTest):
follow_redirects=True)
t = requests[0]['token']
time.sleep(2)
time.sleep(1)
r = self.client.post('/reset/' + t, data={
'password': 'newpassword',
@@ -464,7 +480,8 @@ class ExpiredResetPasswordTest(SecurityTest):
class TrackableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_TRACKABLE': True
'SECURITY_TRACKABLE': True,
'USER_COUNT': 1
}
def test_did_track(self):
@@ -485,7 +502,7 @@ class TrackableTests(SecurityTest):
class PasswordlessTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True,
'SECURITY_PASSWORDLESS': True
}
def test_login_request_for_inactive_user(self):
@@ -544,7 +561,8 @@ class ExpiredLoginTokenTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True,
'SECURITY_LOGIN_WITHIN': '1 seconds'
'SECURITY_LOGIN_WITHIN': '1 milliseconds',
'USER_COUNT': 1
}
def test_expired_login_token_sends_email(self):
@@ -554,19 +572,19 @@ class ExpiredLoginTokenTests(SecurityTest):
self.client.post('/login', data=dict(email=e), follow_redirects=True)
token = requests[0]['login_token']
time.sleep(3)
time.sleep(1.25)
with self.app.extensions['mail'].record_messages() as outbox:
r = self.client.get('/login/' + token, follow_redirects=True)
self.assertEqual(len(outbox), 1)
self.assertIn(e, outbox[0].html)
self.assertNotIn(token, outbox[0].html)
expire_text = self.AUTH_CONFIG['SECURITY_LOGIN_WITHIN']
msg = self.app.config['SECURITY_MSG_LOGIN_EXPIRED'][0] % dict(within=expire_text, email=e)
self.assertIn(msg, r.data)
self.assertEqual(len(outbox), 1)
self.assertIn(e, outbox[0].html)
self.assertNotIn(token, outbox[0].html)
class MongoEngineSecurityTests(DefaultSecurityTests):
@@ -609,6 +627,7 @@ class AsyncMailTaskTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'USER_COUNT': 1
}
def setUp(self):
@@ -620,12 +639,16 @@ class AsyncMailTaskTests(SecurityTest):
def send_email(msg):
self.mail_sent = True
self.client.post('/reset', data=dict(email='joe@lp.com'))
self.client.post('/reset', data=dict(email='matt@lp.com'))
self.assertTrue(self.mail_sent)
class NoBlueprintTests(SecurityTest):
AUTH_CONFIG = {
'USER_COUNT': 1
}
def _create_app(self, auth_config):
return super(NoBlueprintTests, self)._create_app(auth_config, False)
@@ -635,6 +658,6 @@ class NoBlueprintTests(SecurityTest):
def test_http_auth_without_blueprint(self):
r = self._get('/http', headers={
'Authorization': 'Basic ' + base64.b64encode("joe@lp.com:password")
'Authorization': 'Basic ' + base64.b64encode("matt@lp.com:password")
})
self.assertIn('HTTP Authentication', r.data)
+13 -9
View File
@@ -113,19 +113,23 @@ def create_roles():
ds.create_role(name=role)
ds.commit()
def create_users():
for u in (('matt@lp.com', 'password', ['admin'], True),
('joe@lp.com', 'password', ['editor'], True),
('dave@lp.com', 'password', ['admin', 'editor'], True),
('jill@lp.com', 'password', ['author'], True),
('tiya@lp.com', 'password', [], False)):
ds.create_user(email=u[0], password=encrypt_password(u[1]),
def create_users(count=None):
users = [('matt@lp.com', 'password', ['admin'], True),
('joe@lp.com', 'password', ['editor'], True),
('dave@lp.com', 'password', ['admin', 'editor'], True),
('jill@lp.com', 'password', ['author'], True),
('tiya@lp.com', 'password', [], False)]
count = count or len(users)
for u in users[:count]:
pw = encrypt_password(u[1])
ds.create_user(email=u[0], password=pw,
roles=u[2], active=u[3])
ds.commit()
def populate_data():
def populate_data(user_count=None):
create_roles()
create_users()
create_users(user_count)
def add_context_processors(s):
@s.context_processor
+1 -1
View File
@@ -42,7 +42,7 @@ def create_app(config):
def before_first_request():
User.drop_collection()
Role.drop_collection()
populate_data()
populate_data(app.config.get('USER_COUNT', None))
app.security = Security(app, MongoEngineUserDatastore(db, User, Role))
+1 -1
View File
@@ -48,7 +48,7 @@ def create_app(config, register_blueprint=True):
def before_first_request():
db.drop_all()
db.create_all()
populate_data()
populate_data(app.config.get('USER_COUNT', None))
app.security = Security(app, SQLAlchemyUserDatastore(db, User, Role),
register_blueprint=register_blueprint)