from rhodecode.tests import *
from rhodecode.model.db import UsersGroup
from rhodecode.model.db import UsersGroup, UsersGroupToPerm, Permission
TEST_USERS_GROUP = 'admins_test'
class TestAdminUsersGroupsController(TestController):
def test_index(self):
response = self.app.get(url('users_groups'))
# Test response...
def test_index_as_xml(self):
response = self.app.get(url('formatted_users_groups', format='xml'))
def test_create(self):
self.log_user()
users_group_name = TEST_USERS_GROUP
response = self.app.post(url('users_groups'),
{'users_group_name':users_group_name,
{'users_group_name': users_group_name,
'active':True})
response.follow()
self.checkSessionFlash(response,
'created users group %s' % TEST_USERS_GROUP)
def test_new(self):
response = self.app.get(url('new_users_group'))
def test_new_as_xml(self):
response = self.app.get(url('formatted_new_users_group', format='xml'))
def test_update(self):
response = self.app.put(url('users_group', id=1))
def test_update_browser_fakeout(self):
response = self.app.post(url('users_group', id=1),
params=dict(_method='put'))
def test_delete(self):
users_group_name = TEST_USERS_GROUP + 'another'
'created users group %s' % users_group_name)
gr = self.Session.query(UsersGroup)\
.filter(UsersGroup.users_group_name ==
users_group_name).one()
response = self.app.delete(url('users_group', id=gr.users_group_id))
users_group_name).scalar()
self.assertEqual(gr, None)
def test_enable_repository_read_on_group(self):
users_group_name = TEST_USERS_GROUP + 'another2'
ug = UsersGroup.get_by_group_name(users_group_name)
response = self.app.put(url('users_group_perm', id=ug.users_group_id),
{'create_repo_perm': True})
p = Permission.get_by_key('hg.create.repository')
# check if user has this perm
perms = UsersGroupToPerm.query()\
.filter(UsersGroupToPerm.users_group == ug).all()
perms = [[x.__dict__['users_group_id'],
x.__dict__['permission_id'],] for x in perms]
self.assertEqual(
perms,
[[ug.users_group_id, p.permission_id]]
)
# DELETE !
ugid = ug.users_group_id
response = self.app.delete(url('users_group', id=ug.users_group_id))
response = response.follow()
.filter(UsersGroupToPerm.users_group_id == ugid).all()
[]
def test_delete_browser_fakeout(self):
params=dict(_method='delete'))
def test_show(self):
response = self.app.get(url('users_group', id=1))
def test_show_as_xml(self):
response = self.app.get(url('formatted_users_group', id=1, format='xml'))
def test_edit(self):
response = self.app.get(url('edit_users_group', id=1))
def test_edit_as_xml(self):
response = self.app.get(url('formatted_edit_users_group', id=1, format='xml'))
def test_assign_members(self):
pass
def test_add_create_permission(self):
def test_revoke_members(self):
import os
import unittest
from rhodecode.model.repos_group import ReposGroupModel
from rhodecode.model.repo import RepoModel
from rhodecode.model.db import RepoGroup, User, Notification, UserNotification, \
UsersGroup, UsersGroupMember, Permission
from sqlalchemy.exc import IntegrityError
from rhodecode.model.user import UserModel
from rhodecode.model.meta import Session
from rhodecode.model.notification import NotificationModel
from rhodecode.model.users_group import UsersGroupModel
from rhodecode.lib.auth import AuthUser
def _make_group(path, desc='desc', parent_id=None,
skip_if_exists=False):
gr = RepoGroup.get_by_group_name(path)
if gr and skip_if_exists:
return gr
gr = ReposGroupModel().create(path, desc, parent_id)
class TestReposGroups(unittest.TestCase):
def setUp(self):
self.g1 = _make_group('test1', skip_if_exists=True)
Session.commit()
self.g2 = _make_group('test2', skip_if_exists=True)
self.g3 = _make_group('test3', skip_if_exists=True)
def tearDown(self):
print 'out'
def __check_path(self, *path):
"""
Checks the path for existance !
path = [TESTS_TMP_PATH] + list(path)
path = os.path.join(*path)
return os.path.isdir(path)
def _check_folders(self):
print os.listdir(TESTS_TMP_PATH)
def __delete_group(self, id_):
ReposGroupModel().delete(id_)
def __update_group(self, id_, path, desc='desc', parent_id=None):
form_data = dict(
group_name=path,
group_description=desc,
group_parent_id=parent_id,
perms_updates=[],
perms_new=[]
gr = ReposGroupModel().update(id_, form_data)
def test_create_group(self):
g = _make_group('newGroup')
self.assertEqual(g.full_path, 'newGroup')
self.assertTrue(self.__check_path('newGroup'))
def test_create_same_name_group(self):
self.assertRaises(IntegrityError, lambda:_make_group('newGroup'))
Session.rollback()
def test_same_subgroup(self):
sg1 = _make_group('sub1', parent_id=self.g1.group_id)
self.assertEqual(sg1.parent_group, self.g1)
self.assertEqual(sg1.full_path, 'test1/sub1')
self.assertTrue(self.__check_path('test1', 'sub1'))
ssg1 = _make_group('subsub1', parent_id=sg1.group_id)
self.assertEqual(ssg1.parent_group, sg1)
self.assertEqual(ssg1.full_path, 'test1/sub1/subsub1')
self.assertTrue(self.__check_path('test1', 'sub1', 'subsub1'))
def test_remove_group(self):
sg1 = _make_group('deleteme')
self.__delete_group(sg1.group_id)
self.assertEqual(RepoGroup.get(sg1.group_id), None)
self.assertFalse(self.__check_path('deteteme'))
sg1 = _make_group('deleteme', parent_id=self.g1.group_id)
self.assertFalse(self.__check_path('test1', 'deteteme'))
def test_rename_single_group(self):
sg1 = _make_group('initial')
new_sg1 = self.__update_group(sg1.group_id, 'after')
self.assertTrue(self.__check_path('after'))
self.assertEqual(RepoGroup.get_by_group_name('initial'), None)
def test_update_group_parent(self):
sg1 = _make_group('initial', parent_id=self.g1.group_id)
new_sg1 = self.__update_group(sg1.group_id, 'after', parent_id=self.g1.group_id)
self.assertTrue(self.__check_path('test1', 'after'))
self.assertEqual(RepoGroup.get_by_group_name('test1/initial'), None)
new_sg1 = self.__update_group(sg1.group_id, 'after', parent_id=self.g3.group_id)
self.assertTrue(self.__check_path('test3', 'after'))
self.assertEqual(RepoGroup.get_by_group_name('test3/initial'), None)
new_sg1 = self.__update_group(sg1.group_id, 'hello')
self.assertTrue(self.__check_path('hello'))
self.assertEqual(RepoGroup.get_by_group_name('hello'), new_sg1)
def test_subgrouping_with_repo(self):
g1 = _make_group('g1')
g2 = _make_group('g2')
# create new repo
form_data = dict(repo_name='john',
repo_name_full='john',
fork_name=None,
description=None,
repo_group=None,
private=False,
repo_type='hg',
clone_uri=None)
cur_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
r = RepoModel().create(form_data, cur_user)
self.assertEqual(r.repo_name, 'john')
# put repo into group
form_data = form_data
form_data['repo_group'] = g1.group_id
form_data['perms_new'] = []
form_data['perms_updates'] = []
RepoModel().update(r.repo_name, form_data)
self.assertEqual(r.repo_name, 'g1/john')
self.__update_group(g1.group_id, 'g1', parent_id=g2.group_id)
self.assertTrue(self.__check_path('g2', 'g1'))
# test repo
self.assertEqual(r.repo_name, os.path.join('g2', 'g1', r.just_name))
def test_move_to_root(self):
g1 = _make_group('t11')
g2 = _make_group('t22',parent_id=g1.group_id)
self.assertEqual(g2.full_path,'t11/t22')
self.assertTrue(self.__check_path('t11', 't22'))
g2 = self.__update_group(g2.group_id, 'g22', parent_id=None)
self.assertEqual(g2.group_name,'g22')
# we moved out group from t1 to '' so it's full path should be 'g2'
self.assertEqual(g2.full_path,'g22')
self.assertFalse(self.__check_path('t11', 't22'))
self.assertTrue(self.__check_path('g22'))
class TestUser(unittest.TestCase):
def __init__(self, methodName='runTest'):
Session.remove()
super(TestUser, self).__init__(methodName=methodName)
def test_create_and_remove(self):
usr = UserModel().create_or_update(username=u'test_user', password=u'qweqwe',
email=u'u232@rhodecode.org',
name=u'u1', lastname=u'u1')
self.assertEqual(User.get_by_username(u'test_user'), usr)
# make users group
users_group = UsersGroupModel().create('some_example_group')
UsersGroupModel().add_user_to_group(users_group, usr)
self.assertEqual(UsersGroup.get(users_group.users_group_id), users_group)
self.assertEqual(UsersGroupMember.query().count(), 1)
UserModel().delete(usr.user_id)
self.assertEqual(UsersGroupMember.query().all(), [])
class TestNotifications(unittest.TestCase):
self.u1 = UserModel().create_or_update(username=u'u1',
password=u'qweqwe',
email=u'u1@rhodecode.org',
self.u1 = self.u1.user_id
self.u2 = UserModel().create_or_update(username=u'u2',
email=u'u2@rhodecode.org',
name=u'u2', lastname=u'u3')
self.u2 = self.u2.user_id
self.u3 = UserModel().create_or_update(username=u'u3',
email=u'u3@rhodecode.org',
name=u'u3', lastname=u'u3')
self.u3 = self.u3.user_id
super(TestNotifications, self).__init__(methodName=methodName)
def _clean_notifications(self):
for n in Notification.query().all():
Session.delete(n)
self.assertEqual(Notification.query().all(), [])
self._clean_notifications()
def test_create_notification(self):
self.assertEqual([], Notification.query().all())
self.assertEqual([], UserNotification.query().all())
usrs = [self.u1, self.u2]
notification = NotificationModel().create(created_by=self.u1,
subject=u'subj', body=u'hi there',
recipients=usrs)
u1 = User.get(self.u1)
u2 = User.get(self.u2)
u3 = User.get(self.u3)
notifications = Notification.query().all()
self.assertEqual(len(notifications), 1)
unotification = UserNotification.query()\
.filter(UserNotification.notification == notification).all()
self.assertEqual(notifications[0].recipients, [u1, u2])
self.assertEqual(notification.notification_id,
notifications[0].notification_id)
self.assertEqual(len(unotification), len(usrs))
self.assertEqual([x.user.user_id for x in unotification], usrs)
def test_user_notifications(self):
notification1 = NotificationModel().create(created_by=self.u1,
subject=u'subj', body=u'hi there1',
recipients=[self.u3])
notification2 = NotificationModel().create(created_by=self.u1,
subject=u'subj', body=u'hi there2',
u3 = Session.query(User).get(self.u3)
self.assertEqual(sorted([x.notification for x in u3.notifications]),
sorted([notification2, notification1]))
def test_delete_notifications(self):
subject=u'title', body=u'hi there3',
recipients=[self.u3, self.u1, self.u2])
self.assertTrue(notification in notifications)
Notification.delete(notification.notification_id)
self.assertFalse(notification in notifications)
un = UserNotification.query().filter(UserNotification.notification
== notification).all()
self.assertEqual(un, [])
def test_delete_association(self):
.filter(UserNotification.notification ==
notification)\
.filter(UserNotification.user_id == self.u3)\
.scalar()
self.assertEqual(unotification.user_id, self.u3)
NotificationModel().delete(self.u3,
notification.notification_id)
u3notification = UserNotification.query()\
self.assertEqual(u3notification, None)
# notification object is still there
self.assertEqual(Notification.query().all(), [notification])
#u1 and u2 still have assignments
u1notification = UserNotification.query()\
.filter(UserNotification.user_id == self.u1)\
self.assertNotEqual(u1notification, None)
u2notification = UserNotification.query()\
.filter(UserNotification.user_id == self.u2)\
self.assertNotEqual(u2notification, None)
def test_notification_counter(self):
NotificationModel().create(created_by=self.u1,
subject=u'title', body=u'hi there_delete',
recipients=[self.u3, self.u1])
self.assertEqual(NotificationModel()
.get_unread_cnt_for_user(self.u1), 1)
.get_unread_cnt_for_user(self.u2), 0)
.get_unread_cnt_for_user(self.u3), 1)
Status change: