Tests for users API

Makefile

@@ -19,5 +19,5 @@ upload:
 	python bin/release_manager.py $(CIRCLE_SHA1) $(BASE_VERSION) $(FILENAME)
 
 test:
-	nosetests --with-coverage --cover-package=redash tests/*.py
+	nosetests --with-coverage --cover-package=redash tests/
 	#cd rd_ui && grunt test

redash/handlers/users.py

@@ -36,7 +36,6 @@ class UserResource(BaseResource):
 
         req = request.get_json(True)
 
-        # grant admin?
         params = project(req, ('email', 'name', 'password', 'old_password'))
 
         if 'password' in params and 'old_password' not in params:

redash/import_export.py

@@ -7,6 +7,7 @@ from flask.ext.script import Manager
 
 logger = logging.getLogger()
 
+
 class Importer(object):
     def __init__(self, object_mapping=None, data_source=None):
         if object_mapping is None:

redash/permissions.py

@@ -25,5 +25,5 @@ def require_permission(permission):
 
 
 def require_admin_or_owner(object_owner_id):
-    if not (object_owner_id == current_user.id or current_user.has_permission('admin')):
+    if not (int(object_owner_id) == current_user.id or current_user.has_permission('admin')):
         abort(403)

tests/handlers/__init__.py

@@ -0,0 +1,30 @@
+import json
+from contextlib import contextmanager
+
+from tests.factories import user_factory
+from redash.utils import json_dumps
+
+
+@contextmanager
+def authenticated_user(c, user=None):
+    if not user:
+        user = user_factory.create()
+
+    with c.session_transaction() as sess:
+        sess['user_id'] = user.id
+
+    yield user
+
+
+def json_request(method, path, data=None):
+    if data:
+        response = method(path, data=json_dumps(data))
+    else:
+        response = method(path)
+
+    if response.data:
+        response.json = json.loads(response.data)
+    else:
+        response.json = None
+
+    return response

tests/handlers/test_users.py

@@ -0,0 +1,80 @@
+from tests import BaseTestCase
+from tests.factories import user_factory
+from tests.handlers import authenticated_user, json_request
+from redash.wsgi import app
+from redash import models
+
+
+class TestUserListResourcePost(BaseTestCase):
+    def test_returns_403_for_non_admin(self):
+        with app.test_client() as c, authenticated_user(c):
+            rv = c.post("/api/users")
+            self.assertEqual(rv.status_code, 403)
+
+    def test_returns_400_when_missing_fields(self):
+        admin = user_factory.create(groups=['admin', 'default'])
+
+        with app.test_client() as c, authenticated_user(c, user=admin):
+            rv = c.post("/api/users")
+            self.assertEqual(rv.status_code, 400)
+
+            rv = json_request(c.post, '/api/users', data={'name': 'User'})
+
+            self.assertEqual(rv.status_code, 400)
+
+    def test_creates_user(self):
+        admin = user_factory.create(groups=['admin', 'default'])
+
+        with app.test_client() as c, authenticated_user(c, user=admin):
+            test_user = {'name': 'User', 'email': 'user@example.com', 'password': 'test'}
+            rv = json_request(c.post, '/api/users', data=test_user)
+
+            self.assertEqual(rv.status_code, 200)
+            self.assertEqual(rv.json['name'], test_user['name'])
+            self.assertEqual(rv.json['email'], test_user['email'])
+
+
+class TestUserResourcePost(BaseTestCase):
+    def test_returns_403_for_non_admin_changing_not_his_own(self):
+        other_user = user_factory.create()
+        with app.test_client() as c, authenticated_user(c):
+            rv = c.post("/api/users/{}".format(other_user.id), data={"name": "New Name"})
+            self.assertEqual(rv.status_code, 403)
+
+    def test_returns_200_for_non_admin_changing_his_own(self):
+        with app.test_client() as c, authenticated_user(c) as user:
+            rv = json_request(c.post, "/api/users/{}".format(user.id), data={"name": "New Name"})
+            self.assertEqual(rv.status_code, 200)
+
+    def test_returns_200_for_admin_changing_other_user(self):
+        admin = user_factory.create(groups=['admin', 'default'])
+        user = user_factory.create()
+
+        with app.test_client() as c, authenticated_user(c, user=admin):
+            rv = json_request(c.post, "/api/users/{}".format(user.id), data={"name": "New Name"})
+            self.assertEqual(rv.status_code, 200)
+
+    def test_fails_password_change_without_old_password(self):
+        with app.test_client() as c, authenticated_user(c) as user:
+            rv = json_request(c.post, "/api/users/{}".format(user.id), data={"password": "new password"})
+            self.assertEqual(rv.status_code, 403)
+
+    def test_fails_password_change_with_incorrect_old_password(self):
+        with app.test_client() as c, authenticated_user(c) as user:
+            rv = json_request(c.post, "/api/users/{}".format(user.id), data={"password": "new password", "old_password": "wrong"})
+            self.assertEqual(rv.status_code, 403)
+
+    def test_changes_password(self):
+        new_password = "new password"
+        old_password = "old password"
+        with app.test_client() as c, authenticated_user(c) as user:
+            user.hash_password(old_password)
+            user.save()
+
+            rv = json_request(c.post, "/api/users/{}".format(user.id), data={"password": new_password, "old_password": old_password})
+            self.assertEqual(rv.status_code, 200)
+
+            user = models.User.get_by_id(user.id)
+            self.assertTrue(user.verify_password(new_password))
+
+

tests/test_handlers.py

@@ -1,43 +1,18 @@
-from contextlib import contextmanager
 import json
 from unittest import TestCase
 from flask import url_for
 from flask.ext.login import current_user
 from mock import patch
 from tests import BaseTestCase
+from tests.handlers import authenticated_user, json_request
 from tests.factories import dashboard_factory, widget_factory, visualization_factory, query_factory, \
     query_result_factory, user_factory, data_source_factory
 from redash import models, settings
 from redash.wsgi import app
-from redash.utils import json_dumps
 
 
 settings.GOOGLE_APPS_DOMAIN = "example.com"
 
-@contextmanager
-def authenticated_user(c, user=None):
-    if not user:
-        user = user_factory.create()
-
-    with c.session_transaction() as sess:
-        sess['user_id'] = user.id
-
-    yield
-
-
-def json_request(method, path, data=None):
-    if data:
-        response = method(path, data=json_dumps(data))
-    else:
-        response = method(path)
-
-    if response.data:
-        response.json = json.loads(response.data)
-    else:
-        response.json = None
-
-    return response
-
 
 class AuthenticationTestMixin():
     def test_redirects_when_not_authenticated(self):