auth-tokens: fixed tests
This commit is contained in:
parent
b1f8bb4212
commit
d6dfdba2a8
13 changed files with 85 additions and 58 deletions
|
|
@ -22,14 +22,19 @@ import pytest
|
|||
|
||||
from rhodecode.model.meta import Session
|
||||
from rhodecode.model.user import UserModel
|
||||
from rhodecode.model.auth_token import AuthTokenModel
|
||||
from rhodecode.tests import TEST_USER_ADMIN_LOGIN
|
||||
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def testuser_api(request, pylonsapp):
|
||||
cls = request.cls
|
||||
|
||||
# ADMIN USER
|
||||
cls.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
|
||||
cls.apikey = cls.usr.api_key
|
||||
|
||||
# REGULAR USER
|
||||
cls.test_user = UserModel().create_or_update(
|
||||
username='test-api',
|
||||
password='test',
|
||||
|
|
@ -37,6 +42,11 @@ def testuser_api(request, pylonsapp):
|
|||
firstname='first',
|
||||
lastname='last'
|
||||
)
|
||||
# create TOKEN for user, if he doesn't have one
|
||||
if not cls.test_user.api_key:
|
||||
AuthTokenModel().create(
|
||||
user=cls.test_user, description='TEST_USER_TOKEN')
|
||||
|
||||
Session().commit()
|
||||
cls.TEST_USER_LOGIN = cls.test_user.username
|
||||
cls.apikey_regular = cls.test_user.api_key
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ from rhodecode.api.tests.utils import (
|
|||
|
||||
@pytest.mark.usefixtures("testuser_api", "app")
|
||||
class TestClosePullRequest(object):
|
||||
|
||||
@pytest.mark.backends("git", "hg")
|
||||
def test_api_close_pull_request(self, pr_util):
|
||||
pull_request = pr_util.create_pull_request()
|
||||
|
|
|
|||
|
|
@ -53,11 +53,10 @@ class TestApiDeleteRepo(object):
|
|||
}
|
||||
assert_ok(id_, expected, given=response.body)
|
||||
|
||||
def test_api_delete_repo_by_non_admin_no_permission(
|
||||
self, backend, user_regular):
|
||||
def test_api_delete_repo_by_non_admin_no_permission(self, backend):
|
||||
repo = backend.create_repo()
|
||||
id_, params = build_data(
|
||||
user_regular.api_key, 'delete_repo', repoid=repo.repo_name, )
|
||||
self.apikey_regular, 'delete_repo', repoid=repo.repo_name, )
|
||||
response = api_call(self.app, params)
|
||||
expected = 'repository `%s` does not exist' % (repo.repo_name)
|
||||
assert_error(id_, expected, given=response.body)
|
||||
|
|
|
|||
|
|
@ -62,7 +62,6 @@ def get_user(request, apiuser, userid=Optional(OAttr('apiuser'))):
|
|||
"result": {
|
||||
"active": true,
|
||||
"admin": false,
|
||||
"api_key": "api-key",
|
||||
"api_keys": [ list of keys ],
|
||||
"email": "user@example.com",
|
||||
"emails": [
|
||||
|
|
|
|||
|
|
@ -292,13 +292,13 @@ class DbManage(object):
|
|||
TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
|
||||
|
||||
self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
|
||||
TEST_USER_ADMIN_EMAIL, True)
|
||||
TEST_USER_ADMIN_EMAIL, True, api_key=True)
|
||||
|
||||
self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
|
||||
TEST_USER_REGULAR_EMAIL, False)
|
||||
TEST_USER_REGULAR_EMAIL, False, api_key=True)
|
||||
|
||||
self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
|
||||
TEST_USER_REGULAR2_EMAIL, False)
|
||||
TEST_USER_REGULAR2_EMAIL, False, api_key=True)
|
||||
|
||||
def create_ui_settings(self, repo_store_path):
|
||||
"""
|
||||
|
|
@ -561,7 +561,9 @@ class DbManage(object):
|
|||
|
||||
if api_key:
|
||||
log.info('setting a provided api key for the user %s', username)
|
||||
user.api_key = api_key
|
||||
from rhodecode.model.auth_token import AuthTokenModel
|
||||
AuthTokenModel().create(
|
||||
user=user, description='BUILTIN TOKEN')
|
||||
|
||||
def create_default_user(self):
|
||||
log.info('creating default user')
|
||||
|
|
|
|||
|
|
@ -257,6 +257,11 @@ class ExtensionCallback(object):
|
|||
log.debug('Calling extension callback for %s', self._hook_name)
|
||||
|
||||
kwargs_to_pass = dict((key, kwargs[key]) for key in self._kwargs_keys)
|
||||
# backward compat for removed api_key for old hooks. THis was it works
|
||||
# with older rcextensions that require api_key present
|
||||
if self._hook_name in ['CREATE_USER_HOOK', 'DELETE_USER_HOOK']:
|
||||
kwargs_to_pass['api_key'] = '_DEPRECATED_'
|
||||
|
||||
callback = self._get_callback()
|
||||
if callback:
|
||||
return callback(**kwargs_to_pass)
|
||||
|
|
@ -356,7 +361,7 @@ log_create_user = ExtensionCallback(
|
|||
'username', 'full_name_or_username', 'full_contact', 'user_id',
|
||||
'name', 'firstname', 'short_contact', 'admin', 'lastname',
|
||||
'ip_addresses', 'extern_type', 'extern_name',
|
||||
'email', 'api_key', 'api_keys', 'last_login',
|
||||
'email', 'api_keys', 'last_login',
|
||||
'full_name', 'active', 'password', 'emails',
|
||||
'inherit_default_permissions', 'created_by', 'created_on'))
|
||||
|
||||
|
|
@ -367,7 +372,7 @@ log_delete_user = ExtensionCallback(
|
|||
'username', 'full_name_or_username', 'full_contact', 'user_id',
|
||||
'name', 'firstname', 'short_contact', 'admin', 'lastname',
|
||||
'ip_addresses',
|
||||
'email', 'api_key', 'last_login',
|
||||
'email', 'last_login',
|
||||
'full_name', 'active', 'password', 'emails',
|
||||
'inherit_default_permissions', 'deleted_by'))
|
||||
|
||||
|
|
|
|||
|
|
@ -572,6 +572,9 @@ class User(Base, BaseModel):
|
|||
.filter(or_(UserApiKeys.expires == -1,
|
||||
UserApiKeys.expires >= time.time()))\
|
||||
.filter(UserApiKeys.role == UserApiKeys.ROLE_ALL).first()
|
||||
if user_auth_token:
|
||||
user_auth_token = user_auth_token.api_key
|
||||
|
||||
return user_auth_token
|
||||
|
||||
@api_key.setter
|
||||
|
|
@ -962,6 +965,9 @@ class UserApiKeys(Base, BaseModel):
|
|||
|
||||
user = relationship('User', lazy='joined')
|
||||
|
||||
def __unicode__(self):
|
||||
return u"<%s('%s')>" % (self.__class__.__name__, self.role)
|
||||
|
||||
@classmethod
|
||||
def _get_role_name(cls, role):
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ from rhodecode.model.user import UserModel
|
|||
from rhodecode.model.repo_group import RepoGroupModel
|
||||
from rhodecode.model.user_group import UserGroupModel
|
||||
from rhodecode.model.gist import GistModel
|
||||
from rhodecode.model.auth_token import AuthTokenModel
|
||||
|
||||
dn = os.path.dirname
|
||||
FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'tests', 'fixtures')
|
||||
|
|
@ -260,6 +261,11 @@ class Fixture(object):
|
|||
return user
|
||||
form_data = self._get_user_create_params(name, **kwargs)
|
||||
user = UserModel().create(form_data)
|
||||
|
||||
# create token for user
|
||||
AuthTokenModel().create(
|
||||
user=user, description='TEST_USER_TOKEN')
|
||||
|
||||
Session().commit()
|
||||
user = User.get_by_username(user.username)
|
||||
return user
|
||||
|
|
|
|||
|
|
@ -258,39 +258,38 @@ class TestMyAccountController(TestController):
|
|||
usr = self.log_user('test_regular2', 'test12')
|
||||
user = User.get(usr['user_id'])
|
||||
response = self.app.get(url('my_account_auth_tokens'))
|
||||
response.mustcontain(user.api_key)
|
||||
response.mustcontain('expires: never')
|
||||
for token in user.auth_tokens:
|
||||
response.mustcontain(token)
|
||||
response.mustcontain('never')
|
||||
|
||||
@pytest.mark.parametrize("desc, lifetime", [
|
||||
('forever', -1),
|
||||
('5mins', 60*5),
|
||||
('30days', 60*60*24*30),
|
||||
])
|
||||
def test_my_account_add_auth_tokens(self, desc, lifetime):
|
||||
usr = self.log_user('test_regular2', 'test12')
|
||||
user = User.get(usr['user_id'])
|
||||
def test_my_account_add_auth_tokens(self, desc, lifetime, user_util):
|
||||
user = user_util.create_user(password='qweqwe')
|
||||
user_id = user.user_id
|
||||
self.log_user(user.username, 'qweqwe')
|
||||
|
||||
response = self.app.post(url('my_account_auth_tokens'),
|
||||
{'description': desc, 'lifetime': lifetime,
|
||||
'csrf_token': self.csrf_token})
|
||||
assert_session_flash(response, 'Auth token successfully created')
|
||||
try:
|
||||
response = response.follow()
|
||||
user = User.get(usr['user_id'])
|
||||
for auth_token in user.auth_tokens:
|
||||
response.mustcontain(auth_token)
|
||||
finally:
|
||||
for auth_token in UserApiKeys.query().all():
|
||||
Session().delete(auth_token)
|
||||
Session().commit()
|
||||
|
||||
response = response.follow()
|
||||
user = User.get(user_id)
|
||||
for auth_token in user.auth_tokens:
|
||||
response.mustcontain(auth_token)
|
||||
|
||||
def test_my_account_remove_auth_token(self, user_util):
|
||||
user = user_util.create_user(password=self.test_user_1_password)
|
||||
user = user_util.create_user(password='qweqwe')
|
||||
user_id = user.user_id
|
||||
self.log_user(user.username, self.test_user_1_password)
|
||||
self.log_user(user.username, 'qweqwe')
|
||||
|
||||
user = User.get(user_id)
|
||||
keys = user.extra_auth_tokens
|
||||
assert 1 == len(keys)
|
||||
assert 2 == len(keys)
|
||||
|
||||
response = self.app.post(url('my_account_auth_tokens'),
|
||||
{'description': 'desc', 'lifetime': -1,
|
||||
|
|
@ -300,7 +299,7 @@ class TestMyAccountController(TestController):
|
|||
|
||||
user = User.get(user_id)
|
||||
keys = user.extra_auth_tokens
|
||||
assert 2 == len(keys)
|
||||
assert 3 == len(keys)
|
||||
|
||||
response = self.app.post(
|
||||
url('my_account_auth_tokens'),
|
||||
|
|
@ -310,7 +309,7 @@ class TestMyAccountController(TestController):
|
|||
|
||||
user = User.get(user_id)
|
||||
keys = user.extra_auth_tokens
|
||||
assert 1 == len(keys)
|
||||
assert 2 == len(keys)
|
||||
|
||||
def test_valid_change_password(self, user_util):
|
||||
new_password = 'my_new_valid_password'
|
||||
|
|
@ -327,7 +326,7 @@ class TestMyAccountController(TestController):
|
|||
response = self.app.post(url('my_account_password'), form_data).follow()
|
||||
assert 'Successfully updated password' in response
|
||||
|
||||
# check_password depends on user being in session
|
||||
# check_password depends on user being in session
|
||||
Session().add(user)
|
||||
try:
|
||||
assert check_password(new_password, user.password)
|
||||
|
|
|
|||
|
|
@ -573,17 +573,18 @@ class TestAdminUsersController(TestController):
|
|||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
response = self.app.get(
|
||||
url('edit_user_auth_tokens', user_id=user.user_id))
|
||||
response.mustcontain(user.api_key)
|
||||
response.mustcontain('expires: never')
|
||||
for token in user.auth_tokens:
|
||||
response.mustcontain(token)
|
||||
response.mustcontain('never')
|
||||
|
||||
@pytest.mark.parametrize("desc, lifetime", [
|
||||
('forever', -1),
|
||||
('5mins', 60*5),
|
||||
('30days', 60*60*24*30),
|
||||
])
|
||||
def test_add_auth_token(self, desc, lifetime):
|
||||
def test_add_auth_token(self, desc, lifetime, user_util):
|
||||
self.log_user()
|
||||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
|
||||
response = self.app.post(
|
||||
|
|
@ -591,20 +592,15 @@ class TestAdminUsersController(TestController):
|
|||
{'_method': 'put', 'description': desc, 'lifetime': lifetime,
|
||||
'csrf_token': self.csrf_token})
|
||||
assert_session_flash(response, 'Auth token successfully created')
|
||||
try:
|
||||
response = response.follow()
|
||||
user = User.get(user_id)
|
||||
for auth_token in user.auth_tokens:
|
||||
response.mustcontain(auth_token)
|
||||
finally:
|
||||
for api_key in UserApiKeys.query().filter(
|
||||
UserApiKeys.user_id == user_id).all():
|
||||
Session().delete(api_key)
|
||||
Session().commit()
|
||||
|
||||
def test_remove_auth_token(self):
|
||||
response = response.follow()
|
||||
user = User.get(user_id)
|
||||
for auth_token in user.auth_tokens:
|
||||
response.mustcontain(auth_token)
|
||||
|
||||
def test_remove_auth_token(self, user_util):
|
||||
self.log_user()
|
||||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
|
||||
response = self.app.post(
|
||||
|
|
@ -616,7 +612,7 @@ class TestAdminUsersController(TestController):
|
|||
|
||||
# now delete our key
|
||||
keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
|
||||
assert 1 == len(keys)
|
||||
assert 3 == len(keys)
|
||||
|
||||
response = self.app.post(
|
||||
url('edit_user_auth_tokens', user_id=user_id),
|
||||
|
|
@ -624,4 +620,4 @@ class TestAdminUsersController(TestController):
|
|||
'csrf_token': self.csrf_token})
|
||||
assert_session_flash(response, 'Auth token successfully deleted')
|
||||
keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
|
||||
assert 0 == len(keys)
|
||||
assert 2 == len(keys)
|
||||
|
|
|
|||
|
|
@ -366,7 +366,7 @@ class TestLoginController(object):
|
|||
assert ret.email == email
|
||||
assert ret.name == name
|
||||
assert ret.lastname == lastname
|
||||
assert ret.api_key is not None
|
||||
assert ret.auth_tokens is not None
|
||||
assert not ret.admin
|
||||
|
||||
def test_forgot_password_wrong_mail(self):
|
||||
|
|
@ -463,6 +463,7 @@ class TestLoginController(object):
|
|||
|
||||
if test_name == 'proper_auth_token':
|
||||
auth_token = user_admin.api_key
|
||||
assert auth_token
|
||||
|
||||
with fixture.anon_access(False):
|
||||
self.app.get(url(controller='changeset',
|
||||
|
|
|
|||
|
|
@ -71,16 +71,22 @@ def test_create_admin_and_prompt_sets_the_api_key(db_manage):
|
|||
assert create_user.call_args[1]['api_key'] == 'testkey'
|
||||
|
||||
|
||||
def test_create_user_sets_the_api_key(db_manage):
|
||||
@pytest.mark.parametrize('add_keys', [True, False])
|
||||
def test_create_user_sets_the_api_key(db_manage, add_keys):
|
||||
username = 'test_add_keys_{}'.format(add_keys)
|
||||
db_manage.create_user(
|
||||
'test', 'testpassword', 'test@example.com',
|
||||
api_key='testkey')
|
||||
username, 'testpassword', 'test@example.com',
|
||||
api_key=add_keys)
|
||||
|
||||
user = db.User.get_by_username('test')
|
||||
assert user.api_key == 'testkey'
|
||||
user = db.User.get_by_username(username)
|
||||
if add_keys:
|
||||
assert 2 == len(user.auth_tokens)
|
||||
else:
|
||||
# only feed token
|
||||
assert 1 == len(user.auth_tokens)
|
||||
|
||||
|
||||
def test_create_user_without_api_key(db_manage):
|
||||
db_manage.create_user('test', 'testpassword', 'test@example.com')
|
||||
user = db.User.get_by_username('test')
|
||||
assert user.api_key
|
||||
assert user.api_key is None
|
||||
|
|
|
|||
|
|
@ -109,15 +109,12 @@ def test_get_api_data_replaces_secret_data_by_default(test_user):
|
|||
api_key_length = 40
|
||||
expected_replacement = '*' * api_key_length
|
||||
|
||||
assert api_data['api_key'] == expected_replacement
|
||||
for key in api_data['api_keys']:
|
||||
assert key == expected_replacement
|
||||
|
||||
|
||||
def test_get_api_data_includes_secret_data_if_activated(test_user):
|
||||
api_data = test_user.get_api_data(include_secrets=True)
|
||||
|
||||
assert api_data['api_key'] == test_user.api_key
|
||||
assert api_data['api_keys'] == test_user.auth_tokens
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue