From e9ad755887bb800820bdbaf05a449bace85601a6 Mon Sep 17 00:00:00 2001 From: RhodeCode Admin Date: Fri, 10 Jan 2025 10:42:53 +0100 Subject: [PATCH] tests: fixed some flaky tests --- .../apps/login/tests/test_register_captcha.py | 115 ++---- rhodecode/apps/login/views.py | 364 ++++++++--------- rhodecode/model/settings.py | 386 +++++++----------- rhodecode/tests/fixtures/rc_fixture.py | 28 ++ 4 files changed, 393 insertions(+), 500 deletions(-) diff --git a/rhodecode/apps/login/tests/test_register_captcha.py b/rhodecode/apps/login/tests/test_register_captcha.py index c50cd893..9fc81c83 100644 --- a/rhodecode/apps/login/tests/test_register_captcha.py +++ b/rhodecode/apps/login/tests/test_register_captcha.py @@ -24,108 +24,83 @@ from rhodecode.apps._base import ADMIN_PREFIX from rhodecode.apps.login.views import LoginView, CaptchaData from rhodecode.model.settings import SettingsModel from rhodecode.lib.utils2 import AttributeDict +from rhodecode.tests.fixtures.rc_fixture import RhodeCodeSettingFixture from rhodecode.tests.utils import AssertResponse -class RhodeCodeSetting(object): - def __init__(self, name, value): - self.name = name - self.value = value - - def __enter__(self): - from rhodecode.model.settings import SettingsModel - model = SettingsModel() - self.old_setting = model.get_setting_by_name(self.name) - model.create_or_update_setting(name=self.name, val=self.value) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - model = SettingsModel() - if self.old_setting: - model.create_or_update_setting( - name=self.name, val=self.old_setting.app_settings_value) - else: - model.create_or_update_setting(name=self.name) - - class TestRegisterCaptcha(object): - - @pytest.mark.parametrize('private_key, public_key, expected', [ - ('', '', CaptchaData(False, '', '')), - ('', 'pubkey', CaptchaData(False, '', 'pubkey')), - ('privkey', '', CaptchaData(True, 'privkey', '')), - ('privkey', 'pubkey', CaptchaData(True, 'privkey', 'pubkey')), - ]) - def test_get_captcha_data(self, private_key, public_key, expected, - request_stub, user_util): + @pytest.mark.parametrize( + "private_key, public_key, expected", + [ + ("", "", CaptchaData(False, "", "")), + ("", "pubkey", CaptchaData(False, "", "pubkey")), + ("privkey", "", CaptchaData(True, "privkey", "")), + ("privkey", "pubkey", CaptchaData(True, "privkey", "pubkey")), + ], + ) + def test_get_captcha_data(self, private_key, public_key, expected, request_stub, user_util): request_stub.user = user_util.create_user().AuthUser() - request_stub.matched_route = AttributeDict({'name': 'login'}) + request_stub.matched_route = AttributeDict({"name": "login"}) login_view = LoginView(mock.Mock(), request_stub) - with RhodeCodeSetting('captcha_private_key', private_key): - with RhodeCodeSetting('captcha_public_key', public_key): - captcha = login_view._get_captcha_data() + with RhodeCodeSettingFixture("captcha_private_key", private_key): + with RhodeCodeSettingFixture("captcha_public_key", public_key): + captcha = login_view._get_captcha_data(cache=False, from_request=False) assert captcha == expected - @pytest.mark.parametrize('active', [False, True]) - @mock.patch.object(LoginView, '_get_captcha_data') - def test_private_key_does_not_leak_to_html( - self, m_get_captcha_data, active, app): - captcha = CaptchaData( - active=active, private_key='PRIVATE_KEY', public_key='PUBLIC_KEY') + @pytest.mark.parametrize("active", [False, True]) + @mock.patch.object(LoginView, "_get_captcha_data") + def test_private_key_does_not_leak_to_html(self, m_get_captcha_data, active, app): + captcha = CaptchaData(active=active, private_key="PRIVATE_KEY", public_key="PUBLIC_KEY") m_get_captcha_data.return_value = captcha - response = app.get(ADMIN_PREFIX + '/register') - assert 'PRIVATE_KEY' not in response + response = app.get(ADMIN_PREFIX + "/register") + assert "PRIVATE_KEY" not in response - @pytest.mark.parametrize('active', [False, True]) - @mock.patch.object(LoginView, '_get_captcha_data') - def test_register_view_renders_captcha( - self, m_get_captcha_data, active, app): - captcha = CaptchaData( - active=active, private_key='PRIVATE_KEY', public_key='PUBLIC_KEY') + @pytest.mark.parametrize("active", [False, True]) + @mock.patch.object(LoginView, "_get_captcha_data") + def test_register_view_renders_captcha(self, m_get_captcha_data, active, app): + captcha = CaptchaData(active=active, private_key="PRIVATE_KEY", public_key="PUBLIC_KEY") m_get_captcha_data.return_value = captcha - response = app.get(ADMIN_PREFIX + '/register') + response = app.get(ADMIN_PREFIX + "/register") assertr = response.assert_response() if active: - assertr.one_element_exists('#recaptcha_field') + assertr.one_element_exists("#recaptcha_field") else: - assertr.no_element_exists('#recaptcha_field') + assertr.no_element_exists("#recaptcha_field") - @pytest.mark.parametrize('valid', [False, True]) - @mock.patch.object(LoginView, 'validate_captcha') - @mock.patch.object(LoginView, '_get_captcha_data') - def test_register_with_active_captcha( - self, m_get_captcha_data, m_validate_captcha, valid, app, csrf_token): - captcha = CaptchaData( - active=True, private_key='PRIVATE_KEY', public_key='PUBLIC_KEY') + @pytest.mark.parametrize("valid", [False, True]) + @mock.patch.object(LoginView, "validate_captcha") + @mock.patch.object(LoginView, "_get_captcha_data") + def test_register_with_active_captcha(self, m_get_captcha_data, m_validate_captcha, valid, app, csrf_token): + captcha = CaptchaData(active=True, private_key="PRIVATE_KEY", public_key="PUBLIC_KEY") m_get_captcha_data.return_value = captcha m_response = mock.Mock() m_response.is_valid = valid - m_validate_captcha.return_value = valid, 'ok' + m_validate_captcha.return_value = valid, "ok" params = { - 'csrf_token': csrf_token, - 'email': 'pytest@example.com', - 'firstname': 'pytest-firstname', - 'lastname': 'pytest-lastname', - 'password': 'secret', - 'password_confirmation': 'secret', - 'username': 'pytest', + "csrf_token": csrf_token, + "email": "pytest@example.com", + "firstname": "pytest-firstname", + "lastname": "pytest-lastname", + "password": "secret", + "password_confirmation": "secret", + "username": "pytest", } - response = app.post(ADMIN_PREFIX + '/register', params=params) + response = app.post(ADMIN_PREFIX + "/register", params=params) if valid: # If we provided a valid captcha input we expect a successful # registration and redirect to the login page. assert response.status_int == 302 - assert 'location' in response.headers - assert ADMIN_PREFIX + '/login' in response.headers['location'] + assert "location" in response.headers + assert ADMIN_PREFIX + "/login" in response.headers["location"] else: # If captche input is invalid we expect to stay on the registration # page with an error message displayed. assertr = response.assert_response() assert response.status_int == 200 - assertr.one_element_exists('#recaptcha_field ~ span.error-message') + assertr.one_element_exists("#recaptcha_field ~ span.error-message") diff --git a/rhodecode/apps/login/views.py b/rhodecode/apps/login/views.py index 3c5a7202..371cb9a4 100644 --- a/rhodecode/apps/login/views.py +++ b/rhodecode/apps/login/views.py @@ -17,10 +17,9 @@ # and proprietary license terms, please see https://rhodecode.com/licenses/ import time -import json import pyotp import qrcode -import collections +import dataclasses import datetime import formencode import formencode.htmlfill @@ -41,8 +40,7 @@ from rhodecode.authentication.plugins import auth_rhodecode from rhodecode.events import UserRegistered, trigger from rhodecode.lib import helpers as h from rhodecode.lib import audit_logger -from rhodecode.lib.auth import ( - AuthUser, HasPermissionAnyDecorator, CSRFRequired, LoginRequired, NotAnonymous) +from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator, CSRFRequired, LoginRequired, NotAnonymous from rhodecode.lib.base import get_ip_addr from rhodecode.lib.exceptions import UserCreationError from rhodecode.lib.utils2 import safe_str @@ -57,8 +55,12 @@ from rhodecode.translation import _ log = logging.getLogger(__name__) -CaptchaData = collections.namedtuple( - 'CaptchaData', 'active, private_key, public_key') + +@dataclasses.dataclass +class CaptchaData: + active: bool + private_key: str + public_key: str def store_user_in_session(session, user_identifier, remember=False): @@ -66,90 +68,77 @@ def store_user_in_session(session, user_identifier, remember=False): auth_user = AuthUser(user.user_id) auth_user.set_authenticated() cs = auth_user.get_cookie_store() - session['rhodecode_user'] = cs + session["rhodecode_user"] = cs user.update_lastlogin() Session().commit() # If they want to be remembered, update the cookie if remember: - _year = (datetime.datetime.now() + - datetime.timedelta(seconds=60 * 60 * 24 * 365)) + _year = datetime.datetime.now() + datetime.timedelta(seconds=60 * 60 * 24 * 365) session._set_cookie_expires(_year) session.save() safe_cs = cs.copy() - safe_cs['password'] = '****' - log.info('user %s is now authenticated and stored in ' - 'session, session attrs %s', user_identifier, safe_cs) + safe_cs["password"] = "****" + log.info("user %s is now authenticated and stored in " "session, session attrs %s", user_identifier, safe_cs) # dumps session attrs back to cookie session._update_cookie_out() # we set new cookie headers = None - if session.request['set_cookie']: + if session.request["set_cookie"]: # send set-cookie headers back to response to update cookie - headers = [('Set-Cookie', session.request['cookie_out'])] + headers = [("Set-Cookie", session.request["cookie_out"])] return headers def get_came_from(request): - came_from = safe_str(request.GET.get('came_from', '')) + came_from = safe_str(request.GET.get("came_from", "")) parsed = urllib.parse.urlparse(came_from) - allowed_schemes = ['http', 'https'] - default_came_from = h.route_path('home') + allowed_schemes = ["http", "https"] + default_came_from = h.route_path("home") if parsed.scheme and parsed.scheme not in allowed_schemes: - log.error('Suspicious URL scheme detected %s for url %s', - parsed.scheme, parsed) + log.error("Suspicious URL scheme detected %s for url %s", parsed.scheme, parsed) came_from = default_came_from elif parsed.netloc and request.host != parsed.netloc: - log.error('Suspicious NETLOC detected %s for url %s server url ' - 'is: %s', parsed.netloc, parsed, request.host) + log.error("Suspicious NETLOC detected %s for url %s server url " "is: %s", parsed.netloc, parsed, request.host) came_from = default_came_from - elif any(bad_char in came_from for bad_char in ('\r', '\n')): - log.error('Header injection detected `%s` for url %s server url ', - parsed.path, parsed) + elif any(bad_char in came_from for bad_char in ("\r", "\n")): + log.error("Header injection detected `%s` for url %s server url ", parsed.path, parsed) came_from = default_came_from return came_from or default_came_from class LoginView(BaseAppView): - def load_default_context(self): c = self._get_local_tmpl_context() c.came_from = get_came_from(self.request) return c - def _get_captcha_data(self): - settings = SettingsModel().get_all_settings() - private_key = settings.get('rhodecode_captcha_private_key') - public_key = settings.get('rhodecode_captcha_public_key') + def _get_captcha_data(self, *args, **kwargs): + settings = SettingsModel().get_all_settings(*args, **kwargs) + private_key = settings.get("rhodecode_captcha_private_key", "") + public_key = settings.get("rhodecode_captcha_public_key", "") active = bool(private_key) - return CaptchaData( - active=active, private_key=private_key, public_key=public_key) + return CaptchaData(active=active, private_key=private_key, public_key=public_key) def validate_captcha(self, private_key): - - captcha_rs = self.request.POST.get('g-recaptcha-response') + captcha_rs = self.request.POST.get("g-recaptcha-response") url = "https://www.google.com/recaptcha/api/siteverify" - params = { - 'secret': private_key, - 'response': captcha_rs, - 'remoteip': get_ip_addr(self.request.environ) - } + params = {"secret": private_key, "response": captcha_rs, "remoteip": get_ip_addr(self.request.environ)} verify_rs = requests.get(url, params=params, verify=True, timeout=60) verify_rs = verify_rs.json() - captcha_status = verify_rs.get('success', False) - captcha_errors = verify_rs.get('error-codes', []) + captcha_status = verify_rs.get("success", False) + captcha_errors = verify_rs.get("error-codes", []) if not isinstance(captcha_errors, list): captcha_errors = [captcha_errors] - captcha_errors = ', '.join(captcha_errors) - captcha_message = '' + captcha_errors = ", ".join(captcha_errors) + captcha_message = "" if captcha_status is False: - captcha_message = "Bad captcha. Errors: {}".format( - captcha_errors) + captcha_message = "Bad captcha. Errors: {}".format(captcha_errors) return captcha_status, captcha_message @@ -158,22 +147,19 @@ class LoginView(BaseAppView): auth_user = self._rhodecode_user # redirect if already logged in - if (auth_user.is_authenticated and - not auth_user.is_default and auth_user.ip_allowed): + if auth_user.is_authenticated and not auth_user.is_default and auth_user.ip_allowed: raise HTTPFound(c.came_from) # check if we use headers plugin, and try to login using it. try: - log.debug('Running PRE-AUTH for headers based authentication') - auth_info = authenticate( - '', '', self.request.environ, HTTP_TYPE, skip_missing=True) + log.debug("Running PRE-AUTH for headers based authentication") + auth_info = authenticate("", "", self.request.environ, HTTP_TYPE, skip_missing=True) if auth_info: - headers = store_user_in_session( - self.session, auth_info.get('username')) + headers = store_user_in_session(self.session, auth_info.get("username")) raise HTTPFound(c.came_from, headers=headers) except UserCreationError as e: log.error(e) - h.flash(e, category='error') + h.flash(e, category="error") return self._get_template_context(c) @@ -186,41 +172,34 @@ class LoginView(BaseAppView): self.session.invalidate() form_result = login_form.to_python(self.request.POST) # form checks for username/password, now we're authenticated - username = form_result['username'] + username = form_result["username"] if (user := User.get_by_username_or_primary_email(username)).has_enabled_2fa: user.check_2fa_required = True - headers = store_user_in_session( - self.session, - user_identifier=username, - remember=form_result['remember']) + headers = store_user_in_session(self.session, user_identifier=username, remember=form_result["remember"]) log.debug('Redirecting to "%s" after login.', c.came_from) audit_user = audit_logger.UserWrap( - username=self.request.POST.get('username'), - ip_addr=self.request.remote_addr) - action_data = {'user_agent': self.request.user_agent} - audit_logger.store_web( - 'user.login.success', action_data=action_data, - user=audit_user, commit=True) + username=self.request.POST.get("username"), ip_addr=self.request.remote_addr + ) + action_data = {"user_agent": self.request.user_agent} + audit_logger.store_web("user.login.success", action_data=action_data, user=audit_user, commit=True) raise HTTPFound(c.came_from, headers=headers) except formencode.Invalid as errors: defaults = errors.value # remove password from filling in form again - defaults.pop('password', None) + defaults.pop("password", None) render_ctx = { - 'errors': errors.error_dict, - 'defaults': defaults, + "errors": errors.error_dict, + "defaults": defaults, } audit_user = audit_logger.UserWrap( - username=self.request.POST.get('username'), - ip_addr=self.request.remote_addr) - action_data = {'user_agent': self.request.user_agent} - audit_logger.store_web( - 'user.login.failure', action_data=action_data, - user=audit_user, commit=True) + username=self.request.POST.get("username"), ip_addr=self.request.remote_addr + ) + action_data = {"user_agent": self.request.user_agent} + audit_logger.store_web("user.login.failure", action_data=action_data, user=audit_user, commit=True) return self._get_template_context(c, **render_ctx) except UserCreationError as e: @@ -228,125 +207,110 @@ class LoginView(BaseAppView): # the fly can throw this exception signaling that there's issue # with user creation, explanation should be provided in # Exception itself - h.flash(e, category='error') + h.flash(e, category="error") return self._get_template_context(c) @CSRFRequired() def logout(self): auth_user = self._rhodecode_user - log.info('Deleting session for user: `%s`', auth_user) + log.info("Deleting session for user: `%s`", auth_user) - action_data = {'user_agent': self.request.user_agent} - audit_logger.store_web( - 'user.logout', action_data=action_data, - user=auth_user, commit=True) + action_data = {"user_agent": self.request.user_agent} + audit_logger.store_web("user.logout", action_data=action_data, user=auth_user, commit=True) self.session.delete() - return HTTPFound(h.route_path('home')) + return HTTPFound(h.route_path("home")) - @HasPermissionAnyDecorator( - 'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate') + @HasPermissionAnyDecorator("hg.admin", "hg.register.auto_activate", "hg.register.manual_activate") def register(self, defaults=None, errors=None): c = self.load_default_context() defaults = defaults or {} errors = errors or {} settings = SettingsModel().get_all_settings() - register_message = settings.get('rhodecode_register_message') or '' + register_message = settings.get("rhodecode_register_message") or "" captcha = self._get_captcha_data() - auto_active = 'hg.register.auto_activate' in User.get_default_user()\ - .AuthUser().permissions['global'] + auto_active = "hg.register.auto_activate" in User.get_default_user().AuthUser().permissions["global"] render_ctx = self._get_template_context(c) - render_ctx.update({ - 'defaults': defaults, - 'errors': errors, - 'auto_active': auto_active, - 'captcha_active': captcha.active, - 'captcha_public_key': captcha.public_key, - 'register_message': register_message, - }) + render_ctx.update( + { + "defaults": defaults, + "errors": errors, + "auto_active": auto_active, + "captcha_active": captcha.active, + "captcha_public_key": captcha.public_key, + "register_message": register_message, + } + ) return render_ctx - @HasPermissionAnyDecorator( - 'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate') + @HasPermissionAnyDecorator("hg.admin", "hg.register.auto_activate", "hg.register.manual_activate") def register_post(self): from rhodecode.authentication.plugins import auth_rhodecode self.load_default_context() captcha = self._get_captcha_data() - auto_active = 'hg.register.auto_activate' in User.get_default_user()\ - .AuthUser().permissions['global'] + auto_active = "hg.register.auto_activate" in User.get_default_user().AuthUser().permissions["global"] extern_name = auth_rhodecode.RhodeCodeAuthPlugin.uid extern_type = auth_rhodecode.RhodeCodeAuthPlugin.uid register_form = RegisterForm(self.request.translate)() try: - form_result = register_form.to_python(self.request.POST) - form_result['active'] = auto_active - external_identity = self.request.POST.get('external_identity') + form_result["active"] = auto_active + external_identity = self.request.POST.get("external_identity") if external_identity: extern_name = external_identity extern_type = external_identity if captcha.active: - captcha_status, captcha_message = self.validate_captcha( - captcha.private_key) + captcha_status, captcha_message = self.validate_captcha(captcha.private_key) if not captcha_status: _value = form_result - _msg = _('Bad captcha') - error_dict = {'recaptcha_field': captcha_message} - raise formencode.Invalid( - _msg, _value, None, error_dict=error_dict) + _msg = _("Bad captcha") + error_dict = {"recaptcha_field": captcha_message} + raise formencode.Invalid(_msg, _value, None, error_dict=error_dict) - new_user = UserModel().create_registration( - form_result, extern_name=extern_name, extern_type=extern_type) + new_user = UserModel().create_registration(form_result, extern_name=extern_name, extern_type=extern_type) - action_data = {'data': new_user.get_api_data(), - 'user_agent': self.request.user_agent} + action_data = {"data": new_user.get_api_data(), "user_agent": self.request.user_agent} if external_identity: - action_data['external_identity'] = external_identity + action_data["external_identity"] = external_identity audit_user = audit_logger.UserWrap( - username=new_user.username, - user_id=new_user.user_id, - ip_addr=self.request.remote_addr) + username=new_user.username, user_id=new_user.user_id, ip_addr=self.request.remote_addr + ) - audit_logger.store_web( - 'user.register', action_data=action_data, - user=audit_user) + audit_logger.store_web("user.register", action_data=action_data, user=audit_user) event = UserRegistered(user=new_user, session=self.session) trigger(event) - h.flash( - _('You have successfully registered with RhodeCode. You can log-in now.'), - category='success') + h.flash(_("You have successfully registered with RhodeCode. You can log-in now."), category="success") if external_identity: h.flash( - _('Please use the {identity} button to log-in').format( - identity=external_identity), - category='success') + _("Please use the {identity} button to log-in").format(identity=external_identity), + category="success", + ) Session().commit() - redirect_ro = self.request.route_path('login') + redirect_ro = self.request.route_path("login") raise HTTPFound(redirect_ro) except formencode.Invalid as errors: - errors.value.pop('password', None) - errors.value.pop('password_confirmation', None) - return self.register( - defaults=errors.value, errors=errors.error_dict) + errors.value.pop("password", None) + errors.value.pop("password_confirmation", None) + return self.register(defaults=errors.value, errors=errors.error_dict) except UserCreationError as e: # container auth or other auth functions that create users on # the fly can throw this exception signaling that there's issue # with user creation, explanation should be provided in # Exception itself - h.flash(e, category='error') + h.flash(e, category="error") return self.register() def password_reset(self): @@ -354,49 +318,45 @@ class LoginView(BaseAppView): captcha = self._get_captcha_data() template_context = { - 'captcha_active': captcha.active, - 'captcha_public_key': captcha.public_key, - 'defaults': {}, - 'errors': {}, + "captcha_active": captcha.active, + "captcha_public_key": captcha.public_key, + "defaults": {}, + "errors": {}, } # always send implicit message to prevent from discovery of # matching emails - msg = _('If such email exists, a password reset link was sent to it.') + msg = _("If such email exists, a password reset link was sent to it.") def default_response(): - log.debug('faking response on invalid password reset') + log.debug("faking response on invalid password reset") # make this take 2s, to prevent brute forcing. time.sleep(2) - h.flash(msg, category='success') - return HTTPFound(self.request.route_path('reset_password')) + h.flash(msg, category="success") + return HTTPFound(self.request.route_path("reset_password")) if self.request.POST: - if h.HasPermissionAny('hg.password_reset.disabled')(): - _email = self.request.POST.get('email', '') - log.error('Failed attempt to reset password for `%s`.', _email) - h.flash(_('Password reset has been disabled.'), category='error') - return HTTPFound(self.request.route_path('reset_password')) + if h.HasPermissionAny("hg.password_reset.disabled")(): + _email = self.request.POST.get("email", "") + log.error("Failed attempt to reset password for `%s`.", _email) + h.flash(_("Password reset has been disabled."), category="error") + return HTTPFound(self.request.route_path("reset_password")) password_reset_form = PasswordResetForm(self.request.translate)() - description = 'Generated token for password reset from {}'.format( - datetime.datetime.now().isoformat()) + description = "Generated token for password reset from {}".format(datetime.datetime.now().isoformat()) try: - form_result = password_reset_form.to_python( - self.request.POST) - user_email = form_result['email'] + form_result = password_reset_form.to_python(self.request.POST) + user_email = form_result["email"] if captcha.active: - captcha_status, captcha_message = self.validate_captcha( - captcha.private_key) + captcha_status, captcha_message = self.validate_captcha(captcha.private_key) if not captcha_status: _value = form_result - _msg = _('Bad captcha') - error_dict = {'recaptcha_field': captcha_message} - raise formencode.Invalid( - _msg, _value, None, error_dict=error_dict) + _msg = _("Bad captcha") + error_dict = {"recaptcha_field": captcha_message} + raise formencode.Invalid(_msg, _value, None, error_dict=error_dict) # Generate reset URL and send mail. user = User.get_by_email(user_email) @@ -404,42 +364,47 @@ class LoginView(BaseAppView): # only allow rhodecode based users to reset their password # external auth shouldn't allow password reset if user and user.extern_type != auth_rhodecode.RhodeCodeAuthPlugin.uid: - log.warning('User %s with external type `%s` tried a password reset. ' - 'This try was rejected', user, user.extern_type) + log.warning( + "User %s with external type `%s` tried a password reset. " "This try was rejected", + user, + user.extern_type, + ) return default_response() # generate password reset token that expires in 10 minutes reset_token = UserModel().add_auth_token( - user=user, lifetime_minutes=10, + user=user, + lifetime_minutes=10, role=UserModel.auth_token_role.ROLE_PASSWORD_RESET, - description=description) + description=description, + ) Session().commit() - log.debug('Successfully created password recovery token') + log.debug("Successfully created password recovery token") password_reset_url = self.request.route_url( - 'reset_password_confirmation', - _query={'key': reset_token.api_key}) - UserModel().reset_password_link( - form_result, password_reset_url) + "reset_password_confirmation", _query={"key": reset_token.api_key} + ) + UserModel().reset_password_link(form_result, password_reset_url) - action_data = {'email': user_email, - 'user_agent': self.request.user_agent} + action_data = {"email": user_email, "user_agent": self.request.user_agent} audit_logger.store_web( - 'user.password.reset_request', action_data=action_data, - user=self._rhodecode_user, commit=True) + "user.password.reset_request", action_data=action_data, user=self._rhodecode_user, commit=True + ) return default_response() except formencode.Invalid as errors: - template_context.update({ - 'defaults': errors.value, - 'errors': errors.error_dict, - }) - if not self.request.POST.get('email'): + template_context.update( + { + "defaults": errors.value, + "errors": errors.error_dict, + } + ) + if not self.request.POST.get("email"): # case of empty email, we want to report that return self._get_template_context(c, **template_context) - if 'recaptcha_field' in errors.error_dict: + if "recaptcha_field" in errors.error_dict: # case of failed captcha return self._get_template_context(c, **template_context) @@ -450,7 +415,7 @@ class LoginView(BaseAppView): def password_reset_confirmation(self): self.load_default_context() - if key := self.request.GET.get('key'): + if key := self.request.GET.get("key"): # make this take 2s, to prevent brute forcing. time.sleep(2) @@ -458,26 +423,27 @@ class LoginView(BaseAppView): # verify token is the correct role if token is None or token.role != UserApiKeys.ROLE_PASSWORD_RESET: - log.debug('Got token with role:%s expected is %s', - getattr(token, 'role', 'EMPTY_TOKEN'), - UserApiKeys.ROLE_PASSWORD_RESET) - h.flash( - _('Given reset token is invalid'), category='error') - return HTTPFound(self.request.route_path('reset_password')) + log.debug( + "Got token with role:%s expected is %s", + getattr(token, "role", "EMPTY_TOKEN"), + UserApiKeys.ROLE_PASSWORD_RESET, + ) + h.flash(_("Given reset token is invalid"), category="error") + return HTTPFound(self.request.route_path("reset_password")) try: owner = token.user - data = {'email': owner.email, 'token': token.api_key} + data = {"email": owner.email, "token": token.api_key} UserModel().reset_password(data) h.flash( - _('Your password reset was successful, ' - 'a new password has been sent to your email'), - category='success') + _("Your password reset was successful, " "a new password has been sent to your email"), + category="success", + ) except Exception as e: log.error(e) - return HTTPFound(self.request.route_path('reset_password')) + return HTTPFound(self.request.route_path("reset_password")) - return HTTPFound(self.request.route_path('login')) + return HTTPFound(self.request.route_path("login")) @LoginRequired() @NotAnonymous() @@ -487,44 +453,40 @@ class LoginView(BaseAppView): user_instance = self._rhodecode_db_user form = TOTPForm(_, user_instance)() render_ctx = {} - if self.request.method == 'POST': + if self.request.method == "POST": post_items = dict(self.request.POST) try: form_details = form.to_python(post_items) - secret = form_details['secret_totp'] + secret = form_details["secret_totp"] user_instance.init_2fa_recovery_codes(persist=True, force=True) user_instance.secret_2fa = secret Session().commit() - raise HTTPFound(self.request.route_path('my_account_configure_2fa', _query={'show-recovery-codes': 1})) + raise HTTPFound(self.request.route_path("my_account_configure_2fa", _query={"show-recovery-codes": 1})) except formencode.Invalid as errors: defaults = errors.value render_ctx = { - 'errors': errors.error_dict, - 'defaults': defaults, + "errors": errors.error_dict, + "defaults": defaults, } # NOTE: here we DO NOT persist the secret 2FA, since this is only for setup, once a setup is completed # only then we should persist it secret = user_instance.init_secret_2fa(persist=False) - instance_name = rhodecode.ConfigGet().get_str('app.base_url', 'rhodecode') - totp_name = f'{instance_name}:{self.request.user.username}' + instance_name = rhodecode.ConfigGet().get_str("app.base_url", "rhodecode") + totp_name = f"{instance_name}:{self.request.user.username}" qr = qrcode.QRCode(version=1, box_size=5, border=4) qr.add_data(pyotp.totp.TOTP(secret).provisioning_uri(name=totp_name)) qr.make(fit=True) - img = qr.make_image(fill_color='black', back_color='white') + img = qr.make_image(fill_color="black", back_color="white") buffered = BytesIO() img.save(buffered) return self._get_template_context( - c, - qr=b64encode(buffered.getvalue()).decode("utf-8"), - key=secret, - totp_name=totp_name, - ** render_ctx + c, qr=b64encode(buffered.getvalue()).decode("utf-8"), key=secret, totp_name=totp_name, **render_ctx ) @LoginRequired() @@ -535,10 +497,10 @@ class LoginView(BaseAppView): render_ctx = {} user_instance = self._rhodecode_db_user totp_form = TOTPForm(_, user_instance, allow_recovery_code_use=True)() - if self.request.method == 'POST': + if self.request.method == "POST": post_items = dict(self.request.POST) # NOTE: inject secret, as it's a post configured saved item. - post_items['secret_totp'] = user_instance.secret_2fa + post_items["secret_totp"] = user_instance.secret_2fa try: totp_form.to_python(post_items) user_instance.check_2fa_required = False @@ -547,7 +509,7 @@ class LoginView(BaseAppView): except formencode.Invalid as errors: defaults = errors.value render_ctx = { - 'errors': errors.error_dict, - 'defaults': defaults, + "errors": errors.error_dict, + "defaults": defaults, } return self._get_template_context(c, **render_ctx) diff --git a/rhodecode/model/settings.py b/rhodecode/model/settings.py index adcbadb7..7026a638 100644 --- a/rhodecode/model/settings.py +++ b/rhodecode/model/settings.py @@ -27,48 +27,48 @@ from rhodecode.lib.pyramid_utils import get_current_request from rhodecode.lib import rc_cache from rhodecode.lib.hash_utils import sha1_safe from rhodecode.lib.html_filters import sanitize_html -from rhodecode.lib.utils2 import ( - Optional, AttributeDict, safe_str, remove_prefix, str2bool) +from rhodecode.lib.utils2 import Optional, AttributeDict, safe_str, remove_prefix, str2bool from rhodecode.lib.vcs.backends import base from rhodecode.lib.statsd_client import StatsdClient from rhodecode.model import BaseModel -from rhodecode.model.db import ( - RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi, RhodeCodeSetting) +from rhodecode.model.db import RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi, RhodeCodeSetting from rhodecode.model.meta import Session log = logging.getLogger(__name__) -UiSetting = namedtuple( - 'UiSetting', ['section', 'key', 'value', 'active']) +UiSetting = namedtuple("UiSetting", ["section", "key", "value", "active"]) -SOCIAL_PLUGINS_LIST = ['github', 'bitbucket', 'twitter', 'google'] +SOCIAL_PLUGINS_LIST = ["github", "bitbucket", "twitter", "google"] class SettingNotFound(Exception): def __init__(self, setting_id): - msg = f'Setting `{setting_id}` is not found' + msg = f"Setting `{setting_id}` is not found" super().__init__(msg) class SettingsModel(BaseModel): BUILTIN_HOOKS = ( - RhodeCodeUi.HOOK_REPO_SIZE, RhodeCodeUi.HOOK_PUSH, - RhodeCodeUi.HOOK_PRE_PUSH, RhodeCodeUi.HOOK_PRETX_PUSH, - RhodeCodeUi.HOOK_PULL, RhodeCodeUi.HOOK_PRE_PULL, - RhodeCodeUi.HOOK_PUSH_KEY,) - HOOKS_SECTION = 'hooks' + RhodeCodeUi.HOOK_REPO_SIZE, + RhodeCodeUi.HOOK_PUSH, + RhodeCodeUi.HOOK_PRE_PUSH, + RhodeCodeUi.HOOK_PRETX_PUSH, + RhodeCodeUi.HOOK_PULL, + RhodeCodeUi.HOOK_PRE_PULL, + RhodeCodeUi.HOOK_PUSH_KEY, + ) + HOOKS_SECTION = "hooks" def __init__(self, sa=None, repo=None): self.repo = repo self.UiDbModel = RepoRhodeCodeUi if repo else RhodeCodeUi - self.SettingsDbModel = ( - RepoRhodeCodeSetting if repo else RhodeCodeSetting) + self.SettingsDbModel = RepoRhodeCodeSetting if repo else RhodeCodeSetting super().__init__(sa) - def get_keyname(self, key_name, prefix='rhodecode_'): - return f'{prefix}{key_name}' + def get_keyname(self, key_name, prefix="rhodecode_"): + return f"{prefix}{key_name}" def get_ui_by_key(self, key): q = self.UiDbModel.query() @@ -101,8 +101,7 @@ class SettingsModel(BaseModel): # TODO: mikhail: add caching result = [ UiSetting( - section=safe_str(r.ui_section), key=safe_str(r.ui_key), - value=safe_str(r.ui_value), active=r.ui_active + section=safe_str(r.ui_section), key=safe_str(r.ui_key), value=safe_str(r.ui_value), active=r.ui_active ) for r in q.all() ] @@ -124,7 +123,7 @@ class SettingsModel(BaseModel): new_ui.ui_value = val new_ui.ui_active = active - repository_id = '' + repository_id = "" if self.repo: repo = self._get_repo(self.repo) repository_id = repo.repo_id @@ -133,9 +132,9 @@ class SettingsModel(BaseModel): if not key: # keys are unique so they need appended info if self.repo: - key = sha1_safe(f'{section}{val}{repository_id}') + key = sha1_safe(f"{section}{val}{repository_id}") else: - key = sha1_safe(f'{section}{val}') + key = sha1_safe(f"{section}{val}") new_ui.ui_key = key @@ -143,9 +142,7 @@ class SettingsModel(BaseModel): return new_ui def create_or_update_hook(self, key, value): - ui = ( - self.get_ui_by_section_and_key(self.HOOKS_SECTION, key) or - self.UiDbModel()) + ui = self.get_ui_by_section_and_key(self.HOOKS_SECTION, key) or self.UiDbModel() ui.ui_section = self.HOOKS_SECTION ui.ui_active = True ui.ui_key = key @@ -171,7 +168,8 @@ class SettingsModel(BaseModel): return q.scalar() def create_or_update_setting( - self, name, val: Optional | str = Optional(''), type_: Optional | str = Optional('unicode')): + self, name, val: Optional | str = Optional(""), type_: Optional | str = Optional("unicode") + ): """ Creates or updates RhodeCode setting. If updates are triggered, it will only update parameters that are explicitly set Optional instance will @@ -190,9 +188,7 @@ class SettingsModel(BaseModel): val = Optional.extract(val) type_ = Optional.extract(type_) - args = ( - (repo.repo_id, name, val, type_) - if repo else (name, val, type_)) + args = (repo.repo_id, name, val, type_) if repo else (name, val, type_) res = self.SettingsDbModel(*args) else: @@ -213,14 +209,15 @@ class SettingsModel(BaseModel): def get_cache_region(self): repo = self._get_repo(self.repo) if self.repo else None cache_key = f"repo.v1.{repo.repo_id}" if repo else "repo.v1.ALL" - cache_namespace_uid = f'cache_settings.{cache_key}' - region = rc_cache.get_or_create_region('cache_general', cache_namespace_uid) + cache_namespace_uid = f"cache_settings.{cache_key}" + region = rc_cache.get_or_create_region("cache_general", cache_namespace_uid) return region, cache_namespace_uid def invalidate_settings_cache(self, hard=False): region, namespace_key = self.get_cache_region() - log.debug('Invalidation cache [%s] region %s for cache_key: %s', - 'invalidate_settings_cache', region, namespace_key) + log.debug( + "Invalidation cache [%s] region %s for cache_key: %s", "invalidate_settings_cache", region, namespace_key + ) # we use hard cleanup if invalidation is sent rc_cache.clear_cache_namespace(region, namespace_key, method=rc_cache.CLEAR_DELETE) @@ -232,13 +229,11 @@ class SettingsModel(BaseModel): def _get_all_settings(name, key): q = self._get_settings_query() if not q: - raise Exception('Could not get application settings !') + raise Exception("Could not get application settings !") - settings = { - self.get_keyname(res.app_settings_name): res.app_settings_value - for res in q - } + settings = {self.get_keyname(res.app_settings_name): res.app_settings_value for res in q} return settings + return _get_all_settings def get_all_settings(self, cache=False, from_request=True): @@ -250,7 +245,8 @@ class SettingsModel(BaseModel): if from_request and not repo: request = get_current_request() - if request and hasattr(request, 'call_context') and hasattr(request.call_context, 'rc_config'): + if request and hasattr(request, "call_context") and hasattr(request.call_context, "rc_config"): + log.debug("Using cached request settings [%s]", request.call_context.rc_config) rc_config = request.call_context.rc_config if rc_config: return rc_config @@ -259,27 +255,24 @@ class SettingsModel(BaseModel): _get_all_settings = self.get_cache_call_method(cache=cache) start = time.time() - result = _get_all_settings('rhodecode_settings', cache_key) + result = _get_all_settings("rhodecode_settings", cache_key) compute_time = time.time() - start - log.debug('cached method:%s took %.4fs', _get_all_settings.__name__, compute_time) + log.debug("cached method:%s took %.4fs", _get_all_settings.__name__, compute_time) statsd = StatsdClient.statsd if statsd: elapsed_time_ms = round(1000.0 * compute_time) # use ms only - statsd.timing("rhodecode_settings_timing.histogram", elapsed_time_ms, - use_decimals=False) + statsd.timing("rhodecode_settings_timing.histogram", elapsed_time_ms, use_decimals=False) - log.debug('Fetching app settings for key: %s took: %.4fs: cache: %s', cache_key, compute_time, cache) + log.debug("Fetching app settings for key: %s took: %.4fs: cache: %s", cache_key, compute_time, cache) return result def get_auth_settings(self): q = self._get_settings_query() - q = q.filter( - self.SettingsDbModel.app_settings_name.startswith('auth_')) + q = q.filter(self.SettingsDbModel.app_settings_name.startswith("auth_")) rows = q.all() - auth_settings = { - row.app_settings_name: row.app_settings_value for row in rows} + auth_settings = {row.app_settings_name: row.app_settings_value for row in rows} return auth_settings def get_auth_plugins(self): @@ -288,23 +281,21 @@ class SettingsModel(BaseModel): def get_default_repo_settings(self, strip_prefix=False): q = self._get_settings_query() - q = q.filter( - self.SettingsDbModel.app_settings_name.startswith('default_')) + q = q.filter(self.SettingsDbModel.app_settings_name.startswith("default_")) rows = q.all() result = {} for row in rows: key = row.app_settings_name if strip_prefix: - key = remove_prefix(key, prefix='default_') + key = remove_prefix(key, prefix="default_") result.update({key: row.app_settings_value}) return result def get_repo(self): repo = self._get_repo(self.repo) if not repo: - raise Exception( - f'Repository `{self.repo}` cannot be found inside the database') + raise Exception(f"Repository `{self.repo}` cannot be found inside the database") return repo def _filter_by_repo(self, model, query): @@ -325,7 +316,7 @@ class SettingsModel(BaseModel): def list_enabled_social_plugins(self, settings): enabled = [] for plug in SOCIAL_PLUGINS_LIST: - if str2bool(settings.get(f'rhodecode_auth_{plug}_enabled')): + if str2bool(settings.get(f"rhodecode_auth_{plug}_enabled")): enabled.append(plug) return enabled @@ -334,14 +325,15 @@ def assert_repo_settings(func): @functools.wraps(func) def _wrapper(self, *args, **kwargs): if not self.repo_settings: - raise Exception('Repository is not specified') + raise Exception("Repository is not specified") return func(self, *args, **kwargs) + return _wrapper class IssueTrackerSettingsModel(object): - INHERIT_SETTINGS = 'inherit_issue_tracker_settings' - SETTINGS_PREFIX = 'issuetracker_' + INHERIT_SETTINGS = "inherit_issue_tracker_settings" + SETTINGS_PREFIX = "issuetracker_" def __init__(self, sa=None, repo=None): self.global_settings = SettingsModel(sa=sa) @@ -357,48 +349,49 @@ class IssueTrackerSettingsModel(object): @inherit_global_settings.setter def inherit_global_settings(self, value): if self.repo_settings: - settings = self.repo_settings.create_or_update_setting( - self.INHERIT_SETTINGS, value, type_='bool') + settings = self.repo_settings.create_or_update_setting(self.INHERIT_SETTINGS, value, type_="bool") Session().add(settings) - def _get_keyname(self, key, uid, prefix='rhodecode_'): - return f'{prefix}{self.SETTINGS_PREFIX}{key}_{uid}' + def _get_keyname(self, key, uid, prefix="rhodecode_"): + return f"{prefix}{self.SETTINGS_PREFIX}{key}_{uid}" def _make_dict_for_settings(self, qs): - prefix_match = self._get_keyname('pat', '',) + prefix_match = self._get_keyname( + "pat", + "", + ) issuetracker_entries = {} # create keys for k, v in qs.items(): if k.startswith(prefix_match): - uid = k[len(prefix_match):] + uid = k[len(prefix_match) :] issuetracker_entries[uid] = None def url_cleaner(input_str): - input_str = input_str.replace('"', '').replace("'", '') + input_str = input_str.replace('"', "").replace("'", "") input_str = sanitize_html(input_str, strip=True) return input_str # populate for uid in issuetracker_entries: - url_data = qs.get(self._get_keyname('url', uid)) + url_data = qs.get(self._get_keyname("url", uid)) - pat = qs.get(self._get_keyname('pat', uid)) + pat = qs.get(self._get_keyname("pat", uid)) try: - pat_compiled = re.compile(r'%s' % pat) + pat_compiled = re.compile(r"%s" % pat) except re.error: pat_compiled = None - issuetracker_entries[uid] = AttributeDict({ - 'pat': pat, - 'pat_compiled': pat_compiled, - 'url': url_cleaner( - qs.get(self._get_keyname('url', uid)) or ''), - 'pref': sanitize_html( - qs.get(self._get_keyname('pref', uid)) or ''), - 'desc': qs.get( - self._get_keyname('desc', uid)), - }) + issuetracker_entries[uid] = AttributeDict( + { + "pat": pat, + "pat_compiled": pat_compiled, + "url": url_cleaner(qs.get(self._get_keyname("url", uid)) or ""), + "pref": sanitize_html(qs.get(self._get_keyname("pref", uid)) or ""), + "desc": qs.get(self._get_keyname("desc", uid)), + } + ) return issuetracker_entries @@ -415,7 +408,7 @@ class IssueTrackerSettingsModel(object): Returns list of issue tracker settings per repository """ if not self.repo_settings: - raise Exception('Repository is not specified') + raise Exception("Repository is not specified") all_settings = self.repo_settings.get_all_settings(cache=cache) settings = self._make_dict_for_settings(all_settings) return settings @@ -436,76 +429,67 @@ class IssueTrackerSettingsModel(object): entries = all_patterns.get(uid, []) for del_key in entries: - setting_name = self._get_keyname(del_key, uid, prefix='') + setting_name = self._get_keyname(del_key, uid, prefix="") entry = settings_model.get_setting_by_name(setting_name) if entry: Session().delete(entry) Session().commit() - def create_or_update_setting( - self, name, val=Optional(''), type_=Optional('unicode')): + def create_or_update_setting(self, name, val=Optional(""), type_=Optional("unicode")): if self.repo_settings: - setting = self.repo_settings.create_or_update_setting( - name, val, type_) + setting = self.repo_settings.create_or_update_setting(name, val, type_) else: - setting = self.global_settings.create_or_update_setting( - name, val, type_) + setting = self.global_settings.create_or_update_setting(name, val, type_) return setting class VcsSettingsModel(object): - - INHERIT_SETTINGS = 'inherit_vcs_settings' + INHERIT_SETTINGS = "inherit_vcs_settings" GENERAL_SETTINGS = ( - 'use_outdated_comments', - 'pr_merge_enabled', - 'auto_merge_enabled', - 'hg_use_rebase_for_merging', - 'hg_close_branch_before_merging', - 'git_use_rebase_for_merging', - 'git_close_branch_before_merging', - 'diff_cache', + "use_outdated_comments", + "pr_merge_enabled", + "auto_merge_enabled", + "hg_use_rebase_for_merging", + "hg_close_branch_before_merging", + "git_use_rebase_for_merging", + "git_close_branch_before_merging", + "diff_cache", ) HOOKS_SETTINGS = ( - ('hooks', 'changegroup.repo_size'), - ('hooks', 'changegroup.push_logger'), - ('hooks', 'outgoing.pull_logger'), + ("hooks", "changegroup.repo_size"), + ("hooks", "changegroup.push_logger"), + ("hooks", "outgoing.pull_logger"), ) HG_SETTINGS = ( - ('extensions', 'largefiles'), - ('phases', 'publish'), - ('extensions', 'evolve'), - ('extensions', 'topic'), - ('experimental', 'evolution'), - ('experimental', 'evolution.exchange'), - ) - GIT_SETTINGS = ( - ('vcs_git_lfs', 'enabled'), + ("extensions", "largefiles"), + ("phases", "publish"), + ("extensions", "evolve"), + ("extensions", "topic"), + ("experimental", "evolution"), + ("experimental", "evolution.exchange"), ) + GIT_SETTINGS = (("vcs_git_lfs", "enabled"),) GLOBAL_HG_SETTINGS = ( - ('extensions', 'largefiles'), - ('phases', 'publish'), - ('extensions', 'evolve'), - ('extensions', 'topic'), - ('experimental', 'evolution'), - ('experimental', 'evolution.exchange'), + ("extensions", "largefiles"), + ("phases", "publish"), + ("extensions", "evolve"), + ("extensions", "topic"), + ("experimental", "evolution"), + ("experimental", "evolution.exchange"), ) - GLOBAL_GIT_SETTINGS = ( - ('vcs_git_lfs', 'enabled'), - ) + GLOBAL_GIT_SETTINGS = (("vcs_git_lfs", "enabled"),) - SVN_BRANCH_SECTION = 'vcs_svn_branch' - SVN_TAG_SECTION = 'vcs_svn_tag' - PATH_SETTING = ('paths', '/') + SVN_BRANCH_SECTION = "vcs_svn_branch" + SVN_TAG_SECTION = "vcs_svn_tag" + PATH_SETTING = ("paths", "/") def __init__(self, sa=None, repo=None): self.global_settings = SettingsModel(sa=sa) self.repo_settings = SettingsModel(sa=sa, repo=repo) if repo else None - self._ui_settings = ( - self.HG_SETTINGS + self.GIT_SETTINGS + self.HOOKS_SETTINGS) + self._ui_settings = self.HG_SETTINGS + self.GIT_SETTINGS + self.HOOKS_SETTINGS self._svn_sections = (self.SVN_BRANCH_SECTION, self.SVN_TAG_SECTION) @property @@ -517,11 +501,10 @@ class VcsSettingsModel(object): @inherit_global_settings.setter @assert_repo_settings def inherit_global_settings(self, value): - self.repo_settings.create_or_update_setting( - self.INHERIT_SETTINGS, value, type_='bool') + self.repo_settings.create_or_update_setting(self.INHERIT_SETTINGS, value, type_="bool") - def get_keyname(self, key_name, prefix='rhodecode_'): - return f'{prefix}{key_name}' + def get_keyname(self, key_name, prefix="rhodecode_"): + return f"{prefix}{key_name}" def get_global_svn_branch_patterns(self): return self.global_settings.get_ui_by_section(self.SVN_BRANCH_SECTION) @@ -551,24 +534,23 @@ class VcsSettingsModel(object): return global_settings @assert_repo_settings - def create_or_update_repo_settings( - self, data, inherit_global_settings=False): + def create_or_update_repo_settings(self, data, inherit_global_settings=False): from rhodecode.model.scm import ScmModel self.inherit_global_settings = inherit_global_settings repo = self.repo_settings.get_repo() if not inherit_global_settings: - if repo.repo_type == 'svn': + if repo.repo_type == "svn": self.create_repo_svn_settings(data) else: self.create_or_update_repo_hook_settings(data) self.create_or_update_repo_pr_settings(data) - if repo.repo_type == 'hg': + if repo.repo_type == "hg": self.create_or_update_repo_hg_settings(data) - if repo.repo_type == 'git': + if repo.repo_type == "git": self.create_or_update_repo_git_settings(data) ScmModel().mark_for_invalidation(repo.repo_name, delete=True) @@ -578,17 +560,13 @@ class VcsSettingsModel(object): for section, key in self.HOOKS_SETTINGS: data_key = self._get_form_ui_key(section, key) if data_key not in data: - raise ValueError( - f'The given data does not contain {data_key} key') + raise ValueError(f"The given data does not contain {data_key} key") active = data.get(data_key) - repo_setting = self.repo_settings.get_ui_by_section_and_key( - section, key) + repo_setting = self.repo_settings.get_ui_by_section_and_key(section, key) if not repo_setting: - global_setting = self.global_settings.\ - get_ui_by_section_and_key(section, key) - self.repo_settings.create_ui_section_value( - section, global_setting.ui_value, key=key, active=active) + global_setting = self.global_settings.get_ui_by_section_and_key(section, key) + self.repo_settings.create_ui_section_value(section, global_setting.ui_value, key=key, active=active) else: repo_setting.ui_active = active Session().add(repo_setting) @@ -597,22 +575,18 @@ class VcsSettingsModel(object): for section, key in self.HOOKS_SETTINGS: data_key = self._get_form_ui_key(section, key) if data_key not in data: - raise ValueError( - f'The given data does not contain {data_key} key') + raise ValueError(f"The given data does not contain {data_key} key") active = data.get(data_key) - repo_setting = self.global_settings.get_ui_by_section_and_key( - section, key) + repo_setting = self.global_settings.get_ui_by_section_and_key(section, key) repo_setting.ui_active = active Session().add(repo_setting) @assert_repo_settings def create_or_update_repo_pr_settings(self, data): - return self._create_or_update_general_settings( - self.repo_settings, data) + return self._create_or_update_general_settings(self.repo_settings, data) def create_or_update_global_pr_settings(self, data): - return self._create_or_update_general_settings( - self.global_settings, data) + return self._create_or_update_general_settings(self.global_settings, data) @assert_repo_settings def create_repo_svn_settings(self, data): @@ -622,82 +596,54 @@ class VcsSettingsModel(object): if is_enabled: # if evolve is active set evolution=all - self._create_or_update_ui( - settings, *('experimental', 'evolution'), value='all', - active=True) - self._create_or_update_ui( - settings, *('experimental', 'evolution.exchange'), value='yes', - active=True) + self._create_or_update_ui(settings, *("experimental", "evolution"), value="all", active=True) + self._create_or_update_ui(settings, *("experimental", "evolution.exchange"), value="yes", active=True) # if evolve is active set topics server support - self._create_or_update_ui( - settings, *('extensions', 'topic'), value='', - active=True) + self._create_or_update_ui(settings, *("extensions", "topic"), value="", active=True) else: - self._create_or_update_ui( - settings, *('experimental', 'evolution'), value='', - active=False) - self._create_or_update_ui( - settings, *('experimental', 'evolution.exchange'), value='no', - active=False) - self._create_or_update_ui( - settings, *('extensions', 'topic'), value='', - active=False) + self._create_or_update_ui(settings, *("experimental", "evolution"), value="", active=False) + self._create_or_update_ui(settings, *("experimental", "evolution.exchange"), value="no", active=False) + self._create_or_update_ui(settings, *("extensions", "topic"), value="", active=False) @assert_repo_settings def create_or_update_repo_hg_settings(self, data): - largefiles, phases, evolve = \ - self.HG_SETTINGS[:3] - largefiles_key, phases_key, evolve_key = \ - self._get_settings_keys(self.HG_SETTINGS[:3], data) + largefiles, phases, evolve = self.HG_SETTINGS[:3] + largefiles_key, phases_key, evolve_key = self._get_settings_keys(self.HG_SETTINGS[:3], data) - self._create_or_update_ui( - self.repo_settings, *largefiles, value='', - active=data[largefiles_key]) - self._create_or_update_ui( - self.repo_settings, *evolve, value='', - active=data[evolve_key]) + self._create_or_update_ui(self.repo_settings, *largefiles, value="", active=data[largefiles_key]) + self._create_or_update_ui(self.repo_settings, *evolve, value="", active=data[evolve_key]) self._set_evolution(self.repo_settings, is_enabled=data[evolve_key]) - self._create_or_update_ui( - self.repo_settings, *phases, value=safe_str(data[phases_key])) + self._create_or_update_ui(self.repo_settings, *phases, value=safe_str(data[phases_key])) def create_or_update_global_hg_settings(self, data): opts_len = 3 - largefiles, phases, evolve \ - = self.GLOBAL_HG_SETTINGS[:opts_len] - largefiles_key, phases_key, evolve_key \ - = self._get_settings_keys(self.GLOBAL_HG_SETTINGS[:opts_len], data) + largefiles, phases, evolve = self.GLOBAL_HG_SETTINGS[:opts_len] + largefiles_key, phases_key, evolve_key = self._get_settings_keys(self.GLOBAL_HG_SETTINGS[:opts_len], data) - self._create_or_update_ui( - self.global_settings, *largefiles, value='', - active=data[largefiles_key]) - self._create_or_update_ui( - self.global_settings, *phases, value=safe_str(data[phases_key])) - self._create_or_update_ui( - self.global_settings, *evolve, value='', - active=data[evolve_key]) + self._create_or_update_ui(self.global_settings, *largefiles, value="", active=data[largefiles_key]) + self._create_or_update_ui(self.global_settings, *phases, value=safe_str(data[phases_key])) + self._create_or_update_ui(self.global_settings, *evolve, value="", active=data[evolve_key]) self._set_evolution(self.global_settings, is_enabled=data[evolve_key]) def create_or_update_repo_git_settings(self, data): # NOTE(marcink): # comma makes unpack work properly - lfs_enabled, \ - = self.GIT_SETTINGS + (lfs_enabled,) = self.GIT_SETTINGS - lfs_enabled_key, \ - = self._get_settings_keys(self.GIT_SETTINGS, data) + (lfs_enabled_key,) = self._get_settings_keys(self.GIT_SETTINGS, data) self._create_or_update_ui( - self.repo_settings, *lfs_enabled, value=data[lfs_enabled_key], - active=data[lfs_enabled_key]) + self.repo_settings, *lfs_enabled, value=data[lfs_enabled_key], active=data[lfs_enabled_key] + ) def create_or_update_global_git_settings(self, data): lfs_enabled = self.GLOBAL_GIT_SETTINGS[0] lfs_enabled_key = self._get_settings_keys(self.GLOBAL_GIT_SETTINGS, data)[0] self._create_or_update_ui( - self.global_settings, *lfs_enabled, value=data[lfs_enabled_key], - active=data[lfs_enabled_key]) + self.global_settings, *lfs_enabled, value=data[lfs_enabled_key], active=data[lfs_enabled_key] + ) def create_or_update_global_svn_settings(self, data): # branch/tags patterns @@ -722,16 +668,12 @@ class VcsSettingsModel(object): repo_uis = self.repo_settings.get_ui(section, key) filtered_repo_uis = self._filter_ui_settings(repo_uis) - filtered_repo_uis_keys = [ - (s.section, s.key) for s in filtered_repo_uis] + filtered_repo_uis_keys = [(s.section, s.key) for s in filtered_repo_uis] def _is_global_ui_filtered(ui): - return ( - (ui.section, ui.key) in filtered_repo_uis_keys - or ui.section in self._svn_sections) + return (ui.section, ui.key) in filtered_repo_uis_keys or ui.section in self._svn_sections - filtered_global_uis = [ - ui for ui in global_uis if not _is_global_ui_filtered(ui)] + filtered_global_uis = [ui for ui in global_uis if not _is_global_ui_filtered(ui)] return filtered_global_uis + filtered_repo_uis @@ -778,21 +720,16 @@ class VcsSettingsModel(object): return self.get_repo_general_settings() def _filter_ui_settings(self, settings): - filtered_settings = [ - s for s in settings if self._should_keep_setting(s)] + filtered_settings = [s for s in settings if self._should_keep_setting(s)] return filtered_settings def _should_keep_setting(self, setting): - keep = ( - (setting.section, setting.key) in self._ui_settings or - setting.section in self._svn_sections) + keep = (setting.section, setting.key) in self._ui_settings or setting.section in self._svn_sections return keep def _filter_general_settings(self, settings): keys = [self.get_keyname(key) for key in self.GENERAL_SETTINGS] - return { - k: settings[k] - for k in settings if k in keys} + return {k: settings[k] for k in settings if k in keys} def _collect_all_settings(self, global_=False): settings = self.global_settings if global_ else self.repo_settings @@ -803,9 +740,9 @@ class VcsSettingsModel(object): result_key = self._get_form_ui_key(section, key) if ui: - if section in ('hooks', 'extensions'): + if section in ("hooks", "extensions"): result[result_key] = ui.ui_active - elif result_key in ['vcs_git_lfs_enabled']: + elif result_key in ["vcs_git_lfs_enabled"]: result[result_key] = ui.ui_active else: result[result_key] = ui.ui_value @@ -819,16 +756,13 @@ class VcsSettingsModel(object): return result def _get_form_ui_key(self, section, key): - return '{section}_{key}'.format( - section=section, key=key.replace('.', '_')) + return "{section}_{key}".format(section=section, key=key.replace(".", "_")) - def _create_or_update_ui( - self, settings, section, key, value=None, active=None): + def _create_or_update_ui(self, settings, section, key, value=None, active=None): ui = settings.get_ui_by_section_and_key(section, key) if not ui: active = True if active is None else active - settings.create_ui_section_value( - section, value, key=key, active=active) + settings.create_ui_section_value(section, value, key=key, active=active) else: if active is not None: ui.ui_active = active @@ -837,10 +771,7 @@ class VcsSettingsModel(object): Session().add(ui) def _create_svn_settings(self, settings, data): - svn_settings = { - 'new_svn_branch': self.SVN_BRANCH_SECTION, - 'new_svn_tag': self.SVN_TAG_SECTION - } + svn_settings = {"new_svn_branch": self.SVN_BRANCH_SECTION, "new_svn_tag": self.SVN_TAG_SECTION} for key in svn_settings: if data.get(key): settings.create_ui_section_value(svn_settings[key], data[key]) @@ -849,16 +780,13 @@ class VcsSettingsModel(object): for name in self.GENERAL_SETTINGS: data_key = self.get_keyname(name) if data_key not in data: - raise ValueError( - f'The given data does not contain {data_key} key') - setting = settings.create_or_update_setting( - name, data[data_key], 'bool') + raise ValueError(f"The given data does not contain {data_key} key") + setting = settings.create_or_update_setting(name, data[data_key], "bool") Session().add(setting) def _get_settings_keys(self, settings, data): data_keys = [self._get_form_ui_key(*s) for s in settings] for data_key in data_keys: if data_key not in data: - raise ValueError( - f'The given data does not contain {data_key} key') + raise ValueError(f"The given data does not contain {data_key} key") return data_keys diff --git a/rhodecode/tests/fixtures/rc_fixture.py b/rhodecode/tests/fixtures/rc_fixture.py index a7838865..9488d086 100644 --- a/rhodecode/tests/fixtures/rc_fixture.py +++ b/rhodecode/tests/fixtures/rc_fixture.py @@ -420,3 +420,31 @@ class Fixture(object): source = source.strip() return source + + +class RhodeCodeSettingFixture: + def __init__(self, name, value): + self.name = name + self.value = value + self.old_setting = None + + def __enter__(self): + from rhodecode.model.settings import SettingsModel + + model = SettingsModel() + self.old_setting = model.get_setting_by_name(self.name) + setting = model.create_or_update_setting(name=self.name, val=self.value) + Session().add(setting) + Session().commit() + SettingsModel().invalidate_settings_cache() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + model = SettingsModel() + if self.old_setting: + setting = model.create_or_update_setting(name=self.name, val=self.old_setting.app_settings_value) + else: + setting = model.create_or_update_setting(name=self.name) + Session().add(setting) + Session().commit() + SettingsModel().invalidate_settings_cache()