def setUp(self):

        for x in ChangesetComment.query().all():

            Session().delete(x)

        Session().commit()

        for x in Notification.query().all():

            Session().delete(x)

        Session().commit()

    def tearDown(self):

        for x in ChangesetComment.query().all():

            Session().delete(x)

        Session().commit()

        for x in Notification.query().all():

            Session().delete(x)

        Session().commit()

    def test_create(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params)

        # Test response...

        self.assertEqual(response.status, '302 Found')

        response.follow()

        response = self.app.get(url(controller='changeset', action='index',

                                repo_name=HG_REPO, revision=rev))

        # test DB

        self.assertEqual(ChangesetComment.query().count(), 1)

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (0 inline, 1 general)'''

        )

        self.assertEqual(Notification.query().count(), 1)

        self.assertEqual(ChangesetComment.query().count(), 1)

        notification = Notification.query().all()[0]

        ID = ChangesetComment.query().first().comment_id

        self.assertEqual(notification.type_,

                         Notification.TYPE_CHANGESET_COMMENT)

        print "%s vs %s" % (sbj, notification.subject)

        self.assertTrue(sbj in notification.subject)

    def test_create_inline(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        f_path = 'vcs/web/simplevcs/views/repository.py'

        line = 'n1'

        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params)

        # Test response...

        self.assertEqual(response.status, '302 Found')

        response.follow()

        response = self.app.get(url(controller='changeset', action='index',

                                repo_name=HG_REPO, revision=rev))

        #test DB

        self.assertEqual(ChangesetComment.query().count(), 1)

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (1 inline, 0 general)'''

        )

        response.mustcontain(

            '''<div style="display:none" class="inline-comment-placeholder" '''

            '''path="vcs/web/simplevcs/views/repository.py" '''

            '''target_id="vcswebsimplevcsviewsrepositorypy">'''

        )

        self.assertEqual(Notification.query().count(), 1)

        self.assertEqual(ChangesetComment.query().count(), 1)

        notification = Notification.query().all()[0]

        ID = ChangesetComment.query().first().comment_id

        self.assertEqual(notification.type_,

                         Notification.TYPE_CHANGESET_COMMENT)

        print "%s vs %s" % (sbj, notification.subject)

        self.assertTrue(sbj in notification.subject)

    def test_create_with_mention(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params)

        # Test response...

        self.assertEqual(response.status, '302 Found')

        response.follow()

        response = self.app.get(url(controller='changeset', action='index',

                                repo_name=HG_REPO, revision=rev))

        # test DB

        self.assertEqual(ChangesetComment.query().count(), 1)

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (0 inline, 1 general)'''

        )

        self.assertEqual(Notification.query().count(), 2)

        users = [x.user.username for x in UserNotification.query().all()]

        # test_regular gets notification by @mention

        self.assertEqual(sorted(users), [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN])

    def test_delete(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params)

        comments = ChangesetComment.query().all()

        self.assertEqual(len(comments), 1)

        comment_id = comments[0].comment_id

        self.app.delete(url(controller='changeset',

                                    action='delete_comment',

# -*- coding: utf-8 -*-

import os

from kallithea.tests import *

from kallithea.model.db import Repository

from kallithea.model.meta import Session

from kallithea.tests.fixture import Fixture

fixture = Fixture()

ARCHIVE_SPECS = {

    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),

    '.tar.gz': ('application/x-gzip', 'tgz', ''),

    '.zip': ('application/zip', 'zip', ''),

}

HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')

GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')

def _set_downloads(repo_name, set_to):

    repo = Repository.get_by_repo_name(repo_name)

    repo.enable_downloads = set_to

    Session().add(repo)

    Session().commit()

class TestFilesController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=HG_REPO,

                                    revision='tip',

                                    f_path='/'))

        # Test response...

    def test_index_revision(self):

        self.log_user()

        response = self.app.get(

            url(controller='files', action='index',

                repo_name=HG_REPO,

                revision='7ba66bec8d6dbba14a2155be32408c435c5f4492',

                f_path='/')

        )

        #Test response...

        response.mustcontain('1.1 KiB')

    def test_index_different_branch(self):

        self.log_user()

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=HG_REPO,

                                    revision='97e8b885c04894463c51898e14387d80c30ed1ee',

                                    f_path='/'))

        response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")

    def test_index_paging(self):

        self.log_user()

        for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),

                  (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),

                  (109, '75feb4c33e81186c87eac740cee2447330288412'),

                  (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),

                  (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:

            response = self.app.get(url(controller='files', action='index',

                                    repo_name=HG_REPO,

                                    revision=r[1],

                                    f_path='/'))

            response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))

    def test_file_source(self):

        self.log_user()

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=HG_REPO,

                                    revision='8911406ad776fdd3d0b9932a2e89677e57405a48',

                                    f_path='vcs/nodes.py'))

        response.mustcontain("""<div class="commit">Partially implemented <a class="issue-tracker-link" href="https://myissueserver.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.

In addition some other __str__ are unicode as well

Added test for unicode

Improved test to clone into uniq repository.

removed extra unicode conversion in diff.</div>

""")

        response.mustcontain("""<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">default at 8911406ad776</option>""")

    def test_file_source_history(self):

        self.log_user()

        response = self.app.get(url(controller='files', action='history',

                                    repo_name=HG_REPO,

        PermissionModel().create_default_permissions(user=User.DEFAULT_USER,

                                                     force=True)

        Session().commit()

    def setUp(self):

        self.u1 = UserModel().create_or_update(

            username=u'u1', password=u'qweqwe',

            email=u'u1@example.com', firstname=u'u1', lastname=u'u1'

        )

        self.u2 = UserModel().create_or_update(

            username=u'u2', password=u'qweqwe',

            email=u'u2@example.com', firstname=u'u2', lastname=u'u2'

        )

        self.u3 = UserModel().create_or_update(

            username=u'u3', password=u'qweqwe',

            email=u'u3@example.com', firstname=u'u3', lastname=u'u3'

        )

        self.anon = User.get_default_user()

        self.a1 = UserModel().create_or_update(

            username=u'a1', password=u'qweqwe',

            email=u'a1@example.com', firstname=u'a1', lastname=u'a1', admin=True

        )

        Session().commit()

    def tearDown(self):

        if hasattr(self, 'test_repo'):

            RepoModel().delete(repo=self.test_repo)

        UserModel().delete(self.u1)

        UserModel().delete(self.u2)

        UserModel().delete(self.u3)

        UserModel().delete(self.a1)

        if hasattr(self, 'g1'):

            RepoGroupModel().delete(self.g1.group_id)

        if hasattr(self, 'g2'):

            RepoGroupModel().delete(self.g2.group_id)

        if hasattr(self, 'ug1'):

            UserGroupModel().delete(self.ug1, force=True)

        Session().commit()

    def test_default_perms_set(self):

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

        perms = {

            'repositories_groups': {},

            'global': set([u'hg.create.repository', u'repository.read',

                           u'hg.register.manual_activate']),

        }

        self.assertEqual(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)

    def test_default_admin_perms_set(self):

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

        perms = {

            'repositories_groups': {},

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

        }

        self.assertEqual(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])

    def test_default_group_perms(self):

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

        self.g2 = fixture.create_repo_group('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),

        }

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

    def test_default_admin_group_perms(self):

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

        self.g2 = fixture.create_repo_group('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']),

        }

        self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],

                         perms['repositories'][HG_REPO])

        self.assertEqual(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('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)

        # 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([u'hg.create.repository', u'repository.read',

                           u'hg.register.manual_activate']),

        }

        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],

                         new_perm)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                         perms['repositories_groups'])

    def test_propagated_permission_from_users_group(self):

        # make group

        self.ug1 = fixture.create_user_group('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([u'hg.create.repository', u'repository.read',

                           u'hg.register.manual_activate']),

        }

        self.assertEqual(u3_auth.permissions['repositories'][HG_REPO],

                         new_perm_gr)

        self.assertEqual(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('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)

        # 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([u'hg.create.repository', u'repository.read',

                           u'hg.register.manual_activate']),

        }

        self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],

                         new_perm_h)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                         perms['repositories_groups'])

    def test_repo_in_group_permissions(self):

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

        self.g2 = fixture.create_repo_group('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'})

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

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

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

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

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