Changeset - 7483b3f3bea5
[Not reviewed]
default
0 8 0
Thomas De Schampheleire - 10 years ago 2016-05-09 17:44:34
thomas.de.schampheleire@gmail.com
pytest migration: models: switch to standard assert statements

Use unittest2pytest to replace unittest-style assert statements (e.g.
assertEqual) with standard Python assert statements to benefit from pytest's
improved reporting on assert failures.

The conversion by unittest2pytest was correct except for:
- 'import pytest' is not automatically added when needed
- line wrapping in string formatting caused a syntax error in the
transformed code. Reported upstream at
https://github.com/pytest-dev/unittest2pytest/issues/3 .
- in assertRaises with a lambda, the lambda needs to be removed
8 files changed with 166 insertions and 218 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/models/test_changeset_status.py
Show inline comments
 
@@ -28,13 +28,13 @@ class TestChangesetStatusCalculation(Tes
 
        ('approve2_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_APPROVED), None]),
 
        ('approve_reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_REJECTED), None]),
 
        ('approve_underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('approve_notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_UNDER_REVIEW), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), None]),
 
        ('reject_underreview_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_notreviewed_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_NOT_REVIEWED), None]),
 
    ])
 
    def test_result(self, name, expected_result, statuses):
 
        result = self.m._calculate_status(statuses)
 
        self.assertEqual(result, expected_result)
 
        assert result == expected_result
kallithea/tests/models/test_diff_parsers.py
Show inline comments
 
@@ -274,42 +274,42 @@ DIFF_FIXTURES = {
 
class TestDiffLib(TestControllerPytest):
 

	
 
    @parametrize('diff_fixture', DIFF_FIXTURES)
 
    def test_diff(self, diff_fixture):
 
        diff = fixture.load_resource(diff_fixture, strip=False)
 
        vcs = 'hg'
 
        if diff_fixture.startswith('git_'):
 
            vcs = 'git'
 
        diff_proc = DiffProcessor(diff, vcs=vcs)
 
        diff_proc_d = diff_proc.prepare()
 
        data = [(x['filename'], x['operation'], x['stats']) for x in diff_proc_d]
 
        expected_data = DIFF_FIXTURES[diff_fixture]
 
        self.assertListEqual(expected_data, data)
 
        assert expected_data == data
 

	
 
    def test_diff_markup(self):
 
        diff = fixture.load_resource('markuptest.diff', strip=False)
 
        diff_proc = DiffProcessor(diff)
 
        diff_proc_d = diff_proc.prepare()
 
        chunks = diff_proc_d[0]['chunks']
 
        self.assertFalse(chunks[0])
 
        assert not chunks[0]
 
        #from pprint import pprint; pprint(chunks[1])
 
        l = ['\n']
 
        for d in chunks[1]:
 
            l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)
 
        s = ''.join(l)
 
        print s
 
        self.assertEqual(s, r'''
 
        assert s == r'''
 
context ... ... u'@@ -51,5 +51,12 @@\n'
 
unmod    51  51 u'<u>\t</u>begin();\n'
 
unmod    52  52 u'<u>\t</u>\n'
 
add      53     u'<u>\t</u>int foo;<u class="cr"></u>\n'
 
add      54     u'<u>\t</u>int bar; <u class="cr"></u>\n'
 
add      55     u'<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'
 
add      56     u'<u>\t</u>int space; <i></i>'
 
add      57     u'<u>\t</u>int tab;<u>\t</u>\n'
 
add      58     u'<u>\t</u>\n'
 
unmod    59  53 u' <i></i>'
 
del          54 u'<u>\t</u><del>#define MAX_STEPS (48)</del>\n'
 
add      60     u'<u>\t</u><ins><u class="cr"></u></ins>\n'
 
add      61     u'<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'
 
unmod    62  55 u'\n'
 
''')
 
'''
kallithea/tests/models/test_notifications.py
Show inline comments
 
@@ -24,137 +24,128 @@ class TestNotifications(TestControllerPy
 
                                        firstname=u'u2', lastname=u'u3')
 
        Session().commit()
 
        self.u2 = self.u2.user_id
 

	
 
        self.u3 = UserModel().create_or_update(username=u'u3',
 
                                        password=u'qweqwe',
 
                                        email=u'u3@example.com',
 
                                        firstname=u'u3', lastname=u'u3')
 
        Session().commit()
 
        self.u3 = self.u3.user_id
 

	
 
        remove_all_notifications()
 
        self.assertEqual([], Notification.query().all())
 
        self.assertEqual([], UserNotification.query().all())
 
        assert [] == Notification.query().all()
 
        assert [] == UserNotification.query().all()
 

	
 
    def test_create_notification(self):
 
        usrs = [self.u1, self.u2]
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'subj', body=u'hi there',
 
                                           recipients=usrs)
 
        Session().commit()
 
        u1 = User.get(self.u1)
 
        u2 = User.get(self.u2)
 
        u3 = User.get(self.u3)
 
        notifications = Notification.query().all()
 
        self.assertEqual(len(notifications), 1)
 
        assert len(notifications) == 1
 

	
 
        self.assertEqual(notifications[0].recipients, [u1, u2])
 
        self.assertEqual(notification.notification_id,
 
                         notifications[0].notification_id)
 
        assert notifications[0].recipients == [u1, u2]
 
        assert notification.notification_id == notifications[0].notification_id
 

	
 
        unotification = UserNotification.query() \
 
            .filter(UserNotification.notification == notification).all()
 

	
 
        self.assertEqual(len(unotification), len(usrs))
 
        self.assertEqual(set([x.user.user_id for x in unotification]),
 
                         set(usrs))
 
        assert len(unotification) == len(usrs)
 
        assert set([x.user.user_id for x in unotification]) == set(usrs)
 

	
 
    def test_user_notifications(self):
 
        notification1 = NotificationModel().create(created_by=self.u1,
 
                                            subject=u'subj', body=u'hi there1',
 
                                            recipients=[self.u3])
 
        Session().commit()
 
        notification2 = NotificationModel().create(created_by=self.u1,
 
                                            subject=u'subj', body=u'hi there2',
 
                                            recipients=[self.u3])
 
        Session().commit()
 
        u3 = Session().query(User).get(self.u3)
 

	
 
        self.assertEqual(sorted([x.notification for x in u3.notifications]),
 
                         sorted([notification2, notification1]))
 
        assert sorted([x.notification for x in u3.notifications]) == sorted([notification2, notification1])
 

	
 
    def test_delete_notifications(self):
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'title', body=u'hi there3',
 
                                    recipients=[self.u3, self.u1, self.u2])
 
        Session().commit()
 
        notifications = Notification.query().all()
 
        self.assertTrue(notification in notifications)
 
        assert notification in notifications
 

	
 
        Notification.delete(notification.notification_id)
 
        Session().commit()
 

	
 
        notifications = Notification.query().all()
 
        self.assertFalse(notification in notifications)
 
        assert not notification in notifications
 

	
 
        un = UserNotification.query().filter(UserNotification.notification
 
                                             == notification).all()
 
        self.assertEqual(un, [])
 
        assert un == []
 

	
 
    def test_delete_association(self):
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'title', body=u'hi there3',
 
                                    recipients=[self.u3, self.u1, self.u2])
 
        Session().commit()
 

	
 
        unotification = UserNotification.query() \
 
                            .filter(UserNotification.notification ==
 
                                    notification) \
 
                            .filter(UserNotification.user_id == self.u3) \
 
                            .scalar()
 

	
 
        self.assertEqual(unotification.user_id, self.u3)
 
        assert unotification.user_id == self.u3
 

	
 
        NotificationModel().delete(self.u3,
 
                                   notification.notification_id)
 
        Session().commit()
 

	
 
        u3notification = UserNotification.query() \
 
                            .filter(UserNotification.notification ==
 
                                    notification) \
 
                            .filter(UserNotification.user_id == self.u3) \
 
                            .scalar()
 

	
 
        self.assertEqual(u3notification, None)
 
        assert u3notification == None
 

	
 
        # notification object is still there
 
        self.assertEqual(Notification.query().all(), [notification])
 
        assert Notification.query().all() == [notification]
 

	
 
        #u1 and u2 still have assignments
 
        u1notification = UserNotification.query() \
 
                            .filter(UserNotification.notification ==
 
                                    notification) \
 
                            .filter(UserNotification.user_id == self.u1) \
 
                            .scalar()
 
        self.assertNotEqual(u1notification, None)
 
        assert u1notification != None
 
        u2notification = UserNotification.query() \
 
                            .filter(UserNotification.notification ==
 
                                    notification) \
 
                            .filter(UserNotification.user_id == self.u2) \
 
                            .scalar()
 
        self.assertNotEqual(u2notification, None)
 
        assert 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])
 
        Session().commit()
 

	
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u1), 0)
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u2), 0)
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u3), 1)
 
        assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0
 
        assert NotificationModel().get_unread_cnt_for_user(self.u2) == 0
 
        assert NotificationModel().get_unread_cnt_for_user(self.u3) == 1
 

	
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'title', body=u'hi there3',
 
                                    recipients=[self.u3, self.u1, self.u2])
 
        Session().commit()
 

	
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u1), 0)
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u2), 1)
 
        self.assertEqual(NotificationModel()
 
                         .get_unread_cnt_for_user(self.u3), 2)
 
        assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0
 
        assert NotificationModel().get_unread_cnt_for_user(self.u2) == 1
 
        assert NotificationModel().get_unread_cnt_for_user(self.u3) == 2
kallithea/tests/models/test_permissions.py
Show inline comments
 
@@ -63,385 +63,350 @@ class TestPermissions(TestControllerPyte
 
            UserGroupModel().delete(self.ug2, force=True)
 

	
 
        Session().commit()
 

	
 
    def test_default_perms_set(self):
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.create.repository', 'repository.read',
 
                           'hg.register.manual_activate']),
 
            'repositories': {HG_REPO: 'repository.read'}
 
        }
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        assert u1_auth.permissions['repositories'][HG_REPO] == perms['repositories'][HG_REPO]
 
        new_perm = 'repository.write'
 
        RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1,
 
                                          perm=new_perm)
 
        Session().commit()
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm)
 
        assert u1_auth.permissions['repositories'][HG_REPO] == new_perm
 

	
 
    def test_default_admin_perms_set(self):
 
        a1_auth = AuthUser(user_id=self.a1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.admin', 'hg.create.write_on_repogroup.true']),
 
            'repositories': {HG_REPO: 'repository.admin'}
 
        }
 
        self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        assert a1_auth.permissions['repositories'][HG_REPO] == perms['repositories'][HG_REPO]
 
        new_perm = 'repository.write'
 
        RepoModel().grant_user_permission(repo=HG_REPO, user=self.a1,
 
                                          perm=new_perm)
 
        Session().commit()
 
        # cannot really downgrade admins permissions !? they still gets set as
 
        # admin !
 
        u1_auth = AuthUser(user_id=self.a1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        assert u1_auth.permissions['repositories'][HG_REPO] == perms['repositories'][HG_REPO]
 

	
 
    def test_default_group_perms(self):
 
        self.g1 = fixture.create_repo_group(u'test1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group(u'test2', skip_if_exists=True)
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        perms = {
 
            'repositories_groups': {u'test1': 'group.read', u'test2': 'group.read'},
 
            'global': set(Permission.DEFAULT_USER_PERMISSIONS),
 
            'repositories': {HG_REPO: 'repository.read'}
 
        }
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         perms['repositories_groups'])
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         perms['global'])
 
        assert u1_auth.permissions['repositories'][HG_REPO] == perms['repositories'][HG_REPO]
 
        assert u1_auth.permissions['repositories_groups'] == perms['repositories_groups']
 
        assert u1_auth.permissions['global'] == perms['global']
 

	
 
    def test_default_admin_group_perms(self):
 
        self.g1 = fixture.create_repo_group(u'test1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group(u'test2', skip_if_exists=True)
 
        a1_auth = AuthUser(user_id=self.a1.user_id)
 
        perms = {
 
            'repositories_groups': {u'test1': 'group.admin', u'test2': 'group.admin'},
 
            'global': set(['hg.admin', 'hg.create.write_on_repogroup.true']),
 
            'repositories': {HG_REPO: 'repository.admin'}
 
        }
 

	
 
        self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],
 
                         perms['repositories'][HG_REPO])
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         perms['repositories_groups'])
 
        assert a1_auth.permissions['repositories'][HG_REPO] == perms['repositories'][HG_REPO]
 
        assert a1_auth.permissions['repositories_groups'] == perms['repositories_groups']
 

	
 
    def test_propagated_permission_from_users_group_by_explicit_perms_exist(self):
 
        # make group
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
 

	
 
        # set permission to lower
 
        new_perm = 'repository.none'
 
        RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1, perm=new_perm)
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm)
 
        assert u1_auth.permissions['repositories'][HG_REPO] == new_perm
 

	
 
        # grant perm for group this should not override permission from user
 
        # since it has explicitly set
 
        new_perm_gr = 'repository.write'
 
        RepoModel().grant_user_group_permission(repo=HG_REPO,
 
                                                 group_name=self.ug1,
 
                                                 perm=new_perm_gr)
 
        # check perms
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.create.repository', 'repository.read',
 
                           'hg.register.manual_activate']),
 
            'repositories': {HG_REPO: 'repository.read'}
 
        }
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         perms['repositories_groups'])
 
        assert u1_auth.permissions['repositories'][HG_REPO] == new_perm
 
        assert u1_auth.permissions['repositories_groups'] == perms['repositories_groups']
 

	
 
    def test_propagated_permission_from_users_group(self):
 
        # make group
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        UserGroupModel().add_user_to_group(self.ug1, self.u3)
 

	
 
        # grant perm for group this should override default permission from user
 
        new_perm_gr = 'repository.write'
 
        RepoModel().grant_user_group_permission(repo=HG_REPO,
 
                                                 group_name=self.ug1,
 
                                                 perm=new_perm_gr)
 
        # check perms
 
        u3_auth = AuthUser(user_id=self.u3.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.create.repository', 'repository.read',
 
                           'hg.register.manual_activate']),
 
            'repositories': {HG_REPO: 'repository.read'}
 
        }
 
        self.assertEqual(u3_auth.permissions['repositories'][HG_REPO],
 
                         new_perm_gr)
 
        self.assertEqual(u3_auth.permissions['repositories_groups'],
 
                         perms['repositories_groups'])
 
        assert u3_auth.permissions['repositories'][HG_REPO] == new_perm_gr
 
        assert u3_auth.permissions['repositories_groups'] == perms['repositories_groups']
 

	
 
    def test_propagated_permission_from_users_group_lower_weight(self):
 
        # make group
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        # add user to group
 
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
 

	
 
        # set permission to lower
 
        new_perm_h = 'repository.write'
 
        RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1,
 
                                          perm=new_perm_h)
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm_h)
 
        assert u1_auth.permissions['repositories'][HG_REPO] == new_perm_h
 

	
 
        # grant perm for group this should NOT override permission from user
 
        # since it's lower than granted
 
        new_perm_l = 'repository.read'
 
        RepoModel().grant_user_group_permission(repo=HG_REPO,
 
                                                 group_name=self.ug1,
 
                                                 perm=new_perm_l)
 
        # check perms
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        perms = {
 
            'repositories_groups': {},
 
            'global': set(['hg.create.repository', 'repository.read',
 
                           'hg.register.manual_activate']),
 
            'repositories': {HG_REPO: 'repository.write'}
 
        }
 
        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
 
                         new_perm_h)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         perms['repositories_groups'])
 
        assert u1_auth.permissions['repositories'][HG_REPO] == new_perm_h
 
        assert u1_auth.permissions['repositories_groups'] == perms['repositories_groups']
 

	
 
    def test_repo_in_group_permissions(self):
 
        self.g1 = fixture.create_repo_group(u'group1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group(u'group2', skip_if_exists=True)
 
        # both perms should be read !
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.read', u'group2': u'group.read'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.read', u'group2': u'group.read'}
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.read', u'group2': u'group.read'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.read', u'group2': u'group.read'}
 

	
 
        #Change perms to none for both groups
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1,
 
                                               user=self.anon,
 
                                               perm='group.none')
 
        RepoGroupModel().grant_user_permission(repo_group=self.g2,
 
                                               user=self.anon,
 
                                               perm='group.none')
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
        # add repo to group
 
        name = RepoGroup.url_sep().join([self.g1.group_name, 'test_perm'])
 
        self.test_repo = fixture.create_repo(name=name,
 
                                             repo_type='hg',
 
                                             repo_group=self.g1,
 
                                             cur_user=self.u1,)
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
        #grant permission for u2 !
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1, user=self.u2,
 
                                               perm='group.read')
 
        RepoGroupModel().grant_user_permission(repo_group=self.g2, user=self.u2,
 
                                               perm='group.read')
 
        Session().commit()
 
        self.assertNotEqual(self.u1, self.u2)
 
        assert self.u1 != self.u2
 
        #u1 and anon should have not change perms while u2 should !
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
        u2_auth = AuthUser(user_id=self.u2.user_id)
 
        self.assertEqual(u2_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.read', u'group2': u'group.read'})
 
        assert u2_auth.permissions['repositories_groups'] == {u'group1': u'group.read', u'group2': u'group.read'}
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                 {u'group1': u'group.none', u'group2': u'group.none'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.none', u'group2': u'group.none'}
 

	
 
    def test_repo_group_user_as_user_group_member(self):
 
        # create Group1
 
        self.g1 = fixture.create_repo_group(u'group1', skip_if_exists=True)
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 

	
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.read'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.read'}
 

	
 
        # set default permission to none
 
        RepoGroupModel().grant_user_permission(repo_group=self.g1,
 
                                               user=self.anon,
 
                                               perm='group.none')
 
        # make group
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        # add user to group
 
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
 
        Session().commit()
 

	
 
        # check if user is in the group
 
        members = [x.user_id for x in UserGroupModel().get(self.ug1.users_group_id).members]
 
        self.assertEqual(members, [self.u1.user_id])
 
        assert members == [self.u1.user_id]
 
        # add some user to that group
 

	
 
        # check his permissions
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.none'}
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.none'}
 

	
 
        # grant ug1 read permissions for
 
        RepoGroupModel().grant_user_group_permission(repo_group=self.g1,
 
                                                      group_name=self.ug1,
 
                                                      perm='group.read')
 
        Session().commit()
 
        # check if the
 
        obj = Session().query(UserGroupRepoGroupToPerm) \
 
            .filter(UserGroupRepoGroupToPerm.group == self.g1) \
 
            .filter(UserGroupRepoGroupToPerm.users_group == self.ug1) \
 
            .scalar()
 
        self.assertEqual(obj.permission.permission_name, 'group.read')
 
        assert obj.permission.permission_name == 'group.read'
 

	
 
        a1_auth = AuthUser(user_id=self.anon.user_id)
 

	
 
        self.assertEqual(a1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.none'})
 
        assert a1_auth.permissions['repositories_groups'] == {u'group1': u'group.none'}
 

	
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.read'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.read'}
 

	
 
    def test_inherited_permissions_from_default_on_user_enabled(self):
 
        user_model = UserModel()
 
        # enable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.none')
 
        user_model.grant_perm(usr, 'hg.create.repository')
 
        user_model.revoke_perm(usr, 'hg.fork.none')
 
        user_model.grant_perm(usr, 'hg.fork.repository')
 
        # make sure inherit flag is turned on
 
        self.u1.inherit_default_permissions = True
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited permissions from default user
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.repository', 'hg.fork.repository',
 
        assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true']))
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
 

	
 
    def test_inherited_permissions_from_default_on_user_disabled(self):
 
        user_model = UserModel()
 
        # disable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.repository')
 
        user_model.grant_perm(usr, 'hg.create.none')
 
        user_model.revoke_perm(usr, 'hg.fork.repository')
 
        user_model.grant_perm(usr, 'hg.fork.none')
 
        # make sure inherit flag is turned on
 
        self.u1.inherit_default_permissions = True
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have inherited permissions from default user
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.none', 'hg.fork.none',
 
        assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true']))
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
 

	
 
    def test_non_inherited_permissions_from_default_on_user_enabled(self):
 
        user_model = UserModel()
 
        # enable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.none')
 
        user_model.grant_perm(usr, 'hg.create.repository')
 
        user_model.revoke_perm(usr, 'hg.fork.none')
 
        user_model.grant_perm(usr, 'hg.fork.repository')
 

	
 
        #disable global perms on specific user
 
        user_model.revoke_perm(self.u1, 'hg.create.repository')
 
        user_model.grant_perm(self.u1, 'hg.create.none')
 
        user_model.revoke_perm(self.u1, 'hg.fork.repository')
 
        user_model.grant_perm(self.u1, 'hg.fork.none')
 

	
 
        # make sure inherit flag is turned off
 
        self.u1.inherit_default_permissions = False
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have non inherited permissions from he's
 
        # explicitly set permissions
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.none', 'hg.fork.none',
 
        assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true']))
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
 

	
 
    def test_non_inherited_permissions_from_default_on_user_disabled(self):
 
        user_model = UserModel()
 
        # disable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.repository')
 
        user_model.grant_perm(usr, 'hg.create.none')
 
        user_model.revoke_perm(usr, 'hg.fork.repository')
 
        user_model.grant_perm(usr, 'hg.fork.none')
 

	
 
        #enable global perms on specific user
 
        user_model.revoke_perm(self.u1, 'hg.create.none')
 
        user_model.grant_perm(self.u1, 'hg.create.repository')
 
        user_model.revoke_perm(self.u1, 'hg.fork.none')
 
        user_model.grant_perm(self.u1, 'hg.fork.repository')
 

	
 
        # make sure inherit flag is turned off
 
        self.u1.inherit_default_permissions = False
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        # this user will have non inherited permissions from he's
 
        # explicitly set permissions
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.repository', 'hg.fork.repository',
 
        assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true']))
 
                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
 

	
 
    def test_inactive_user_group_does_not_affect_global_permissions(self):
 
        # Issue #138: Inactive User Groups affecting permissions
 
        # Add user to inactive user group, set specific permissions on user
 
        # group and disable inherit-from-default. User permissions should still
 
        # inherit from default.
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
@@ -453,31 +418,30 @@ class TestPermissions(TestControllerPyte
 

	
 
        user_model = UserModel()
 
        # disable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.repository')
 
        user_model.grant_perm(usr, 'hg.create.none')
 
        user_model.revoke_perm(usr, 'hg.fork.repository')
 
        user_model.grant_perm(usr, 'hg.fork.none')
 

	
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 

	
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.none', 'hg.fork.none',
 
        assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read',
 
                              'hg.create.write_on_repogroup.true']))
 
                              'hg.create.write_on_repogroup.true'])
 

	
 
    def test_inactive_user_group_does_not_affect_global_permissions_inverse(self):
 
        # Issue #138: Inactive User Groups affecting permissions
 
        # Add user to inactive user group, set specific permissions on user
 
        # group and disable inherit-from-default. User permissions should still
 
        # inherit from default.
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
@@ -489,219 +453,208 @@ class TestPermissions(TestControllerPyte
 

	
 
        user_model = UserModel()
 
        # enable fork and create on default user
 
        usr = 'default'
 
        user_model.revoke_perm(usr, 'hg.create.none')
 
        user_model.grant_perm(usr, 'hg.create.repository')
 
        user_model.revoke_perm(usr, 'hg.fork.none')
 
        user_model.grant_perm(usr, 'hg.fork.repository')
 

	
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 

	
 
        self.assertEqual(u1_auth.permissions['global'],
 
                         set(['hg.create.repository', 'hg.fork.repository',
 
        assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
 
                              'hg.register.manual_activate',
 
                              'hg.extern_activate.auto',
 
                              'repository.read', 'group.read',
 
                              'usergroup.read',
 
                              'hg.create.write_on_repogroup.true']))
 
                              'hg.create.write_on_repogroup.true'])
 

	
 
    def test_inactive_user_group_does_not_affect_repo_permissions(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        # note: make u2 repo owner rather than u1, because the owner always has
 
        # admin permissions
 
        self.test_repo = fixture.create_repo(name=u'myownrepo',
 
                                             repo_type='hg',
 
                                             cur_user=self.u2)
 

	
 
        # enable admin access for user group on repo
 
        RepoModel().grant_user_group_permission(self.test_repo,
 
                                                group_name=self.ug1,
 
                                                perm='repository.admin')
 
        # enable only write access for default user on repo
 
        RepoModel().grant_user_permission(self.test_repo,
 
                                          user='default',
 
                                          perm='repository.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.write')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.write'
 

	
 
    def test_inactive_user_group_does_not_affect_repo_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        # note: make u2 repo owner rather than u1, because the owner always has
 
        # admin permissions
 
        self.test_repo = fixture.create_repo(name=u'myownrepo',
 
                                             repo_type='hg',
 
                                             cur_user=self.u2)
 

	
 
        # enable only write access for user group on repo
 
        RepoModel().grant_user_group_permission(self.test_repo,
 
                                                group_name=self.ug1,
 
                                                perm='repository.write')
 
        # enable admin access for default user on repo
 
        RepoModel().grant_user_permission(self.test_repo,
 
                                          user='default',
 
                                          perm='repository.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 

	
 
    def test_inactive_user_group_does_not_affect_repo_group_permissions(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        self.g1 = fixture.create_repo_group(u'group1', skip_if_exists=True)
 

	
 
        # enable admin access for user group on repo group
 
        RepoGroupModel().grant_user_group_permission(self.g1,
 
                                                     group_name=self.ug1,
 
                                                     perm='group.admin')
 
        # enable only write access for default user on repo group
 
        RepoGroupModel().grant_user_permission(self.g1,
 
                                               user='default',
 
                                               perm='group.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.write'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.write'}
 

	
 
    def test_inactive_user_group_does_not_affect_repo_group_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        self.g1 = fixture.create_repo_group(u'group1', skip_if_exists=True)
 

	
 
        # enable only write access for user group on repo group
 
        RepoGroupModel().grant_user_group_permission(self.g1,
 
                                                     group_name=self.ug1,
 
                                                     perm='group.write')
 
        # enable admin access for default user on repo group
 
        RepoGroupModel().grant_user_permission(self.g1,
 
                                               user='default',
 
                                               perm='group.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories_groups'],
 
                         {u'group1': u'group.admin'})
 
        assert u1_auth.permissions['repositories_groups'] == {u'group1': u'group.admin'}
 

	
 
    def test_inactive_user_group_does_not_affect_user_group_permissions(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        self.ug2 = fixture.create_user_group(u'G2')
 

	
 
        # enable admin access for user group on user group
 
        UserGroupModel().grant_user_group_permission(self.ug2,
 
                                                     user_group=self.ug1,
 
                                                     perm='usergroup.admin')
 
        # enable only write access for default user on user group
 
        UserGroupModel().grant_user_permission(self.ug2,
 
                                               user='default',
 
                                               perm='usergroup.write')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['user_groups'][u'G1'], u'usergroup.read')
 
        self.assertEqual(u1_auth.permissions['user_groups'][u'G2'], u'usergroup.write')
 
        assert u1_auth.permissions['user_groups'][u'G1'] == u'usergroup.read'
 
        assert u1_auth.permissions['user_groups'][u'G2'] == u'usergroup.write'
 

	
 
    def test_inactive_user_group_does_not_affect_user_group_permissions_inverse(self):
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        self.ug1.inherit_default_permissions = False
 
        user_group_model = UserGroupModel()
 
        user_group_model.add_user_to_group(self.ug1, self.u1)
 
        user_group_model.update(self.ug1, {'users_group_active': False})
 

	
 
        self.ug2 = fixture.create_user_group(u'G2')
 

	
 
        # enable only write access for user group on user group
 
        UserGroupModel().grant_user_group_permission(self.ug2,
 
                                                     user_group=self.ug1,
 
                                                     perm='usergroup.write')
 
        # enable admin access for default user on user group
 
        UserGroupModel().grant_user_permission(self.ug2,
 
                                               user='default',
 
                                               perm='usergroup.admin')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['user_groups'][u'G1'], u'usergroup.read')
 
        self.assertEqual(u1_auth.permissions['user_groups'][u'G2'], u'usergroup.admin')
 
        assert u1_auth.permissions['user_groups'][u'G1'] == u'usergroup.read'
 
        assert u1_auth.permissions['user_groups'][u'G2'] == u'usergroup.admin'
 

	
 
    def test_owner_permissions_doesnot_get_overwritten_by_group(self):
 
        #create repo as USER,
 
        self.test_repo = fixture.create_repo(name=u'myownrepo',
 
                                             repo_type='hg',
 
                                             cur_user=self.u1)
 

	
 
        #he has permissions of admin as owner
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        #set his permission as user group, he should still be admin
 
        self.ug1 = fixture.create_user_group(u'G1')
 
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
 
        RepoModel().grant_user_group_permission(self.test_repo,
 
                                                 group_name=self.ug1,
 
                                                 perm='repository.none')
 

	
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 

	
 
    def test_owner_permissions_doesnot_get_overwritten_by_others(self):
 
        #create repo as USER,
 
        self.test_repo = fixture.create_repo(name=u'myownrepo',
 
                                             repo_type='hg',
 
                                             cur_user=self.u1)
 

	
 
        #he has permissions of admin as owner
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 
        #set his permission as user, he should still be admin
 
        RepoModel().grant_user_permission(self.test_repo, user=self.u1,
 
                                          perm='repository.none')
 
        Session().commit()
 
        u1_auth = AuthUser(user_id=self.u1.user_id)
 
        self.assertEqual(u1_auth.permissions['repositories']['myownrepo'],
 
                         'repository.admin')
 
        assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
 

	
 
    def _test_def_perm_equal(self, user, change_factor=0):
 
        perms = UserToPerm.query() \
 
                .filter(UserToPerm.user == user) \
 
                .all()
 
        self.assertEqual(len(perms),
 
                         len(Permission.DEFAULT_USER_PERMISSIONS,)+change_factor,
 
                         msg=perms)
 
        assert len(perms) == len(Permission.DEFAULT_USER_PERMISSIONS,)+change_factor, perms
 

	
 
    def test_set_default_permissions(self):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
    def test_set_default_permissions_after_one_is_missing(self):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 
        #now we delete one, it should be re-created after another call
 
        perms = UserToPerm.query() \
 
                .filter(UserToPerm.user == self.u1) \
 
                .all()
 
@@ -719,26 +672,26 @@ class TestPermissions(TestControllerPyte
 
        ('group.read', 'group.none'),
 
        ('usergroup.read', 'usergroup.none'),
 
        ('hg.create.repository', 'hg.create.none'),
 
        ('hg.fork.repository', 'hg.fork.none'),
 
        ('hg.register.manual_activate', 'hg.register.auto_activate',)
 
    ])
 
    def test_set_default_permissions_after_modification(self, perm, modify_to):
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
        old = Permission.get_by_key(perm)
 
        new = Permission.get_by_key(modify_to)
 
        self.assertNotEqual(old, None)
 
        self.assertNotEqual(new, None)
 
        assert old != None
 
        assert new != None
 

	
 
        #now modify permissions
 
        p = UserToPerm.query() \
 
                .filter(UserToPerm.user == self.u1) \
 
                .filter(UserToPerm.permission == old) \
 
                .one()
 
        p.permission = new
 
        Session().add(p)
 
        Session().commit()
 

	
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
kallithea/tests/models/test_repo_groups.py
Show inline comments
 
import os
 
import pytest
 
from sqlalchemy.exc import IntegrityError
 

	
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 
@@ -51,149 +52,150 @@ class TestRepoGroups(TestControllerPytes
 
        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_):
 
        RepoGroupModel().delete(id_)
 

	
 
    def test_create_group(self):
 
        g = fixture.create_repo_group(u'newGroup')
 
        Session().commit()
 
        self.assertEqual(g.full_path, 'newGroup')
 
        assert g.full_path == 'newGroup'
 

	
 
        self.assertTrue(self.__check_path('newGroup'))
 
        assert self.__check_path('newGroup')
 

	
 
    def test_create_same_name_group(self):
 
        self.assertRaises(IntegrityError, lambda: fixture.create_repo_group(u'newGroup'))
 
        with pytest.raises(IntegrityError):
 
            fixture.create_repo_group(u'newGroup')
 
        Session().rollback()
 

	
 
    def test_same_subgroup(self):
 
        sg1 = fixture.create_repo_group(u'sub1', group_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'))
 
        assert sg1.parent_group == self.g1
 
        assert sg1.full_path == 'test1/sub1'
 
        assert self.__check_path('test1', 'sub1')
 

	
 
        ssg1 = fixture.create_repo_group(u'subsub1', group_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'))
 
        assert ssg1.parent_group == sg1
 
        assert ssg1.full_path == 'test1/sub1/subsub1'
 
        assert self.__check_path('test1', 'sub1', 'subsub1')
 

	
 
    def test_remove_group(self):
 
        sg1 = fixture.create_repo_group(u'deleteme')
 
        self.__delete_group(sg1.group_id)
 

	
 
        self.assertEqual(RepoGroup.get(sg1.group_id), None)
 
        self.assertFalse(self.__check_path('deteteme'))
 
        assert RepoGroup.get(sg1.group_id) == None
 
        assert not self.__check_path('deteteme')
 

	
 
        sg1 = fixture.create_repo_group(u'deleteme', group_parent_id=self.g1.group_id)
 
        self.__delete_group(sg1.group_id)
 

	
 
        self.assertEqual(RepoGroup.get(sg1.group_id), None)
 
        self.assertFalse(self.__check_path('test1', 'deteteme'))
 
        assert RepoGroup.get(sg1.group_id) == None
 
        assert not self.__check_path('test1', 'deteteme')
 

	
 
    def test_rename_single_group(self):
 
        sg1 = fixture.create_repo_group(u'initial')
 

	
 
        new_sg1 = _update_group(sg1.group_id, u'after')
 
        self.assertTrue(self.__check_path('after'))
 
        self.assertEqual(RepoGroup.get_by_group_name(u'initial'), None)
 
        assert self.__check_path('after')
 
        assert RepoGroup.get_by_group_name(u'initial') == None
 

	
 
    def test_update_group_parent(self):
 

	
 
        sg1 = fixture.create_repo_group(u'initial', group_parent_id=self.g1.group_id)
 

	
 
        new_sg1 = _update_group(sg1.group_id, u'after', parent_id=self.g1.group_id)
 
        self.assertTrue(self.__check_path('test1', 'after'))
 
        self.assertEqual(RepoGroup.get_by_group_name(u'test1/initial'), None)
 
        assert self.__check_path('test1', 'after')
 
        assert RepoGroup.get_by_group_name(u'test1/initial') == None
 

	
 
        new_sg1 = _update_group(sg1.group_id, u'after', parent_id=self.g3.group_id)
 
        self.assertTrue(self.__check_path('test3', 'after'))
 
        self.assertEqual(RepoGroup.get_by_group_name(u'test3/initial'), None)
 
        assert self.__check_path('test3', 'after')
 
        assert RepoGroup.get_by_group_name(u'test3/initial') == None
 

	
 
        new_sg1 = _update_group(sg1.group_id, u'hello')
 
        self.assertTrue(self.__check_path('hello'))
 
        assert self.__check_path('hello')
 

	
 
        self.assertEqual(RepoGroup.get_by_group_name(u'hello'), new_sg1)
 
        assert RepoGroup.get_by_group_name(u'hello') == new_sg1
 

	
 
    def test_subgrouping_with_repo(self):
 

	
 
        g1 = fixture.create_repo_group(u'g1')
 
        g2 = fixture.create_repo_group(u'g2')
 
        # create new repo
 
        r = fixture.create_repo(u'john')
 

	
 
        self.assertEqual(r.repo_name, 'john')
 
        assert r.repo_name == 'john'
 
        # put repo into group
 
        r = _update_repo(u'john', repo_group=g1.group_id)
 
        Session().commit()
 
        self.assertEqual(r.repo_name, 'g1/john')
 
        assert r.repo_name == 'g1/john'
 

	
 
        _update_group(g1.group_id, u'g1', parent_id=g2.group_id)
 
        self.assertTrue(self.__check_path('g2', 'g1'))
 
        assert self.__check_path('g2', 'g1')
 

	
 
        # test repo
 
        self.assertEqual(r.repo_name, RepoGroup.url_sep().join(['g2', 'g1',
 
                                                                r.just_name]))
 
        assert r.repo_name == RepoGroup.url_sep().join(['g2', 'g1',
 
                                                                r.just_name])
 

	
 
    def test_move_to_root(self):
 
        g1 = fixture.create_repo_group(u't11')
 
        g2 = fixture.create_repo_group(u't22', group_parent_id=g1.group_id)
 

	
 
        self.assertEqual(g2.full_path, 't11/t22')
 
        self.assertTrue(self.__check_path('t11', 't22'))
 
        assert g2.full_path == 't11/t22'
 
        assert self.__check_path('t11', 't22')
 

	
 
        g2 = _update_group(g2.group_id, u'g22', parent_id=None)
 
        Session().commit()
 

	
 
        self.assertEqual(g2.group_name, 'g22')
 
        assert 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'))
 
        assert g2.full_path == 'g22'
 
        assert not self.__check_path('t11', 't22')
 
        assert self.__check_path('g22')
 

	
 
    def test_rename_top_level_group_in_nested_setup(self):
 
        g1 = fixture.create_repo_group(u'L1')
 
        g2 = fixture.create_repo_group(u'L2', group_parent_id=g1.group_id)
 
        g3 = fixture.create_repo_group(u'L3', group_parent_id=g2.group_id)
 

	
 
        r = fixture.create_repo(u'L1/L2/L3/L3_REPO', repo_group=g3.group_id)
 

	
 
        ##rename L1 all groups should be now changed
 
        _update_group(g1.group_id, u'L1_NEW')
 
        Session().commit()
 
        self.assertEqual(g1.full_path, 'L1_NEW')
 
        self.assertEqual(g2.full_path, 'L1_NEW/L2')
 
        self.assertEqual(g3.full_path, 'L1_NEW/L2/L3')
 
        self.assertEqual(r.repo_name,  'L1_NEW/L2/L3/L3_REPO')
 
        assert g1.full_path == 'L1_NEW'
 
        assert g2.full_path == 'L1_NEW/L2'
 
        assert g3.full_path == 'L1_NEW/L2/L3'
 
        assert r.repo_name == 'L1_NEW/L2/L3/L3_REPO'
 

	
 
    def test_change_parent_of_top_level_group_in_nested_setup(self):
 
        g1 = fixture.create_repo_group(u'R1')
 
        g2 = fixture.create_repo_group(u'R2', group_parent_id=g1.group_id)
 
        g3 = fixture.create_repo_group(u'R3', group_parent_id=g2.group_id)
 
        g4 = fixture.create_repo_group(u'R1_NEW')
 

	
 
        r = fixture.create_repo(u'R1/R2/R3/R3_REPO', repo_group=g3.group_id)
 
        ##rename L1 all groups should be now changed
 
        _update_group(g1.group_id, u'R1', parent_id=g4.group_id)
 
        Session().commit()
 
        self.assertEqual(g1.full_path, 'R1_NEW/R1')
 
        self.assertEqual(g2.full_path, 'R1_NEW/R1/R2')
 
        self.assertEqual(g3.full_path, 'R1_NEW/R1/R2/R3')
 
        self.assertEqual(r.repo_name,  'R1_NEW/R1/R2/R3/R3_REPO')
 
        assert g1.full_path == 'R1_NEW/R1'
 
        assert g2.full_path == 'R1_NEW/R1/R2'
 
        assert g3.full_path == 'R1_NEW/R1/R2/R3'
 
        assert r.repo_name == 'R1_NEW/R1/R2/R3/R3_REPO'
 

	
 
    def test_change_parent_of_top_level_group_in_nested_setup_with_rename(self):
 
        g1 = fixture.create_repo_group(u'X1')
 
        g2 = fixture.create_repo_group(u'X2', group_parent_id=g1.group_id)
 
        g3 = fixture.create_repo_group(u'X3', group_parent_id=g2.group_id)
 
        g4 = fixture.create_repo_group(u'X1_NEW')
 

	
 
        r = fixture.create_repo(u'X1/X2/X3/X3_REPO', repo_group=g3.group_id)
 

	
 
        ##rename L1 all groups should be now changed
 
        _update_group(g1.group_id, u'X1_PRIM', parent_id=g4.group_id)
 
        Session().commit()
 
        self.assertEqual(g1.full_path, 'X1_NEW/X1_PRIM')
 
        self.assertEqual(g2.full_path, 'X1_NEW/X1_PRIM/X2')
 
        self.assertEqual(g3.full_path, 'X1_NEW/X1_PRIM/X2/X3')
 
        self.assertEqual(r.repo_name,  'X1_NEW/X1_PRIM/X2/X3/X3_REPO')
 
        assert g1.full_path == 'X1_NEW/X1_PRIM'
 
        assert g2.full_path == 'X1_NEW/X1_PRIM/X2'
 
        assert g3.full_path == 'X1_NEW/X1_PRIM/X2/X3'
 
        assert r.repo_name == 'X1_NEW/X1_PRIM/X2/X3/X3_REPO'
kallithea/tests/models/test_repos.py
Show inline comments
 
import pytest
 
from kallithea.tests import *
 

	
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import Repository
 
from kallithea.lib.exceptions import AttachedForksError
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestRepos(TestControllerPytest):
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_remove_repo(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        RepoModel().delete(repo=repo)
 
        Session().commit()
 

	
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-1'))
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-1')
 

	
 
    def test_remove_repo_repo_raises_exc_when_attached_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        fixture.create_fork(repo.repo_name, u'test-repo-fork-1')
 
        Session().commit()
 

	
 
        self.assertRaises(AttachedForksError, lambda: RepoModel().delete(repo=repo))
 
        with pytest.raises(AttachedForksError):
 
            RepoModel().delete(repo=repo)
 
        # cleanup
 
        RepoModel().delete(repo=u'test-repo-fork-1')
 
        RepoModel().delete(repo=u'test-repo-1')
 
        Session().commit()
 

	
 
    def test_remove_repo_delete_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        fork = fixture.create_fork(repo.repo_name, u'test-repo-fork-1')
 
        Session().commit()
 

	
 
        #fork of fork
 
        fixture.create_fork(fork.repo_name, u'test-repo-fork-fork-1')
 
        Session().commit()
 

	
 
        RepoModel().delete(repo=repo, forks='delete')
 
        Session().commit()
 

	
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-1'))
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-1'))
 
        self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1'))
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-1')
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-fork-1')
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1')
 

	
 
    def test_remove_repo_detach_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        fork = fixture.create_fork(repo.repo_name, u'test-repo-fork-1')
 
        Session().commit()
 

	
 
        #fork of fork
 
        fixture.create_fork(fork.repo_name, u'test-repo-fork-fork-1')
 
        Session().commit()
 

	
 
        RepoModel().delete(repo=repo, forks='detach')
 
        Session().commit()
 

	
 
        try:
 
            self.assertEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-1'))
 
            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-1'))
 
            self.assertNotEqual(None, Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1'))
 
            assert None == Repository.get_by_repo_name(repo_name=u'test-repo-1')
 
            assert None != Repository.get_by_repo_name(repo_name=u'test-repo-fork-1')
 
            assert None != Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1')
 
        finally:
 
            RepoModel().delete(repo=u'test-repo-fork-fork-1')
 
            RepoModel().delete(repo=u'test-repo-fork-1')
 
            Session().commit()
kallithea/tests/models/test_user_groups.py
Show inline comments
 
@@ -49,13 +49,13 @@ class TestUserGroups(TestControllerPytes
 
        # now special external groups created by auth plugins
 
        for gr in external_should_be:
 
            gr = fixture.create_user_group(gr, user_group_data={'extern_type': 'container'})
 
            Session().commit()
 
            UserGroupModel().add_user_to_group(gr, user)
 
            Session().commit()
 

	
 
        UserGroupModel().enforce_groups(user, groups, 'container')
 
        Session().commit()
 

	
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        in_groups = user.group_member
 
        self.assertEqual(expected, [x.users_group.users_group_name for x in in_groups])
 
        assert expected == [x.users_group.users_group_name for x in in_groups]
kallithea/tests/models/test_users.py
Show inline comments
 
import pytest
 
from kallithea.tests import *
 

	
 
from kallithea.model.db import User, UserGroup, UserGroupMember, UserEmailMap, \
 
    Permission
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 
@@ -17,91 +18,90 @@ class TestUser(TestControllerPytest):
 
    def setup_class(cls):
 
        Session.remove()
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_create_and_remove(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                           email=u'u232@example.com',
 
                                           firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 
        self.assertEqual(User.get_by_username(u'test_user'), usr)
 
        self.assertEqual(User.get_by_username(u'test_USER', case_insensitive=True), usr)
 
        self.assertEqual(User.get_by_username(u'test_USER', case_insensitive=False), None)
 
        assert User.get_by_username(u'test_user') == usr
 
        assert User.get_by_username(u'test_USER', case_insensitive=True) == usr
 
        assert User.get_by_username(u'test_USER', case_insensitive=False) == None
 

	
 
        # make user group
 
        user_group = fixture.create_user_group(u'some_example_group')
 
        Session().commit()
 

	
 
        UserGroupModel().add_user_to_group(user_group, usr)
 
        Session().commit()
 

	
 
        self.assertEqual(UserGroup.get(user_group.users_group_id), user_group)
 
        self.assertEqual(UserGroupMember.query().count(), 1)
 
        assert UserGroup.get(user_group.users_group_id) == user_group
 
        assert UserGroupMember.query().count() == 1
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
        self.assertEqual(UserGroupMember.query().all(), [])
 
        assert UserGroupMember.query().all() == []
 

	
 
    def test_additional_email_as_main(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        def do():
 
        with pytest.raises(AttributeError):
 
            m = UserEmailMap()
 
            m.email = u'main_email@example.com'
 
            m.user = usr
 
            Session().add(m)
 
            Session().commit()
 
        self.assertRaises(AttributeError, do)
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
    def test_extra_email_map(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        m = UserEmailMap()
 
        m.email = u'main_email2@example.com'
 
        m.user = usr
 
        Session().add(m)
 
        Session().commit()
 

	
 
        u = User.get_by_email(email='MAIN_email@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 

	
 
        u = User.get_by_email(email='main_email@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 

	
 
        u = User.get_by_email(email='main_email2@example.com')
 
        self.assertEqual(usr.user_id, u.user_id)
 
        self.assertEqual(usr.username, u.username)
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 
        u = User.get_by_email(email='main_email3@example.com')
 
        self.assertEqual(None, u)
 
        assert None == u
 

	
 
        u = User.get_by_email(email='main_e%ail@example.com')
 
        self.assertEqual(None, u)
 
        assert None == u
 
        u = User.get_by_email(email='main_emai_@example.com')
 
        self.assertEqual(None, u)
 
        assert None == u
 

	
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 

	
 
class TestUsers(TestControllerPytest):
 

	
 
    def setup_method(self, method):
 
        self.u1 = UserModel().create_or_update(username=u'u1',
 
                                        password=u'qweqwe',
 
                                        email=u'u1@example.com',
 
@@ -111,30 +111,30 @@ class TestUsers(TestControllerPytest):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            UserModel().revoke_perm(self.u1, p)
 

	
 
        UserModel().delete(self.u1)
 
        Session().commit()
 
        Session.remove()
 

	
 
    def test_add_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), True)
 
        assert UserModel().has_perm(self.u1, perm) == True
 

	
 
    def test_has_perm(self):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            has_p = UserModel().has_perm(self.u1, p)
 
            self.assertEqual(False, has_p)
 
            assert False == has_p
 

	
 
    def test_revoke_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), True)
 
        assert UserModel().has_perm(self.u1, perm) == True
 

	
 
        #revoke
 
        UserModel().revoke_perm(self.u1, perm)
 
        Session().commit()
 
        self.assertEqual(UserModel().has_perm(self.u1, perm), False)
 
        assert UserModel().has_perm(self.u1, perm) == False
0 comments (0 inline, 0 general)