kallithea Changeset - 48a00daba2f2

Changeset - 48a00daba2f2

Parent rev.

Child rev.

[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 38 insertions and 40 deletions:

kallithea/lib/db_manage.py

kallithea/lib/ipaddr.py

kallithea/tests/api/api_base.py

kallithea/tests/functional/test_admin_repos.py

kallithea/tests/functional/test_admin_user_groups.py

kallithea/tests/functional/test_login.py

kallithea/tests/functional/test_pullrequests.py

kallithea/tests/models/test_permissions.py

kallithea/tests/models/test_repo_groups.py

kallithea/tests/models/test_repos.py

kallithea/tests/models/test_users.py

kallithea/tests/other/test_validators.py

0 comments (0 inline, 0 general)

kallithea/lib/db_manage.py

➞

Show inline comments

@@ @@ -274,98 +274,98 @@ class DbManage(object): @@
         self.sa.add(hooks5)
         hooks6 = Ui()
         hooks6.ui_section = 'hooks'
         hooks6.ui_key = Ui.HOOK_PRE_PULL
         hooks6.ui_value = 'python:kallithea.lib.hooks.pre_pull'
         self.sa.add(hooks6)
         # enable largefiles
         largefiles = Ui()
         largefiles.ui_section = 'extensions'
         largefiles.ui_key = 'largefiles'
         largefiles.ui_value = ''
         self.sa.add(largefiles)
         # set default largefiles cache dir, defaults to
         # /repo location/.cache/largefiles
         largefiles = Ui()
         largefiles.ui_section = 'largefiles'
         largefiles.ui_key = 'usercache'
         largefiles.ui_value = os.path.join(repo_store_path, '.cache',
                                            'largefiles')
         self.sa.add(largefiles)
         # enable hgsubversion disabled by default
         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:
                         ('auth_internal_enabled', 'True', 'bool')]:
             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():
             g.group_name = g.get_new_name(g.name)
             # get default perm
             default = UserRepoGroupToPerm.query() \
                 .filter(UserRepoGroupToPerm.group == g) \
                 .filter(UserRepoGroupToPerm.user == def_usr) \
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding', g)
                 RepoGroupModel()._create_default_perms(g)
     def reset_permissions(self, username):
         """
         Resets permissions to default state, useful when old systems had
         bad permissions, we must clean them up
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
         u2p = UserToPerm.query() \
             .filter(UserToPerm.user == default_user).all()

kallithea/lib/ipaddr.py

➞

Show inline comments

@@ @@ -724,108 +724,107 @@ class _BaseNet(_IPAddrBase): @@
                 IPNetwork('::2/127'),
                 IPNetwork('::4/126'),
                 IPNetwork('::8/125'),
                 ...
                 IPNetwork('0:0:8000::/33')]
         Args:
             other: An IPvXNetwork object of the same type.
         Returns:
             A sorted list of IPvXNetwork objects addresses which is self
             minus other.
         Raises:
             TypeError: If self and other are of differing address
               versions, or if other is not a network object.
             ValueError: If other is not completely contained by self.
         """
         if not self._version == other._version:
             raise TypeError("%s and %s are not of the same version" % (
                 str(self), str(other)))
         if not isinstance(other, _BaseNet):
             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: '
                                        '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)))
         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')
 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')
 if self > other
               eg: IPv4('1.1.1.0/24') > IPv4('1.1.0.0/24')
               IPv6('1080::1:200C:417A/112') >
               IPv6('1080::0:200C:417A/112')
             If the IP versions of self and other are different, returns:
             -1 if self._version < other._version
               eg: IPv4('10.0.0.1/24') < IPv6('::1/128')
 if self._version > other._version
               eg: IPv6('::1/128') > IPv4('255.255.255.0/24')
         """
         if self._version < other._version:
             return -1
         if self._version > other._version:
             return 1
         # self._version == other._version below here:
         if self.network < other.network:
             return -1
         if self.network > other.network:
             return 1
         # self.network == other.network below here:
         if self.netmask < other.netmask:

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -97,97 +97,97 @@ class _BaseTestApi(object): @@
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@example.com',
             firstname=u'first',
             lastname=u'last'
+        )
         Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
     @classmethod
     def teardown_class(cls):
         pass
     def setup_method(self, method):
         make_user_group()
         make_repo_group()
     def teardown_method(self, method):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
     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)
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_args_is_null(self):
@@ @@ -953,202 +953,202 @@ class _BaseTestApi(object): @@
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         expected = ('ret_type must be one of %s'
                     % (','.join(['files', 'dirs', 'all'])))
         self._compare_error(id_, expected, given=response.body)
     @parametrize('name,ret_type,grant_perm', [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         Session().commit()
         rev = 'tip'
         path = '/'
         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,
+        )
         response = api_call(self, params)
         # Current result when API access control is different from Web:
         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)
         # Expected and arguably more correct result:
         #expected = 'failed to create repository `%s`' % repo_name
         #self._compare_error(id_, expected, given=response.body)
         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
         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)
         expected = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create', crash)
     def test_api_create_repo_exception_occurred(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)
         expected = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
     @parametrize('changing_attr,updates', [
         ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': u'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}),
         ('clone_uri', {'clone_uri': None}),

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

@@ @@ -344,147 +344,147 @@ class _BaseTestCase(TestController): @@
                                                 clone_uri='svn+http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('Invalid repository URL')
     def test_delete(self):
         self.log_user()
         repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_name=repo_name,
                                                 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))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         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) \
             .filter(Repository.repo_name == repo_name_unicode).one()
         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'
         assert Repository.get_by_repo_name(self.REPO).private == False
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == True
         # now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
@@ @@ -567,78 +567,78 @@ class _BaseTestCase(TestController): @@
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO + u'no_perms'
         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()))
         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
         # 400 Bad Request - Unable to detect pull/push action
         self.app.get(url('summary_home', repo_name=permanent_name),
             extra_environ={'HTTP_ACCEPT': 'application/mercurial'},
             status=400,
+        )

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

@@ @@ -15,190 +15,190 @@ class TestAdminUsersGroupsController(Tes @@
     def test_create(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP
         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 <a href="/_admin/user_groups/')
         self.checkSessionFlash(response,
                                '/edit">%s</a>' % TEST_USER_GROUP)
     def test_new(self):
         response = self.app.get(url('new_users_group'))
     def test_update(self):
         response = self.app.post(url('update_users_group', id=1), status=403)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('update_users_group', id=1),
                                  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')
         # 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]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(
             url('edit_user_group_default_perms_update', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         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()})
         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.repository')
         # 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]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(url('edit_user_group_default_perms_update', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         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

@@ @@ -298,97 +298,97 @@ class TestLoginController(TestController @@
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test'})
         with test_context(self.app):
             msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         with test_context(self.app):
             msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     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'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
         token = "bad"

kallithea/tests/functional/test_pullrequests.py

➞

Show inline comments

@@ @@ -112,133 +112,133 @@ class TestPullrequestsController(TestCon @@
                                  status=302)
         pull_request2_id = re.search('/pull-request/(\d+)/', response.location).group(1)
         assert pull_request2_id != pull_request1_id
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewer was added
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         # update without creating new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request2_id),
+                                 {
                                   'pullrequest_title': 'Title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'org_review_members': [admin_user.user_id], # fake - just to get some 'meanwhile' warning ... but it is also added ...
                                   'review_members': [regular_user2.user_id, admin_user.user_id],
                                  },
                                  status=302)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewers were added / removed
         response.mustcontain('Meanwhile, the following reviewers have been added: test_regular')
         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
         # : :
         # o :   3d1091ee5a53 default
         # :/
         # o     948da46b29c1 default
         self.log_user()
         # create initial PR
         response = self.app.post(
             url(controller='pullrequests', action='create', repo_name=HG_REPO),
+            {
                 'org_repo': HG_REPO,
                 'org_ref': 'rev:9e6119747791:9e6119747791ff886a5abe1193a730b6bf874e1c',
                 'other_repo': HG_REPO,
                 'other_ref': 'branch:default:3d1091ee5a533b1f4577ec7d8a226bb315fb1336',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 '_authentication_token': self.authentication_token(),
             },
             status=302)
         pr1_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         assert pr1.org_ref == 'branch:webvcs:9e6119747791ff886a5abe1193a730b6bf874e1c'

kallithea/tests/models/test_permissions.py

➞

Show inline comments

@@ @@ -639,61 +639,61 @@ class TestPermissions(TestController): @@
         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)
         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()
         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()
         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

@@ @@ -40,119 +40,119 @@ class TestRepoGroups(TestController): @@
         self.g1 = fixture.create_repo_group(u'test1', skip_if_exists=True)
         self.g2 = fixture.create_repo_group(u'test2', skip_if_exists=True)
         self.g3 = fixture.create_repo_group(u'test3', skip_if_exists=True)
     def teardown_method(self, method):
         Session.remove()
     def __check_path(self, *path):
         """
         Checks the path for existence !
         """
         path = [TESTS_TMP_PATH] + list(path)
         path = os.path.join(*path)
         return os.path.isdir(path)
     def _check_folders(self):
         print os.listdir(TESTS_TMP_PATH)
     def __delete_group(self, id_):
         RepoGroupModel().delete(id_)
     def test_create_group(self):
         g = fixture.create_repo_group(u'newGroup')
         Session().commit()
         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)
         assert self.__check_path('g2', 'g1')
         # test repo
         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', parent_group_id=g1.group_id)
         assert g2.full_path == 't11/t22'
         assert self.__check_path('t11', 't22')
         g2 = _update_repo_group(g2.group_id, u'g22', parent_id=None)
         Session().commit()
         assert g2.group_name == 'g22'
         # we moved out group from t1 to '' so it's full path should be 'g2'
         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')

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

@@ @@ -45,99 +45,98 @@ class TestUser(TestController): @@
         assert UserGroupMember.query().count() == 1
         UserModel().delete(usr.user_id)
         Session().commit()
         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()
         with pytest.raises(AttributeError):
             m = UserEmailMap()
             m.email = u'main_email@example.com'
             m.user = usr
             Session().add(m)
             Session().commit()
         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')
         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]
         UserModel().grant_perm(self.u1, perm)
         Session().commit()
         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)
             assert False == has_p
     def test_revoke_perm(self):
         perm = Permission.query().all()[0]
         UserModel().grant_perm(self.u1, perm)
         Session().commit()
         assert UserModel().has_perm(self.u1, perm) == True
         # revoke
         UserModel().revoke_perm(self.u1, perm)
         Session().commit()
         assert UserModel().has_perm(self.u1, perm) == False

kallithea/tests/other/test_validators.py

➞

Show inline comments

@@ @@ -58,97 +58,97 @@ class TestRepoGroups(TestController): @@
             validator.to_python(u'default')
         with pytest.raises(formencode.Invalid):
             validator.to_python(u'.,')
         gr = fixture.create_user_group(u'test')
         gr2 = fixture.create_user_group(u'tes2')
         Session().commit()
         with pytest.raises(formencode.Invalid):
             validator.to_python(u'test')
         assert gr.users_group_id is not None
         validator = v.ValidUserGroup(edit=True,
                                     old_data={'users_group_id':
                                               gr2.users_group_id})
         with pytest.raises(formencode.Invalid):
             validator.to_python(u'test')
         with pytest.raises(formencode.Invalid):
             validator.to_python(u'TesT')
         with pytest.raises(formencode.Invalid):
             validator.to_python(u'TEST')
         UserGroupModel().delete(gr)
         UserGroupModel().delete(gr2)
         Session().commit()
     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 = {
             'username': TEST_USER_REGULAR2_LOGIN,
             'password': TEST_USER_REGULAR2_PASS,
+        }
         invalid_creds = {
             'username': 'err',
             'password': 'err',
+        }
         assert valid_creds == validator.to_python(valid_creds)
         with pytest.raises(formencode.Invalid):
             validator.to_python(invalid_creds)
     def test_ValidAuthToken(self):
         validator = v.ValidAuthToken()
         # this is untestable without a threadlocal
 #        self.assertRaises(formencode.Invalid,
 #                          validator.to_python, 'BadToken')
         validator
     def test_ValidRepoName(self):
         validator = v.ValidRepoName()
         with pytest.raises(formencode.Invalid):
             validator.to_python({'repo_name': ''})

0 comments (0 inline, 0 general)