core: avoid using rhodecode.test packages inside main packages as tests are removed during build which can cause some problems in some edge case calls

rhodecode/bootstrap.py

@@ -0,0 +1,41 b''
+# Copyright (C) 2010-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+# bootstrap data available for tests and setup clean install
+
+TEST_USER_ADMIN_LOGIN = 'test_admin'
+TEST_USER_ADMIN_PASS = 'test12'
+TEST_USER_ADMIN_EMAIL = '[e-mail]'
+
+TEST_USER_REGULAR_LOGIN = 'test_regular'
+TEST_USER_REGULAR_PASS = 'test12'
+TEST_USER_REGULAR_EMAIL = '[e-mail]'
+
+TEST_USER_REGULAR2_LOGIN = 'test_regular2'
+TEST_USER_REGULAR2_PASS = 'test12'
+TEST_USER_REGULAR2_EMAIL = '[e-mail]'
+
+HG_REPO = 'vcs_test_hg'
+GIT_REPO = 'vcs_test_git'
+SVN_REPO = 'vcs_test_svn'
+
+NEW_HG_REPO = 'vcs_test_hg_new'
+NEW_GIT_REPO = 'vcs_test_git_new'
+
+HG_FORK = 'vcs_test_hg_fork'
+GIT_FORK = 'vcs_test_git_fork'

rhodecode/api/tests/conftest.py

@@ -1,50 +1,50 b''
 # Copyright (C) 2010-2024 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.user import UserModel
 from rhodecode.model.auth_token import AuthTokenModel
-from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
 
 @pytest.fixture(scope="class")
 def testuser_api(request, baseapp):
     cls = request.cls
+    from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
     # ADMIN USER
     cls.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
     cls.apikey = cls.usr.api_key
 
     # REGULAR USER
     cls.test_user = UserModel().create_or_update(
         username='test-api',
         password='test',
         email='[e-mail]',
         firstname='first',
         lastname='last'
     )
     # create TOKEN for user, if he doesn't have one
     if not cls.test_user.api_key:
         AuthTokenModel().create(
             user=cls.test_user, description='TEST_USER_TOKEN')
 
     Session().commit()
     cls.TEST_USER_LOGIN = cls.test_user.username
     cls.apikey_regular = cls.test_user.api_key

rhodecode/api/views/server_api.py

@@ -1,423 +1,479 b''
 # Copyright (C) 2011-2024 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import itertools
 import base64
 
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, find_methods)
 
 from rhodecode.api.utils import (
     Optional, OAttr, has_superadmin_permission, get_user_or_error)
 from rhodecode.lib.utils import repo2db_mapper, get_rhodecode_repo_store_path
 from rhodecode.lib import system_info
 from rhodecode.lib import user_sessions
 from rhodecode.lib import exc_tracking
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils2 import safe_int
 from rhodecode.model.db import UserIpMap
 from rhodecode.model.scm import ScmModel
 from rhodecode.apps.file_store import utils as store_utils
 from rhodecode.apps.file_store.exceptions import FileNotAllowedException, \
     FileOverSizeException
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_server_info(request, apiuser):
     """
     Returns the |RCE| server information.
 
     This includes the running version of |RCE| and all installed
     packages. This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'modules': [<module name>,...]
         'py_version': <python version>,
         'platform': <platform type>,
         'rhodecode_version': <rhodecode version>
       }
       error :  null
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     server_info = ScmModel().get_server_info(request.environ)
     # rhodecode-index requires those
 
     server_info['index_storage'] = server_info['search']['value']['location']
     server_info['storage'] = server_info['storage']['value']['path']
 
     return server_info
 
 
 @jsonrpc_method()
 def get_repo_store(request, apiuser):
     """
     Returns the |RCE| repository storage information.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'modules': [<module name>,...]
         'py_version': <python version>,
         'platform': <platform type>,
         'rhodecode_version': <rhodecode version>
       }
       error :  null
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     path = get_rhodecode_repo_store_path()
     return {"path": path}
 
 
 @jsonrpc_method()
 def get_ip(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Displays the IP Address as seen from the |RCE| server.
 
     * This command displays the IP Address, as well as all the defined IP
       addresses for the specified user. If the ``userid`` is not set, the
       data returned is for the user calling the method.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from |authtoken|.
     :type apiuser: AuthUser
     :param userid: Sets the userid for which associated IP Address data
         is returned.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
                      "server_ip_addr": "<ip_from_clien>",
                      "user_ips": [
                                     {
                                        "ip_addr": "<ip_with_mask>",
                                        "ip_range": ["<start_ip>", "<end_ip>"],
                                     },
                                     ...
                                  ]
         }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     userid = Optional.extract(userid, evaluate_locals=locals())
     userid = getattr(userid, 'user_id', userid)
 
     user = get_user_or_error(userid)
     ips = UserIpMap.query().filter(UserIpMap.user == user).all()
     return {
         'server_ip_addr': request.rpc_ip_addr,
         'user_ips': ips
     }
 
 
 @jsonrpc_method()
 def rescan_repos(request, apiuser, remove_obsolete=Optional(False)):
     """
     Triggers a rescan of the specified repositories.
 
     * If the ``remove_obsolete`` option is set, it also deletes repositories
       that are found in the database but not on the file system, so called
       "clean zombies".
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param remove_obsolete: Deletes repositories from the database that
         are not found on the filesystem.
     :type remove_obsolete: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'added': [<added repository name>,...]
         'removed': [<removed repository name>,...]
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred during rescan repositories action'
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     try:
         rm_obsolete = Optional.extract(remove_obsolete)
         added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                         remove_obsolete=rm_obsolete, force_hooks_rebuild=True)
         return {'added': added, 'removed': removed}
     except Exception:
         log.exception('Failed to run repo rescann')
         raise JSONRPCError(
             'Error occurred during rescan repositories action'
         )
 
+@jsonrpc_method()
+def cleanup_repos(request, apiuser, remove_obsolete=Optional(False)):
+    """
+    Triggers a rescan of the specified repositories.
+
+    * If the ``remove_obsolete`` option is set, it also deletes repositories
+      that are found in the database but not on the file system, so called
+      "clean zombies".
+
+    This command can only be run using an |authtoken| with admin rights to
+    the specified repository.
+
+    This command takes the following options:
+
+    :param apiuser: This is filled automatically from the |authtoken|.
+    :type apiuser: AuthUser
+    :param remove_obsolete: Deletes repositories from the database that
+        are not found on the filesystem.
+    :type remove_obsolete: Optional(``True`` | ``False``)
+
+    Example output:
+
+    .. code-block:: bash
+
+      id : <id_given_in_input>
+      result : {
+        'added': [<added repository name>,...]
+        'removed': [<removed repository name>,...]
+      }
+      error :  null
+
+    Example error output:
+
+    .. code-block:: bash
+
+      id : <id_given_in_input>
+      result : null
+      error :  {
+        'Error occurred during rescan repositories action'
+      }
+
+    """
+    if not has_superadmin_permission(apiuser):
+        raise JSONRPCForbidden()
+
+    try:
+        rm_obsolete = Optional.extract(remove_obsolete)
+        added, removed = repo2db_mapper(ScmModel().repo_scan(),
+                                        remove_obsolete=rm_obsolete, force_hooks_rebuild=True)
+        return {'added': added, 'removed': removed}
+    except Exception:
+        log.exception('Failed to run repo rescann')
+        raise JSONRPCError(
+            'Error occurred during rescan repositories action'
+        )
+
 
 @jsonrpc_method()
 def cleanup_sessions(request, apiuser, older_then=Optional(60)):
     """
     Triggers a session cleanup action.
 
     If the ``older_then`` option is set, only sessions that hasn't been
     accessed in the given number of days will be removed.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param older_then: Deletes session that hasn't been accessed
         in given number of days.
     :type older_then: Optional(int)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result: {
         "backend": "<type of backend>",
         "sessions_removed": <number_of_removed_sessions>
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred during session cleanup'
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     older_then = safe_int(Optional.extract(older_then)) or 60
     older_than_seconds = 60 * 60 * 24 * older_then
 
     config = system_info.rhodecode_config().get_value()['value']['config']
     session_model = user_sessions.get_session_handler(
         config.get('beaker.session.type', 'memory'))(config)
 
     backend = session_model.SESSION_TYPE
     try:
         cleaned = session_model.clean_sessions(
             older_than_seconds=older_than_seconds)
         return {'sessions_removed': cleaned, 'backend': backend}
     except user_sessions.CleanupCommand as msg:
         return {'cleanup_command': str(msg), 'backend': backend}
     except Exception as e:
         log.exception('Failed session cleanup')
         raise JSONRPCError(
             'Error occurred during session cleanup'
         )
 
 
 @jsonrpc_method()
 def get_method(request, apiuser, pattern=Optional('*')):
     """
     Returns list of all available API methods. By default match pattern
     os "*" but any other pattern can be specified. eg *comment* will return
     all methods with comment inside them. If just single method is matched
     returned data will also include method specification
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param pattern: pattern to match method names against
     :type pattern: Optional("*")
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": [
         "changeset_comment",
         "comment_pull_request",
         "comment_commit"
       ]
       error :  null
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": [
         "comment_commit",
         {
           "apiuser": "<RequiredType>",
           "comment_type": "<Optional:u'note'>",
           "commit_id": "<RequiredType>",
           "message": "<RequiredType>",
           "repoid": "<RequiredType>",
           "request": "<RequiredType>",
           "resolves_comment_id": "<Optional:None>",
           "status": "<Optional:None>",
           "userid": "<Optional:<OptionalAttr:apiuser>>"
         }
       ]
       error :  null
     """
     from rhodecode.config import patches
     inspect = patches.inspect_getargspec()
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     pattern = Optional.extract(pattern)
 
     matches = find_methods(request.registry.jsonrpc_methods, pattern)
 
     args_desc = []
     matches_keys = list(matches.keys())
     if len(matches_keys) == 1:
         func = matches[matches_keys[0]]
 
         argspec = inspect.getargspec(func)
         arglist = argspec[0]
         defaults = list(map(repr, argspec[3] or []))
 
         default_empty = '<RequiredType>'
 
         # kw arguments required by this method
         func_kwargs = dict(itertools.zip_longest(
             reversed(arglist), reversed(defaults), fillvalue=default_empty))
         args_desc.append(func_kwargs)
 
     return matches_keys + args_desc
 
 
 @jsonrpc_method()
 def store_exception(request, apiuser, exc_data_json, prefix=Optional('rhodecode')):
     """
     Stores sent exception inside the built-in exception tracker in |RCE| server.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     :param exc_data_json: JSON data with exception e.g
         {"exc_traceback": "Value `1` is not allowed", "exc_type_name": "ValueError"}
     :type exc_data_json: JSON data
 
     :param prefix: prefix for error type, e.g 'rhodecode', 'vcsserver', 'rhodecode-tools'
     :type prefix: Optional("rhodecode")
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": {
         "exc_id": 139718459226384,
         "exc_url": "http://localhost:8080/_admin/settings/exceptions/139718459226384"
       }
       error :  null
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     prefix = Optional.extract(prefix)
     exc_id = exc_tracking.generate_id()
 
     try:
         exc_data = json.loads(exc_data_json)
     except Exception:
         log.error('Failed to parse JSON: %r', exc_data_json)
         raise JSONRPCError('Failed to parse JSON data from exc_data_json field. '
                            'Please make sure it contains a valid JSON.')
 
     try:
         exc_traceback = exc_data['exc_traceback']
         exc_type_name = exc_data['exc_type_name']
         exc_value = ''
     except KeyError as err:
         raise JSONRPCError(
             f'Missing exc_traceback, or exc_type_name '
             f'in exc_data_json field. Missing: {err}')
 
     class ExcType:
         __name__ = exc_type_name
 
     exc_info = (ExcType(), exc_value, exc_traceback)
 
     exc_tracking._store_exception(
         exc_id=exc_id, exc_info=exc_info, prefix=prefix)
 
     exc_url = request.route_url(
         'admin_settings_exception_tracker_show', exception_id=exc_id)
     return {'exc_id': exc_id, 'exc_url': exc_url}

rhodecode/lib/db_manage.py

@@ -1,658 +1,658 b''
 # Copyright (C) 2010-2024 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Database creation, and setup module for RhodeCode Enterprise. Used for creation
 of database as well as for migration operations
 """
 
 import os
 import sys
 import time
 import uuid
 import logging
 import getpass
 from os.path import dirname as dn, join as jn
 
 from sqlalchemy.engine import create_engine
 
 from rhodecode import __dbversion__
 from rhodecode.model import init_model
 from rhodecode.model.user import UserModel
 from rhodecode.model.db import (
     User, Permission, RhodeCodeUi, RhodeCodeSetting, UserToPerm,
     DbMigrateVersion, RepoGroup, UserRepoGroupToPerm, CacheKey, Repository)
 from rhodecode.model.meta import Session, Base
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import SettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 def notify(msg):
     """
     Notification for migrations messages
     """
     ml = len(msg) + (4 * 2)
     print((('\n%s\n*** %s ***\n%s' % ('*' * ml, msg, '*' * ml)).upper()))
 
 
 class DbManage(object):
 
     def __init__(self, log_sql, dbconf, root, tests=False,
                  SESSION=None, cli_args=None, enc_key=b''):
 
         self.dbname = dbconf.split('/')[-1]
         self.tests = tests
         self.root = root
         self.dburi = dbconf
         self.log_sql = log_sql
         self.cli_args = cli_args or {}
         self.sa = None
         self.engine = None
         self.enc_key = enc_key
         # sets .sa .engine
         self.init_db(SESSION=SESSION)
 
         self.ask_ok = self.get_ask_ok_func(self.cli_args.get('force_ask'))
 
     def db_exists(self):
         if not self.sa:
             self.init_db()
         try:
             self.sa.query(RhodeCodeUi)\
                 .filter(RhodeCodeUi.ui_key == '/')\
                 .scalar()
             return True
         except Exception:
             return False
         finally:
             self.sa.rollback()
 
     def get_ask_ok_func(self, param):
         if param not in [None]:
             # return a function lambda that has a default set to param
             return lambda *args, **kwargs: param
         else:
             from rhodecode.lib.utils import ask_ok
             return ask_ok
 
     def init_db(self, SESSION=None):
 
         if SESSION:
             self.sa = SESSION
             self.engine = SESSION.bind
         else:
             # init new sessions
             engine = create_engine(self.dburi, echo=self.log_sql)
             init_model(engine, encryption_key=self.enc_key)
             self.sa = Session()
             self.engine = engine
 
     def create_tables(self, override=False):
         """
         Create a auth database
         """
 
         log.info("Existing database with the same name is going to be destroyed.")
         log.info("Setup command will run DROP ALL command on that database.")
         engine = self.engine
 
         if self.tests:
             destroy = True
         else:
             destroy = self.ask_ok('Are you sure that you want to destroy the old database? [y/n]')
         if not destroy:
             log.info('db tables bootstrap: Nothing done.')
             sys.exit(0)
         if destroy:
             Base.metadata.drop_all(bind=engine)
 
         checkfirst = not override
         Base.metadata.create_all(bind=engine, checkfirst=checkfirst)
         log.info('Created tables for %s', self.dbname)
 
     def set_db_version(self):
         ver = DbMigrateVersion()
         ver.version = __dbversion__
         ver.repository_id = 'rhodecode_db_migrations'
         ver.repository_path = 'versions'
         self.sa.add(ver)
         log.info('db version set to: %s', __dbversion__)
 
     def run_post_migration_tasks(self):
         """
         Run various tasks before actually doing migrations
         """
         # delete cache keys on each upgrade
         total = CacheKey.query().count()
         log.info("Deleting (%s) cache keys now...", total)
         CacheKey.delete_all_cache()
 
     def upgrade(self, version=None):
         """
         Upgrades given database schema to given revision following
         all needed steps, to perform the upgrade
 
         """
 
         from rhodecode.lib.dbmigrate.migrate.versioning import api
         from rhodecode.lib.dbmigrate.migrate.exceptions import DatabaseNotControlledError
 
         if 'sqlite' in self.dburi:
             print(
                '********************** WARNING **********************\n'
                'Make sure your version of sqlite is at least 3.7.X.  \n'
                'Earlier versions are known to fail on some migrations\n'
                '*****************************************************\n')
 
         upgrade = self.ask_ok(
             'You are about to perform a database upgrade. Make '
             'sure you have backed up your database. '
             'Continue ? [y/n]')
         if not upgrade:
             log.info('No upgrade performed')
             sys.exit(0)
 
         repository_path = jn(dn(dn(dn(os.path.realpath(__file__)))),
                              'rhodecode/lib/dbmigrate')
         db_uri = self.dburi
 
         if version:
             DbMigrateVersion.set_version(version)
 
         try:
             curr_version = api.db_version(db_uri, repository_path)
             msg = (f'Found current database db_uri under version '
                    f'control with version {curr_version}')
 
         except (RuntimeError, DatabaseNotControlledError):
             curr_version = 1
             msg = f'Current database is not under version control. ' \
                   f'Setting as version {curr_version}'
             api.version_control(db_uri, repository_path, curr_version)
 
         notify(msg)
 
         if curr_version == __dbversion__:
             log.info('This database is already at the newest version')
             sys.exit(0)
 
         upgrade_steps = list(range(curr_version + 1, __dbversion__ + 1))
         notify(f'attempting to upgrade database from '
                f'version {curr_version} to version {__dbversion__}')
 
         # CALL THE PROPER ORDER OF STEPS TO PERFORM FULL UPGRADE
         final_step = 'latest'
         for step in upgrade_steps:
             notify(f'performing upgrade step {step}')
             time.sleep(0.5)
 
             api.upgrade(db_uri, repository_path, step)
             self.sa.rollback()
             notify(f'schema upgrade for step {step} completed')
 
             final_step = step
 
         self.run_post_migration_tasks()
         notify(f'upgrade to version {final_step} successful')
 
     def fix_repo_paths(self):
         """
         Fixes an old RhodeCode version path into new one without a '*'
         """
 
         paths = self.sa.query(RhodeCodeUi)\
                 .filter(RhodeCodeUi.ui_key == '/')\
                 .scalar()
 
         paths.ui_value = paths.ui_value.replace('*', '')
 
         try:
             self.sa.add(paths)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
 
     def fix_default_user(self):
         """
         Fixes an old default user with some 'nicer' default values,
         used mostly for anonymous access
         """
         def_user = self.sa.query(User)\
             .filter(User.username == User.DEFAULT_USER)\
             .one()
 
         def_user.name = 'Anonymous'
         def_user.lastname = 'User'
         def_user.email = User.DEFAULT_USER_EMAIL
 
         try:
             self.sa.add(def_user)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
 
     def fix_settings(self):
         """
         Fixes rhodecode settings and adds ga_code key for google analytics
         """
 
         hgsettings3 = RhodeCodeSetting('ga_code', '')
 
         try:
             self.sa.add(hgsettings3)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
 
     def create_admin_and_prompt(self):
 
         # defaults
         defaults = self.cli_args
         username = defaults.get('username')
         password = defaults.get('password')
         email = defaults.get('email')
 
         if username is None:
             username = input('Specify admin username:')
         if password is None:
             password = self._get_admin_password()
             if not password:
                 # second try
                 password = self._get_admin_password()
                 if not password:
                     sys.exit()
         if email is None:
             email = input('Specify admin email:')
         api_key = self.cli_args.get('api_key')
         self.create_user(username, password, email, True,
                          strict_creation_check=False,
                          api_key=api_key)
 
     def _get_admin_password(self):
         password = getpass.getpass('Specify admin password '
                                    '(min 6 chars):')
         confirm = getpass.getpass('Confirm password:')
 
         if password != confirm:
             log.error('passwords mismatch')
             return False
         if len(password) < 6:
             log.error('password is too short - use at least 6 characters')
             return False
 
         return password
 
     def create_test_admin_and_users(self):
         log.info('creating admin and regular test users')
-        from rhodecode.tests import TEST_USER_ADMIN_LOGIN, \
+        from rhodecode.bootstrap import TEST_USER_ADMIN_LOGIN, \
             TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
             TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
             TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
             TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
 
         self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                          TEST_USER_ADMIN_EMAIL, True, api_key=True)
 
         self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                          TEST_USER_REGULAR_EMAIL, False, api_key=True)
 
         self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                          TEST_USER_REGULAR2_EMAIL, False, api_key=True)
 
     def create_ui_settings(self, repo_store_path):
         """
         Creates ui settings, fills out hooks
         and disables dotencode
         """
         settings_model = SettingsModel(sa=self.sa)
 
         # Build HOOKS
         hooks = [
             (RhodeCodeUi.HOOK_REPO_SIZE, 'python:vcsserver.hooks.repo_size'),
 
             # HG
             (RhodeCodeUi.HOOK_PRE_PULL, 'python:vcsserver.hooks.pre_pull'),
             (RhodeCodeUi.HOOK_PULL, 'python:vcsserver.hooks.log_pull_action'),
             (RhodeCodeUi.HOOK_PRE_PUSH, 'python:vcsserver.hooks.pre_push'),
             (RhodeCodeUi.HOOK_PRETX_PUSH, 'python:vcsserver.hooks.pre_push'),
             (RhodeCodeUi.HOOK_PUSH, 'python:vcsserver.hooks.log_push_action'),
             (RhodeCodeUi.HOOK_PUSH_KEY, 'python:vcsserver.hooks.key_push'),
 
         ]
 
         for key, value in hooks:
             hook_obj = settings_model.get_ui_by_key(key)
             hooks2 = hook_obj if hook_obj else RhodeCodeUi()
             hooks2.ui_section = 'hooks'
             hooks2.ui_key = key
             hooks2.ui_value = value
             self.sa.add(hooks2)
 
         # enable largefiles
         largefiles = RhodeCodeUi()
         largefiles.ui_section = 'extensions'
         largefiles.ui_key = 'largefiles'
         largefiles.ui_value = ''
         self.sa.add(largefiles)
 
         # enable hgevolve disabled by default
         hgevolve = RhodeCodeUi()
         hgevolve.ui_section = 'extensions'
         hgevolve.ui_key = 'evolve'
         hgevolve.ui_value = ''
         hgevolve.ui_active = False
         self.sa.add(hgevolve)
 
         hgevolve = RhodeCodeUi()
         hgevolve.ui_section = 'experimental'
         hgevolve.ui_key = 'evolution'
         hgevolve.ui_value = ''
         hgevolve.ui_active = False
         self.sa.add(hgevolve)
 
         hgevolve = RhodeCodeUi()
         hgevolve.ui_section = 'experimental'
         hgevolve.ui_key = 'evolution.exchange'
         hgevolve.ui_value = ''
         hgevolve.ui_active = False
         self.sa.add(hgevolve)
 
         hgevolve = RhodeCodeUi()
         hgevolve.ui_section = 'extensions'
         hgevolve.ui_key = 'topic'
         hgevolve.ui_value = ''
         hgevolve.ui_active = False
         self.sa.add(hgevolve)
 
         # enable hggit disabled by default
         hggit = RhodeCodeUi()
         hggit.ui_section = 'extensions'
         hggit.ui_key = 'hggit'
         hggit.ui_value = ''
         hggit.ui_active = False
         self.sa.add(hggit)
 
         # set svn branch defaults
         branches = ["/branches/*", "/trunk"]
         tags = ["/tags/*"]
 
         for branch in branches:
             settings_model.create_ui_section_value(
                 RhodeCodeUi.SVN_BRANCH_ID, branch)
 
         for tag in tags:
             settings_model.create_ui_section_value(RhodeCodeUi.SVN_TAG_ID, tag)
 
     def create_auth_plugin_options(self, skip_existing=False):
         """
         Create default auth plugin settings, and make it active
 
         :param skip_existing:
         """
         defaults = [
             ('auth_plugins',
              'egg:rhodecode-enterprise-ce#token,egg:rhodecode-enterprise-ce#rhodecode',
              'list'),
 
             ('auth_authtoken_enabled',
              'True',
              'bool'),
 
             ('auth_rhodecode_enabled',
              'True',
              'bool'),
         ]
         for k, v, t in defaults:
             if (skip_existing and
                     SettingsModel().get_setting_by_name(k) is not None):
                 log.debug('Skipping option %s', k)
                 continue
             setting = RhodeCodeSetting(k, v, t)
             self.sa.add(setting)
 
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
 
         for k, v, t in [
             ('default_repo_enable_locking',  False, 'bool'),
             ('default_repo_enable_downloads', False, 'bool'),
             ('default_repo_enable_statistics', False, 'bool'),
             ('default_repo_private', False, 'bool'),
             ('default_repo_type', 'hg', 'unicode')]:
 
             if (skip_existing and
                     SettingsModel().get_setting_by_name(k) is not None):
                 log.debug('Skipping option %s', k)
                 continue
             setting = RhodeCodeSetting(k, v, t)
             self.sa.add(setting)
 
     def fixup_groups(self):
         def_usr = User.get_default_user()
         for g in RepoGroup.query().all():
             g.group_name = g.get_new_name(g.name)
             self.sa.add(g)
             # get default perm
             default = UserRepoGroupToPerm.query()\
                 .filter(UserRepoGroupToPerm.group == g)\
                 .filter(UserRepoGroupToPerm.user == def_usr)\
                 .scalar()
 
             if default is None:
                 log.debug('missing default permission for group %s adding', g)
                 perm_obj = RepoGroupModel()._create_default_perms(g)
                 self.sa.add(perm_obj)
 
     def reset_permissions(self, username):
         """
         Resets permissions to default state, useful when old systems had
         bad permissions, we must clean them up
 
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
 
         u2p = UserToPerm.query()\
             .filter(UserToPerm.user == default_user).all()
         fixed = False
         if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):
             for p in u2p:
                 Session().delete(p)
             fixed = True
             self.populate_default_permissions()
         return fixed
 
     def config_prompt(self, test_repo_path='', retries=3):
         defaults = self.cli_args
         _path = defaults.get('repos_location')
         if retries == 3:
             log.info('Setting up repositories config')
 
         if _path is not None:
             path = _path
         elif not self.tests and not test_repo_path:
             path = input(
                  'Enter a valid absolute path to store repositories. '
                  'All repositories in that path will be added automatically:'
             )
         else:
             path = test_repo_path
         path_ok = True
 
         # check proper dir
         if not os.path.isdir(path):
             path_ok = False
             log.error('Given path %s is not a valid directory', path)
 
         elif not os.path.isabs(path):
             path_ok = False
             log.error('Given path %s is not an absolute path', path)
 
         # check if path is at least readable.
         if not os.access(path, os.R_OK):
             path_ok = False
             log.error('Given path %s is not readable', path)
 
         # check write access, warn user about non writeable paths
         elif not os.access(path, os.W_OK) and path_ok:
             log.warning('No write permission to given path %s', path)
 
             q = (f'Given path {path} is not writeable, do you want to '
                  f'continue with read only mode ? [y/n]')
             if not self.ask_ok(q):
                 log.error('Canceled by user')
                 sys.exit(-1)
 
         if retries == 0:
             sys.exit('max retries reached')
         if not path_ok:
             retries -= 1
             return self.config_prompt(test_repo_path, retries)
 
         real_path = os.path.normpath(os.path.realpath(path))
 
         if real_path != os.path.normpath(path):
             q = (f'Path looks like a symlink, RhodeCode Enterprise will store '
                  f'given path as {real_path} ? [y/n]')
             if not self.ask_ok(q):
                 log.error('Canceled by user')
                 sys.exit(-1)
 
         return real_path
 
     def create_settings(self, path):
 
         self.create_ui_settings(path)
 
         ui_config = [
             ('web', 'allow_archive', 'gz zip bz2'),
             ('web', 'allow_push', '*'),
             ('web', 'baseurl', '/'),
             ('paths', '/', path),
             ('phases', 'publish', 'True')
         ]
         for section, key, value in ui_config:
             ui_conf = RhodeCodeUi()
             setattr(ui_conf, 'ui_section', section)
             setattr(ui_conf, 'ui_key', key)
             setattr(ui_conf, 'ui_value', value)
             self.sa.add(ui_conf)
 
         # rhodecode app settings
         settings = [
             ('realm', 'RhodeCode', 'unicode'),
             ('title', '', 'unicode'),
             ('pre_code', '', 'unicode'),
             ('post_code', '', 'unicode'),
 
             # Visual
             ('show_public_icon', True, 'bool'),
             ('show_private_icon', True, 'bool'),
             ('stylify_metatags', True, 'bool'),
             ('dashboard_items', 100, 'int'),
             ('admin_grid_items', 25, 'int'),
 
             ('markup_renderer', 'markdown', 'unicode'),
 
             ('repository_fields', True, 'bool'),
             ('show_version', True, 'bool'),
             ('show_revision_number', True, 'bool'),
             ('show_sha_length', 12, 'int'),
 
             ('use_gravatar', False, 'bool'),
             ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'),
 
             ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'),
             ('clone_uri_id_tmpl', Repository.DEFAULT_CLONE_URI_ID, 'unicode'),
             ('clone_uri_ssh_tmpl', Repository.DEFAULT_CLONE_URI_SSH, 'unicode'),
             ('support_url', '', 'unicode'),
             ('update_url', RhodeCodeSetting.DEFAULT_UPDATE_URL, 'unicode'),
 
             # VCS Settings
             ('pr_merge_enabled', True, 'bool'),
             ('use_outdated_comments', True, 'bool'),
             ('diff_cache', True, 'bool'),
         ]
 
         for key, val, type_ in settings:
             sett = RhodeCodeSetting(key, val, type_)
             self.sa.add(sett)
 
         self.create_auth_plugin_options()
         self.create_default_options()
 
         log.info('created ui config')
 
     def create_user(self, username, password, email='', admin=False,
                     strict_creation_check=True, api_key=None):
         log.info('creating user `%s`', username)
         user = UserModel().create_or_update(
             username, password, email, firstname='RhodeCode', lastname='Admin',
             active=True, admin=admin, extern_type="rhodecode",
             strict_creation_check=strict_creation_check)
 
         if api_key:
             log.info('setting a new default auth token for user `%s`', username)
             UserModel().add_auth_token(
                 user=user, lifetime_minutes=-1,
                 role=UserModel.auth_token_role.ROLE_ALL,
                 description='BUILTIN TOKEN')
 
     def create_default_user(self):
         log.info('creating default user')
         # create default user for handling default permissions.
         user = UserModel().create_or_update(username=User.DEFAULT_USER,
                                             password=str(uuid.uuid1())[:20],
                                             email=User.DEFAULT_USER_EMAIL,
                                             firstname='Anonymous',
                                             lastname='User',
                                             strict_creation_check=False)
         # based on configuration options activate/de-activate this user which
         # controls anonymous access
         if self.cli_args.get('public_access') is False:
             log.info('Public access disabled')
             user.active = False
             Session().add(user)
             Session().commit()
 
     def create_permissions(self):
         """
         Creates all permissions defined in the system
         """
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         log.info('creating permissions')
         PermissionModel(self.sa).create_permissions()
 
     def populate_default_permissions(self):
         """
         Populate default permissions. It will create only the default
         permissions that are missing, and not alter already defined ones
         """
         log.info('creating default user permissions')
         PermissionModel(self.sa).create_default_user_permissions(user=User.DEFAULT_USER)

rhodecode/tests/__init__.py

@@ -1,253 +1,224 b''
 # Copyright (C) 2010-2024 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import time
 import logging
 import datetime
 import tempfile
 from os.path import join as jn
 import urllib.parse
 
 import pytest
 
 import rhodecode
 from rhodecode.model.db import User
 from rhodecode.lib import auth
 from rhodecode.lib import helpers as h
 from rhodecode.lib.helpers import flash
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.hash_utils import sha1_safe
+from rhodecode.bootstrap import \
+    TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
+    TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, TEST_USER_REGULAR_EMAIL, \
+    TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL,\
+    HG_REPO, GIT_REPO, SVN_REPO,\
+    NEW_HG_REPO, NEW_GIT_REPO,\
+    HG_FORK, GIT_FORK
 
 log = logging.getLogger(__name__)
 
-__all__ = [
-    'get_new_dir', 'TestController', 'console_printer',
-    'clear_cache_regions',
-    'assert_session_flash', 'login_user', 'no_newline_id_generator',
-    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'SVN_REPO',
-    'NEW_HG_REPO', 'NEW_GIT_REPO',
-    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
-    'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
-    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
-    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
-    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
-    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'SCM_TESTS',
-]
-
 
 # SOME GLOBALS FOR TESTS
 TEST_DIR = tempfile.gettempdir()
 
-TEST_USER_ADMIN_LOGIN = 'test_admin'
-TEST_USER_ADMIN_PASS = 'test12'
-TEST_USER_ADMIN_EMAIL = '[e-mail]'
-
-TEST_USER_REGULAR_LOGIN = 'test_regular'
-TEST_USER_REGULAR_PASS = 'test12'
-TEST_USER_REGULAR_EMAIL = '[e-mail]'
-
-TEST_USER_REGULAR2_LOGIN = 'test_regular2'
-TEST_USER_REGULAR2_PASS = 'test12'
-TEST_USER_REGULAR2_EMAIL = '[e-mail]'
-
-HG_REPO = 'vcs_test_hg'
-GIT_REPO = 'vcs_test_git'
-SVN_REPO = 'vcs_test_svn'
-
-NEW_HG_REPO = 'vcs_test_hg_new'
-NEW_GIT_REPO = 'vcs_test_git_new'
-
-HG_FORK = 'vcs_test_hg_fork'
-GIT_FORK = 'vcs_test_git_fork'
-
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 
 TESTS_TMP_PATH = tempfile.mkdtemp(prefix='rc_test_', dir=TEST_DIR)
 
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, f'vcsgitclone{uniq_suffix}')
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, f'vcsgitpull{uniq_suffix}')
 
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, f'vcshgclone{uniq_suffix}')
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, f'vcshgpull{uniq_suffix}')
 
 TEST_REPO_PREFIX = 'vcs-test'
 
 
 def clear_cache_regions(regions=None):
     # dogpile
     from rhodecode.lib.rc_cache import region_meta
     for region_name, region in region_meta.dogpile_cache_regions.items():
         if not regions or region_name in regions:
             region.invalidate()
 
 
 def get_new_dir(title):
     """
     Returns always new directory path.
     """
     from rhodecode.tests.vcs.utils import get_normalized_path
     name_parts = [TEST_REPO_PREFIX]
     if title:
         name_parts.append(title)
     hex_str = sha1_safe(f'{os.getpid()} {time.time()}')
     name_parts.append(hex_str)
     name = '-'.join(name_parts)
     path = jn(TEST_DIR, name)
     return get_normalized_path(path)
 
 
 def repo_id_generator(name):
     numeric_hash = 0
     for char in name:
         numeric_hash += (ord(char))
     return numeric_hash
 
 
 @pytest.mark.usefixtures('app', 'index_location')
 class TestController(object):
 
     maxDiff = None
 
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         self._session = login_user_session(self.app, username, password)
         self.csrf_token = auth.get_csrf_token(self._session)
 
         return self._session['rhodecode_user']
 
     def logout_user(self):
         logout_user_session(self.app, auth.get_csrf_token(self._session))
         self.csrf_token = None
         self._logged_username = None
         self._session = None
 
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
 
 
 def login_user_session(
         app, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):
 
     response = app.post(
         h.route_path('login'),
         {'username': username, 'password': password})
     if 'invalid user name' in response.text:
         pytest.fail(f'could not login using {username} {password}')
 
     assert response.status == '302 Found'
     response = response.follow()
     assert response.status == '200 OK'
 
     session = response.get_session_from_response()
     assert 'rhodecode_user' in session
     rc_user = session['rhodecode_user']
     assert rc_user.get('username') == username
     assert rc_user.get('is_authenticated')
 
     return session
 
 
 def logout_user_session(app, csrf_token):
     app.post(h.route_path('logout'), {'csrf_token': csrf_token}, status=302)
 
 
 def login_user(app, username=TEST_USER_ADMIN_LOGIN,
                password=TEST_USER_ADMIN_PASS):
     return login_user_session(app, username, password)['rhodecode_user']
 
 
 def assert_session_flash(response, msg=None, category=None, no_=None):
     """
     Assert on a flash message in the current session.
 
     :param response: Response from give calll, it will contain flash
         messages or bound session with them.
     :param msg: The expected message. Will be evaluated if a
         :class:`LazyString` is passed in.
     :param category: Optional. If passed, the message category will be
         checked as well.
     :param no_: Optional. If passed, the message will be checked to NOT
         be in the flash session
     """
     if msg is None and no_ is None:
         raise ValueError("Parameter msg or no_ is required.")
 
     if msg and no_:
         raise ValueError("Please specify either msg or no_, but not both")
 
     session = response.get_session_from_response()
     messages = flash.pop_messages(session=session)
     msg = _eval_if_lazy(msg)
 
     if no_:
         error_msg = f'unable to detect no_ message `{no_}` in empty flash list'
     else:
         error_msg = f'unable to find message `{msg}` in empty flash list'
     assert messages, error_msg
     message = messages[0]
 
     message_text = _eval_if_lazy(message.message) or ''
 
     if no_:
         if no_ in message_text:
             msg = f'msg `{no_}` found in session flash.'
             pytest.fail(safe_str(msg))
     else:
 
         if msg not in message_text:
             fail_msg = f'msg `{msg}` not found in ' \
                        f'session flash: got `{message_text}` (type:{type(message_text)}) instead'
 
             pytest.fail(safe_str(fail_msg))
 
         if category:
             assert category == message.category
 
 
 def _eval_if_lazy(value):
     return value.eval() if hasattr(value, 'eval') else value
 
 
 def no_newline_id_generator(test_name):
     """
     Generates a test name without spaces or newlines characters. Used for
     nicer output of progress of test
     """
 
     test_name = safe_str(test_name)\
         .replace('\n', '_N') \
         .replace('\r', '_N') \
         .replace('\t', '_T') \
         .replace(' ', '_S')
 
     return test_name or 'test-with-empty-name'
 
 def console_printer(*msg):
     print_func = print
     try:
         from rich import print as print_func
     except ImportError:
         pass
 
     print_func(*msg)