])

    def test_result(self, name, expected_result, statuses):

        result = self.m._calculate_status(statuses)

        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]

    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']

        #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

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      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'

        self.u3 = self.u3.user_id

        remove_all_notifications()

    def test_create_notification(self):

        usrs = [self.u1, self.u2]

        u2 = User.get(self.u2)

        u3 = User.get(self.u3)

        notifications = Notification.query().all()

        unotification = UserNotification.query() \

            .filter(UserNotification.notification == notification).all()

    def test_user_notifications(self):

        notification1 = NotificationModel().create(created_by=self.u1,

        Session().commit()

        u3 = Session().query(User).get(self.u3)

    def test_delete_notifications(self):

        notification = NotificationModel().create(created_by=self.u1,

                                    recipients=[self.u3, self.u1, self.u2])

        Session().commit()

        notifications = Notification.query().all()

        Notification.delete(notification.notification_id)

        Session().commit()

        notifications = Notification.query().all()

        un = UserNotification.query().filter(UserNotification.notification

                                             == notification).all()

    def test_delete_association(self):

        notification = NotificationModel().create(created_by=self.u1,

                            .filter(UserNotification.user_id == self.u3) \

                            .scalar()

        NotificationModel().delete(self.u3,

                                   notification.notification_id)

                            .filter(UserNotification.user_id == self.u3) \

                            .scalar()

        # notification object is still there

        #u1 and u2 still have assignments

        u1notification = UserNotification.query() \

                                    notification) \

                            .filter(UserNotification.user_id == self.u1) \

                            .scalar()

        u2notification = UserNotification.query() \

                            .filter(UserNotification.notification ==

                                    notification) \

                            .filter(UserNotification.user_id == self.u2) \

                            .scalar()

    def test_notification_counter(self):

        NotificationModel().create(created_by=self.u1,

                            recipients=[self.u3, self.u1])

        Session().commit()

        notification = NotificationModel().create(created_by=self.u1,

                                           subject=u'title', body=u'hi there3',

                                    recipients=[self.u3, self.u1, self.u2])

        Session().commit()

                           'hg.register.manual_activate']),

            'repositories': {HG_REPO: 'repository.read'}

        }

        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)

    def test_default_admin_perms_set(self):

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

            'global': set(['hg.admin', 'hg.create.write_on_repogroup.true']),

            'repositories': {HG_REPO: 'repository.admin'}

        }

        new_perm = 'repository.write'

        RepoModel().grant_user_permission(repo=HG_REPO, user=self.a1,

                                          perm=new_perm)

        # cannot really downgrade admins permissions !? they still gets set as

        # admin !

        u1_auth = AuthUser(user_id=self.a1.user_id)

    def test_default_group_perms(self):

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

            'global': set(Permission.DEFAULT_USER_PERMISSIONS),

            'repositories': {HG_REPO: 'repository.read'}

        }

    def test_default_admin_group_perms(self):

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

            'repositories': {HG_REPO: 'repository.admin'}

        }

    def test_propagated_permission_from_users_group_by_explicit_perms_exist(self):

        # make group

        RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1, perm=new_perm)

        Session().commit()

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

        # grant perm for group this should not override permission from user

        # since it has explicitly set

                           'hg.register.manual_activate']),

            'repositories': {HG_REPO: 'repository.read'}

        }

    def test_propagated_permission_from_users_group(self):

        # make group

                           'hg.register.manual_activate']),

            'repositories': {HG_REPO: 'repository.read'}

        }

    def test_propagated_permission_from_users_group_lower_weight(self):

        # make group

                                          perm=new_perm_h)

        Session().commit()

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

        # grant perm for group this should NOT override permission from user

        # since it's lower than granted

                           'hg.register.manual_activate']),

            'repositories': {HG_REPO: 'repository.write'}

        }

    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)

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

        #Change perms to none for both groups

        RepoGroupModel().grant_user_permission(repo_group=self.g1,

                                               perm='group.none')

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

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

        # add repo to group

        name = RepoGroup.url_sep().join([self.g1.group_name, 'test_perm'])

                                             cur_user=self.u1,)

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

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

        #grant permission for u2 !

        RepoGroupModel().grant_user_permission(repo_group=self.g1, user=self.u2,

        RepoGroupModel().grant_user_permission(repo_group=self.g2, user=self.u2,

                                               perm='group.read')

        Session().commit()

        #u1 and anon should have not change perms while u2 should !

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

        u2_auth = AuthUser(user_id=self.u2.user_id)

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

    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)

        # set default permission to none

        RepoGroupModel().grant_user_permission(repo_group=self.g1,

        # check if user is in the group

        members = [x.user_id for x in UserGroupModel().get(self.ug1.users_group_id).members]

        # add some user to that group

        # check his permissions

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

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

        # grant ug1 read permissions for

        RepoGroupModel().grant_user_group_permission(repo_group=self.g1,

            .filter(UserGroupRepoGroupToPerm.group == self.g1) \

            .filter(UserGroupRepoGroupToPerm.users_group == self.ug1) \

            .scalar()

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

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

    def test_inherited_permissions_from_default_on_user_enabled(self):

        user_model = UserModel()

        Session().commit()

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

        # this user will have inherited permissions from default user

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

    def test_inherited_permissions_from_default_on_user_disabled(self):

        user_model = UserModel()

        Session().commit()

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

        # this user will have inherited permissions from default user

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

    def test_non_inherited_permissions_from_default_on_user_enabled(self):

        user_model = UserModel()

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

        # this user will have non inherited permissions from he's

        # explicitly set permissions

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

    def test_non_inherited_permissions_from_default_on_user_disabled(self):

        user_model = UserModel()

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

        # this user will have non inherited permissions from he's

        # explicitly set permissions

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

    def test_inactive_user_group_does_not_affect_global_permissions(self):

        # Issue #138: Inactive User Groups affecting permissions

        Session().commit()

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

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

                              'usergroup.read',

    def test_inactive_user_group_does_not_affect_global_permissions_inverse(self):

        # Issue #138: Inactive User Groups affecting permissions

        Session().commit()

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

                              'hg.register.manual_activate',

                              'hg.extern_activate.auto',

                              'repository.read', 'group.read',

                              'usergroup.read',

    def test_inactive_user_group_does_not_affect_repo_permissions(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                          perm='repository.write')

        Session().commit()

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

    def test_inactive_user_group_does_not_affect_repo_permissions_inverse(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                          perm='repository.admin')

        Session().commit()

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

    def test_inactive_user_group_does_not_affect_repo_group_permissions(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                               perm='group.write')

        Session().commit()

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

    def test_inactive_user_group_does_not_affect_repo_group_permissions_inverse(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                               perm='group.admin')

        Session().commit()

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

    def test_inactive_user_group_does_not_affect_user_group_permissions(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                               perm='usergroup.write')

        Session().commit()

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

    def test_inactive_user_group_does_not_affect_user_group_permissions_inverse(self):

        self.ug1 = fixture.create_user_group(u'G1')

                                               perm='usergroup.admin')

        Session().commit()

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

    def test_owner_permissions_doesnot_get_overwritten_by_group(self):

        #create repo as USER,

        #he has permissions of admin as owner

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

        #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)

        Session().commit()

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

    def test_owner_permissions_doesnot_get_overwritten_by_others(self):

        #create repo as USER,

        #he has permissions of admin as owner

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

        #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)

    def _test_def_perm_equal(self, user, change_factor=0):

        perms = UserToPerm.query() \

                .filter(UserToPerm.user == user) \

                .all()

    def test_set_default_permissions(self):

        PermissionModel().create_default_permissions(user=self.u1)

        old = Permission.get_by_key(perm)

        new = Permission.get_by_key(modify_to)

        #now modify permissions

        p = UserToPerm.query() \

import os

from sqlalchemy.exc import IntegrityError

from kallithea.tests import *

    def test_create_group(self):

        g = fixture.create_repo_group(u'newGroup')

        Session().commit()

    def test_create_same_name_group(self):

        Session().rollback()

    def test_same_subgroup(self):

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

        ssg1 = fixture.create_repo_group(u'subsub1', group_parent_id=sg1.group_id)

    def test_remove_group(self):

        sg1 = fixture.create_repo_group(u'deleteme')

        self.__delete_group(sg1.group_id)

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

        self.__delete_group(sg1.group_id)

    def test_rename_single_group(self):

        sg1 = fixture.create_repo_group(u'initial')

        new_sg1 = _update_group(sg1.group_id, u'after')

    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)

        new_sg1 = _update_group(sg1.group_id, u'after', parent_id=self.g3.group_id)

        new_sg1 = _update_group(sg1.group_id, u'hello')

    def test_subgrouping_with_repo(self):

        # create new repo

        r = fixture.create_repo(u'john')

        # put repo into group

        r = _update_repo(u'john', repo_group=g1.group_id)

        Session().commit()

        _update_group(g1.group_id, u'g1', parent_id=g2.group_id)

        # test repo

    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)

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

        Session().commit()

        # we moved out group from t1 to '' so it's full path should be 'g2'

    def test_rename_top_level_group_in_nested_setup(self):

        g1 = fixture.create_repo_group(u'L1')

        ##rename L1 all groups should be now changed

        _update_group(g1.group_id, u'L1_NEW')

        Session().commit()

    def test_change_parent_of_top_level_group_in_nested_setup(self):

        g1 = fixture.create_repo_group(u'R1')

        ##rename L1 all groups should be now changed

        _update_group(g1.group_id, u'R1', parent_id=g4.group_id)

        Session().commit()

    def test_change_parent_of_top_level_group_in_nested_setup_with_rename(self):

        g1 = fixture.create_repo_group(u'X1')

        ##rename L1 all groups should be now changed

        _update_group(g1.group_id, u'X1_PRIM', parent_id=g4.group_id)

        Session().commit()

from kallithea.tests import *

from kallithea.model.meta import Session

        RepoModel().delete(repo=repo)

        Session().commit()

    def test_remove_repo_repo_raises_exc_when_attached_forks(self):

        repo = fixture.create_repo(name=u'test-repo-1')

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

        Session().commit()

        # cleanup

        RepoModel().delete(repo=u'test-repo-fork-1')

        RepoModel().delete(repo=u'test-repo-1')

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

        Session().commit()

    def test_remove_repo_detach_forks(self):

        repo = fixture.create_repo(name=u'test-repo-1')

        Session().commit()

        try:

        finally:

            RepoModel().delete(repo=u'test-repo-fork-fork-1')

            RepoModel().delete(repo=u'test-repo-fork-1')