mirror of
https://github.com/wassname/flask-security.git
synced 2026-07-12 00:50:18 +08:00
Full test coverage!
This commit is contained in:
+40
-2
@@ -18,6 +18,7 @@ from flask.ext.security.datastore import SQLAlchemyUserDatastore, \
|
||||
MongoEngineUserDatastore
|
||||
from flask.ext.security.decorators import http_auth_required, \
|
||||
auth_token_required
|
||||
from flask.ext.security.exceptions import RoleNotFoundError
|
||||
|
||||
|
||||
def create_roles():
|
||||
@@ -45,12 +46,12 @@ def create_app(auth_config):
|
||||
app.debug = True
|
||||
app.config['SECRET_KEY'] = 'secret'
|
||||
|
||||
app.mail = Mail(app)
|
||||
|
||||
if auth_config:
|
||||
for key, value in auth_config.items():
|
||||
app.config[key] = value
|
||||
|
||||
app.mail = Mail(app)
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
return render_template('index.html', content='Home Page')
|
||||
@@ -115,6 +116,43 @@ def create_app(auth_config):
|
||||
def unauthorized():
|
||||
return render_template('unauthorized.html')
|
||||
|
||||
@app.route('/coverage/add_role_to_user')
|
||||
def add_role_to_user():
|
||||
ds = app.security.datastore
|
||||
u = ds.find_user(email='joe@lp.com')
|
||||
r = ds.find_role('admin')
|
||||
ds.add_role_to_user(u, r)
|
||||
return 'success'
|
||||
|
||||
@app.route('/coverage/remove_role_from_user')
|
||||
def remove_role_from_user():
|
||||
ds = app.security.datastore
|
||||
u = ds.find_user(email='matt@lp.com')
|
||||
ds.remove_role_from_user(u, 'admin')
|
||||
return 'success'
|
||||
|
||||
@app.route('/coverage/deactivate_user')
|
||||
def deactivate_user():
|
||||
ds = app.security.datastore
|
||||
u = ds.find_user(email='matt@lp.com')
|
||||
ds.deactivate_user(u)
|
||||
return 'success'
|
||||
|
||||
@app.route('/coverage/activate_user')
|
||||
def activate_user():
|
||||
ds = app.security.datastore
|
||||
u = ds.find_user(email='tiya@lp.com')
|
||||
ds.activate_user(u)
|
||||
return 'success'
|
||||
|
||||
@app.route('/coverage/invalid_role')
|
||||
def invalid_role():
|
||||
ds = app.security.datastore
|
||||
try:
|
||||
ds.find_role('bogus')
|
||||
except RoleNotFoundError:
|
||||
return 'success'
|
||||
|
||||
return app
|
||||
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
{% include "_messages.html" %}
|
||||
{% include "_nav.html" %}
|
||||
<h1>Register</h1>
|
||||
<form action="{{ url_for('flask_security.register') }}" method="POST" name="register_form">
|
||||
{{ register_user_form.hidden_tag() }}
|
||||
{{ register_user_form.email.label }} {{ register_user_form.email }}<br/>
|
||||
|
||||
@@ -20,7 +20,7 @@ _security = LocalProxy(lambda: current_app.extensions['security'])
|
||||
|
||||
class UserDatastore(object):
|
||||
"""Abstracted user datastore. Always extend this class and implement the
|
||||
:attr:`_save_model`, :attr:`_do_with_id`, :attr:`_do_find_user`, and
|
||||
:attr:`_save_model`, :attr:`_delete_model`, :attr:`_do_find_user`, and
|
||||
:attr:`_do_find_role` methods.
|
||||
|
||||
:param db: An instance of a configured databse manager from a Flask
|
||||
@@ -41,10 +41,6 @@ class UserDatastore(object):
|
||||
raise NotImplementedError(
|
||||
"User datastore does not implement _delete_model method")
|
||||
|
||||
def _do_with_id(self, id):
|
||||
raise NotImplementedError(
|
||||
"User datastore does not implement _do_with_id method")
|
||||
|
||||
def _do_find_user(self, **kwargs):
|
||||
raise NotImplementedError(
|
||||
"User datastore does not implement _do_find_user method")
|
||||
@@ -65,11 +61,9 @@ class UserDatastore(object):
|
||||
user.roles.remove(role)
|
||||
return user
|
||||
|
||||
def _do_toggle_active(self, user, active=None):
|
||||
def _do_toggle_active(self, user, active):
|
||||
user = self.find_user(email=user.email)
|
||||
if active is None:
|
||||
user.active = not user.active
|
||||
elif active != user.active:
|
||||
if active != user.active:
|
||||
user.active = active
|
||||
return user
|
||||
|
||||
@@ -80,30 +74,13 @@ class UserDatastore(object):
|
||||
return self._do_toggle_active(user, True)
|
||||
|
||||
def _prepare_role_modify_args(self, user, role):
|
||||
if isinstance(role, self.role_model):
|
||||
role = role.name
|
||||
|
||||
role = role.name if isinstance(role, self.role_model) else role
|
||||
return self.find_user(email=user.email), self.find_role(role)
|
||||
|
||||
def _prepare_create_role_args(self, kwargs):
|
||||
if kwargs['name'] is None:
|
||||
raise exceptions.RoleCreationError("Missing name argument")
|
||||
|
||||
return kwargs
|
||||
|
||||
def _prepare_create_user_args(self, kwargs):
|
||||
email = kwargs.get('email', None)
|
||||
password = kwargs.get('password', None)
|
||||
|
||||
def _prepare_create_user_args(self, **kwargs):
|
||||
kwargs.setdefault('active', True)
|
||||
kwargs.setdefault('roles', _security.default_roles)
|
||||
|
||||
if email is None:
|
||||
raise exceptions.UserCreationError('Missing email argument')
|
||||
|
||||
if password is None:
|
||||
raise exceptions.UserCreationError('Missing password argument')
|
||||
|
||||
roles = kwargs.get('roles', [])
|
||||
|
||||
for i, role in enumerate(roles):
|
||||
@@ -124,15 +101,6 @@ class UserDatastore(object):
|
||||
|
||||
return kwargs
|
||||
|
||||
def with_id(self, id):
|
||||
"""Returns a user with the specified ID.
|
||||
|
||||
:param id: User ID"""
|
||||
user = self._do_with_id(id)
|
||||
if user:
|
||||
return user
|
||||
raise exceptions.UserIdNotFoundError()
|
||||
|
||||
def find_user(self, **kwargs):
|
||||
"""Returns a user based on the specified identifier.
|
||||
|
||||
@@ -158,7 +126,7 @@ class UserDatastore(object):
|
||||
|
||||
:param name: Role name
|
||||
"""
|
||||
role = self.role_model(**self._prepare_create_role_args(kwargs))
|
||||
role = self.role_model(**kwargs)
|
||||
return self._save_model(role)
|
||||
|
||||
def create_user(self, **kwargs):
|
||||
@@ -168,7 +136,7 @@ class UserDatastore(object):
|
||||
:param password: Unencrypted password
|
||||
:param active: The optional active state
|
||||
"""
|
||||
user = self.user_model(**self._prepare_create_user_args(kwargs))
|
||||
user = self.user_model(**self._prepare_create_user_args(**kwargs))
|
||||
return self._save_model(user)
|
||||
|
||||
def delete_user(self, user):
|
||||
@@ -260,9 +228,6 @@ class SQLAlchemyUserDatastore(UserDatastore):
|
||||
self.db.session.delete(model)
|
||||
self.db.session.commit()
|
||||
|
||||
def _do_with_id(self, id):
|
||||
return self.user_model.query.get(id)
|
||||
|
||||
def _do_find_user(self, **kwargs):
|
||||
return self.user_model.query.filter_by(**kwargs).first()
|
||||
|
||||
@@ -307,12 +272,6 @@ class MongoEngineUserDatastore(UserDatastore):
|
||||
def _delete_model(self, model):
|
||||
model.delete()
|
||||
|
||||
def _do_with_id(self, id):
|
||||
try:
|
||||
return self.user_model.objects.get(id=id)
|
||||
except:
|
||||
return None
|
||||
|
||||
def _do_find_user(self, **kwargs):
|
||||
return self.user_model.objects(**kwargs).first()
|
||||
|
||||
|
||||
@@ -40,12 +40,6 @@ class RoleNotFoundError(SecurityError):
|
||||
"""
|
||||
|
||||
|
||||
class UserIdNotFoundError(SecurityError):
|
||||
"""Raised by a user datastore when there is an attempt to find a user by
|
||||
ID and the user is not found.
|
||||
"""
|
||||
|
||||
|
||||
class UserDatastoreError(SecurityError):
|
||||
"""Raised when a user datastore experiences an unexpected error
|
||||
"""
|
||||
|
||||
+4
-15
@@ -49,11 +49,12 @@ def login_user(user, remember=True):
|
||||
user.last_login_at = old_current or new_current
|
||||
user.current_login_at = new_current
|
||||
|
||||
old_current, new_current = user.current_login_ip, request.remote_addr
|
||||
remote_addr = request.remote_addr or 'untrackable'
|
||||
old_current, new_current = user.current_login_ip, remote_addr
|
||||
user.last_login_ip = old_current or new_current
|
||||
user.current_login_ip = new_current
|
||||
|
||||
user.login_count = user.login_count + 1 if user.login_count else 0
|
||||
user.login_count = user.login_count + 1 if user.login_count else 1
|
||||
|
||||
_datastore._save_model(user)
|
||||
|
||||
@@ -102,11 +103,6 @@ def md5(data):
|
||||
return hashlib.md5(data).hexdigest()
|
||||
|
||||
|
||||
def generate_token():
|
||||
"""Generate an arbitrary URL safe token."""
|
||||
return base64.urlsafe_b64encode(os.urandom(30))
|
||||
|
||||
|
||||
def do_flash(message, category=None):
|
||||
"""Flash a message depending on if the `FLASH_MESSAGES` configuration
|
||||
value is set.
|
||||
@@ -216,16 +212,9 @@ def send_mail(subject, recipient, template, context=None):
|
||||
:param template: The name of the email template
|
||||
:param context: The context to render the template with
|
||||
"""
|
||||
mail = current_app.extensions.get('mail', None)
|
||||
current_app.logger.debug('%s' % current_app.extensions)
|
||||
|
||||
if mail is None:
|
||||
raise RuntimeError('You need to install and configure the '
|
||||
'Flask-Mail extension in order to send '
|
||||
'emails with Flask-Security')
|
||||
|
||||
from flask.ext.mail import Message
|
||||
|
||||
mail = current_app.extensions.get('mail')
|
||||
context = context or {}
|
||||
|
||||
msg = Message(subject,
|
||||
|
||||
@@ -145,10 +145,6 @@ def send_confirmation():
|
||||
|
||||
do_flash(msg, cat)
|
||||
|
||||
else:
|
||||
for key, value in form.errors.items():
|
||||
do_flash(value[0], 'error')
|
||||
|
||||
return render_template('security/confirmations/new.html',
|
||||
reset_confirmation_form=form)
|
||||
|
||||
|
||||
+3
-2
@@ -23,10 +23,11 @@ class SecurityTest(TestCase):
|
||||
content_type=content_type or 'text/html',
|
||||
headers=headers)
|
||||
|
||||
def _post(self, route, data=None, content_type=None, follow_redirects=True):
|
||||
def _post(self, route, data=None, content_type=None, follow_redirects=True, headers=None):
|
||||
return self.client.post(route, data=data,
|
||||
follow_redirects=follow_redirects,
|
||||
content_type=content_type or 'application/x-www-form-urlencoded')
|
||||
content_type=content_type or 'application/x-www-form-urlencoded',
|
||||
headers=headers)
|
||||
|
||||
def register(self, email, password='password'):
|
||||
data = dict(email=email, password=password, password_confirm=password)
|
||||
|
||||
@@ -36,6 +36,8 @@ class DefaultSecurityTests(SecurityTest):
|
||||
r = self._get('/login')
|
||||
self.assertIn('Login Page', r.data)
|
||||
|
||||
|
||||
|
||||
def test_authenticate(self):
|
||||
r = self.authenticate()
|
||||
self.assertIn('Hello matt@lp.com', r.data)
|
||||
@@ -221,6 +223,10 @@ class ConfiguredSecurityTests(SecurityTest):
|
||||
r = self.logout(endpoint="/custom_logout")
|
||||
self.assertIn('Post Logout', r.data)
|
||||
|
||||
def test_register_view(self):
|
||||
r = self._get('/register')
|
||||
self.assertIn('<h1>Register</h1>', r.data)
|
||||
|
||||
def test_register(self):
|
||||
data = dict(email='dude@lp.com',
|
||||
password='password',
|
||||
@@ -432,7 +438,57 @@ class ExpiredResetPasswordTest(SecurityTest):
|
||||
self.assertIn('You did not reset your password within', r.data)
|
||||
|
||||
|
||||
class TrackableTests(SecurityTest):
|
||||
|
||||
AUTH_CONFIG = {
|
||||
'SECURITY_TRACKABLE': True
|
||||
}
|
||||
|
||||
def test_did_track(self):
|
||||
e = 'matt@lp.com'
|
||||
self.authenticate(email=e)
|
||||
self.logout()
|
||||
self.authenticate(email=e)
|
||||
|
||||
with self.app.test_request_context('/profile'):
|
||||
user = self.app.security.datastore.find_user(email=e)
|
||||
self.assertIsNotNone(user.last_login_at)
|
||||
self.assertIsNotNone(user.current_login_at)
|
||||
self.assertEquals('untrackable', user.last_login_ip)
|
||||
self.assertEquals('untrackable', user.current_login_ip)
|
||||
self.assertEquals(2, user.login_count)
|
||||
|
||||
|
||||
class MongoEngineSecurityTests(DefaultSecurityTests):
|
||||
|
||||
def _create_app(self, auth_config):
|
||||
return app.create_mongoengine_app(auth_config)
|
||||
|
||||
|
||||
class DefaultDatastoreTests(SecurityTest):
|
||||
|
||||
def test_add_role_to_user(self):
|
||||
r = self._get('/coverage/add_role_to_user')
|
||||
self.assertIn('success', r.data)
|
||||
|
||||
def test_remove_role_from_user(self):
|
||||
r = self._get('/coverage/remove_role_from_user')
|
||||
self.assertIn('success', r.data)
|
||||
|
||||
def test_activate_user(self):
|
||||
r = self._get('/coverage/activate_user')
|
||||
self.assertIn('success', r.data)
|
||||
|
||||
def test_deactivate_user(self):
|
||||
r = self._get('/coverage/deactivate_user')
|
||||
self.assertIn('success', r.data)
|
||||
|
||||
def test_invalid_role(self):
|
||||
r = self._get('/coverage/invalid_role')
|
||||
self.assertIn('success', r.data)
|
||||
|
||||
|
||||
class MongoEngineDatastoreTests(DefaultDatastoreTests):
|
||||
|
||||
def _create_app(self, auth_config):
|
||||
return app.create_mongoengine_app(auth_config)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import unittest
|
||||
|
||||
from flask_security import RoleMixin, UserMixin, AnonymousUser
|
||||
from flask_security.datastore import UserDatastore
|
||||
|
||||
|
||||
class Role(RoleMixin):
|
||||
@@ -41,3 +42,13 @@ class SecurityEntityTests(unittest.TestCase):
|
||||
au = AnonymousUser()
|
||||
self.assertEqual(0, len(au.roles))
|
||||
self.assertFalse(au.has_role('admin'))
|
||||
|
||||
|
||||
class UserDatastoreTests(unittest.TestCase):
|
||||
|
||||
def test_unimplemented(self):
|
||||
ds = UserDatastore(None, None, None)
|
||||
self.assertRaises(NotImplementedError, ds._save_model, None)
|
||||
self.assertRaises(NotImplementedError, ds._delete_model, None)
|
||||
self.assertRaises(NotImplementedError, ds._do_find_user)
|
||||
self.assertRaises(NotImplementedError, ds._do_find_role)
|
||||
|
||||
Reference in New Issue
Block a user