lint: ruff run

rhodecode/api/tests/test_create_user.py

@@ -1,206 +1,206 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.lib.auth import check_password
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (
     TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_EMAIL)
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, jsonify, crash)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.model.db import RepoGroup
 
 
 # TODO: mikhail: remove fixture from here
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateUser(object):
     def test_api_create_existing_user(self):
         id_, params = build_data(
             self.apikey, 'create_user',
             username=TEST_USER_ADMIN_LOGIN,
             email='[e-mail]',
             password='trololo')
         response = api_call(self.app, params)
 
         expected = "user `%s` already exist" % (TEST_USER_ADMIN_LOGIN,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_with_existing_email(self):
         id_, params = build_data(
             self.apikey, 'create_user',
             username=TEST_USER_ADMIN_LOGIN + 'new',
             email=TEST_USER_REGULAR_EMAIL,
             password='trololo')
         response = api_call(self.app, params)
 
         expected = "email `%s` already exist" % (TEST_USER_REGULAR_EMAIL,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user_with_wrong_username(self):
         bad_username = '<> HELLO WORLD <>'
         id_, params = build_data(
             self.apikey, 'create_user',
             username=bad_username,
             email='[e-mail]',
             password='trololo')
         response = api_call(self.app, params)
 
         expected = {'username':
                         "Username may only contain alphanumeric characters "
                         "underscores, periods or dashes and must begin with "
                         "alphanumeric character or underscore"}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email,
             description='CTO of Things',
             password='example')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert check_password('example', usr.password)
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode')
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_password_change(self):
         username = 'test_new_api_user_password_change'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode',
             force_password_change=True)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_personal_repo_group(self):
         username = 'test_new_api_user_personal_group'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email, extern_name='rhodecode',
             create_personal_repo_group=True)
         response = api_call(self.app, params)
 
         usr = UserModel().get_by_username(username)
         ret = {
             'msg': 'created new user `%s`' % (username,),
             'user': jsonify(usr.get_api_data(include_secrets=True)),
         }
 
         personal_group = RepoGroup.get_by_group_name(username)
         assert personal_group
-        assert personal_group.personal == True
+        assert personal_group.personal is True
         assert personal_group.user.username == username
 
         try:
             expected = ret
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(username)
             fixture.destroy_user(usr.user_id)
 
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
 
         username = 'test_new_api_user'
         email = username + "@foo.com"
 
         id_, params = build_data(
             self.apikey, 'create_user',
             username=username,
             email=email,
             password='trololo')
         response = api_call(self.app, params)
         expected = 'failed to create user `%s`' % (username,)
         assert_error(id_, expected, given=response.body)

rhodecode/apps/user_group/tests/test_user_groups.py

@@ -1,255 +1,255 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
 from rhodecode.tests import (
     TestController, assert_session_flash, TEST_USER_ADMIN_LOGIN)
 from rhodecode.model.db import UserGroup
 from rhodecode.model.meta import Session
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib.request
     import urllib.parse
     import urllib.error
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'user_groups': ADMIN_PREFIX + '/user_groups',
         'user_groups_data': ADMIN_PREFIX + '/user_groups_data',
         'user_group_members_data': ADMIN_PREFIX + '/user_groups/{user_group_id}/members',
         'user_groups_new': ADMIN_PREFIX + '/user_groups/new',
         'user_groups_create': ADMIN_PREFIX + '/user_groups/create',
         'edit_user_group': ADMIN_PREFIX + '/user_groups/{user_group_id}/edit',
         'edit_user_group_advanced_sync': ADMIN_PREFIX + '/user_groups/{user_group_id}/edit/advanced/sync',
         'edit_user_group_global_perms_update': ADMIN_PREFIX + '/user_groups/{user_group_id}/edit/global_permissions/update',
         'user_groups_update': ADMIN_PREFIX + '/user_groups/{user_group_id}/update',
         'user_groups_delete': ADMIN_PREFIX + '/user_groups/{user_group_id}/delete',
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.parse.urlencode(params))
     return base_url
 
 
 class TestUserGroupsView(TestController):
 
     def test_set_synchronization(self, user_util):
         self.log_user()
         user_group_name = user_util.create_user_group().users_group_name
 
         group = Session().query(UserGroup).filter(
             UserGroup.users_group_name == user_group_name).one()
 
         assert group.group_data.get('extern_type') is None
 
         # enable
         self.app.post(
             route_path('edit_user_group_advanced_sync',
                        user_group_id=group.users_group_id),
             params={'csrf_token': self.csrf_token}, status=302)
 
         group = Session().query(UserGroup).filter(
             UserGroup.users_group_name == user_group_name).one()
         assert group.group_data.get('extern_type') == 'manual'
         assert group.group_data.get('extern_type_set_by') == TEST_USER_ADMIN_LOGIN
 
         # disable
         self.app.post(
             route_path('edit_user_group_advanced_sync',
                 user_group_id=group.users_group_id),
             params={'csrf_token': self.csrf_token}, status=302)
 
         group = Session().query(UserGroup).filter(
             UserGroup.users_group_name == user_group_name).one()
         assert group.group_data.get('extern_type') is None
         assert group.group_data.get('extern_type_set_by') == TEST_USER_ADMIN_LOGIN
 
     def test_delete_user_group(self, user_util):
         self.log_user()
         user_group_id = user_util.create_user_group().users_group_id
 
         group = Session().query(UserGroup).filter(
             UserGroup.users_group_id == user_group_id).one()
 
         self.app.post(
             route_path('user_groups_delete', user_group_id=group.users_group_id),
             params={'csrf_token': self.csrf_token})
 
         group = Session().query(UserGroup).filter(
             UserGroup.users_group_id == user_group_id).scalar()
 
         assert group is None
 
     @pytest.mark.parametrize('repo_create, repo_create_write, user_group_create, repo_group_create, fork_create, inherit_default_permissions, expect_error, expect_form_error', [
         ('hg.create.none', 'hg.create.write_on_repogroup.false', 'hg.usergroup.create.false', 'hg.repogroup.create.false', 'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
         ('hg.create.repository', 'hg.create.write_on_repogroup.true', 'hg.usergroup.create.true', 'hg.repogroup.create.true', 'hg.fork.repository', 'hg.inherit_default_perms.false', False, False),
         ('hg.create.XXX', 'hg.create.write_on_repogroup.true', 'hg.usergroup.create.true', 'hg.repogroup.create.true', 'hg.fork.repository', 'hg.inherit_default_perms.false', False, True),
         ('', '', '', '', '', '', True, False),
     ])
     def test_global_permissions_on_user_group(
             self, repo_create, repo_create_write, user_group_create,
             repo_group_create, fork_create, expect_error, expect_form_error,
             inherit_default_permissions, user_util):
 
         self.log_user()
         user_group = user_util.create_user_group()
 
         user_group_name = user_group.users_group_name
         user_group_id = user_group.users_group_id
 
         # ENABLE REPO CREATE ON A GROUP
         perm_params = {
             'inherit_default_permissions': False,
             'default_repo_create': repo_create,
             'default_repo_create_on_write': repo_create_write,
             'default_user_group_create': user_group_create,
             'default_repo_group_create': repo_group_create,
             'default_fork_create': fork_create,
             'default_inherit_default_permissions': inherit_default_permissions,
 
             'csrf_token': self.csrf_token,
         }
         response = self.app.post(
             route_path('edit_user_group_global_perms_update',
                        user_group_id=user_group_id),
             params=perm_params)
 
         if expect_form_error:
             assert response.status_int == 200
             response.mustcontain('Value must be one of')
         else:
             if expect_error:
                 msg = 'An error occurred during permissions saving'
             else:
                 msg = 'User Group global permissions updated successfully'
                 ug = UserGroup.get_by_group_name(user_group_name)
                 del perm_params['csrf_token']
                 del perm_params['inherit_default_permissions']
                 assert perm_params == ug.get_default_perms()
             assert_session_flash(response, msg)
 
     def test_edit_view(self, user_util):
         self.log_user()
 
         user_group = user_util.create_user_group()
         self.app.get(
             route_path('edit_user_group',
                        user_group_id=user_group.users_group_id),
             status=200)
 
     def test_update_user_group(self, user_util):
         user = self.log_user()
 
         user_group = user_util.create_user_group()
         users_group_id = user_group.users_group_id
         new_name = user_group.users_group_name + '_CHANGE'
 
         params = [
             ('users_group_active', False),
             ('user_group_description', 'DESC'),
             ('users_group_name', new_name),
             ('user', user['username']),
             ('csrf_token', self.csrf_token),
             ('__start__', 'user_group_members:sequence'),
             ('__start__', 'member:mapping'),
             ('member_user_id', user['user_id']),
             ('type', 'existing'),
             ('__end__', 'member:mapping'),
             ('__end__', 'user_group_members:sequence'),
         ]
 
         self.app.post(
             route_path('user_groups_update',
                        user_group_id=users_group_id),
             params=params,
             status=302)
 
         user_group = UserGroup.get(users_group_id)
         assert user_group
 
         assert user_group.users_group_name == new_name
         assert user_group.user_group_description == 'DESC'
-        assert user_group.users_group_active == False
+        assert user_group.users_group_active is False
 
     def test_update_user_group_name_conflicts(self, user_util):
         self.log_user()
         user_group_old = user_util.create_user_group()
         new_name = user_group_old.users_group_name
 
         user_group = user_util.create_user_group()
 
         params = dict(
             users_group_active=False,
             user_group_description='DESC',
             users_group_name=new_name,
             csrf_token=self.csrf_token)
 
         response = self.app.post(
             route_path('user_groups_update',
                        user_group_id=user_group.users_group_id),
             params=params,
             status=200)
 
         response.mustcontain('User group `{}` already exists'.format(
             new_name))
 
     def test_update_members_from_user_ids(self, user_regular):
         uid = user_regular.user_id
         username = user_regular.username
         self.log_user()
 
         user_group = fixture.create_user_group('test_gr_ids')
         assert user_group.members == []
         assert user_group.user != user_regular
         expected_active_state = not user_group.users_group_active
 
         form_data = [
             ('csrf_token', self.csrf_token),
             ('user', username),
             ('users_group_name', 'changed_name'),
             ('users_group_active', expected_active_state),
             ('user_group_description', 'changed_description'),
 
             ('__start__', 'user_group_members:sequence'),
             ('__start__', 'member:mapping'),
             ('member_user_id', uid),
             ('type', 'existing'),
             ('__end__', 'member:mapping'),
             ('__end__', 'user_group_members:sequence'),
         ]
         ugid = user_group.users_group_id
         self.app.post(
             route_path('user_groups_update', user_group_id=ugid), form_data)
 
         user_group = UserGroup.get(ugid)
         assert user_group
 
         assert user_group.members[0].user_id == uid
         assert user_group.user_id == uid
         assert 'changed_name' in user_group.users_group_name
         assert 'changed_description' in user_group.user_group_description
         assert user_group.users_group_active == expected_active_state
 
         fixture.destroy_user_group(user_group)

rhodecode/authentication/plugins/auth_crowd.py

@@ -1,297 +1,297 b''
 
 
 # Copyright (C) 2012-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 RhodeCode authentication plugin for Atlassian CROWD
 """
 
 
 import colander
 import base64
 import logging
 import urllib.request
 import urllib.error
 import urllib.parse
 
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, hybrid_property)
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.lib.colander_utils import strip_whitespace
 from rhodecode.lib.ext_json import json, formatted_json
 from rhodecode.model.db import User
 
 log = logging.getLogger(__name__)
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     """
     Factory function that is called during plugin discovery.
     It returns the plugin instance.
     """
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class CrowdAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class CrowdSettingsSchema(AuthnPluginSettingsSchemaBase):
     host = colander.SchemaNode(
         colander.String(),
         default='127.0.0.1',
         description=_('The FQDN or IP of the Atlassian CROWD Server'),
         preparer=strip_whitespace,
         title=_('Host'),
         widget='string')
     port = colander.SchemaNode(
         colander.Int(),
         default=8095,
         description=_('The Port in use by the Atlassian CROWD Server'),
         preparer=strip_whitespace,
         title=_('Port'),
         validator=colander.Range(min=0, max=65536),
         widget='int')
     app_name = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('The Application Name to authenticate to CROWD'),
         preparer=strip_whitespace,
         title=_('Application Name'),
         widget='string')
     app_password = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('The password to authenticate to CROWD'),
         preparer=strip_whitespace,
         title=_('Application Password'),
         widget='password')
     admin_groups = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('A comma separated list of group names that identify '
                       'users as RhodeCode Administrators'),
         missing='',
         preparer=strip_whitespace,
         title=_('Admin Groups'),
         widget='string')
 
 
 class CrowdServer(object):
     def __init__(self, *args, **kwargs):
         """
         Create a new CrowdServer object that points to IP/Address 'host',
         on the given port, and using the given method (https/http). user and
         passwd can be set here or with set_credentials. If unspecified,
         "version" defaults to "latest".
 
         example::
 
             cserver = CrowdServer(host="127.0.0.1",
                                   port="8095",
                                   user="some_app",
                                   passwd="some_passwd",
                                   version="1")
         """
-        if not "port" in kwargs:
+        if 'port' not in kwargs:
             kwargs["port"] = "8095"
         self._logger = kwargs.get("logger", logging.getLogger(__name__))
         self._uri = "%s://%s:%s/crowd" % (kwargs.get("method", "http"),
                                     kwargs.get("host", "127.0.0.1"),
                                     kwargs.get("port", "8095"))
         self.set_credentials(kwargs.get("user", ""),
                              kwargs.get("passwd", ""))
         self._version = kwargs.get("version", "latest")
         self._url_list = None
         self._appname = "crowd"
 
     def set_credentials(self, user, passwd):
         self.user = user
         self.passwd = passwd
         self._make_opener()
 
     def _make_opener(self):
         mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
         mgr.add_password(None, self._uri, self.user, self.passwd)
         handler = urllib.request.HTTPBasicAuthHandler(mgr)
         self.opener = urllib.request.build_opener(handler)
 
     def _request(self, url, body=None, headers=None,
                  method=None, noformat=False,
                  empty_response_ok=False):
         _headers = {"Content-type": "application/json",
                     "Accept": "application/json"}
         if self.user and self.passwd:
             authstring = base64.b64encode("%s:%s" % (self.user, self.passwd))
             _headers["Authorization"] = "Basic %s" % authstring
         if headers:
             _headers.update(headers)
         log.debug("Sent crowd: \n%s"
                   % (formatted_json({"url": url, "body": body,
                                            "headers": _headers})))
         request = urllib.request.Request(url, body, _headers)
         if method:
             request.get_method = lambda: method
 
         global msg
         msg = ""
         try:
             ret_doc = self.opener.open(request)
             msg = ret_doc.read()
             if not msg and empty_response_ok:
                 ret_val = {}
                 ret_val["status"] = True
                 ret_val["error"] = "Response body was empty"
             elif not noformat:
                 ret_val = json.loads(msg)
                 ret_val["status"] = True
             else:
                 ret_val = msg
         except Exception as e:
             if not noformat:
                 ret_val = {"status": False,
                            "body": body,
                            "error": "{}\n{}".format(e, msg)}
             else:
                 ret_val = None
         return ret_val
 
     def user_auth(self, username, password):
         """Authenticate a user against crowd. Returns brief information about
         the user."""
         url = ("%s/rest/usermanagement/%s/authentication?username=%s"
                % (self._uri, self._version, username))
         body = json.dumps({"value": password})
         return self._request(url, body)
 
     def user_groups(self, username):
         """Retrieve a list of groups to which this user belongs."""
         url = ("%s/rest/usermanagement/%s/user/group/nested?username=%s"
                % (self._uri, self._version, username))
         return self._request(url)
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'crowd'
     _settings_unsafe_keys = ['app_password']
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), CrowdAuthnResource(self))
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_get',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='GET',
             route_name='auth_home',
             context=CrowdAuthnResource)
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_post',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='POST',
             route_name='auth_home',
             context=CrowdAuthnResource)
 
     def get_settings_schema(self):
         return CrowdSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('CROWD')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-crowd.html"
 
     @hybrid_property
     def name(self):
         return u"crowd"
 
     def use_fake_password(self):
         return True
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
 
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Given a user object (which may be null), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
 
         Return None on failure. On success, return a dictionary of the form:
 
             see: RhodeCodeAuthPluginBase.auth_func_attrs
         This is later validated for correctness
         """
         if not username or not password:
             log.debug('Empty username or password skipping...')
             return None
 
         log.debug("Crowd settings: \n%s", formatted_json(settings))
         server = CrowdServer(**settings)
         server.set_credentials(settings["app_name"], settings["app_password"])
         crowd_user = server.user_auth(username, password)
         log.debug("Crowd returned: \n%s", formatted_json(crowd_user))
         if not crowd_user["status"]:
             return None
 
         res = server.user_groups(crowd_user["name"])
         log.debug("Crowd groups: \n%s", formatted_json(res))
         crowd_user["groups"] = [x["name"] for x in res["groups"]]
 
         # old attrs fetched from RhodeCode database
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = getattr(userobj, 'email', '')
         username = getattr(userobj, 'username', username)
         firstname = getattr(userobj, 'firstname', '')
         lastname = getattr(userobj, 'lastname', '')
         extern_type = getattr(userobj, 'extern_type', '')
 
         user_attrs = {
             'username': username,
             'firstname': crowd_user["first-name"] or firstname,
             'lastname': crowd_user["last-name"] or lastname,
             'groups': crowd_user["groups"],
             'user_group_sync': True,
             'email': crowd_user["email"] or email,
             'admin': admin,
             'active': active,
             'active_from_extern': crowd_user.get('active'),
             'extern_name': crowd_user["name"],
             'extern_type': extern_type,
         }
 
         # set an admin if we're in admin_groups of crowd
         for group in settings["admin_groups"]:
             if group in user_attrs["groups"]:
                 user_attrs["admin"] = True
         log.debug("Final crowd user object: \n%s", formatted_json(user_attrs))
         log.info('user `%s` authenticated correctly', user_attrs['username'])
         return user_attrs
 
 
 def includeme(config):
     plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
     plugin_factory(plugin_id).includeme(config)

rhodecode/lib/dbmigrate/migrate/versioning/version.py

@@ -1,263 +1,263 b''
 #!/usr/bin/env python
 
 
 import os
 import re
 import shutil
 import logging
 
 from rhodecode.lib.dbmigrate.migrate import exceptions
 from rhodecode.lib.dbmigrate.migrate.versioning import pathed, script
 from datetime import datetime
 
 
 log = logging.getLogger(__name__)
 
 class VerNum(object):
     """A version number that behaves like a string and int at the same time"""
 
     _instances = {}
 
     def __new__(cls, value):
         val = str(value)
         if val not in cls._instances:
             cls._instances[val] = super(VerNum, cls).__new__(cls)
         ret = cls._instances[val]
         return ret
 
     def __init__(self,value):
         self.value = str(int(value))
         if self < 0:
             raise ValueError("Version number cannot be negative")
 
     def __add__(self, value):
         ret = int(self) + int(value)
         return VerNum(ret)
 
     def __sub__(self, value):
         return self + (int(value) * -1)
 
     def __eq__(self, value):
         return int(self) == int(value)
 
     def __ne__(self, value):
         return int(self) != int(value)
 
     def __lt__(self, value):
         return int(self) < int(value)
 
     def __gt__(self, value):
         return int(self) > int(value)
 
     def __ge__(self, value):
         return int(self) >= int(value)
 
     def __le__(self, value):
         return int(self) <= int(value)
 
     def __repr__(self):
         return "<VerNum(%s)>" % self.value
 
     def __str__(self):
         return str(self.value)
 
     def __int__(self):
         return int(self.value)
 
 
 class Collection(pathed.Pathed):
     """A collection of versioning scripts in a repository"""
 
     FILENAME_WITH_VERSION = re.compile(r'^(\d{3,}).*')
 
     def __init__(self, path):
         """Collect current version scripts in repository
         and store them in self.versions
         """
         super(Collection, self).__init__(path)
 
         # Create temporary list of files, allowing skipped version numbers.
         files = os.listdir(path)
         if '1' in files:
             # deprecation
             raise Exception('It looks like you have a repository in the old '
                 'format (with directories for each version). '
                 'Please convert repository before proceeding.')
 
         tempVersions = {}
         for filename in files:
             match = self.FILENAME_WITH_VERSION.match(filename)
             if match:
                 num = int(match.group(1))
                 tempVersions.setdefault(num, []).append(filename)
             else:
                 pass  # Must be a helper file or something, let's ignore it.
 
         # Create the versions member where the keys
         # are VerNum's and the values are Version's.
         self.versions = {}
         for num, files in list(tempVersions.items()):
             self.versions[VerNum(num)] = Version(num, path, files)
 
     @property
     def latest(self):
         """:returns: Latest version in Collection"""
         return max([VerNum(0)] + list(self.versions.keys()))
 
     def _next_ver_num(self, use_timestamp_numbering):
-        if use_timestamp_numbering == True:
+        if use_timestamp_numbering is True:
             return VerNum(int(datetime.utcnow().strftime('%Y%m%d%H%M%S')))
         else:
             return self.latest + 1
 
     def create_new_python_version(self, description, **k):
         """Create Python files for new version"""
         ver = self._next_ver_num(k.pop('use_timestamp_numbering', False))
         extra = str_to_filename(description)
 
         if extra:
             if extra == '_':
                 extra = ''
             elif not extra.startswith('_'):
                 extra = '_%s' % extra
 
         filename = '%03d%s.py' % (ver, extra)
         filepath = self._version_path(filename)
 
         script.PythonScript.create(filepath, **k)
         self.versions[ver] = Version(ver, self.path, [filename])
 
     def create_new_sql_version(self, database, description, **k):
         """Create SQL files for new version"""
         ver = self._next_ver_num(k.pop('use_timestamp_numbering', False))
         self.versions[ver] = Version(ver, self.path, [])
 
         extra = str_to_filename(description)
 
         if extra:
             if extra == '_':
                 extra = ''
             elif not extra.startswith('_'):
                 extra = '_%s' % extra
 
         # Create new files.
         for op in ('upgrade', 'downgrade'):
             filename = '%03d%s_%s_%s.sql' % (ver, extra, database, op)
             filepath = self._version_path(filename)
             script.SqlScript.create(filepath, **k)
             self.versions[ver].add_script(filepath)
 
     def version(self, vernum=None):
         """Returns latest Version if vernum is not given.
         Otherwise, returns wanted version"""
         if vernum is None:
             vernum = self.latest
         return self.versions[VerNum(vernum)]
 
     @classmethod
     def clear(cls):
         super(Collection, cls).clear()
 
     def _version_path(self, ver):
         """Returns path of file in versions repository"""
         return os.path.join(self.path, str(ver))
 
 
 class Version(object):
     """A single version in a collection
     :param vernum: Version Number
     :param path: Path to script files
     :param filelist: List of scripts
     :type vernum: int, VerNum
     :type path: string
     :type filelist: list
     """
 
     def __init__(self, vernum, path, filelist):
         self.version = VerNum(vernum)
 
         # Collect scripts in this folder
         self.sql = {}
         self.python = None
 
         for script in filelist:
             self.add_script(os.path.join(path, script))
 
     def script(self, database=None, operation=None):
         """Returns SQL or Python Script"""
         for db in (database, 'default'):
             # Try to return a .sql script first
             try:
                 return self.sql[db][operation]
             except KeyError:
                 continue  # No .sql script exists
 
         # TODO: maybe add force Python parameter?
         ret = self.python
 
         assert ret is not None, \
             "There is no script for %d version" % self.version
         return ret
 
     def add_script(self, path):
         """Add script to Collection/Version"""
         if path.endswith(Extensions.py):
             self._add_script_py(path)
         elif path.endswith(Extensions.sql):
             self._add_script_sql(path)
 
     SQL_FILENAME = re.compile(r'^.*\.sql')
 
     def _add_script_sql(self, path):
         basename = os.path.basename(path)
         match = self.SQL_FILENAME.match(basename)
 
         if match:
             basename = basename.replace('.sql', '')
             parts = basename.split('_')
             if len(parts) < 3:
                 raise exceptions.ScriptError(
                     "Invalid SQL script name %s " % basename + \
                     "(needs to be ###_description_database_operation.sql)")
             version = parts[0]
             op = parts[-1]
             # NOTE(mriedem): check for ibm_db_sa as the database in the name
             if 'ibm_db_sa' in basename:
                 if len(parts) == 6:
                     dbms = '_'.join(parts[-4: -1])
                 else:
                     raise exceptions.ScriptError(
                         "Invalid ibm_db_sa SQL script name '%s'; "
                         "(needs to be "
                         "###_description_ibm_db_sa_operation.sql)" % basename)
             else:
                 dbms = parts[-2]
         else:
             raise exceptions.ScriptError(
                 "Invalid SQL script name %s " % basename + \
                 "(needs to be ###_description_database_operation.sql)")
 
         # File the script into a dictionary
         self.sql.setdefault(dbms, {})[op] = script.SqlScript(path)
 
     def _add_script_py(self, path):
         if self.python is not None:
             raise exceptions.ScriptError('You can only have one Python script '
                 'per version, but you have: %s and %s' % (self.python, path))
         self.python = script.PythonScript(path)
 
 
 class Extensions:
     """A namespace for file extensions"""
     py = 'py'
     sql = 'sql'
 
 def str_to_filename(s):
     """Replaces spaces, (double and single) quotes
     and double underscores to underscores
     """
 
     s = s.replace(' ', '_').replace('"', '_').replace("'", '_').replace(".", "_")
     while '__' in s:
         s = s.replace('__', '_')
     return s

rhodecode/lib/logging_formatter.py

@@ -1,186 +1,185 b''
-
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import sys
 import logging
 
 
 BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = list(range(30, 38))
 
 # Sequences
 RESET_SEQ = "\033[0m"
 COLOR_SEQ = "\033[0;%dm"
 BOLD_SEQ = "\033[1m"
 
 COLORS = {
     'CRITICAL': MAGENTA,
     'ERROR': RED,
     'WARNING': CYAN,
     'INFO': GREEN,
     'DEBUG': BLUE,
     'SQL': YELLOW
 }
 
 
 def _inject_req_id(record, with_prefix=True):
     from pyramid.threadlocal import get_current_request
     dummy = '00000000-0000-0000-0000-000000000000'
     req_id = None
 
     req = get_current_request()
     if req:
         req_id = getattr(req, 'req_id', None)
     if with_prefix:
         req_id = 'req_id:%-36s' % (req_id or dummy)
     else:
         req_id = (req_id or dummy)
     record.req_id = req_id
 
 
 def _add_log_to_debug_bucket(formatted_record):
     from pyramid.threadlocal import get_current_request
     req = get_current_request()
     if req:
         req.req_id_bucket.append(formatted_record)
 
 
 def one_space_trim(s):
     if s.find("  ") == -1:
         return s
     else:
         s = s.replace('  ', ' ')
         return one_space_trim(s)
 
 
 def format_sql(sql):
     sql = sql.replace('\n', '')
     sql = one_space_trim(sql)
     sql = sql\
         .replace(',', ',\n\t')\
         .replace('SELECT', '\n\tSELECT \n\t')\
         .replace('UPDATE', '\n\tUPDATE \n\t')\
         .replace('DELETE', '\n\tDELETE \n\t')\
         .replace('FROM', '\n\tFROM')\
         .replace('ORDER BY', '\n\tORDER BY')\
         .replace('LIMIT', '\n\tLIMIT')\
         .replace('WHERE', '\n\tWHERE')\
         .replace('AND', '\n\tAND')\
         .replace('LEFT', '\n\tLEFT')\
         .replace('INNER', '\n\tINNER')\
         .replace('INSERT', '\n\tINSERT')\
         .replace('DELETE', '\n\tDELETE')
     return sql
 
 
 class ExceptionAwareFormatter(logging.Formatter):
     """
     Extended logging formatter which prints out remote tracebacks.
     """
 
     def formatException(self, ei):
         ex_type, ex_value, ex_tb = ei
 
         local_tb = logging.Formatter.formatException(self, ei)
         if hasattr(ex_value, '_vcs_server_traceback'):
 
             def formatRemoteTraceback(remote_tb_lines):
                 result = ["\n +--- This exception occured remotely on VCSServer - Remote traceback:\n\n"]
                 result.append(remote_tb_lines)
                 result.append("\n +--- End of remote traceback\n")
                 return result
 
             try:
                 if ex_type is not None and ex_value is None and ex_tb is None:
                     # possible old (3.x) call syntax where caller is only
                     # providing exception object
                     if type(ex_type) is not type:
                         raise TypeError(
                             "invalid argument: ex_type should be an exception "
                             "type, or just supply no arguments at all")
                 if ex_type is None and ex_tb is None:
                     ex_type, ex_value, ex_tb = sys.exc_info()
 
                 remote_tb = getattr(ex_value, "_vcs_server_traceback", None)
 
                 if remote_tb:
                     remote_tb = formatRemoteTraceback(remote_tb)
                     return local_tb + ''.join(remote_tb)
             finally:
                 # clean up cycle to traceback, to allow proper GC
                 del ex_type, ex_value, ex_tb
 
         return local_tb
 
 
 class RequestTrackingFormatter(ExceptionAwareFormatter):
     def format(self, record):
         _inject_req_id(record)
         def_record = logging.Formatter.format(self, record)
         _add_log_to_debug_bucket(def_record)
         return def_record
 
 
 class ColorFormatter(ExceptionAwareFormatter):
 
     def format(self, record):
         """
         Changes record's levelname to use with COLORS enum
         """
         def_record = super(ColorFormatter, self).format(record)
 
         levelname = record.levelname
         start = COLOR_SEQ % (COLORS[levelname])
         end = RESET_SEQ
 
         colored_record = ''.join([start, def_record, end])
         return colored_record
 
 
 class ColorRequestTrackingFormatter(RequestTrackingFormatter):
 
     def format(self, record):
         """
         Changes record's levelname to use with COLORS enum
         """
         def_record = super(ColorRequestTrackingFormatter, self).format(record)
 
         levelname = record.levelname
         start = COLOR_SEQ % (COLORS[levelname])
         end = RESET_SEQ
 
         colored_record = ''.join([start, def_record, end])
         return colored_record
 
 
 class ColorFormatterSql(logging.Formatter):
 
     def format(self, record):
         """
         Changes record's levelname to use with COLORS enum
         """
 
         start = COLOR_SEQ % (COLORS['SQL'])
         def_record = format_sql(logging.Formatter.format(self, record))
         end = RESET_SEQ
 
         colored_record = ''.join([start, def_record, end])
         return colored_record
 
-# marcink: needs to stay with this name for backward .ini compatability
-Pyro4AwareFormatter = ExceptionAwareFormatter
+# NOTE (marcink): needs to stay with this name for backward .ini compatability
+Pyro4AwareFormatter = ExceptionAwareFormatter  # noqa

rhodecode/lib/markdown_ext.py

@@ -1,132 +1,132 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import re
 import markdown
 import xml.etree.ElementTree as etree
 
 from markdown.extensions import Extension
 from markdown.extensions.fenced_code import FencedCodeExtension
 from markdown.extensions.tables import TableExtension
 from markdown.extensions.nl2br import Nl2BrExtension as _Nl2BrExtension
 from markdown.extensions.wikilinks import WikiLinkExtension
 from markdown.inlinepatterns import Pattern
 
 import gfm
 
 
 class InlineProcessor(Pattern):
     """
     Base class that inline patterns subclass.
     This is the newer style inline processor that uses a more
     efficient and flexible search approach.
     """
 
     def __init__(self, pattern, md=None):
         """
         Create an instant of an inline pattern.
         Keyword arguments:
         * pattern: A regular expression that matches a pattern
         """
         self.pattern = pattern
         self.compiled_re = re.compile(pattern, re.DOTALL | re.UNICODE)
 
         # Api for Markdown to pass safe_mode into instance
         self.safe_mode = False
         self.md = md
 
     def handleMatch(self, m, data):
         """Return a ElementTree element from the given match and the
         start and end index of the matched text.
         If `start` and/or `end` are returned as `None`, it will be
         assumed that the processor did not find a valid region of text.
         Subclasses should override this method.
         Keyword arguments:
         * m: A re match object containing a match of the pattern.
         * data: The buffer current under analysis
         Returns:
         * el: The ElementTree element, text or None.
         * start: The start of the region that has been matched or None.
         * end: The end of the region that has been matched or None.
         """
         pass  # pragma: no cover
 
 
 class SimpleTagInlineProcessor(InlineProcessor):
     """
     Return element of type `tag` with a text attribute of group(2)
     of a Pattern.
     """
     def __init__(self, pattern, tag):
         InlineProcessor.__init__(self, pattern)
         self.tag = tag
 
     def handleMatch(self, m, data):  # pragma: no cover
         el = etree.Element(self.tag)
         el.text = m.group(2)
         return el, m.start(0), m.end(0)
 
 
 class SubstituteTagInlineProcessor(SimpleTagInlineProcessor):
     """ Return an element of type `tag` with no children. """
     def handleMatch(self, m, data):
         return etree.Element(self.tag), m.start(0), m.end(0)
 
 
 class Nl2BrExtension(_Nl2BrExtension):
     pass
 
 
 # Global Vars
 URLIZE_RE = '(%s)' % '|'.join([
     r'<(?:f|ht)tps?://[^>]*>',
     r'\b(?:f|ht)tps?://[^)<>\s]+[^.,)<>\s]',
     r'\bwww\.[^)<>\s]+[^.,)<>\s]',
     r'[^(<\s]+\.(?:com|net|org)\b',
 ])
 
 
 class UrlizePattern(markdown.inlinepatterns.Pattern):
     """ Return a link Element given an autolink (`http://example/com`). """
     def handleMatch(self, m):
         url = m.group(2)
 
         if url.startswith('<'):
             url = url[1:-1]
 
         text = url
 
-        if not url.split('://')[0] in ('http','https','ftp'):
-            if '@' in url and not '/' in url:
+        if url.split('://')[0] not in ('http', 'https', 'ftp'):
+            if '@' in url and '/' not in url:
                 url = 'mailto:' + url
             else:
                 url = 'http://' + url
 
         el = markdown.util.etree.Element("a")
         el.set('href', url)
         el.text = markdown.util.AtomicString(text)
         return el
 
 
 class UrlizeExtension(Extension):
     """ Urlize Extension for Python-Markdown. """
 
     def extendMarkdown(self, md):
         """ Replace autolink with UrlizePattern """
         md.inlinePatterns['autolink'] = UrlizePattern(URLIZE_RE, md)

rhodecode/tests/models/test_user_groups.py

@@ -1,218 +1,218 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.model.db import User
 from rhodecode.tests import TEST_USER_REGULAR_LOGIN
 from rhodecode.tests.fixture import Fixture
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.meta import Session
 
 
 fixture = Fixture()
 
 
 def teardown_module(self):
     _delete_all_user_groups()
 
 
 class TestGetUserGroups(object):
     def test_returns_filtered_list(self, backend, user_util):
         created_groups = []
         for i in range(4):
             created_groups.append(
                 user_util.create_user_group(users_group_active=True))
 
         group_filter = created_groups[-1].users_group_name[-2:]
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with self._patch_user_group_list():
                 groups = UserGroupModel().get_user_groups(group_filter)
 
         fake_groups = [
             u for u in groups if u['value'].startswith('test_returns')]
         assert len(fake_groups) == 1
         assert fake_groups[0]['value'] == created_groups[-1].users_group_name
         assert fake_groups[0]['value_display'].startswith(
             'Group: test_returns')
 
     def test_returns_limited_list(self, backend, user_util):
         created_groups = []
         for i in range(3):
             created_groups.append(
                 user_util.create_user_group(users_group_active=True))
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with self._patch_user_group_list():
                 groups = UserGroupModel().get_user_groups('test_returns')
 
         fake_groups = [
             u for u in groups if u['value'].startswith('test_returns')]
         assert len(fake_groups) == 3
 
     def test_returns_active_user_groups(self, backend, user_util):
         for i in range(4):
             is_active = i % 2 == 0
             user_util.create_user_group(users_group_active=is_active)
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with self._patch_user_group_list():
                 groups = UserGroupModel().get_user_groups()
         expected = ('id', 'icon_link', 'value_display', 'value', 'value_type')
         for group in groups:
-            assert group['value_type'] is 'user_group'
+            assert group['value_type'] == 'user_group'
             for key in expected:
                 assert key in group
 
         fake_groups = [
             u for u in groups if u['value'].startswith('test_returns')]
         assert len(fake_groups) == 2
         for user in fake_groups:
             assert user['value_display'].startswith('Group: test_returns')
 
     def _patch_user_group_list(self):
         def side_effect(group_list, perm_set):
             return group_list
         return mock.patch(
             'rhodecode.model.user_group.UserGroupList', side_effect=side_effect)
 
 
 @pytest.mark.parametrize(
     "pre_existing, regular_should_be, external_should_be, groups, "
     "expected", [
         ([], [], [], [], []),
         # no changes of regular
         ([], ['regular'], [], [], ['regular']),
         # not added to regular group
         (['some_other'], [], [], ['some_other'], []),
         (
             [], ['regular'], ['container'], ['container'],
             ['regular', 'container']
         ),
         (
             [], ['regular'], [], ['container', 'container2'],
             ['regular', 'container', 'container2']
         ),
         # remove not used
         ([], ['regular'], ['other'], [], ['regular']),
         (
             ['some_other'], ['regular'], ['other', 'container'],
             ['container', 'container2'],
             ['regular', 'container', 'container2']
         ),
     ])
 def test_enforce_groups(pre_existing, regular_should_be,
                         external_should_be, groups, expected, backend_hg):
     # TODO: anderson: adding backend_hg fixture so it sets up the database
     # for when running this file alone
     _delete_all_user_groups()
 
     user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
     for gr in pre_existing:
         gr = fixture.create_user_group(gr)
     Session().commit()
 
     # make sure use is just in those groups
     for gr in regular_should_be:
         gr = fixture.create_user_group(gr)
         Session().commit()
         UserGroupModel().add_user_to_group(gr, user)
         Session().commit()
 
     # now special external groups created by auth plugins
     for gr in external_should_be:
         gr = fixture.create_user_group(
             gr, user_group_data={'extern_type': 'container'})
         Session().commit()
         UserGroupModel().add_user_to_group(gr, user)
         Session().commit()
 
     UserGroupModel().enforce_groups(user, groups, 'container')
     Session().commit()
 
     user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
     in_groups = user.group_member
 
     expected.sort()
     assert (
         expected == sorted(x.users_group.users_group_name for x in in_groups))
 
 
 def _delete_all_user_groups():
     for gr in UserGroupModel.get_all():
         fixture.destroy_user_group(gr)
     Session().commit()
 
 
 def test_add_and_remove_user_from_group(user_regular, user_util):
     user_group = user_util.create_user_group()
     assert user_group.members == []
     UserGroupModel().add_user_to_group(user_group, user_regular)
     Session().commit()
     assert user_group.members[0].user == user_regular
     UserGroupModel().remove_user_from_group(user_group, user_regular)
     Session().commit()
     assert user_group.members == []
 
 
 @pytest.mark.parametrize('data, expected', [
     ([], []),
     ([{"member_user_id": 1, "type": "new"}], [1]),
     ([{"member_user_id": 1, "type": "new"},
       {"member_user_id": 1, "type": "existing"}], [1]),
     ([{"member_user_id": 1, "type": "new"},
       {"member_user_id": 2, "type": "new"},
       {"member_user_id": 3, "type": "remove"}], [1, 2])
 ])
 def test_clean_members_data(data, expected):
     cleaned = UserGroupModel()._clean_members_data(data)
     assert cleaned == expected
 
 
 def _create_test_members():
     members = []
     for member_number in range(3):
         member = mock.Mock()
         member.user_id = member_number + 1
         member.user.user_id = member_number + 1
         members.append(member)
     return members
 
 
 def test_get_added_and_removed_users():
     members = _create_test_members()
     mock_user_group = mock.Mock()
     mock_user_group.members = [members[0], members[1]]
     new_users_list = [members[1].user.user_id, members[2].user.user_id]
     model = UserGroupModel()
 
     added, removed = model._get_added_and_removed_user_ids(
         mock_user_group, new_users_list)
 
     assert added == [members[2].user.user_id]
     assert removed == [members[0].user.user_id]
 
 
 def test_set_users_as_members_and_find_user_in_group(
         user_util, user_regular, user_admin):
     user_group = user_util.create_user_group()
     assert len(user_group.members) == 0
     user_list = [user_regular.user_id, user_admin.user_id]
     UserGroupModel()._set_users_as_members(user_group, user_list)
     assert len(user_group.members) == 2
     assert UserGroupModel()._find_user_in_group(user_regular, user_group)

rhodecode/tests/models/test_users.py

@@ -1,328 +1,328 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 import mock
 
 from rhodecode.model.db import (
     true, User, UserGroup, UserGroupMember, UserEmailMap, Permission, UserIpMap)
 from rhodecode.model.meta import Session
 from rhodecode.model.user import UserModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.tests.fixture import Fixture
 from rhodecode.lib.str_utils import safe_str
 
 
 fixture = Fixture()
 
 
 class TestGetUsers(object):
     def test_returns_active_users(self, backend, user_util):
         for i in range(4):
             is_active = i % 2 == 0
             user_util.create_user(active=is_active, lastname='Fake user')
 
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with mock.patch('rhodecode.lib.helpers.link_to_user'):
                 users = UserModel().get_users()
         fake_users = [u for u in users if u['last_name'] == 'Fake user']
         assert len(fake_users) == 2
 
         expected_keys = (
             'id', 'first_name', 'last_name', 'username', 'icon_link',
             'value_display', 'value', 'value_type')
         for user in users:
-            assert user['value_type'] is 'user'
+            assert user['value_type'] == 'user'
             for key in expected_keys:
                 assert key in user
 
     def test_returns_user_filtered_by_last_name(self, backend, user_util):
         keywords = ('aBc', u'ünicode')
         for keyword in keywords:
             for i in range(2):
                 user_util.create_user(
                     active=True, lastname=u'Fake {} user'.format(keyword))
 
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with mock.patch('rhodecode.lib.helpers.link_to_user'):
                 keyword = keywords[1].lower()
                 users = UserModel().get_users(name_contains=keyword)
 
         fake_users = [u for u in users if u['last_name'].startswith('Fake')]
         assert len(fake_users) == 2
         for user in fake_users:
             assert user['last_name'] == safe_str('Fake ünicode user')
 
     def test_returns_user_filtered_by_first_name(self, backend, user_util):
         created_users = []
         keywords = ('aBc', u'ünicode')
         for keyword in keywords:
             for i in range(2):
                 created_users.append(user_util.create_user(
                     active=True, lastname='Fake user',
                     firstname=u'Fake {} user'.format(keyword)))
 
         keyword = keywords[1].lower()
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with mock.patch('rhodecode.lib.helpers.link_to_user'):
                 users = UserModel().get_users(name_contains=keyword)
 
         fake_users = [u for u in users if u['last_name'].startswith('Fake')]
         assert len(fake_users) == 2
         for user in fake_users:
             assert user['first_name'] == safe_str('Fake ünicode user')
 
     def test_returns_user_filtered_by_username(self, backend, user_util):
         created_users = []
         for i in range(5):
             created_users.append(user_util.create_user(
                 active=True, lastname='Fake user'))
 
         user_filter = created_users[-1].username[-2:]
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with mock.patch('rhodecode.lib.helpers.link_to_user'):
                 users = UserModel().get_users(name_contains=user_filter)
 
         fake_users = [u for u in users if u['last_name'].startswith('Fake')]
         assert len(fake_users) == 1
         assert fake_users[0]['username'] == created_users[-1].username
 
     def test_returns_limited_user_list(self, backend, user_util):
         created_users = []
         for i in range(5):
             created_users.append(user_util.create_user(
                 active=True, lastname='Fake user'))
 
         with mock.patch('rhodecode.lib.helpers.gravatar_url'):
             with mock.patch('rhodecode.lib.helpers.link_to_user'):
                 users = UserModel().get_users(name_contains='Fake', limit=3)
 
         fake_users = [u for u in users if u['last_name'].startswith('Fake')]
         assert len(fake_users) == 3
 
 
 @pytest.fixture()
 def test_user(request, baseapp):
     usr = UserModel().create_or_update(
         username='test_user',
         password='qweqwe',
         email='[e-mail]',
         firstname='u1', lastname=u'u1')
     Session().commit()
     assert User.get_by_username(u'test_user') == usr
 
     @request.addfinalizer
     def cleanup():
         if UserModel().get_user(usr.user_id) is None:
             return
 
         perm = Permission.query().all()
         for p in perm:
             UserModel().revoke_perm(usr, p)
 
         UserModel().delete(usr.user_id)
         Session().commit()
 
     return usr
 
 
 def test_create_and_remove(test_user):
     usr = test_user
 
     # make user group
     user_group = fixture.create_user_group('some_example_group')
     Session().commit()
 
     UserGroupModel().add_user_to_group(user_group, usr)
     Session().commit()
 
     assert UserGroup.get(user_group.users_group_id) == user_group
     assert UserGroupMember.query().count() == 1
     UserModel().delete(usr.user_id)
     Session().commit()
 
     assert UserGroupMember.query().all() == []
 
 
 def test_additonal_email_as_main(test_user):
     with pytest.raises(AttributeError):
         m = UserEmailMap()
         m.email = test_user.email
         m.user = test_user
         Session().add(m)
         Session().commit()
 
 
 def test_extra_email_map(test_user):
 
     m = UserEmailMap()
     m.email = u'[e-mail]'
     m.user = test_user
     Session().add(m)
     Session().commit()
 
     u = User.get_by_email(email='[e-mail]')
     assert test_user.user_id == u.user_id
     assert test_user.username == u.username
 
     u = User.get_by_email(email='[e-mail]')
     assert test_user.user_id == u.user_id
     assert test_user.username == u.username
     u = User.get_by_email(email='[e-mail]')
     assert u is None
 
 
 def test_get_api_data_replaces_secret_data_by_default(test_user):
     api_data = test_user.get_api_data()
     api_key_length = 40
     expected_replacement = '*' * api_key_length
 
     for key in api_data['auth_tokens']:
         assert key == expected_replacement
 
 
 def test_get_api_data_includes_secret_data_if_activated(test_user):
     api_data = test_user.get_api_data(include_secrets=True)
     assert api_data['auth_tokens'] == test_user.auth_tokens
 
 
 def test_add_perm(test_user):
     perm = Permission.query().all()[0]
     UserModel().grant_perm(test_user, perm)
     Session().commit()
     assert UserModel().has_perm(test_user, perm)
 
 
 def test_has_perm(test_user):
     perm = Permission.query().all()
     for p in perm:
         assert not UserModel().has_perm(test_user, p)
 
 
 def test_revoke_perm(test_user):
     perm = Permission.query().all()[0]
     UserModel().grant_perm(test_user, perm)
     Session().commit()
     assert UserModel().has_perm(test_user, perm)
 
     # revoke
     UserModel().revoke_perm(test_user, perm)
     Session().commit()
     assert not UserModel().has_perm(test_user, perm)
 
 
 @pytest.mark.parametrize("ip_range, expected, expect_errors", [
     ('', [], False),
     ('127.0.0.1', ['127.0.0.1'], False),
     ('127.0.0.1,127.0.0.2', ['127.0.0.1', '127.0.0.2'], False),
     ('127.0.0.1 ,  127.0.0.2', ['127.0.0.1', '127.0.0.2'], False),
     (
         '127.0.0.1,172.172.172.0,127.0.0.2',
         ['127.0.0.1', '172.172.172.0', '127.0.0.2'], False),
     (
         '127.0.0.1-127.0.0.5',
         ['127.0.0.1', '127.0.0.2', '127.0.0.3', '127.0.0.4', '127.0.0.5'],
         False),
     (
         '127.0.0.1 - 127.0.0.5',
         ['127.0.0.1', '127.0.0.2', '127.0.0.3', '127.0.0.4', '127.0.0.5'],
         False
     ),
     ('-', [], True),
     ('127.0.0.1-32', [], True),
     (
         '127.0.0.1,127.0.0.1,127.0.0.1,127.0.0.1-127.0.0.2,127.0.0.2',
         ['127.0.0.1', '127.0.0.2'], False),
     (
         '127.0.0.1-127.0.0.2,127.0.0.4-127.0.0.6,',
         ['127.0.0.1', '127.0.0.2', '127.0.0.4', '127.0.0.5', '127.0.0.6'],
         False
     ),
     (
         '127.0.0.1-127.0.0.2,127.0.0.1-127.0.0.6,',
         ['127.0.0.1', '127.0.0.2', '127.0.0.3', '127.0.0.4', '127.0.0.5',
          '127.0.0.6'],
         False
     ),
 ])
 def test_ip_range_generator(ip_range, expected, expect_errors):
     func = UserModel().parse_ip_range
     if expect_errors:
         pytest.raises(ValueError, func, ip_range)
     else:
         parsed_list = func(ip_range)
         assert parsed_list == expected
 
 
 def test_user_delete_cascades_ip_whitelist(test_user):
     sample_ip = '1.1.1.1'
     uid_map = UserIpMap(user_id=test_user.user_id, ip_addr=sample_ip)
     Session().add(uid_map)
     Session().delete(test_user)
     try:
         Session().flush()
     finally:
         Session().rollback()
 
 
 def test_account_for_deactivation_generation(test_user):
     accounts = UserModel().get_accounts_in_creation_order(
         current_user=test_user)
     # current user should be #1 in the list
     assert accounts[0] == test_user.user_id
     active_users = User.query().filter(User.active == true()).count()
     assert active_users == len(accounts)
 
 
 def test_user_delete_cascades_permissions_on_repo(backend, test_user):
     test_repo = backend.create_repo()
     RepoModel().grant_user_permission(
         test_repo, test_user, 'repository.write')
     Session().commit()
 
     assert test_user.repo_to_perm
 
     UserModel().delete(test_user)
     Session().commit()
 
 
 def test_user_delete_cascades_permissions_on_repo_group(
         test_repo_group, test_user):
     RepoGroupModel().grant_user_permission(
         test_repo_group, test_user, 'group.write')
     Session().commit()
 
     assert test_user.repo_group_to_perm
 
     Session().delete(test_user)
     Session().commit()
 
 
 def test_user_delete_cascades_permissions_on_user_group(
         test_user_group, test_user):
     UserGroupModel().grant_user_permission(
         test_user_group, test_user, 'usergroup.write')
     Session().commit()
 
     assert test_user.user_group_to_perm
 
     Session().delete(test_user)
     Session().commit()

rhodecode/tests/other/test_validators.py

@@ -1,295 +1,295 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import formencode
 import pytest
 
 from rhodecode.tests import (
     HG_REPO, TEST_USER_REGULAR2_EMAIL, TEST_USER_REGULAR2_LOGIN,
     TEST_USER_REGULAR2_PASS, TEST_USER_ADMIN_LOGIN, TESTS_TMP_PATH)
 
 from rhodecode.model import validators as v
 from rhodecode.model.user_group import UserGroupModel
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.db import ChangesetStatus, Repository
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 pytestmark = pytest.mark.usefixtures('baseapp')
 
 
 @pytest.fixture()
 def localizer():
     def func(msg):
         return msg
     return func
 
 
 def test_Message_extractor(localizer):
     validator = v.ValidUsername(localizer)
     pytest.raises(formencode.Invalid, validator.to_python, 'default')
 
     class StateObj(object):
         pass
 
     pytest.raises(
         formencode.Invalid, validator.to_python, 'default', StateObj)
 
 
 def test_ValidUsername(localizer):
     validator = v.ValidUsername(localizer)
 
     pytest.raises(formencode.Invalid, validator.to_python, 'default')
     pytest.raises(formencode.Invalid, validator.to_python, 'new_user')
     pytest.raises(formencode.Invalid, validator.to_python, '.,')
     pytest.raises(
         formencode.Invalid, validator.to_python, TEST_USER_ADMIN_LOGIN)
     assert 'test' == validator.to_python('test')
 
     validator = v.ValidUsername(localizer, edit=True, old_data={'user_id': 1})
 
 
 def test_ValidRepoUser(localizer):
     validator = v.ValidRepoUser(localizer)
     pytest.raises(formencode.Invalid, validator.to_python, 'nouser')
     assert TEST_USER_ADMIN_LOGIN == \
         validator.to_python(TEST_USER_ADMIN_LOGIN)
 
 
 def test_ValidUserGroup(localizer):
     validator = v.ValidUserGroup(localizer)
     pytest.raises(formencode.Invalid, validator.to_python, 'default')
     pytest.raises(formencode.Invalid, validator.to_python, '.,')
 
     gr = fixture.create_user_group('test')
     gr2 = fixture.create_user_group('tes2')
     Session().commit()
     pytest.raises(formencode.Invalid, validator.to_python, 'test')
     assert gr.users_group_id is not None
     validator = v.ValidUserGroup(localizer,
         edit=True,
         old_data={'users_group_id': gr2.users_group_id})
 
     pytest.raises(formencode.Invalid, validator.to_python, 'test')
     pytest.raises(formencode.Invalid, validator.to_python, 'TesT')
     pytest.raises(formencode.Invalid, validator.to_python, 'TEST')
     UserGroupModel().delete(gr)
     UserGroupModel().delete(gr2)
     Session().commit()
 
 
 @pytest.fixture(scope='function')
 def repo_group(request):
     model = RepoGroupModel()
     gr = model.create(
         group_name='test_gr', group_description='desc', just_db=True,
         owner=TEST_USER_ADMIN_LOGIN)
 
     def cleanup():
         model.delete(gr)
 
     request.addfinalizer(cleanup)
 
     return gr
 
 
 def test_ValidRepoGroup_same_name_as_repo(localizer):
     validator = v.ValidRepoGroup(localizer)
     with pytest.raises(formencode.Invalid) as excinfo:
         validator.to_python({'group_name': HG_REPO})
     expected_msg = 'Repository with name "vcs_test_hg" already exists'
     assert expected_msg in str(excinfo.value)
 
 
 def test_ValidRepoGroup_group_exists(localizer, repo_group):
     validator = v.ValidRepoGroup(localizer)
     with pytest.raises(formencode.Invalid) as excinfo:
         validator.to_python({'group_name': repo_group.group_name})
     expected_msg = 'Group "test_gr" already exists'
     assert expected_msg in str(excinfo.value)
 
 
 def test_ValidRepoGroup_invalid_parent(localizer, repo_group):
     validator = v.ValidRepoGroup(localizer, edit=True,
                                  old_data={'group_id': repo_group.group_id})
     with pytest.raises(formencode.Invalid) as excinfo:
         validator.to_python({
             'group_name': repo_group.group_name + 'n',
             'group_parent_id': repo_group.group_id,
         })
     expected_msg = 'Cannot assign this group as parent'
     assert expected_msg in str(excinfo.value)
 
 
 def test_ValidRepoGroup_edit_group_no_root_permission(localizer, repo_group):
     validator = v.ValidRepoGroup(localizer,
         edit=True, old_data={'group_id': repo_group.group_id},
         can_create_in_root=False)
 
     # Cannot change parent
     with pytest.raises(formencode.Invalid) as excinfo:
         validator.to_python({'group_parent_id': '25'})
     expected_msg = 'no permission to store repository group in root location'
     assert expected_msg in str(excinfo.value)
 
     # Chaning all the other fields is allowed
     validator.to_python({'group_name': 'foo', 'group_parent_id': '-1'})
     validator.to_python(
         {'user': TEST_USER_REGULAR2_LOGIN, 'group_parent_id': '-1'})
     validator.to_python({'group_description': 'bar', 'group_parent_id': '-1'})
     validator.to_python({'enable_locking': 'true', 'group_parent_id': '-1'})
 
 
 def test_ValidPassword(localizer):
     validator = v.ValidPassword(localizer)
     assert 'lol' == validator.to_python('lol')
-    assert None == validator.to_python(None)
+    assert None is validator.to_python(None)
     pytest.raises(formencode.Invalid, validator.to_python, 'ąćżź')
 
 
 def test_ValidPasswordsMatch(localizer):
     validator = v.ValidPasswordsMatch(localizer)
     pytest.raises(
         formencode.Invalid,
         validator.to_python, {'password': 'pass',
                               'password_confirmation': 'pass2'})
 
     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 {'password': 'pass', 'password_confirmation': 'pass'} == \
         validator.to_python({'password': 'pass',
                              'password_confirmation': 'pass'})
 
 
 def test_ValidAuth(localizer, config_stub):
     config_stub.testing_securitypolicy()
     config_stub.include('rhodecode.authentication')
     config_stub.include('rhodecode.authentication.plugins.auth_rhodecode')
     config_stub.include('rhodecode.authentication.plugins.auth_token')
 
     validator = v.ValidAuth(localizer)
     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)
     pytest.raises(
         formencode.Invalid, validator.to_python, invalid_creds)
 
 
 def test_ValidRepoName(localizer):
     validator = v.ValidRepoName(localizer)
 
     pytest.raises(
         formencode.Invalid, validator.to_python, {'repo_name': ''})
 
     pytest.raises(
         formencode.Invalid, validator.to_python, {'repo_name': HG_REPO})
 
     gr = RepoGroupModel().create(group_name='group_test',
                                  group_description='desc',
                                  owner=TEST_USER_ADMIN_LOGIN)
     pytest.raises(
         formencode.Invalid, validator.to_python, {'repo_name': gr.group_name})
 
     #TODO: write an error case for that ie. create a repo withinh a group
 #        pytest.raises(formencode.Invalid,
 #                          validator.to_python, {'repo_name': 'some',
 #                                                'repo_group': gr.group_id})
 
 
 def test_ValidForkName(localizer):
     # this uses ValidRepoName validator
     assert True
 
 @pytest.mark.parametrize("name, expected", [
     ('test', 'test'), ('lolz!', 'lolz'), ('  aavv', 'aavv'),
     ('ala ma kota', 'ala-ma-kota'), ('@nooo', 'nooo'),
     ('$!haha lolz !', 'haha-lolz'), ('$$$$$', ''), ('{}OK!', 'OK'),
     ('/]re po', 're-po')])
 def test_SlugifyName(name, expected, localizer):
     validator = v.SlugifyName(localizer)
     assert expected == validator.to_python(name)
 
 
 def test_ValidForkType(localizer):
         validator = v.ValidForkType(localizer, old_data={'repo_type': 'hg'})
         assert 'hg' == validator.to_python('hg')
         pytest.raises(formencode.Invalid, validator.to_python, 'git')
 
 
 def test_ValidPath(localizer):
         validator = v.ValidPath(localizer)
         assert TESTS_TMP_PATH == validator.to_python(TESTS_TMP_PATH)
         pytest.raises(
             formencode.Invalid, validator.to_python, '/no_such_dir')
 
 
 def test_UniqSystemEmail(localizer):
     validator = v.UniqSystemEmail(localizer, old_data={})
 
     assert '[e-mail]' == validator.to_python('[e-mail]')
 
     email = TEST_USER_REGULAR2_EMAIL
     pytest.raises(formencode.Invalid, validator.to_python, email)
 
 
 def test_ValidSystemEmail(localizer):
     validator = v.ValidSystemEmail(localizer)
     email = TEST_USER_REGULAR2_EMAIL
 
     assert email == validator.to_python(email)
     pytest.raises(formencode.Invalid, validator.to_python, 'err')
 
 
 def test_NotReviewedRevisions(localizer):
     repo_id = Repository.get_by_repo_name(HG_REPO).repo_id
     validator = v.NotReviewedRevisions(localizer, repo_id)
     rev = '0' * 40
     # add status for a rev, that should throw an error because it is already
     # reviewed
     new_status = ChangesetStatus()
     new_status.author = ChangesetStatusModel()._get_user(TEST_USER_ADMIN_LOGIN)
     new_status.repo = ChangesetStatusModel()._get_repo(HG_REPO)
     new_status.status = ChangesetStatus.STATUS_APPROVED
     new_status.comment = None
     new_status.revision = rev
     Session().add(new_status)
     Session().commit()
     try:
         pytest.raises(formencode.Invalid, validator.to_python, [rev])
     finally:
         Session().delete(new_status)
         Session().commit()