engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests: fixed some flaky tests

Browse source
This commit is contained in:
RhodeCode Admin 2025-01-10 10:42:53 +01:00
parent 159899a475
commit e9ad755887
4 changed files with 393 additions and 500 deletions
Download patch file Download diff file Expand all files Collapse all files

115
rhodecode/apps/login/tests/test_register_captcha.py
View file

@ -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")

364
rhodecode/apps/login/views.py
View file

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

386
rhodecode/model/settings.py
View file

@ -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

28
rhodecode/tests/fixtures/rc_fixture.py vendored
View file

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