Changeset - 48a00daba2f2
[Not reviewed]
default
0 12 0
Lars Kruse - 8 years ago 2017-08-25 14:35:28
devel@sumpfralle.de
codingstyle: replace comparison for equality against None with "is" expression

Both style and correctness.

Reported by flake8.
12 files changed with 35 insertions and 37 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/db_manage.py
Show inline comments
 
@@ -299,49 +299,49 @@ class DbManage(object):
 
        hgsubversion = Ui()
 
        hgsubversion.ui_section = 'extensions'
 
        hgsubversion.ui_key = 'hgsubversion'
 
        hgsubversion.ui_value = ''
 
        hgsubversion.ui_active = False
 
        self.sa.add(hgsubversion)
 

	
 
        # enable hggit disabled by default
 
        hggit = Ui()
 
        hggit.ui_section = 'extensions'
 
        hggit.ui_key = 'hggit'
 
        hggit.ui_value = ''
 
        hggit.ui_active = False
 
        self.sa.add(hggit)
 

	
 
    def create_auth_plugin_options(self, skip_existing=False):
 
        """
 
        Create default auth plugin settings, and make it active
 

	
 
        :param skip_existing:
 
        """
 

	
 
        for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),
 
                     ('auth_internal_enabled', 'True', 'bool')]:
 
            if skip_existing and Setting.get_by_name(k) != None:
 
            if skip_existing and Setting.get_by_name(k) is not None:
 
                log.debug('Skipping option %s', k)
 
                continue
 
            setting = Setting(k, v, t)
 
            self.sa.add(setting)
 

	
 
    def create_default_options(self, skip_existing=False):
 
        """Creates default settings"""
 

	
 
        for k, v, t in [
 
            ('default_repo_enable_locking',  False, 'bool'),
 
            ('default_repo_enable_downloads', False, 'bool'),
 
            ('default_repo_enable_statistics', False, 'bool'),
 
            ('default_repo_private', False, 'bool'),
 
            ('default_repo_type', 'hg', 'unicode')]:
 

	
 
            if skip_existing and Setting.get_by_name(k) is not None:
 
                log.debug('Skipping option %s', k)
 
                continue
 
            setting = Setting(k, v, t)
 
            self.sa.add(setting)
 

	
 
    def fixup_groups(self):
 
        def_usr = User.get_default_user()
 
        for g in RepoGroup.query().all():
kallithea/lib/ipaddr.py
Show inline comments
 
@@ -748,60 +748,59 @@ class _BaseNet(_IPAddrBase):
 
            raise TypeError("%s is not a network object" % str(other))
 

	
 
        if other not in self:
 
            raise ValueError('%s not contained in %s' % (str(other),
 
                                                         str(self)))
 
        if other == self:
 
            return []
 

	
 
        ret_addrs = []
 

	
 
        # Make sure we're comparing the network of other.
 
        other = IPNetwork('%s/%s' % (str(other.network), str(other.prefixlen)),
 
                   version=other._version)
 

	
 
        s1, s2 = self.subnet()
 
        while s1 != other and s2 != other:
 
            if other in s1:
 
                ret_addrs.append(s2)
 
                s1, s2 = s1.subnet()
 
            elif other in s2:
 
                ret_addrs.append(s1)
 
                s1, s2 = s2.subnet()
 
            else:
 
                # If we got here, there's a bug somewhere.
 
                assert True == False, ('Error performing exclusion: '
 
                assert False, ('Error performing exclusion: '
 
                                       's1: %s s2: %s other: %s' %
 
                                       (str(s1), str(s2), str(other)))
 
        if s1 == other:
 
            ret_addrs.append(s2)
 
        elif s2 == other:
 
            ret_addrs.append(s1)
 
        else:
 
            # If we got here, there's a bug somewhere.
 
            assert True == False, ('Error performing exclusion: '
 
                                   's1: %s s2: %s other: %s' %
 
                                   (str(s1), str(s2), str(other)))
 
            assert False, ('Error performing exclusion: s1: %s s2: %s other: %s'
 
                           % (str(s1), str(s2), str(other)))
 

	
 
        return sorted(ret_addrs, key=_BaseNet._get_networks_key)
 

	
 
    def compare_networks(self, other):
 
        """Compare two IP objects.
 

	
 
        This is only concerned about the comparison of the integer
 
        representation of the network addresses.  This means that the
 
        host bits aren't considered at all in this method.  If you want
 
        to compare host bits, you can easily enough do a
 
        'HostA._ip < HostB._ip'
 

	
 
        Args:
 
            other: An IP object.
 

	
 
        Returns:
 
            If the IP versions of self and other are the same, returns:
 

	
 
            -1 if self < other:
 
              eg: IPv4('1.1.1.0/24') < IPv4('1.1.2.0/24')
 
              IPv6('1080::200C:417A') < IPv6('1080::200B:417B')
 
            0 if self == other
 
              eg: IPv4('1.1.1.1/24') == IPv4('1.1.1.2/24')
 
              IPv6('1080::200C:417A/96') == IPv6('1080::200C:417B/96')
kallithea/tests/api/api_base.py
Show inline comments
 
@@ -121,49 +121,49 @@ class _BaseTestApi(object):
 

	
 
    def _compare_ok(self, id_, expected, given):
 
        expected = jsonify({
 
            'id': id_,
 
            'error': None,
 
            'result': expected
 
        })
 
        given = json.loads(given)
 
        assert expected == given
 

	
 
    def _compare_error(self, id_, expected, given):
 
        expected = jsonify({
 
            'id': id_,
 
            'error': expected,
 
            'result': None
 
        })
 
        given = json.loads(given)
 
        assert expected == given
 

	
 
    def test_Optional_object(self):
 
        from kallithea.controllers.api.api import Optional
 

	
 
        option1 = Optional(None)
 
        assert '<Optional:%s>' % None == repr(option1)
 
        assert option1() == None
 
        assert option1() is None
 

	
 
        assert 1 == Optional.extract(Optional(1))
 
        assert 'trololo' == Optional.extract('trololo')
 

	
 
    def test_Optional_OAttr(self):
 
        from kallithea.controllers.api.api import Optional, OAttr
 

	
 
        option1 = Optional(OAttr('apiuser'))
 
        assert 'apiuser' == Optional.extract(option1)
 

	
 
    def test_OAttr_object(self):
 
        from kallithea.controllers.api.api import OAttr
 

	
 
        oattr1 = OAttr('apiuser')
 
        assert '<OptionalAttr:apiuser>' == repr(oattr1)
 
        assert oattr1() == oattr1
 

	
 
    def test_api_wrong_key(self):
 
        id_, params = _build_data('trololo', 'get_user')
 
        response = api_call(self, params)
 

	
 
        expected = 'Invalid API key'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
@@ -977,68 +977,68 @@ class _BaseTestApi(object):
 
        id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
 
                                  repoid=self.REPO, revision=rev,
 
                                  root_path=path,
 
                                  ret_type=ret_type)
 
        response = api_call(self, params)
 

	
 
        # we don't the actual return types here since it's tested somewhere
 
        # else
 
        expected = response.json['result']
 
        try:
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
 

	
 
    def test_api_create_repo(self):
 
        repo_name = u'api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo != None
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_and_repo_group(self):
 
        repo_name = u'my_gr/api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        print params
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo != None
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 
        fixture.destroy_repo_group(u'my_gr')
 

	
 
    def test_api_create_repo_in_repo_group_without_permission(self):
 
        repo_group_name = u'%s/api-repo-repo' % TEST_REPO_GROUP
 
        repo_name = u'%s/api-repo' % repo_group_name
 

	
 
        rg = fixture.create_repo_group(repo_group_name)
 
        Session().commit()
 
        RepoGroupModel().grant_user_permission(repo_group_name,
 
                                               self.TEST_USER_LOGIN,
 
                                               'group.none')
 
        Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
@@ -1062,69 +1062,69 @@ class _BaseTestApi(object):
 
        fixture.destroy_repo_group(repo_group_name)
 

	
 
    def test_api_create_repo_unknown_owner(self):
 
        repo_name = u'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=owner,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 
        expected = 'user `%s` does not exist' % owner
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_repo_dont_specify_owner(self):
 
        repo_name = u'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo != None
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_by_non_admin(self):
 
        repo_name = u'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
        )
 
        response = api_call(self, params)
 

	
 
        repo = RepoModel().get_by_repo_name(repo_name)
 
        assert repo != None
 
        assert repo is not None
 
        ret = {
 
            'msg': 'Created new repository `%s`' % repo_name,
 
            'success': True,
 
            'task': None,
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_by_non_admin_specify_owner(self):
 
        repo_name = u'api-repo'
 
        owner = 'i-dont-exist'
 
        id_, params = _build_data(self.apikey_regular, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  repo_type=self.REPO_TYPE,
 
                                  owner=owner)
 
        response = api_call(self, params)
 

	
 
        expected = 'Only Kallithea admin can specify `owner` param'
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_create_repo_exists(self):
 
        repo_name = self.REPO
kallithea/tests/functional/test_admin_repos.py
Show inline comments
 
@@ -368,49 +368,49 @@ class _BaseTestCase(TestController):
 
        assert new_repo.description == description
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name))
 
        response.mustcontain(repo_name)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
 

	
 
        response.follow()
 

	
 
        # check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name).scalar()
 

	
 
        assert deleted_repo == None
 
        assert deleted_repo is None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
 

	
 
    def test_delete_non_ascii(self):
 
        self.log_user()
 
        non_ascii = "ąęł"
 
        repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
 
        repo_name_unicode = safe_unicode(repo_name)
 
        description = 'description for newly created repo' + non_ascii
 
        description_unicode = safe_unicode(description)
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _authentication_token=self.authentication_token()))
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=repo_name))
 
        assert response.json == {u'result': True}
 
        self.checkSessionFlash(response,
 
                               u'Created repository <a href="/%s">%s</a>'
 
                               % (urllib.quote(repo_name), repo_name_unicode))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
@@ -418,49 +418,49 @@ class _BaseTestCase(TestController):
 

	
 
        assert new_repo.repo_name == repo_name_unicode
 
        assert new_repo.description == description_unicode
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name))
 
        response.mustcontain(repo_name)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
 
        response.follow()
 

	
 
        # check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_unicode).scalar()
 

	
 
        assert deleted_repo == None
 
        assert deleted_repo is None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
 

	
 
    def test_delete_repo_with_group(self):
 
        # TODO:
 
        pass
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('delete_repo', repo_name=self.REPO),
 
                                 params=dict(_authentication_token=self.authentication_token()))
 

	
 
    def test_show(self):
 
        self.log_user()
 
        response = self.app.get(url('summary_home', repo_name=self.REPO))
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_repo', repo_name=self.REPO))
 

	
 
    def test_set_private_flag_sets_default_to_none(self):
 
        self.log_user()
 
        # initially repository perm should be read
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.read'
 
@@ -591,49 +591,49 @@ class _BaseTestCase(TestController):
 
                                                _authentication_token=self.authentication_token()))
 

	
 
        response.mustcontain('<span class="error-message">Invalid value</span>')
 

	
 
        RepoModel().delete(repo_name)
 
        Session().commit()
 

	
 
    @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
 
    def test_create_repo_when_filesystem_op_fails(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = 'description for newly created repo'
 

	
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _authentication_token=self.authentication_token()))
 

	
 
        self.checkSessionFlash(response,
 
                               'Error creating repository %s' % repo_name)
 
        # repo must not be in db
 
        repo = Repository.get_by_repo_name(repo_name)
 
        assert repo == None
 
        assert repo is None
 

	
 
        # repo must not be in filesystem !
 
        assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 

	
 

	
 
class TestAdminReposControllerGIT(_BaseTestCase):
 
    REPO = GIT_REPO
 
    REPO_TYPE = 'git'
 
    NEW_REPO = NEW_GIT_REPO
 
    OTHER_TYPE_REPO = HG_REPO
 
    OTHER_TYPE = 'hg'
 

	
 

	
 
class TestAdminReposControllerHG(_BaseTestCase):
 
    REPO = HG_REPO
 
    REPO_TYPE = 'hg'
 
    NEW_REPO = NEW_HG_REPO
 
    OTHER_TYPE_REPO = GIT_REPO
 
    OTHER_TYPE = 'git'
 

	
 
    def test_permanent_url_protocol_access(self):
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        permanent_name = '_%d' % repo.repo_id
 

	
kallithea/tests/functional/test_admin_user_groups.py
Show inline comments
 
@@ -39,49 +39,49 @@ class TestAdminUsersGroupsController(Tes
 
                                 params=dict(_authentication_token=self.authentication_token()))
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 

	
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).one()
 

	
 
        response = self.app.post(url('delete_users_group', id=gr.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).scalar()
 

	
 
        assert gr == None
 
        assert gr is None
 

	
 
    def test_default_perms_enable_repository_read_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.post(url('edit_user_group_default_perms_update',
 
                                     id=ug.users_group_id),
 
                                 {'create_repo_perm': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.repository')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
@@ -102,49 +102,49 @@ class TestAdminUsersGroupsController(Tes
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 

	
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('delete_users_group', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).scalar()
 

	
 
        assert gr == None
 
        assert gr is None
 
        p = Permission.get_by_key('hg.create.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_default_perms_enable_repository_fork_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.post(url('edit_user_group_default_perms_update',
 
                                     id=ug.users_group_id),
 
                                 {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
 
@@ -170,35 +170,35 @@ class TestAdminUsersGroupsController(Tes
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('delete_users_group', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
                           .filter(UserGroup.users_group_name ==
 
                                   users_group_name).scalar()
 

	
 
        assert gr == None
 
        assert gr is None
 
        p = Permission.get_by_key('hg.fork.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('delete_users_group', id=1),
 
                                 params=dict(_authentication_token=self.authentication_token()))
kallithea/tests/functional/test_login.py
Show inline comments
 
@@ -322,49 +322,49 @@ class TestLoginController(TestController
 
    def test_register_ok(self):
 
        username = 'test_regular4'
 
        password = 'qweqwe'
 
        email = 'user4@example.com'
 
        name = 'testname'
 
        lastname = 'testlastname'
 

	
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': username,
 
                                             'password': password,
 
                                             'password_confirmation': password,
 
                                             'email': email,
 
                                             'firstname': name,
 
                                             'lastname': lastname,
 
                                             'admin': True})  # This should be overridden
 
        assert response.status == '302 Found'
 
        self.checkSessionFlash(response, 'You have successfully registered with Kallithea')
 

	
 
        ret = Session().query(User).filter(User.username == 'test_regular4').one()
 
        assert ret.username == username
 
        assert check_password(password, ret.password) == True
 
        assert ret.email == email
 
        assert ret.name == name
 
        assert ret.lastname == lastname
 
        assert ret.api_key != None
 
        assert ret.api_key is not None
 
        assert ret.admin == False
 

	
 
    #==========================================================================
 
    # PASSWORD RESET
 
    #==========================================================================
 

	
 
    def test_forgot_password_wrong_mail(self):
 
        bad_email = 'username%wrongmail.org'
 
        response = self.app.post(
 
                        url(controller='login', action='password_reset'),
 
                            {'email': bad_email, }
 
        )
 

	
 
        response.mustcontain('An email address must contain a single @')
 

	
 
    def test_forgot_password(self):
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset'))
 
        assert response.status == '200 OK'
 

	
 
        username = 'test_password_reset_1'
 
        password = 'qweqwe'
 
        email = 'username@example.com'
 
        name = u'passwd'
kallithea/tests/functional/test_pullrequests.py
Show inline comments
 
@@ -136,85 +136,85 @@ class TestPullrequestsController(TestCon
 
        response.mustcontain('Meanwhile, the following reviewers have been removed: test_admin')
 
        response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
 
        response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user2.user_id)
 
        response.mustcontain(no='<input type="hidden" value="%s" name="review_members" />' % admin_user.user_id)
 

	
 
    def test_update_with_invalid_reviewer(self):
 
        invalid_user_id = 99999
 
        self.log_user()
 
        # create a valid pull request
 
        response = self.app.post(url(controller='pullrequests', action='create',
 
                                     repo_name=HG_REPO),
 
                                 {
 
                                  'org_repo': HG_REPO,
 
                                  'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
 
                                  'other_repo': HG_REPO,
 
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  '_authentication_token': self.authentication_token(),
 
                                 },
 
                                status=302)
 
        # location is of the form:
 
        # http://localhost/vcs_test_hg/pull-request/54/_/title
 
        m = re.search('/pull-request/(\d+)/', response.location)
 
        assert m != None
 
        assert m is not None
 
        pull_request_id = m.group(1)
 

	
 
        # update it
 
        response = self.app.post(url(controller='pullrequests', action='post',
 
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
 
                                 {
 
                                  'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  'owner': TEST_USER_ADMIN_LOGIN,
 
                                  '_authentication_token': self.authentication_token(),
 
                                  'review_members': [str(invalid_user_id)],
 
                                 },
 
                                 status=400)
 
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
 

	
 
    def test_edit_with_invalid_reviewer(self):
 
        invalid_user_id = 99999
 
        self.log_user()
 
        # create a valid pull request
 
        response = self.app.post(url(controller='pullrequests', action='create',
 
                                     repo_name=HG_REPO),
 
                                 {
 
                                  'org_repo': HG_REPO,
 
                                  'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
 
                                  'other_repo': HG_REPO,
 
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  '_authentication_token': self.authentication_token(),
 
                                 },
 
                                status=302)
 
        # location is of the form:
 
        # http://localhost/vcs_test_hg/pull-request/54/_/title
 
        m = re.search('/pull-request/(\d+)/', response.location)
 
        assert m != None
 
        assert m is not None
 
        pull_request_id = m.group(1)
 

	
 
        # edit it
 
        response = self.app.post(url(controller='pullrequests', action='post',
 
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
 
                                 {
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  'owner': TEST_USER_ADMIN_LOGIN,
 
                                  '_authentication_token': self.authentication_token(),
 
                                  'review_members': [str(invalid_user_id)],
 
                                 },
 
                                 status=400)
 
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
 

	
 
    def test_iteration_refs(self):
 
        # Repo graph excerpt:
 
        #   o   fb95b340e0d0 webvcs
 
        #  /:
 
        # o :   41d2568309a0 default
 
        # : :
 
        # : o   5ec21f21aafe webvcs
 
        # : :
 
        # : o   9e6119747791 webvcs
kallithea/tests/models/test_permissions.py
Show inline comments
 
@@ -663,37 +663,37 @@ class TestPermissions(TestController):
 
                .all()
 
        Session().delete(perms[0])
 
        Session().commit()
 

	
 
        self._test_def_perm_equal(user=self.u1, change_factor=-1)
 

	
 
        # create missing one !
 
        PermissionModel().create_default_permissions(user=self.u1)
 
        self._test_def_perm_equal(user=self.u1)
 

	
 
    @parametrize('perm,modify_to', [
 
        ('repository.read', 'repository.none'),
 
        ('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)
 
        assert old != None
 
        assert new != None
 
        assert old is not None
 
        assert new is not None
 

	
 
        # now modify permissions
 
        p = UserToPerm.query() \
 
                .filter(UserToPerm.user == self.u1) \
 
                .filter(UserToPerm.permission == old) \
 
                .one()
 
        p.permission = new
 
        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
 
@@ -64,71 +64,71 @@ class TestRepoGroups(TestController):
 
        assert g.full_path == 'newGroup'
 

	
 
        assert self.__check_path('newGroup')
 

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

	
 
    def test_same_subgroup(self):
 
        sg1 = fixture.create_repo_group(u'sub1', parent_group_id=self.g1.group_id)
 
        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', parent_group_id=sg1.group_id)
 
        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)
 

	
 
        assert RepoGroup.get(sg1.group_id) == None
 
        assert RepoGroup.get(sg1.group_id) is None
 
        assert not self.__check_path('deteteme')
 

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

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

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

	
 
        new_sg1 = _update_repo_group(sg1.group_id, u'after')
 
        assert self.__check_path('after')
 
        assert RepoGroup.get_by_group_name(u'initial') == None
 
        assert RepoGroup.get_by_group_name(u'initial') is None
 

	
 
    def test_update_group_parent(self):
 

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

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

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

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

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

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

	
 
        _update_repo_group(g1.group_id, u'g1', parent_id=g2.group_id)
kallithea/tests/models/test_repos.py
Show inline comments
 
import pytest
 
from kallithea.tests.base 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(TestController):
 

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

	
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-1')
 
        assert Repository.get_by_repo_name(repo_name=u'test-repo-1') is None
 

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

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

	
 
        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')
 
        assert Repository.get_by_repo_name(repo_name=u'test-repo-1') is None
 
        assert Repository.get_by_repo_name(repo_name=u'test-repo-fork-1') is None
 
        assert Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1') is None
 

	
 
    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:
 
            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')
 
            assert Repository.get_by_repo_name(repo_name=u'test-repo-1') is None
 
            assert Repository.get_by_repo_name(repo_name=u'test-repo-fork-1') is not None
 
            assert Repository.get_by_repo_name(repo_name=u'test-repo-fork-fork-1') is not None
 
        finally:
 
            RepoModel().delete(repo=u'test-repo-fork-fork-1')
 
            RepoModel().delete(repo=u'test-repo-fork-1')
 
            Session().commit()
kallithea/tests/models/test_users.py
Show inline comments
 
@@ -69,55 +69,54 @@ class TestUser(TestController):
 
        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')
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 

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

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

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

	
 
        assert u is None
 

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

	
 

	
 
class TestUsers(TestController):
 

	
 
    def setup_method(self, method):
 
        self.u1 = UserModel().create_or_update(username=u'u1',
 
                                        password=u'qweqwe',
 
                                        email=u'u1@example.com',
 
                                        firstname=u'u1', lastname=u'u1')
 

	
 
    def teardown_method(self, method):
 
        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]
kallithea/tests/other/test_validators.py
Show inline comments
 
@@ -82,49 +82,49 @@ class TestRepoGroups(TestController):
 
    def test_ValidRepoGroup(self):
 
        validator = v.ValidRepoGroup()
 
        model = RepoGroupModel()
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python({'group_name': HG_REPO, })
 
        gr = model.create(group_name=u'test_gr', group_description=u'desc',
 
                          parent=None,
 
                          just_db=True,
 
                          owner=TEST_USER_ADMIN_LOGIN)
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python({'group_name': gr.group_name, })
 

	
 
        validator = v.ValidRepoGroup(edit=True,
 
                                      old_data={'group_id':  gr.group_id})
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python({
 
                                        'group_name': gr.group_name + 'n',
 
                                        'parent_group_id': gr.group_id
 
                                        })
 
        model.delete(gr)
 

	
 
    def test_ValidPassword(self):
 
        validator = v.ValidPassword()
 
        assert 'lol' == validator.to_python('lol')
 
        assert None == validator.to_python(None)
 
        assert validator.to_python(None) is None
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('ąćżź')
 

	
 
    def test_ValidPasswordsMatch(self):
 
        validator = v.ValidPasswordsMatch('new_password', 'password_confirmation')
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python({'new_password': 'pass',
 
                                          'password_confirmation': 'pass2'})
 

	
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python({'new_password': 'pass',
 
                                          'password_confirmation': 'pass2'})
 

	
 
        assert {'new_password': 'pass',
 
                          'password_confirmation': 'pass'} == validator.to_python({'new_password': 'pass',
 
                                         'password_confirmation': 'pass'})
 

	
 
        assert {'new_password': 'pass',
 
                          'password_confirmation': 'pass'} == validator.to_python({'new_password': 'pass',
 
                                         'password_confirmation': 'pass'})
 

	
 
    def test_ValidAuth(self):
 
        validator = v.ValidAuth()
 
        valid_creds = {
0 comments (0 inline, 0 general)