# -*- coding: utf-8 -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sqlalchemy.orm.exc import NoResultFound
from kallithea.tests import *
from kallithea.tests.fixture import Fixture
from kallithea.model.db import User, Permission, UserIpMap, UserApiKeys
from kallithea.lib.auth import check_password
from kallithea.model.user import UserModel
from kallithea.model import validators
from kallithea.lib import helpers as h
from kallithea.model.meta import Session
fixture = Fixture()
class TestAdminUsersController(TestController):
test_user_1 = 'testme'
@classmethod
def teardown_class(cls):
if User.get_by_username(cls.test_user_1):
UserModel().delete(cls.test_user_1)
Session().commit()
def test_index(self):
self.log_user()
response = self.app.get(url('users'))
# Test response...
def test_create(self):
username = 'newtestuser'
password = 'test12'
password_confirmation = password
name = 'name'
lastname = 'lastname'
email = 'mail@mail.com'
response = self.app.post(url('users'),
{'username': username,
'password': password,
'password_confirmation': password_confirmation,
'firstname': name,
'active': True,
'lastname': lastname,
'extern_name': 'internal',
'extern_type': 'internal',
'email': email})
self.checkSessionFlash(response, '''Created user <a href="/_admin/users/''')
self.checkSessionFlash(response, '''/edit">%s</a>''' % (username))
new_user = Session().query(User).\
filter(User.username == username).one()
self.assertEqual(new_user.username, username)
self.assertEqual(check_password(password, new_user.password), True)
self.assertEqual(new_user.name, name)
self.assertEqual(new_user.lastname, lastname)
self.assertEqual(new_user.email, email)
response.follow()
response = response.follow()
response.mustcontain("""newtestuser""")
def test_create_err(self):
username = 'new_user'
password = ''
email = 'errmail.com'
response = self.app.post(url('users'), {'username': username,
'name': name,
'active': False,
msg = validators.ValidUsername(False, {})._messages['system_invalid_username']
msg = h.html_escape(msg % {'username': 'new_user'})
response.mustcontain("""<span class="error-message">%s</span>""" % msg)
response.mustcontain("""<span class="error-message">Please enter a value</span>""")
response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
def get_user():
Session().query(User).filter(User.username == username).one()
self.assertRaises(NoResultFound, get_user), 'found user in database'
def test_new(self):
response = self.app.get(url('new_user'))
@parameterized.expand(
[('firstname', {'firstname': 'new_username'}),
('lastname', {'lastname': 'new_username'}),
('admin', {'admin': True}),
('admin', {'admin': False}),
('extern_type', {'extern_type': 'ldap'}),
('extern_type', {'extern_type': None}),
('extern_name', {'extern_name': 'test'}),
('extern_name', {'extern_name': None}),
('active', {'active': False}),
('active', {'active': True}),
('email', {'email': 'some@email.com'}),
# ('new_password', {'new_password': 'foobar123',
# 'password_confirmation': 'foobar123'})
])
def test_update(self, name, attrs):
usr = fixture.create_user(self.test_user_1, password='qweqwe',
email='testme@example.com',
extern_type='internal',
extern_name=self.test_user_1,
skip_if_exists=True)
params = usr.get_api_data()
params.update({'password_confirmation': ''})
params.update({'new_password': ''})
params.update(attrs)
if name == 'email':
params['emails'] = [attrs['email']]
if name == 'extern_type':
#cannot update this via form, expected value is original one
params['extern_type'] = "internal"
if name == 'extern_name':
params['extern_name'] = self.test_user_1
# special case since this user is not
# logged in yet his data is not filled
# so we use creation data
response = self.app.put(url('user', id=usr.user_id), params)
self.checkSessionFlash(response, 'User updated successfully')
updated_user = User.get_by_username(self.test_user_1)
updated_params = updated_user.get_api_data()
updated_params.update({'password_confirmation': ''})
updated_params.update({'new_password': ''})
self.assertEqual(params, updated_params)
def test_delete(self):
username = 'newtestuserdeleteme'
fixture.create_user(name=username)
new_user = Session().query(User)\
.filter(User.username == username).one()
response = self.app.delete(url('user', id=new_user.user_id))
self.checkSessionFlash(response, 'Successfully deleted user')
def test_delete_repo_err(self):
username = 'repoerr'
reponame = 'repoerr_fail'
fixture.create_repo(name=reponame, cur_user=username)
self.checkSessionFlash(response, 'User "%s" still '
'owns 1 repositories and cannot be removed. '
'Switch owners or remove those repositories: '
'%s' % (username, reponame))
response = self.app.delete(url('repo', repo_name=reponame))
self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
def test_delete_repo_group_err(self):
username = 'repogrouperr'
groupname = 'repogroup_fail'
fixture.create_repo_group(name=groupname, cur_user=username)
'owns 1 repository groups and cannot be removed. '
'Switch owners or remove those repository groups: '
'%s' % (username, groupname))
# Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
# rg = RepoGroup.get_by_group_name(group_name=groupname)
# response = self.app.get(url('repos_groups', id=rg.group_id))
response = self.app.delete(url('delete_repo_group', group_name=groupname))
self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
def test_show(self):
response = self.app.get(url('user', id=1))
def test_edit(self):
user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
response = self.app.get(url('edit_user', id=user.user_id))
def test_add_perm_create_repo(self):
perm_none = Permission.get_by_key('hg.create.none')
perm_create = Permission.get_by_key('hg.create.repository')
user = UserModel().create_or_update(username='dummy', password='qwe',
email='dummy', firstname='a',
lastname='b')
uid = user.user_id
try:
#User should have None permission on creation repository
self.assertEqual(UserModel().has_perm(user, perm_none), False)
self.assertEqual(UserModel().has_perm(user, perm_create), False)
response = self.app.post(url('edit_user_perms', id=uid),
params=dict(_method='put',
create_repo_perm=True))
self.assertEqual(UserModel().has_perm(uid, perm_none), False)
self.assertEqual(UserModel().has_perm(uid, perm_create), True)
finally:
UserModel().delete(uid)
def test_revoke_perm_create_repo(self):
params=dict(_method='put'))
self.assertEqual(UserModel().has_perm(uid, perm_none), True)
self.assertEqual(UserModel().has_perm(uid, perm_create), False)
def test_add_perm_fork_repo(self):
perm_none = Permission.get_by_key('hg.fork.none')
perm_fork = Permission.get_by_key('hg.fork.repository')
self.assertEqual(UserModel().has_perm(user, perm_fork), False)
def test_revoke_perm_fork_repo(self):
def test_ips(self):
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
response = self.app.get(url('edit_user_ips', id=user.user_id))
response.mustcontain('All IP addresses are allowed')
@parameterized.expand([
('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),
('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),
('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),
('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),
('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),
('127_bad_ip', 'foobar', 'foobar', True),
def test_add_ip(self, test_name, ip, ip_range, failure):
user_id = user.user_id
response = self.app.put(url('edit_user_ips', id=user_id),
params=dict(new_ip=ip))
if failure:
self.checkSessionFlash(response, 'Please enter a valid IPv4 or IpV6 address')
response = self.app.get(url('edit_user_ips', id=user_id))
response.mustcontain(no=[ip])
response.mustcontain(no=[ip_range])
else:
response.mustcontain(ip)
response.mustcontain(ip_range)
## cleanup
for del_ip in UserIpMap.query().filter(UserIpMap.user_id == user_id).all():
Session().delete(del_ip)
def test_delete_ip(self):
ip = '127.0.0.1/32'
ip_range = '127.0.0.1 - 127.0.0.1'
new_ip = UserModel().add_extra_ip(user_id, ip)
new_ip_id = new_ip.ip_id
self.app.post(url('edit_user_ips', id=user_id),
params=dict(_method='delete', del_ip_id=new_ip_id))
def test_api_keys(self):
response = self.app.get(url('edit_user_api_keys', id=user.user_id))
response.mustcontain(user.api_key)
Status change: