code: import fix/pep8

rhodecode/api/views/repo_api.py

@@ -1,2534 +1,2533 b''
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import time
 
 import rhodecode
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     has_superadmin_permission, Optional, OAttr, get_repo_or_error,
     get_user_group_or_error, get_user_or_error, validate_repo_permissions,
     get_perm_or_error, parse_args, get_origin, build_commit_data,
     validate_set_owner_permissions)
 from rhodecode.lib import audit_logger, rc_cache, channelstream
 from rhodecode.lib import repo_maintenance
 from rhodecode.lib.auth import (
     HasPermissionAnyApi, HasUserGroupPermissionAnyApi,
     HasRepoPermissionAnyApi)
 from rhodecode.lib.celerylib.utils import get_task_id
 from rhodecode.lib.utils2 import (
     str2bool, time_to_datetime, safe_str, safe_int)
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.exceptions import (
     StatusChangeOnClosedPullRequestError, CommentVersionMismatch)
 from rhodecode.lib.vcs import RepositoryError
 from rhodecode.lib.vcs.exceptions import NodeDoesNotExistError
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     Session, ChangesetStatus, RepositoryField, Repository, RepoGroup,
     ChangesetComment)
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.scm import ScmModel, RepoList
 from rhodecode.model.settings import SettingsModel, VcsSettingsModel
 from rhodecode.model import validation_schema
 from rhodecode.model.validation_schema.schemas import repo_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_repo(request, apiuser, repoid, cache=Optional(True)):
     """
     Gets an existing repository by its name or repository_id.
 
     The members section so the output returns users groups or users
     associated with that repository.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param cache: use the cached value for last changeset
     :type: cache: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         {
           "error": null,
           "id": <repo_id>,
           "result": {
             "clone_uri": null,
             "created_on": "timestamp",
             "description": "repo description",
             "enable_downloads": false,
             "enable_locking": false,
             "enable_statistics": false,
             "followers": [
               {
                 "active": true,
                 "admin": false,
                 "api_key": "****************************************",
                 "api_keys": [
                   "****************************************"
                 ],
                 "email": "[e-mail]",
                 "emails": [
                   "[e-mail]"
                 ],
                 "extern_name": "rhodecode",
                 "extern_type": "rhodecode",
                 "firstname": "username",
                 "ip_addresses": [],
                 "language": null,
                 "last_login": "2015-09-16T17:16:35.854",
                 "lastname": "surname",
                 "user_id": <user_id>,
                 "username": "name"
               }
             ],
             "fork_of": "parent-repo",
             "landing_rev": [
               "rev",
               "tip"
             ],
             "last_changeset": {
               "author": "User <[e-mail]>",
               "branch": "default",
               "date": "timestamp",
               "message": "last commit message",
               "parents": [
                 {
                   "raw_id": "commit-id"
                 }
               ],
               "raw_id": "commit-id",
               "revision": <revision number>,
               "short_id": "short id"
             },
             "lock_reason": null,
             "locked_by": null,
             "locked_date": null,
             "owner": "owner-name",
             "permissions": [
               {
                 "name": "super-admin-name",
                 "origin": "super-admin",
                 "permission": "repository.admin",
                 "type": "user"
               },
               {
                 "name": "owner-name",
                 "origin": "owner",
                 "permission": "repository.admin",
                 "type": "user"
               },
               {
                 "name": "user-group-name",
                 "origin": "permission",
                 "permission": "repository.write",
                 "type": "user_group"
               }
             ],
             "private": true,
             "repo_id": 676,
             "repo_name": "user-group/repo-name",
             "repo_type": "hg"
           }
         }
     """
 
     repo = get_repo_or_error(repoid)
     cache = Optional.extract(cache)
 
     include_secrets = False
     if has_superadmin_permission(apiuser):
         include_secrets = True
     else:
         # check if we have at least read permission for this repo !
         _perms = (
             'repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     permissions = []
     for _user in repo.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in repo.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
 
     following_users = [
         user.user.get_api_data(include_secrets=include_secrets)
         for user in repo.followers]
 
     if not cache:
         repo.update_commit_cache()
     data = repo.get_api_data(include_secrets=include_secrets)
     data['permissions'] = permissions
     data['followers'] = following_users
 
     return data
 
 
 @jsonrpc_method()
 def get_repos(request, apiuser, root=Optional(None), traverse=Optional(True)):
     """
     Lists all existing repositories.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param root: specify root repository group to fetch repositories.
         filters the returned repositories to be members of given root group.
     :type root: Optional(None)
     :param traverse: traverse given root into subrepositories. With this flag
         set to False, it will only return top-level repositories from `root`.
         if root is empty it will return just top-level repositories.
     :type traverse: Optional(True)
 
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: [
                   {
                     "repo_id" :          "<repo_id>",
                     "repo_name" :        "<reponame>"
                     "repo_type" :        "<repo_type>",
                     "clone_uri" :        "<clone_uri>",
                     "private": :         "<bool>",
                     "created_on" :       "<datetimecreated>",
                     "description" :      "<description>",
                     "landing_rev":       "<landing_rev>",
                     "owner":             "<repo_owner>",
                     "fork_of":           "<name_of_fork_parent>",
                     "enable_downloads":  "<bool>",
                     "enable_locking":    "<bool>",
                     "enable_statistics": "<bool>",
                   },
                   ...
                 ]
         error:  null
     """
 
     include_secrets = has_superadmin_permission(apiuser)
     _perms = ('repository.read', 'repository.write', 'repository.admin',)
     extras = {'user': apiuser}
 
     root = Optional.extract(root)
     traverse = Optional.extract(traverse, binary=True)
 
     if root:
         # verify parent existance, if it's empty return an error
         parent = RepoGroup.get_by_group_name(root)
         if not parent:
             raise JSONRPCError(
                 f'Root repository group `{root}` does not exist')
 
         if traverse:
             repos = RepoModel().get_repos_for_root(root=root, traverse=traverse)
         else:
             repos = RepoModel().get_repos_for_root(root=parent)
     else:
         if traverse:
             repos = RepoModel().get_all()
         else:
             # return just top-level
             repos = RepoModel().get_repos_for_root(root=None)
 
     repo_list = RepoList(repos, perm_set=_perms, extra_kwargs=extras)
     return [repo.get_api_data(include_secrets=include_secrets)
             for repo in repo_list]
 
 
 @jsonrpc_method()
 def get_repo_changeset(request, apiuser, repoid, revision,
                        details=Optional('basic')):
     """
     Returns information about a changeset.
 
     Additionally parameters define the amount of details returned by
     this function.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id
     :type repoid: str or int
     :param revision: revision for which listing should be done
     :type revision: str
     :param details: details can be 'basic|extended|full' full gives diff
         info details like the diff itself, and number of changed files etc.
     :type details: Optional(str)
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     changes_details = Optional.extract(details)
     _changes_details_types = ['basic', 'extended', 'full']
     if changes_details not in _changes_details_types:
         raise JSONRPCError(
             'ret_type must be one of %s' % (
                 ','.join(_changes_details_types)))
 
     vcs_repo = repo.scm_instance()
     pre_load = ['author', 'branch', 'date', 'message', 'parents',
                 'status', '_commit', '_file_paths']
 
     try:
         commit = repo.get_commit(commit_id=revision, pre_load=pre_load)
     except TypeError as e:
         raise JSONRPCError(safe_str(e))
     _cs_json = commit.__json__()
     _cs_json['diff'] = build_commit_data(vcs_repo, commit, changes_details)
     if changes_details == 'full':
         _cs_json['refs'] = commit._get_refs()
     return _cs_json
 
 
 @jsonrpc_method()
 def get_repo_changesets(request, apiuser, repoid, start_rev, limit,
                         details=Optional('basic')):
     """
     Returns a set of commits limited by the number starting
     from the `start_rev` option.
 
     Additional parameters define the amount of details returned by this
     function.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param start_rev: The starting revision from where to get changesets.
     :type start_rev: str
     :param limit: Limit the number of commits to this amount
     :type limit: str or int
     :param details: Set the level of detail returned. Valid option are:
         ``basic``, ``extended`` and ``full``.
     :type details: Optional(str)
 
     .. note::
 
        Setting the parameter `details` to the value ``full`` is extensive
        and returns details like the diff itself, and the number
        of changed files.
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     changes_details = Optional.extract(details)
     _changes_details_types = ['basic', 'extended', 'full']
     if changes_details not in _changes_details_types:
         raise JSONRPCError(
             'ret_type must be one of %s' % (
                 ','.join(_changes_details_types)))
 
     limit = int(limit)
     pre_load = ['author', 'branch', 'date', 'message', 'parents',
                 'status', '_commit', '_file_paths']
 
     vcs_repo = repo.scm_instance()
     # SVN needs a special case to distinguish its index and commit id
     if vcs_repo and vcs_repo.alias == 'svn' and (start_rev == '0'):
         start_rev = vcs_repo.commit_ids[0]
 
     try:
         commits = vcs_repo.get_commits(
             start_id=start_rev, pre_load=pre_load, translate_tags=False)
     except TypeError as e:
         raise JSONRPCError(safe_str(e))
     except Exception:
         log.exception('Fetching of commits failed')
         raise JSONRPCError('Error occurred during commit fetching')
 
     ret = []
     for cnt, commit in enumerate(commits):
         if cnt >= limit != -1:
             break
         _cs_json = commit.__json__()
         _cs_json['diff'] = build_commit_data(vcs_repo, commit, changes_details)
         if changes_details == 'full':
             _cs_json['refs'] = {
                 'branches': [commit.branch],
                 'bookmarks': getattr(commit, 'bookmarks', []),
                 'tags': commit.tags
             }
         ret.append(_cs_json)
     return ret
 
 
 @jsonrpc_method()
 def get_repo_nodes(request, apiuser, repoid, revision, root_path,
                    ret_type=Optional('all'), details=Optional('basic'),
                    max_file_bytes=Optional(None)):
     """
     Returns a list of nodes and children in a flat list for a given
     path at given revision.
 
     It's possible to specify ret_type to show only `files` or `dirs`.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param revision: The revision for which listing should be done.
     :type revision: str
     :param root_path: The path from which to start displaying.
     :type root_path: str
     :param ret_type: Set the return type. Valid options are
         ``all`` (default), ``files`` and ``dirs``.
     :type ret_type: Optional(str)
     :param details: Returns extended information about nodes, such as
         md5, binary, and or content.
         The valid options are ``basic`` and ``full``.
     :type details: Optional(str)
     :param max_file_bytes: Only return file content under this file size bytes
     :type details: Optional(int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: [
                     {
                       "binary": false,
                       "content": "File line",
                       "extension": "md",
                       "lines": 2,
                       "md5": "059fa5d29b19c0657e384749480f6422",
                       "mimetype": "text/x-minidsrc",
                       "name": "file.md",
                       "size": 580,
                       "type": "file"
                     },
                   ...
                 ]
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     ret_type = Optional.extract(ret_type)
     details = Optional.extract(details)
     max_file_bytes = Optional.extract(max_file_bytes)
 
     _extended_types = ['basic', 'full']
     if details not in _extended_types:
         ret_types = ','.join(_extended_types)
         raise JSONRPCError(f'ret_type must be one of {ret_types}')
 
     extended_info = False
     content = False
     if details == 'basic':
         extended_info = True
 
     if details == 'full':
         extended_info = content = True
 
     _map = {}
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if _scm.is_empty():
             return []
 
         _d, _f = ScmModel().get_nodes(
             repo, revision, root_path, flat=False,
             extended_info=extended_info, content=content,
             max_file_bytes=max_file_bytes)
 
         _map = {
             'all': _d + _f,
             'files': _f,
             'dirs': _d,
         }
 
         return _map[ret_type]
     except KeyError:
         keys = ','.join(sorted(_map.keys()))
         raise JSONRPCError(f'ret_type must be one of {keys}')
     except Exception:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` nodes')
 
 
 @jsonrpc_method()
 def get_repo_file(request, apiuser, repoid, commit_id, file_path,
                   max_file_bytes=Optional(0), details=Optional('basic'),
                   cache=Optional(True)):
     """
     Returns a single file from repository at given revision.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param commit_id: The revision for which listing should be done.
     :type commit_id: str
     :param file_path: The path from which to start displaying.
     :type file_path: str
     :param details: Returns different set of information about nodes.
         The valid options are ``minimal`` ``basic`` and ``full``.
     :type details: Optional(str)
     :param max_file_bytes: Only return file content under this file size bytes
     :type max_file_bytes: Optional(int)
     :param cache: Use internal caches for fetching files. If disabled fetching
         files is slower but more memory efficient
     :type cache: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
             "binary": false,
             "extension": "py",
             "lines": 35,
             "content": "....",
             "md5": "76318336366b0f17ee249e11b0c99c41",
             "mimetype": "text/x-python",
             "name": "python.py",
             "size": 817,
             "type": "file",
         }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     cache = Optional.extract(cache, binary=True)
     details = Optional.extract(details)
     max_file_bytes = Optional.extract(max_file_bytes)
 
     _extended_types = ['minimal', 'minimal+search', 'basic', 'full']
     if details not in _extended_types:
         ret_types = ','.join(_extended_types)
         raise JSONRPCError(f'ret_type must be one of %s, got {ret_types}', details)
     extended_info = False
     content = False
 
     if details == 'minimal':
         extended_info = False
 
     elif details == 'basic':
         extended_info = True
 
     elif details == 'full':
         extended_info = content = True
 
     file_path = safe_str(file_path)
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if _scm.is_empty():
             return None
 
         node = ScmModel().get_node(
             repo, commit_id, file_path, extended_info=extended_info,
             content=content, max_file_bytes=max_file_bytes, cache=cache)
 
     except NodeDoesNotExistError:
         raise JSONRPCError(
             f'There is no file in repo: `{repo.repo_name}` at path `{file_path}` for commit: `{commit_id}`')
     except Exception:
         log.exception("Exception occurred while trying to get repo %s file",
                       repo.repo_name)
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` file at path {file_path}')
 
     return node
 
 
 @jsonrpc_method()
 def get_repo_fts_tree(request, apiuser, repoid, commit_id, root_path):
     """
     Returns a list of tree nodes for path at given revision. This api is built
     strictly for usage in full text search building, and shouldn't be consumed
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     repo_id = repo.repo_id
     cache_seconds = rhodecode.ConfigGet().get_int('rc_cache.cache_repo.expiration_time')
     cache_on = cache_seconds > 0
 
     cache_namespace_uid = f'repo.{rc_cache.FILE_TREE_CACHE_VER}.{repo_id}'
     rc_cache.get_or_create_region('cache_repo', cache_namespace_uid)
 
     def compute_fts_tree(repo_id, commit_id, root_path):
         return ScmModel().get_fts_data(repo_id, commit_id, root_path)
 
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if not _scm or _scm.is_empty():
             return []
     except RepositoryError:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` nodes')
 
     try:
         # we need to resolve commit_id to a FULL sha for cache to work correctly.
         # sending 'master' is a pointer that needs to be translated to current commit.
         commit_id = _scm.get_commit(commit_id=commit_id).raw_id
         log.debug(
             'Computing FTS REPO TREE for repo_id %s commit_id `%s` '
             'with caching: %s[TTL: %ss]' % (
                 repo_id, commit_id, cache_on, cache_seconds or 0))
 
         tree_files = compute_fts_tree(repo_id, commit_id, root_path)
 
         return tree_files
 
     except Exception:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError('failed to get repo: `%s` nodes' % repo.repo_name)
 
 
 @jsonrpc_method()
 def get_repo_refs(request, apiuser, repoid):
     """
     Returns a dictionary of current references. It returns
     bookmarks, branches, closed_branches, and tags for given repository
 
     It's possible to specify ret_type to show only `files` or `dirs`.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         "result": {
             "bookmarks": {
               "dev": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "master": "367f590445081d8ec8c2ea0456e73ae1f1c3d6cf"
             },
             "branches": {
               "default": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "stable": "367f590445081d8ec8c2ea0456e73ae1f1c3d6cf"
             },
             "branches_closed": {},
             "tags": {
               "tip": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "v4.4.0": "1232313f9e6adac5ce5399c2a891dc1e72b79022",
               "v4.4.1": "cbb9f1d329ae5768379cdec55a62ebdd546c4e27",
               "v4.4.2": "24ffe44a27fcd1c5b6936144e176b9f6dd2f3a17",
             }
         }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         vcs_instance = repo.scm_instance()
         refs = vcs_instance.refs()
         return refs
     except Exception:
         log.exception("Exception occurred while trying to get repo refs")
         raise JSONRPCError(
             'failed to get repo: `%s` references' % repo.repo_name
         )
 
 
 @jsonrpc_method()
 def create_repo(
         request, apiuser, repo_name, repo_type,
         owner=Optional(OAttr('apiuser')),
         description=Optional(''),
         private=Optional(False),
         clone_uri=Optional(None),
         push_uri=Optional(None),
         landing_rev=Optional(None),
         enable_statistics=Optional(False),
         enable_locking=Optional(False),
         enable_downloads=Optional(False),
         copy_permissions=Optional(False)):
     """
     Creates a repository.
 
     * If the repository name contains "/", repository will be created inside
       a repository group or nested repository groups
 
       For example "foo/bar/repo1" will create |repo| called "repo1" inside
       group "foo/bar". You have to have permissions to access and write to
       the last repository group ("bar" in this example)
 
     This command can only be run using an |authtoken| with at least
     permissions to create repositories, or write permissions to
     parent repository groups.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repo_name: Set the repository name.
     :type repo_name: str
     :param repo_type: Set the repository type; 'hg','git', or 'svn'.
     :type repo_type: str
     :param owner: user_id or username
     :type owner: Optional(str)
     :param description: Set the repository description.
     :type description: Optional(str)
     :param private: set repository as private
     :type private: bool
     :param clone_uri: set clone_uri
     :type clone_uri: str
     :param push_uri: set push_uri
     :type push_uri: str
     :param landing_rev: <rev_type>:<rev>, e.g branch:default, book:dev, rev:abcd
     :type landing_rev: str
     :param enable_locking:
     :type enable_locking: bool
     :param enable_downloads:
     :type enable_downloads: bool
     :param enable_statistics:
     :type enable_statistics: bool
     :param copy_permissions: Copy permission from group in which the
         repository is being created.
     :type copy_permissions: bool
 
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Created new repository `<reponame>`",
                   "success": true,
                   "task": "<celery task id or None if done sync>"
                 }
         error:  null
 
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
          'failed to create repository `<repo_name>`'
       }
 
     """
 
     owner = validate_set_owner_permissions(apiuser, owner)
 
     description = Optional.extract(description)
     copy_permissions = Optional.extract(copy_permissions)
     clone_uri = Optional.extract(clone_uri)
     push_uri = Optional.extract(push_uri)
 
     defs = SettingsModel().get_default_repo_settings(strip_prefix=True)
     if isinstance(private, Optional):
         private = defs.get('repo_private') or Optional.extract(private)
     if isinstance(repo_type, Optional):
         repo_type = defs.get('repo_type')
     if isinstance(enable_statistics, Optional):
         enable_statistics = defs.get('repo_enable_statistics')
     if isinstance(enable_locking, Optional):
         enable_locking = defs.get('repo_enable_locking')
     if isinstance(enable_downloads, Optional):
         enable_downloads = defs.get('repo_enable_downloads')
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(request.translate)
     ref_choices = list(set(ref_choices + [landing_ref]))
 
     landing_commit_ref = Optional.extract(landing_rev) or landing_ref
 
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo_type,
         # user caller
         user=apiuser)
 
     try:
         schema_data = schema.deserialize(dict(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_owner=owner.username,
             repo_description=description,
             repo_landing_commit_ref=landing_commit_ref,
             repo_clone_uri=clone_uri,
             repo_push_uri=push_uri,
             repo_private=private,
             repo_copy_permissions=copy_permissions,
             repo_enable_statistics=enable_statistics,
             repo_enable_downloads=enable_downloads,
             repo_enable_locking=enable_locking))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         data = {
             'owner': owner,
             'repo_name': schema_data['repo_group']['repo_name_without_group'],
             'repo_name_full': schema_data['repo_name'],
             'repo_group': schema_data['repo_group']['repo_group_id'],
             'repo_type': schema_data['repo_type'],
             'repo_description': schema_data['repo_description'],
             'repo_private': schema_data['repo_private'],
             'clone_uri': schema_data['repo_clone_uri'],
             'push_uri': schema_data['repo_push_uri'],
             'repo_landing_rev': schema_data['repo_landing_commit_ref'],
             'enable_statistics': schema_data['repo_enable_statistics'],
             'enable_locking': schema_data['repo_enable_locking'],
             'enable_downloads': schema_data['repo_enable_downloads'],
             'repo_copy_permissions': schema_data['repo_copy_permissions'],
         }
 
         task = RepoModel().create(form_data=data, cur_user=owner.user_id)
         task_id = get_task_id(task)
         # no commit, it's done in RepoModel, or async via celery
         return {
             'msg': "Created new repository `{}`".format(schema_data['repo_name']),
             'success': True,  # cannot return the repo data here since fork
             # can be done async
             'task': task_id
         }
     except Exception:
         log.exception(
             "Exception while trying to create the repository %s",
             schema_data['repo_name'])
         raise JSONRPCError(
             'failed to create repository `{}`'.format(schema_data['repo_name']))
 
 
 @jsonrpc_method()
 def add_field_to_repo(request, apiuser, repoid, key, label=Optional(''),
                       description=Optional('')):
     """
     Adds an extra field to a repository.
 
     This command can only be run using an |authtoken| with at least
     write permissions to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository id.
     :type repoid: str or int
     :param key: Create a unique field key for this repository.
     :type key: str
     :param label:
     :type label: Optional(str)
     :param description:
     :type description: Optional(str)
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     label = Optional.extract(label) or key
     description = Optional.extract(description)
 
     field = RepositoryField.get_by_key_name(key, repo)
     if field:
-        raise JSONRPCError('Field with key '
-                           '`%s` exists for repo `%s`' % (key, repoid))
+        raise JSONRPCError(f'Field with key `{key}` exists for repo `{repoid}`')
 
     try:
         RepoModel().add_repo_field(repo, key, field_label=label,
                                    field_desc=description)
         Session().commit()
         return {
             'msg': f"Added new repository field `{key}`",
             'success': True,
         }
     except Exception:
         log.exception("Exception occurred while trying to add field to repo")
         raise JSONRPCError(
             f'failed to create new field for repository `{repoid}`')
 
 
 @jsonrpc_method()
 def remove_field_from_repo(request, apiuser, repoid, key):
     """
     Removes an extra field from a repository.
 
     This command can only be run using an |authtoken| with at least
     write permissions to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param key: Set the unique field key for this repository.
     :type key: str
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     field = RepositoryField.get_by_key_name(key, repo)
     if not field:
         raise JSONRPCError('Field with key `%s` does not '
                            'exists for repo `%s`' % (key, repoid))
 
     try:
         RepoModel().delete_repo_field(repo, field_key=key)
         Session().commit()
         return {
             'msg': f"Deleted repository field `{key}`",
             'success': True,
         }
     except Exception:
         log.exception(
             "Exception occurred while trying to delete field from repo")
         raise JSONRPCError(
             f'failed to delete field for repository `{repoid}`')
 
 
 @jsonrpc_method()
 def update_repo(
         request, apiuser, repoid, repo_name=Optional(None),
         owner=Optional(OAttr('apiuser')), description=Optional(''),
         private=Optional(False),
         clone_uri=Optional(None), push_uri=Optional(None),
         landing_rev=Optional(None), fork_of=Optional(None),
         enable_statistics=Optional(False),
         enable_locking=Optional(False),
         enable_downloads=Optional(False), fields=Optional('')):
     r"""
     Updates a repository with the given information.
 
     This command can only be run using an |authtoken| with at least
     admin permissions to the |repo|.
 
     * If the repository name contains "/", repository will be updated
       accordingly with a repository group or nested repository groups
 
       For example repoid=repo-test name="foo/bar/repo-test" will update |repo|
       called "repo-test" and place it inside group "foo/bar".
       You have to have permissions to access and write to the last repository
       group ("bar" in this example)
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: repository name or repository ID.
     :type repoid: str or int
     :param repo_name: Update the |repo| name, including the
         repository group it's in.
     :type repo_name: str
     :param owner: Set the |repo| owner.
     :type owner: str
     :param fork_of: Set the |repo| as fork of another |repo|.
     :type fork_of: str
     :param description: Update the |repo| description.
     :type description: str
     :param private: Set the |repo| as private. (True | False)
     :type private: bool
     :param clone_uri: Update the |repo| clone URI.
     :type clone_uri: str
     :param landing_rev: Set the |repo| landing revision. e.g branch:default, book:dev, rev:abcd
     :type landing_rev: str
     :param enable_statistics: Enable statistics on the |repo|, (True | False).
     :type enable_statistics: bool
     :param enable_locking: Enable |repo| locking.
     :type enable_locking: bool
     :param enable_downloads: Enable downloads from the |repo|, (True | False).
     :type enable_downloads: bool
     :param fields: Add extra fields to the |repo|. Use the following
         example format: ``field_key=field_val,field_key2=fieldval2``.
         Escape ', ' with \,
     :type fields: str
     """
 
     repo = get_repo_or_error(repoid)
 
     include_secrets = False
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
     else:
         include_secrets = True
 
     updates = dict(
         repo_name=repo_name
         if not isinstance(repo_name, Optional) else repo.repo_name,
 
         fork_id=fork_of
         if not isinstance(fork_of, Optional) else repo.fork.repo_name if repo.fork else None,
 
         user=owner
         if not isinstance(owner, Optional) else repo.user.username,
 
         repo_description=description
         if not isinstance(description, Optional) else repo.description,
 
         repo_private=private
         if not isinstance(private, Optional) else repo.private,
 
         clone_uri=clone_uri
         if not isinstance(clone_uri, Optional) else repo.clone_uri,
 
         push_uri=push_uri
         if not isinstance(push_uri, Optional) else repo.push_uri,
 
         repo_landing_rev=landing_rev
         if not isinstance(landing_rev, Optional) else repo._landing_revision,
 
         repo_enable_statistics=enable_statistics
         if not isinstance(enable_statistics, Optional) else repo.enable_statistics,
 
         repo_enable_locking=enable_locking
         if not isinstance(enable_locking, Optional) else repo.enable_locking,
 
         repo_enable_downloads=enable_downloads
         if not isinstance(enable_downloads, Optional) else repo.enable_downloads)
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo.repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(
         request.translate, repo=repo)
     ref_choices = list(set(ref_choices + [landing_ref]))
 
     old_values = repo.get_api_data()
     repo_type = repo.repo_type
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo_type,
         # user caller
         user=apiuser,
         old_values=old_values)
     try:
         schema_data = schema.deserialize(dict(
             # we save old value, users cannot change type
             repo_type=repo_type,
 
             repo_name=updates['repo_name'],
             repo_owner=updates['user'],
             repo_description=updates['repo_description'],
             repo_clone_uri=updates['clone_uri'],
             repo_push_uri=updates['push_uri'],
             repo_fork_of=updates['fork_id'],
             repo_private=updates['repo_private'],
             repo_landing_commit_ref=updates['repo_landing_rev'],
             repo_enable_statistics=updates['repo_enable_statistics'],
             repo_enable_downloads=updates['repo_enable_downloads'],
             repo_enable_locking=updates['repo_enable_locking']))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     # save validated data back into the updates dict
     validated_updates = dict(
         repo_name=schema_data['repo_group']['repo_name_without_group'],
         repo_group=schema_data['repo_group']['repo_group_id'],
 
         user=schema_data['repo_owner'],
         repo_description=schema_data['repo_description'],
         repo_private=schema_data['repo_private'],
         clone_uri=schema_data['repo_clone_uri'],
         push_uri=schema_data['repo_push_uri'],
         repo_landing_rev=schema_data['repo_landing_commit_ref'],
         repo_enable_statistics=schema_data['repo_enable_statistics'],
         repo_enable_locking=schema_data['repo_enable_locking'],
         repo_enable_downloads=schema_data['repo_enable_downloads'],
     )
 
     if schema_data['repo_fork_of']:
         fork_repo = get_repo_or_error(schema_data['repo_fork_of'])
         validated_updates['fork_id'] = fork_repo.repo_id
 
     # extra fields
     fields = parse_args(Optional.extract(fields), key_prefix='ex_')
     if fields:
         validated_updates.update(fields)
 
     try:
         RepoModel().update(repo, **validated_updates)
         audit_logger.store_api(
             'repo.edit', action_data={'old_data': old_values},
             user=apiuser, repo=repo)
         Session().commit()
         return {
             'msg': f'updated repo ID:{repo.repo_id} {repo.repo_name}',
             'repository': repo.get_api_data(include_secrets=include_secrets)
         }
     except Exception:
         log.exception(
             "Exception while trying to update the repository %s",
             repoid)
         raise JSONRPCError('failed to update repo `%s`' % repoid)
 
 
 @jsonrpc_method()
 def fork_repo(request, apiuser, repoid, fork_name,
               owner=Optional(OAttr('apiuser')),
               description=Optional(''),
               private=Optional(False),
               clone_uri=Optional(None),
               landing_rev=Optional(None),
               copy_permissions=Optional(False)):
     """
     Creates a fork of the specified |repo|.
 
     * If the fork_name contains "/", fork will be created inside
       a repository group or nested repository groups
 
       For example "foo/bar/fork-repo" will create fork called "fork-repo"
       inside group "foo/bar". You have to have permissions to access and
       write to the last repository group ("bar" in this example)
 
     This command can only be run using an |authtoken| with minimum
     read permissions of the forked repo, create fork permissions for an user.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set repository name or repository ID.
     :type repoid: str or int
     :param fork_name: Set the fork name, including it's repository group membership.
     :type fork_name: str
     :param owner: Set the fork owner.
     :type owner: str
     :param description: Set the fork description.
     :type description: str
     :param copy_permissions: Copy permissions from parent |repo|. The
         default is False.
     :type copy_permissions: bool
     :param private: Make the fork private. The default is False.
     :type private: bool
     :param landing_rev: Set the landing revision. E.g branch:default, book:dev, rev:abcd
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_for_response>
         api_key : "<api_key>"
         args:     {
                     "repoid" :          "<reponame or repo_id>",
                     "fork_name":        "<forkname>",
                     "owner":            "<username or user_id = Optional(=apiuser)>",
                     "description":      "<description>",
                     "copy_permissions": "<bool>",
                     "private":          "<bool>",
                     "landing_rev":      "<landing_rev>"
                   }
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Created fork of `<reponame>` as `<forkname>`",
                   "success": true,
                   "task": "<celery task id or None if done sync>"
                 }
         error:  null
 
     """
 
     repo = get_repo_or_error(repoid)
     repo_name = repo.repo_name
 
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for
         # this repo that we fork !
         _perms = ('repository.admin', 'repository.write', 'repository.read')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
         # check if the regular user has at least fork permissions as well
         if not HasPermissionAnyApi(PermissionModel.FORKING_ENABLED)(user=apiuser):
             raise JSONRPCForbidden()
 
     # check if user can set owner parameter
     owner = validate_set_owner_permissions(apiuser, owner)
 
     description = Optional.extract(description)
     copy_permissions = Optional.extract(copy_permissions)
     clone_uri = Optional.extract(clone_uri)
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo.repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(request.translate)
     ref_choices = list(set(ref_choices + [landing_ref]))
     landing_commit_ref = Optional.extract(landing_rev) or landing_ref
 
     private = Optional.extract(private)
 
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo.repo_type,
         # user caller
         user=apiuser)
 
     try:
         schema_data = schema.deserialize(dict(
             repo_name=fork_name,
             repo_type=repo.repo_type,
             repo_owner=owner.username,
             repo_description=description,
             repo_landing_commit_ref=landing_commit_ref,
             repo_clone_uri=clone_uri,
             repo_private=private,
             repo_copy_permissions=copy_permissions))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         data = {
             'fork_parent_id': repo.repo_id,
 
             'repo_name': schema_data['repo_group']['repo_name_without_group'],
             'repo_name_full': schema_data['repo_name'],
             'repo_group': schema_data['repo_group']['repo_group_id'],
             'repo_type': schema_data['repo_type'],
             'description': schema_data['repo_description'],
             'private': schema_data['repo_private'],
             'copy_permissions': schema_data['repo_copy_permissions'],
             'landing_rev': schema_data['repo_landing_commit_ref'],
         }
 
         task = RepoModel().create_fork(data, cur_user=owner.user_id)
         # no commit, it's done in RepoModel, or async via celery
         task_id = get_task_id(task)
 
         return {
             'msg': 'Created fork of `{}` as `{}`'.format(
                 repo.repo_name, schema_data['repo_name']),
             'success': True,  # cannot return the repo data here since fork
             # can be done async
             'task': task_id
         }
     except Exception:
         log.exception(
             "Exception while trying to create fork %s",
             schema_data['repo_name'])
         raise JSONRPCError(
             'failed to fork repository `{}` as `{}`'.format(
                 repo_name, schema_data['repo_name']))
 
 
 @jsonrpc_method()
 def delete_repo(request, apiuser, repoid, forks=Optional('')):
     """
     Deletes a repository.
 
     * When the `forks` parameter is set it's possible to detach or delete
       forks of deleted repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param forks: Set to `detach` or `delete` forks from the |repo|.
     :type forks: Optional(str)
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Deleted repository `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     repo_name = repo.repo_name
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         handle_forks = Optional.extract(forks)
         _forks_msg = ''
         _forks = [f for f in repo.forks]
         if handle_forks == 'detach':
             _forks_msg = ' ' + 'Detached %s forks' % len(_forks)
         elif handle_forks == 'delete':
             _forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
         elif _forks:
             raise JSONRPCError(
                 'Cannot delete `%s` it still contains attached forks' %
                 (repo.repo_name,)
             )
         old_data = repo.get_api_data()
         RepoModel().delete(repo, forks=forks)
 
         repo = audit_logger.RepoWrap(repo_id=None,
                                      repo_name=repo.repo_name)
 
         audit_logger.store_api(
             'repo.delete', action_data={'old_data': old_data},
             user=apiuser, repo=repo)
 
         ScmModel().mark_for_invalidation(repo_name, delete=True)
         Session().commit()
         return {
             'msg': f'Deleted repository `{repo_name}`{_forks_msg}',
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying to delete repo")
         raise JSONRPCError(
             f'failed to delete repository `{repo_name}`'
         )
 
 
 #TODO: marcink, change name ?
 @jsonrpc_method()
 def invalidate_cache(request, apiuser, repoid, delete_keys=Optional(False)):
     """
     Invalidates the cache for the specified repository.
 
     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 repoid: Sets the repository name or repository ID.
     :type repoid: str or int
     :param delete_keys: This deletes the invalidated keys instead of
         just flagging them.
     :type delete_keys: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'msg': Cache for repository `<repository name>` was invalidated,
         'repository': <repository name>
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error : {
          'Error occurred during cache invalidation action'
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     delete = Optional.extract(delete_keys)
     try:
         ScmModel().mark_for_invalidation(repo.repo_name, delete=delete)
         return {
             'msg': f'Cache for repository `{repoid}` was invalidated',
             'repository': repo.repo_name
         }
     except Exception:
         log.exception(
             "Exception occurred while trying to invalidate repo cache")
         raise JSONRPCError(
             'Error occurred during cache invalidation action'
         )
 
 
 #TODO: marcink, change name ?
 @jsonrpc_method()
 def lock(request, apiuser, repoid, locked=Optional(None),
          userid=Optional(OAttr('apiuser'))):
     """
     Sets the lock state of the specified |repo| by the given user.
     From more information, see :ref:`repo-locking`.
 
     * If the ``userid`` option is not set, the repository is locked to the
       user who called the method.
     * If the ``locked`` parameter is not set, the current lock state of the
       repository is displayed.
 
     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 repoid: Sets the repository name or repository ID.
     :type repoid: str or int
     :param locked: Sets the lock state.
     :type locked: Optional(``True`` | ``False``)
     :param userid: Set the repository lock to this user.
     :type userid: Optional(str or int)
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'repo': '<reponame>',
         'locked': <bool: lock state>,
         'locked_since': <int: lock timestamp>,
         'locked_by': <username of person who made the lock>,
         'lock_reason': <str: reason for locking>,
         'lock_state_changed': <bool: True if lock state has been changed in this request>,
         'msg': 'Repo `<reponame>` locked by `<username>` on <timestamp>.'
         or
         'msg': 'Repo `<repository name>` not locked.'
         or
         'msg': 'User `<user name>` set lock state for repo `<repository name>` to `<new lock state>`'
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred locking repository `<reponame>`'
       }
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least write permission for this repo !
         _perms = ('repository.admin', 'repository.write',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError('userid is not the same as your user')
 
     if isinstance(userid, Optional):
         userid = apiuser.user_id
 
     user = get_user_or_error(userid)
 
     if isinstance(locked, Optional):
         lockobj = repo.locked
 
         if lockobj[0] is None:
             _d = {
                 'repo': repo.repo_name,
                 'locked': False,
                 'locked_since': None,
                 'locked_by': None,
                 'lock_reason': None,
                 'lock_state_changed': False,
                 'msg': 'Repo `%s` not locked.' % repo.repo_name
             }
             return _d
         else:
             _user_id, _time, _reason = lockobj
             lock_user = get_user_or_error(userid)
             _d = {
                 'repo': repo.repo_name,
                 'locked': True,
                 'locked_since': _time,
                 'locked_by': lock_user.username,
                 'lock_reason': _reason,
                 'lock_state_changed': False,
                 'msg': ('Repo `%s` locked by `%s` on `%s`.'
                         % (repo.repo_name, lock_user.username,
                            json.dumps(time_to_datetime(_time))))
             }
             return _d
 
     # force locked state through a flag
     else:
         locked = str2bool(locked)
         lock_reason = Repository.LOCK_API
         try:
             if locked:
                 lock_time = time.time()
                 Repository.lock(repo, user.user_id, lock_time, lock_reason)
             else:
                 lock_time = None
                 Repository.unlock(repo)
             _d = {
                 'repo': repo.repo_name,
                 'locked': locked,
                 'locked_since': lock_time,
                 'locked_by': user.username,
                 'lock_reason': lock_reason,
                 'lock_state_changed': True,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (user.username, repo.repo_name, locked))
             }
             return _d
         except Exception:
             log.exception(
                 "Exception occurred while trying to lock repository")
             raise JSONRPCError(
                 'Error occurred locking repository `%s`' % repo.repo_name
             )
 
 
 @jsonrpc_method()
 def comment_commit(
         request, apiuser, repoid, commit_id, message, status=Optional(None),
         comment_type=Optional(ChangesetComment.COMMENT_TYPE_NOTE),
         resolves_comment_id=Optional(None), extra_recipients=Optional([]),
         userid=Optional(OAttr('apiuser')), send_email=Optional(True)):
     """
     Set a commit comment, and optionally change the status of the commit.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param commit_id: Specify the commit_id for which to set a comment.
     :type commit_id: str
     :param message: The comment text.
     :type message: str
     :param status: (**Optional**) status of commit, one of: 'not_reviewed',
         'approved', 'rejected', 'under_review'
     :type status: str
     :param comment_type: Comment type, one of: 'note', 'todo'
     :type comment_type: Optional(str), default: 'note'
     :param resolves_comment_id: id of comment which this one will resolve
     :type resolves_comment_id: Optional(int)
     :param extra_recipients: list of user ids or usernames to add
         notifications for this comment. Acts like a CC for notification
     :type extra_recipients: Optional(list)
     :param userid: Set the user name of the comment creator.
     :type userid: Optional(str or int)
     :param send_email: Define if this comment should also send email notification
     :type send_email: Optional(bool)
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : {
                 "msg": "Commented on commit `<commit_id>` for repository `<repoid>`",
                 "status_change": null or <status>,
                 "success": true
             },
             "error" :  null
         }
 
     """
     _ = request.translate
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.read', 'repository.write', 'repository.admin')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
     db_repo_name = repo.repo_name
 
     try:
         commit = repo.scm_instance().get_commit(commit_id=commit_id)
         commit_id = commit.raw_id
     except Exception as e:
         log.exception('Failed to fetch commit')
         raise JSONRPCError(safe_str(e))
 
     if isinstance(userid, Optional):
         userid = apiuser.user_id
 
     user = get_user_or_error(userid)
     status = Optional.extract(status)
     comment_type = Optional.extract(comment_type)
     resolves_comment_id = Optional.extract(resolves_comment_id)
     extra_recipients = Optional.extract(extra_recipients)
     send_email = Optional.extract(send_email, binary=True)
 
     allowed_statuses = [x[0] for x in ChangesetStatus.STATUSES]
     if status and status not in allowed_statuses:
         raise JSONRPCError('Bad status, must be on '
                            'of %s got %s' % (allowed_statuses, status,))
 
     if resolves_comment_id:
         comment = ChangesetComment.get(resolves_comment_id)
         if not comment:
             raise JSONRPCError(
                 'Invalid resolves_comment_id `%s` for this commit.'
                 % resolves_comment_id)
         if comment.comment_type != ChangesetComment.COMMENT_TYPE_TODO:
             raise JSONRPCError(
                 'Comment `%s` is wrong type for setting status to resolved.'
                 % resolves_comment_id)
 
     try:
         rc_config = SettingsModel().get_all_settings()
         renderer = rc_config.get('rhodecode_markup_renderer', 'rst')
         status_change_label = ChangesetStatus.get_status_lbl(status)
         comment = CommentsModel().create(
             message, repo, user, commit_id=commit_id,
             status_change=status_change_label,
             status_change_type=status,
             renderer=renderer,
             comment_type=comment_type,
             resolves_comment_id=resolves_comment_id,
             auth_user=apiuser,
             extra_recipients=extra_recipients,
             send_email=send_email
         )
         is_inline = comment.is_inline
 
         if status:
             # also do a status change
             try:
                 ChangesetStatusModel().set_status(
                     repo, status, user, comment, revision=commit_id,
                     dont_allow_on_closed_pull_request=True
                 )
             except StatusChangeOnClosedPullRequestError:
                 log.exception(
                     "Exception occurred while trying to change repo commit status")
                 msg = ('Changing status on a commit associated with '
                        'a closed pull request is not allowed')
                 raise JSONRPCError(msg)
 
         CommentsModel().trigger_commit_comment_hook(
             repo, apiuser, 'create',
             data={'comment': comment, 'commit': commit})
 
         Session().commit()
 
         comment_broadcast_channel = channelstream.comment_channel(
             db_repo_name, commit_obj=commit)
 
         comment_data = {'comment': comment, 'comment_id': comment.comment_id}
         comment_type = 'inline' if is_inline else 'general'
         channelstream.comment_channelstream_push(
             request, comment_broadcast_channel, apiuser,
             _('posted a new {} comment').format(comment_type),
             comment_data=comment_data)
 
         return {
             'msg': (
                 'Commented on commit `{}` for repository `{}`'.format(
                     comment.revision, repo.repo_name)),
             'status_change': status,
             'success': True,
         }
     except JSONRPCError:
         # catch any inside errors, and re-raise them to prevent from
         # below global catch to silence them
         raise
     except Exception:
         log.exception("Exception occurred while trying to comment on commit")
         raise JSONRPCError(
             f'failed to set comment on repository `{repo.repo_name}`'
         )
 
 
 @jsonrpc_method()
 def get_repo_comments(request, apiuser, repoid,
                       commit_id=Optional(None), comment_type=Optional(None),
                       userid=Optional(None)):
     """
     Get all comments for a repository
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param commit_id: Optionally filter the comments by the commit_id
     :type commit_id: Optional(str), default: None
     :param comment_type: Optionally filter the comments by the comment_type
         one of: 'note', 'todo'
     :type comment_type: Optional(str), default: None
     :param userid: Optionally filter the comments by the author of comment
     :type userid: Optional(str or int), Default: None
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : [
                 {
                   "comment_author": <USER_DETAILS>,
                   "comment_created_on": "2017-02-01T14:38:16.309",
                   "comment_f_path": "file.txt",
                   "comment_id": 282,
                   "comment_lineno": "n1",
                   "comment_resolved_by": null,
                   "comment_status": [],
                   "comment_text": "This file needs a header",
                   "comment_type": "todo",
                   "comment_last_version: 0
                 }
             ],
             "error" :  null
         }
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.read', 'repository.write', 'repository.admin')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     commit_id = Optional.extract(commit_id)
 
     userid = Optional.extract(userid)
     if userid:
         user = get_user_or_error(userid)
     else:
         user = None
 
     comment_type = Optional.extract(comment_type)
     if comment_type and comment_type not in ChangesetComment.COMMENT_TYPES:
         raise JSONRPCError(
                 'comment_type must be one of `{}` got {}'.format(
                     ChangesetComment.COMMENT_TYPES, comment_type)
             )
 
     comments = CommentsModel().get_repository_comments(
         repo=repo, comment_type=comment_type, user=user, commit_id=commit_id)
     return comments
 
 
 @jsonrpc_method()
 def get_comment(request, apiuser, comment_id):
     """
     Get single comment from repository or pull_request
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param comment_id: comment id found in the URL of comment
     :type comment_id: str or int
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : {
                   "comment_author": <USER_DETAILS>,
                   "comment_created_on": "2017-02-01T14:38:16.309",
                   "comment_f_path": "file.txt",
                   "comment_id": 282,
                   "comment_lineno": "n1",
                   "comment_resolved_by": null,
                   "comment_status": [],
                   "comment_text": "This file needs a header",
                   "comment_type": "todo",
                   "comment_last_version: 0
             },
             "error" :  null
         }
 
     """
 
     comment = ChangesetComment.get(comment_id)
     if not comment:
         raise JSONRPCError(f'comment `{comment_id}` does not exist')
 
     perms = ('repository.read', 'repository.write', 'repository.admin')
     has_comment_perm = HasRepoPermissionAnyApi(*perms)\
         (user=apiuser, repo_name=comment.repo.repo_name)
 
     if not has_comment_perm:
         raise JSONRPCError(f'comment `{comment_id}` does not exist')
 
     return comment
 
 
 @jsonrpc_method()
 def edit_comment(request, apiuser, message, comment_id, version,
                  userid=Optional(OAttr('apiuser'))):
     """
     Edit comment on the pull request or commit,
     specified by the `comment_id` and version. Initially version should be 0
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param comment_id: Specify the comment_id for editing
     :type comment_id: int
     :param version: version of the comment that will be created, starts from 0
     :type version: int
     :param message: The text content of the comment.
     :type message: str
     :param userid: Comment on the pull request as this user
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             "comment": "<comment data>",
             "version": "<Integer>",
         },
         error :  null
     """
 
     auth_user = apiuser
     comment = ChangesetComment.get(comment_id)
     if not comment:
         raise JSONRPCError(f'comment `{comment_id}` does not exist')
 
     is_super_admin = has_superadmin_permission(apiuser)
     is_repo_admin = HasRepoPermissionAnyApi('repository.admin')\
         (user=apiuser, repo_name=comment.repo.repo_name)
 
     if not isinstance(userid, Optional):
         if is_super_admin or is_repo_admin:
             apiuser = get_user_or_error(userid)
             auth_user = apiuser.AuthUser()
         else:
             raise JSONRPCError('userid is not the same as your user')
 
     comment_author = comment.author.user_id == auth_user.user_id
 
     if comment.immutable:
         raise JSONRPCError("Immutable comment cannot be edited")
 
     if not (is_super_admin or is_repo_admin or comment_author):
         raise JSONRPCError("you don't have access to edit this comment")
 
     try:
         comment_history = CommentsModel().edit(
             comment_id=comment_id,
             text=message,
             auth_user=auth_user,
             version=version,
         )
         Session().commit()
     except CommentVersionMismatch:
         raise JSONRPCError(
             f'comment ({comment_id}) version ({version}) mismatch'
         )
     if not comment_history and not message:
         raise JSONRPCError(
             f"comment ({comment_id}) can't be changed with empty string"
         )
 
     if comment.pull_request:
         pull_request = comment.pull_request
         PullRequestModel().trigger_pull_request_hook(
             pull_request, apiuser, 'comment_edit',
             data={'comment': comment})
     else:
         db_repo = comment.repo
         commit_id = comment.revision
         commit = db_repo.get_commit(commit_id)
         CommentsModel().trigger_commit_comment_hook(
             db_repo, apiuser, 'edit',
             data={'comment': comment, 'commit': commit})
 
     data = {
         'comment': comment,
         'version': comment_history.version if comment_history else None,
     }
     return data
 
 
 # TODO(marcink): write this with all required logic for deleting a comments in PR or commits
 # @jsonrpc_method()
 # def delete_comment(request, apiuser, comment_id):
 #     auth_user = apiuser
 #
 #     comment = ChangesetComment.get(comment_id)
 #     if not comment:
 #         raise JSONRPCError('comment `%s` does not exist' % (comment_id,))
 #
 #     is_super_admin = has_superadmin_permission(apiuser)
 #     is_repo_admin = HasRepoPermissionAnyApi('repository.admin')\
 #         (user=apiuser, repo_name=comment.repo.repo_name)
 #
 #     comment_author = comment.author.user_id == auth_user.user_id
 #     if not (comment.immutable is False and (is_super_admin or is_repo_admin) or comment_author):
 #         raise JSONRPCError("you don't have access to edit this comment")
 
 @jsonrpc_method()
 def grant_user_permission(request, apiuser, repoid, userid, perm):
     """
     Grant permissions for the specified user on the given repository,
     or update existing permissions if found.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param userid: Set the user name.
     :type userid: str
     :param perm: Set the user permissions, using the following format
         ``(repository.(none|read|write|admin))``
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     user = get_user_or_error(userid)
     perm = get_perm_or_error(perm)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     perm_additions = [[user.user_id, perm.permission_name, "user"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_additions=perm_additions, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Granted perm: `{}` for user: `{}` in repo: `{}`'.format(
                 perm.permission_name, user.username, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying edit permissions for repo")
         raise JSONRPCError(
             'failed to edit permission for user: `{}` in repo: `{}`'.format(
                 userid, repoid
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_permission(request, apiuser, repoid, userid):
     """
     Revoke permission for a user on the specified repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param userid: Set the user name of revoked user.
     :type userid: str or int
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     user = get_user_or_error(userid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     perm_deletions = [[user.user_id, None, "user"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_deletions=perm_deletions, cur_user=user)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Revoked perm for user: `{}` in repo: `{}`'.format(
                 user.username, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying revoke permissions to repo")
         raise JSONRPCError(
             'failed to edit permission for user: `{}` in repo: `{}`'.format(
                 userid, repoid
             )
         )
 
 
 @jsonrpc_method()
 def grant_user_group_permission(request, apiuser, repoid, usergroupid, perm):
     """
     Grant permission for a user group on the specified repository,
     or update existing permissions.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param usergroupid: Specify the ID of the user group.
     :type usergroupid: str or int
     :param perm: Set the user group permissions using the following
         format: (repository.(none|read|write|admin))
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
         "success": true
 
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     perm = get_perm_or_error(perm)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
                 f'user group `{usergroupid}` does not exist')
 
     perm_additions = [[user_group.users_group_id, perm.permission_name, "user_group"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_additions=perm_additions, cur_user=apiuser)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Granted perm: `%s` for user group: `%s` in '
                    'repo: `%s`' % (
                        perm.permission_name, user_group.users_group_name,
                        repo.repo_name
                    ),
             'success': True
         }
     except Exception:
         log.exception(
             "Exception occurred while trying change permission on repo")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'repo: `%s`' % (
                 usergroupid, repo.repo_name
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_group_permission(request, apiuser, repoid, usergroupid):
     """
     Revoke the permissions of a user group on a given repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param usergroupid: Specify the user group ID.
     :type usergroupid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
                 f'user group `{usergroupid}` does not exist')
 
     perm_deletions = [[user_group.users_group_id, None, "user_group"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_deletions=perm_deletions, cur_user=apiuser)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Revoked perm for user group: `{}` in repo: `{}`'.format(
                 user_group.users_group_name, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying revoke "
                       "user group permission on repo")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'repo: `%s`' % (
                 user_group.users_group_name, repo.repo_name
             )
         )
 
 
 @jsonrpc_method()
 def pull(request, apiuser, repoid, remote_uri=Optional(None)):
     """
     Triggers a pull on the given repository from a remote location. You
     can use this to keep remote repositories up-to-date.
 
     This command can only be run using an |authtoken| with admin
     rights to the specified repository. For more information,
     see :ref:`config-token-ref`.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param remote_uri: Optional remote URI to pass in for pull
     :type remote_uri: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Pulled from url `<remote_url>` on repo `<repository name>`"
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to push changes from `<remote_url>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     remote_uri = Optional.extract(remote_uri)
     remote_uri_display = remote_uri or repo.clone_uri_hidden
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         ScmModel().pull_changes(
             repo.repo_name, apiuser.username, remote_uri=remote_uri)
         return {
             'msg': 'Pulled from url `{}` on repo `{}`'.format(
                 remote_uri_display, repo.repo_name),
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception occurred while trying to "
                       "pull changes from remote location")
         raise JSONRPCError(
             'Unable to pull changes from `%s`' % remote_uri_display
         )
 
 
 @jsonrpc_method()
 def strip(request, apiuser, repoid, revision, branch):
     """
     Strips the given revision from the specified repository.
 
     * This will remove the revision and all of its decendants.
 
     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 repoid: The repository name or repository ID.
     :type repoid: str or int
     :param revision: The revision you wish to strip.
     :type revision: str
     :param branch: The branch from which to strip the revision.
     :type branch: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "'Stripped commit <commit_hash> from repo `<repository name>`'"
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to strip commit <commit_hash> from repo `<repository name>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         ScmModel().strip(repo, revision, branch)
         audit_logger.store_api(
             'repo.commit.strip', action_data={'commit_id': revision},
             repo=repo,
             user=apiuser, commit=True)
 
         return {
             'msg': 'Stripped commit {} from repo `{}`'.format(
                 revision, repo.repo_name),
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception while trying to strip")
         raise JSONRPCError(
             'Unable to strip commit {} from repo `{}`'.format(
                 revision, repo.repo_name)
         )
 
 
 @jsonrpc_method()
 def get_repo_settings(request, apiuser, repoid, key=Optional(None)):
     """
     Returns all settings for a repository. If key is given it only returns the
     setting identified by the key or null.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param key: Key of the setting to return.
     :type: key: Optional(str)
 
     Example output:
 
     .. code-block:: bash
 
         {
             "error": null,
             "id": 237,
             "result": {
                 "extensions_largefiles": true,
                 "extensions_evolve": true,
                 "hooks_changegroup_push_logger": true,
                 "hooks_changegroup_repo_size": false,
                 "hooks_outgoing_pull_logger": true,
                 "phases_publish": "True",
                 "rhodecode_hg_use_rebase_for_merging": true,
                 "rhodecode_pr_merge_enabled": true,
                 "rhodecode_use_outdated_comments": true
             }
         }
     """
 
     # Restrict access to this api method to super-admins, and repo admins only.
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         settings_model = VcsSettingsModel(repo=repo)
         settings = settings_model.get_global_settings()
         settings.update(settings_model.get_repo_settings())
 
         # If only a single setting is requested fetch it from all settings.
         key = Optional.extract(key)
         if key is not None:
             settings = settings.get(key, None)
     except Exception:
         msg = f'Failed to fetch settings for repository `{repoid}`'
         log.exception(msg)
         raise JSONRPCError(msg)
 
     return settings
 
 
 @jsonrpc_method()
 def set_repo_settings(request, apiuser, repoid, settings):
     """
     Update repository settings. Returns true on success.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param settings: The new settings for the repository.
     :type: settings: dict
 
     Example output:
 
     .. code-block:: bash
 
         {
             "error": null,
             "id": 237,
             "result": true
         }
     """
     # Restrict access to this api method to super-admins, and repo admins only.
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     if type(settings) is not dict:
         raise JSONRPCError('Settings have to be a JSON Object.')
 
     try:
         settings_model = VcsSettingsModel(repo=repoid)
 
         # Merge global, repo and incoming settings.
         new_settings = settings_model.get_global_settings()
         new_settings.update(settings_model.get_repo_settings())
         new_settings.update(settings)
 
         # Update the settings.
         inherit_global_settings = new_settings.get(
             'inherit_global_settings', False)
         settings_model.create_or_update_repo_settings(
             new_settings, inherit_global_settings=inherit_global_settings)
         Session().commit()
     except Exception:
         msg = f'Failed to update settings for repository `{repoid}`'
         log.exception(msg)
         raise JSONRPCError(msg)
 
     # Indicate success.
     return True
 
 
 @jsonrpc_method()
 def maintenance(request, apiuser, repoid):
     """
     Triggers a maintenance on the given repository.
 
     This command can only be run using an |authtoken| with admin
     rights to the specified repository. For more information,
     see :ref:`config-token-ref`.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "executed maintenance command",
         "executed_actions": [
            <action_message>, <action_message2>...
         ],
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to execute maintenance on `<reponame>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         maintenance = repo_maintenance.RepoMaintenance()
         executed_actions = maintenance.execute(repo)
 
         return {
             'msg': 'executed maintenance command',
             'executed_actions': executed_actions,
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception occurred while trying to run maintenance")
         raise JSONRPCError(
             'Unable to execute maintenance on `%s`' % repo.repo_name)

rhodecode/apps/my_account/tests/test_my_account_password.py

@@ -1,134 +1,132 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 import mock
 
-from rhodecode.apps._base import ADMIN_PREFIX
-from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import check_password
 from rhodecode.model.meta import Session
 from rhodecode.model.user import UserModel
-from rhodecode.tests import assert_session_flash
-from rhodecode.tests.fixture import Fixture, TestController, error_function
+from rhodecode.tests import assert_session_flash, TestController
+from rhodecode.tests.fixture import Fixture, error_function
 from rhodecode.tests.routes import route_path
 
 fixture = Fixture()
 
 
 test_user_1 = 'testme'
 test_user_1_password = '0jd83nHNS/d23n'
 
 
 class TestMyAccountPassword(TestController):
     def test_valid_change_password(self, user_util):
         new_password = 'my_new_valid_password'
         user = user_util.create_user(password=test_user_1_password)
         self.log_user(user.username, test_user_1_password)
 
         form_data = [
             ('current_password', test_user_1_password),
             ('__start__', 'new_password:mapping'),
             ('new_password', new_password),
             ('new_password-confirm', new_password),
             ('__end__', 'new_password:mapping'),
             ('csrf_token', self.csrf_token),
         ]
         response = self.app.post(
             route_path('my_account_password_update'), form_data).follow()
         assert 'Successfully updated password' in response
 
         # check_password depends on user being in session
         Session().add(user)
         try:
             assert check_password(new_password, user.password)
         finally:
             Session().expunge(user)
 
     @pytest.mark.parametrize('current_pw, new_pw, confirm_pw', [
         ('', 'abcdef123', 'abcdef123'),
         ('wrong_pw', 'abcdef123', 'abcdef123'),
         (test_user_1_password, test_user_1_password, test_user_1_password),
         (test_user_1_password, '', ''),
         (test_user_1_password, 'abcdef123', ''),
         (test_user_1_password, '', 'abcdef123'),
         (test_user_1_password, 'not_the', 'same_pw'),
         (test_user_1_password, 'short', 'short'),
     ])
     def test_invalid_change_password(self, current_pw, new_pw, confirm_pw,
                                      user_util):
         user = user_util.create_user(password=test_user_1_password)
         self.log_user(user.username, test_user_1_password)
 
         form_data = [
             ('current_password', current_pw),
             ('__start__', 'new_password:mapping'),
             ('new_password', new_pw),
             ('new_password-confirm', confirm_pw),
             ('__end__', 'new_password:mapping'),
             ('csrf_token', self.csrf_token),
         ]
         response = self.app.post(
             route_path('my_account_password_update'), form_data)
 
         assert_response = response.assert_response()
         assert assert_response.get_elements('.error-block')
 
     @mock.patch.object(UserModel, 'update_user', error_function)
     def test_invalid_change_password_exception(self, user_util):
         user = user_util.create_user(password=test_user_1_password)
         self.log_user(user.username, test_user_1_password)
 
         form_data = [
             ('current_password', test_user_1_password),
             ('__start__', 'new_password:mapping'),
             ('new_password', '123456'),
             ('new_password-confirm', '123456'),
             ('__end__', 'new_password:mapping'),
             ('csrf_token', self.csrf_token),
         ]
         response = self.app.post(
             route_path('my_account_password_update'), form_data)
         assert_session_flash(
             response, 'Error occurred during update of user password')
 
     def test_password_is_updated_in_session_on_password_change(self, user_util):
         old_password = 'abcdef123'
         new_password = 'abcdef124'
 
         user = user_util.create_user(password=old_password)
         session = self.log_user(user.username, old_password)
         old_password_hash = session['password']
 
         form_data = [
             ('current_password', old_password),
             ('__start__', 'new_password:mapping'),
             ('new_password', new_password),
             ('new_password-confirm', new_password),
             ('__end__', 'new_password:mapping'),
             ('csrf_token', self.csrf_token),
         ]
         self.app.post(
             route_path('my_account_password_update'), form_data)
 
         response = self.app.get(route_path('home'))
         session = response.get_session_from_response()
         new_password_hash = session['rhodecode_user']['password']
 
         assert old_password_hash != new_password_hash
  No newline at end of file

rhodecode/lib/system_info.py

@@ -1,847 +1,845 b''
 # Copyright (C) 2017-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import os
 import sys
 import time
 import platform
 import collections
 import psutil
 from functools import wraps
 
 import pkg_resources
 import logging
 import resource
 
 import configparser
 
 from rc_license.models import LicenseModel
 from rhodecode.lib.str_utils import safe_str
 
 log = logging.getLogger(__name__)
 
 
 _NA = 'NOT AVAILABLE'
 _NA_FLOAT = 0.0
 
 STATE_OK = 'ok'
 STATE_ERR = 'error'
 STATE_WARN = 'warning'
 
 STATE_OK_DEFAULT = {'message': '', 'type': STATE_OK}
 
 
 registered_helpers = {}
 
 
 def register_sysinfo(func):
     """
     @register_helper
     def db_check():
         pass
 
     db_check == registered_helpers['db_check']
     """
     global registered_helpers
     registered_helpers[func.__name__] = func
 
     @wraps(func)
     def _wrapper(*args, **kwargs):
         return func(*args, **kwargs)
     return _wrapper
 
 
 # HELPERS
 def percentage(part: (int, float), whole: (int, float)):
     whole = float(whole)
     if whole > 0:
         return round(100 * float(part) / whole, 1)
     return 0.0
 
 
 def get_storage_size(storage_path):
     sizes = []
     for file_ in os.listdir(storage_path):
         storage_file = os.path.join(storage_path, file_)
         if os.path.isfile(storage_file):
             try:
                 sizes.append(os.path.getsize(storage_file))
             except OSError:
                 log.exception('Failed to get size of storage file %s', storage_file)
                 pass
 
     return sum(sizes)
 
 
 def get_resource(resource_type):
     try:
         return resource.getrlimit(resource_type)
     except Exception:
         return 'NOT_SUPPORTED'
 
 
 def get_cert_path(ini_path):
     default = '/etc/ssl/certs/ca-certificates.crt'
     control_ca_bundle = os.path.join(
         os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(ini_path)))),
         '.rccontrol-profile/etc/ca-bundle.crt')
     if os.path.isfile(control_ca_bundle):
         default = control_ca_bundle
 
     return default
 
 
 class SysInfoRes(object):
     def __init__(self, value, state=None, human_value=None):
         self.value = value
         self.state = state or STATE_OK_DEFAULT
         self.human_value = human_value or value
 
     def __json__(self):
         return {
             'value': self.value,
             'state': self.state,
             'human_value': self.human_value,
         }
 
     def get_value(self):
         return self.__json__()
 
     def __str__(self):
         return f'<SysInfoRes({self.__json__()})>'
 
 
 class SysInfo(object):
 
     def __init__(self, func_name, **kwargs):
         self.function_name = func_name
         self.value = _NA
         self.state = None
         self.kwargs = kwargs or {}
 
     def __call__(self):
         computed = self.compute(**self.kwargs)
         if not isinstance(computed, SysInfoRes):
             raise ValueError(
                 'computed value for {} is not instance of '
                 '{}, got {} instead'.format(
                     self.function_name, SysInfoRes, type(computed)))
         return computed.__json__()
 
     def __str__(self):
         return f'<SysInfo({self.function_name})>'
 
     def compute(self, **kwargs):
         return self.function_name(**kwargs)
 
 
 # SysInfo functions
 @register_sysinfo
 def python_info():
     value = dict(version=f'{platform.python_version()}:{platform.python_implementation()}',
                  executable=sys.executable)
     return SysInfoRes(value=value)
 
 
 @register_sysinfo
 def py_modules():
     mods = dict([(p.project_name, {'version': p.version, 'location': p.location})
                  for p in pkg_resources.working_set])
 
     value = sorted(mods.items(), key=lambda k: k[0].lower())
     return SysInfoRes(value=value)
 
 
 @register_sysinfo
 def platform_type():
     from rhodecode.lib.utils import generate_platform_uuid
 
     value = dict(
         name=safe_str(platform.platform()),
         uuid=generate_platform_uuid()
     )
     return SysInfoRes(value=value)
 
 
 @register_sysinfo
 def locale_info():
     import locale
 
     def safe_get_locale(locale_name):
         try:
             locale.getlocale(locale_name)
         except TypeError:
             return f'FAILED_LOCALE_GET:{locale_name}'
 
     value = dict(
         locale_default=locale.getlocale(),
         locale_lc_all=safe_get_locale(locale.LC_ALL),
         locale_lc_ctype=safe_get_locale(locale.LC_CTYPE),
         lang_env=os.environ.get('LANG'),
         lc_all_env=os.environ.get('LC_ALL'),
         local_archive_env=os.environ.get('LOCALE_ARCHIVE'),
     )
     human_value = \
         f"LANG: {value['lang_env']}, \
           locale LC_ALL: {value['locale_lc_all']},  \
           locale LC_CTYPE: {value['locale_lc_ctype']},  \
           Default locales: {value['locale_default']}"
 
     return SysInfoRes(value=value, human_value=human_value)
 
 
 @register_sysinfo
 def ulimit_info():
     data = collections.OrderedDict([
         ('cpu time (seconds)', get_resource(resource.RLIMIT_CPU)),
         ('file size', get_resource(resource.RLIMIT_FSIZE)),
         ('stack size', get_resource(resource.RLIMIT_STACK)),
         ('core file size', get_resource(resource.RLIMIT_CORE)),
         ('address space size', get_resource(resource.RLIMIT_AS)),
         ('locked in mem size', get_resource(resource.RLIMIT_MEMLOCK)),
         ('heap size', get_resource(resource.RLIMIT_DATA)),
         ('rss size', get_resource(resource.RLIMIT_RSS)),
         ('number of processes', get_resource(resource.RLIMIT_NPROC)),
         ('open files', get_resource(resource.RLIMIT_NOFILE)),
     ])
 
     text = ', '.join(f'{k}:{v}' for k, v in data.items())
 
     value = {
         'limits': data,
         'text': text,
     }
     return SysInfoRes(value=value)
 
 
 @register_sysinfo
 def uptime():
     from rhodecode.lib.helpers import age, time_to_datetime
     from rhodecode.translation import TranslationString
 
     value = dict(boot_time=0, uptime=0, text='')
     state = STATE_OK_DEFAULT
 
     boot_time = psutil.boot_time()
     value['boot_time'] = boot_time
     value['uptime'] = time.time() - boot_time
 
     date_or_age = age(time_to_datetime(boot_time))
     if isinstance(date_or_age, TranslationString):
         date_or_age = date_or_age.interpolate()
 
     human_value = value.copy()
     human_value['boot_time'] = time_to_datetime(boot_time)
     human_value['uptime'] = age(time_to_datetime(boot_time), show_suffix=False)
 
     human_value['text'] = f'Server started {date_or_age}'
     return SysInfoRes(value=value, human_value=human_value)
 
 
 @register_sysinfo
 def memory():
     from rhodecode.lib.helpers import format_byte_size_binary
     value = dict(available=0, used=0, used_real=0, cached=0, percent=0,
                  percent_used=0, free=0, inactive=0, active=0, shared=0,
                  total=0, buffers=0, text='')
 
     state = STATE_OK_DEFAULT
 
     value.update(dict(psutil.virtual_memory()._asdict()))
     value['used_real'] = value['total'] - value['available']
-    value['percent_used'] = psutil._common.usage_percent(
-        value['used_real'], value['total'], 1)
+    value['percent_used'] = psutil._common.usage_percent(value['used_real'], value['total'], 1)
 
     human_value = value.copy()
     human_value['text'] = '{}/{}, {}% used'.format(
         format_byte_size_binary(value['used_real']),
         format_byte_size_binary(value['total']),
         value['percent_used'])
 
     keys = list(value.keys())[::]
     keys.pop(keys.index('percent'))
     keys.pop(keys.index('percent_used'))
     keys.pop(keys.index('text'))
     for k in keys:
         human_value[k] = format_byte_size_binary(value[k])
 
     if state['type'] == STATE_OK and value['percent_used'] > 90:
         msg = 'Critical: your available RAM memory is very low.'
         state = {'message': msg, 'type': STATE_ERR}
 
     elif state['type'] == STATE_OK and value['percent_used'] > 70:
         msg = 'Warning: your available RAM memory is running low.'
         state = {'message': msg, 'type': STATE_WARN}
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def machine_load():
     value = {'1_min': _NA_FLOAT, '5_min': _NA_FLOAT, '15_min': _NA_FLOAT, 'text': ''}
     state = STATE_OK_DEFAULT
 
     # load averages
     if hasattr(psutil.os, 'getloadavg'):
         value.update(dict(
             list(zip(['1_min', '5_min', '15_min'], psutil.os.getloadavg()))
         ))
 
     human_value = value.copy()
     human_value['text'] = '1min: {}, 5min: {}, 15min: {}'.format(
         value['1_min'], value['5_min'], value['15_min'])
 
     if state['type'] == STATE_OK and value['15_min'] > 5.0:
         msg = 'Warning: your machine load is very high.'
         state = {'message': msg, 'type': STATE_WARN}
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def cpu():
     value = {'cpu': 0, 'cpu_count': 0, 'cpu_usage': []}
     state = STATE_OK_DEFAULT
 
     value['cpu'] = psutil.cpu_percent(0.5)
     value['cpu_usage'] = psutil.cpu_percent(0.5, percpu=True)
     value['cpu_count'] = psutil.cpu_count()
 
     human_value = value.copy()
-    human_value['text'] = '{} cores at {} %'.format(
-        value['cpu_count'], value['cpu'])
+    human_value['text'] = '{} cores at {} %'.format(value['cpu_count'], value['cpu'])
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def storage():
     from rhodecode.lib.helpers import format_byte_size_binary
     from rhodecode.model.settings import VcsSettingsModel
     path = VcsSettingsModel().get_repos_location()
 
     value = dict(percent=0, used=0, total=0, path=path, text='')
     state = STATE_OK_DEFAULT
 
     try:
         value.update(dict(psutil.disk_usage(path)._asdict()))
     except Exception as e:
         log.exception('Failed to fetch disk info')
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['used'] = format_byte_size_binary(value['used'])
     human_value['total'] = format_byte_size_binary(value['total'])
     human_value['text'] = "{}/{}, {}% used".format(
         format_byte_size_binary(value['used']),
         format_byte_size_binary(value['total']),
         value['percent'])
 
     if state['type'] == STATE_OK and value['percent'] > 90:
         msg = 'Critical: your disk space is very low.'
         state = {'message': msg, 'type': STATE_ERR}
 
     elif state['type'] == STATE_OK and value['percent'] > 70:
         msg = 'Warning: your disk space is running low.'
         state = {'message': msg, 'type': STATE_WARN}
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def storage_inodes():
     from rhodecode.model.settings import VcsSettingsModel
     path = VcsSettingsModel().get_repos_location()
 
     value = dict(percent=0.0, free=0, used=0, total=0, path=path, text='')
     state = STATE_OK_DEFAULT
 
     try:
         i_stat = os.statvfs(path)
         value['free'] = i_stat.f_ffree
         value['used'] = i_stat.f_files-i_stat.f_favail
         value['total'] = i_stat.f_files
         value['percent'] = percentage(value['used'], value['total'])
     except Exception as e:
         log.exception('Failed to fetch disk inodes info')
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['text'] = "{}/{}, {}% used".format(
         value['used'], value['total'], value['percent'])
 
     if state['type'] == STATE_OK and value['percent'] > 90:
         msg = 'Critical: your disk free inodes are very low.'
         state = {'message': msg, 'type': STATE_ERR}
 
     elif state['type'] == STATE_OK and value['percent'] > 70:
         msg = 'Warning: your disk free inodes are running low.'
         state = {'message': msg, 'type': STATE_WARN}
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def storage_archives():
     import rhodecode
     from rhodecode.lib.utils import safe_str
     from rhodecode.lib.helpers import format_byte_size_binary
 
     msg = 'Archive cache storage is controlled by ' \
           'archive_cache.store_dir=/path/to/cache option in the .ini file'
     path = safe_str(rhodecode.CONFIG.get('archive_cache.store_dir', msg))
 
     value = dict(percent=0, used=0, total=0, items=0, path=path, text='')
     state = STATE_OK_DEFAULT
     try:
         items_count = 0
         used = 0
         for root, dirs, files in os.walk(path):
             if root == path:
                 items_count = len(dirs)
 
             for f in files:
                 try:
                     used += os.path.getsize(os.path.join(root, f))
                 except OSError:
                     pass
         value.update({
             'percent': 100,
             'used': used,
             'total': used,
             'items': items_count
         })
 
     except Exception as e:
         log.exception('failed to fetch archive cache storage')
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['used'] = format_byte_size_binary(value['used'])
     human_value['total'] = format_byte_size_binary(value['total'])
     human_value['text'] = "{} ({} items)".format(
         human_value['used'], value['items'])
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def storage_gist():
     from rhodecode.model.gist import GIST_STORE_LOC
     from rhodecode.model.settings import VcsSettingsModel
     from rhodecode.lib.utils import safe_str
     from rhodecode.lib.helpers import format_byte_size_binary
     path = safe_str(os.path.join(
         VcsSettingsModel().get_repos_location(), GIST_STORE_LOC))
 
     # gist storage
     value = dict(percent=0, used=0, total=0, items=0, path=path, text='')
     state = STATE_OK_DEFAULT
 
     try:
         items_count = 0
         used = 0
         for root, dirs, files in os.walk(path):
             if root == path:
                 items_count = len(dirs)
 
             for f in files:
                 try:
                     used += os.path.getsize(os.path.join(root, f))
                 except OSError:
                     pass
         value.update({
             'percent': 100,
             'used': used,
             'total': used,
             'items': items_count
         })
     except Exception as e:
         log.exception('failed to fetch gist storage items')
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['used'] = format_byte_size_binary(value['used'])
     human_value['total'] = format_byte_size_binary(value['total'])
     human_value['text'] = "{} ({} items)".format(
         human_value['used'], value['items'])
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def storage_temp():
     import tempfile
     from rhodecode.lib.helpers import format_byte_size_binary
 
     path = tempfile.gettempdir()
     value = dict(percent=0, used=0, total=0, items=0, path=path, text='')
     state = STATE_OK_DEFAULT
 
     if not psutil:
         return SysInfoRes(value=value, state=state)
 
     try:
         value.update(dict(psutil.disk_usage(path)._asdict()))
     except Exception as e:
         log.exception('Failed to fetch temp dir info')
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['used'] = format_byte_size_binary(value['used'])
     human_value['total'] = format_byte_size_binary(value['total'])
     human_value['text'] = "{}/{}, {}% used".format(
         format_byte_size_binary(value['used']),
         format_byte_size_binary(value['total']),
         value['percent'])
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def search_info():
     import rhodecode
     from rhodecode.lib.index import searcher_from_config
 
     backend = rhodecode.CONFIG.get('search.module', '')
     location = rhodecode.CONFIG.get('search.location', '')
 
     try:
         searcher = searcher_from_config(rhodecode.CONFIG)
         searcher = searcher.__class__.__name__
     except Exception:
         searcher = None
 
     value = dict(
         backend=backend, searcher=searcher, location=location, text='')
     state = STATE_OK_DEFAULT
 
     human_value = value.copy()
     human_value['text'] = "backend:`{}`".format(human_value['backend'])
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def git_info():
     from rhodecode.lib.vcs.backends import git
     state = STATE_OK_DEFAULT
     value = human_value = ''
     try:
         value = git.discover_git_version(raise_on_exc=True)
         human_value = f'version reported from VCSServer: {value}'
     except Exception as e:
         state = {'message': str(e), 'type': STATE_ERR}
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def hg_info():
     from rhodecode.lib.vcs.backends import hg
     state = STATE_OK_DEFAULT
     value = human_value = ''
     try:
         value = hg.discover_hg_version(raise_on_exc=True)
         human_value = f'version reported from VCSServer: {value}'
     except Exception as e:
         state = {'message': str(e), 'type': STATE_ERR}
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def svn_info():
     from rhodecode.lib.vcs.backends import svn
     state = STATE_OK_DEFAULT
     value = human_value = ''
     try:
         value = svn.discover_svn_version(raise_on_exc=True)
         human_value = f'version reported from VCSServer: {value}'
     except Exception as e:
         state = {'message': str(e), 'type': STATE_ERR}
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def vcs_backends():
     import rhodecode
     value = rhodecode.CONFIG.get('vcs.backends')
     human_value = 'Enabled backends in order: {}'.format(','.join(value))
     return SysInfoRes(value=value, human_value=human_value)
 
 
 @register_sysinfo
 def vcs_server():
     import rhodecode
     from rhodecode.lib.vcs.backends import get_vcsserver_service_data
 
     server_url = rhodecode.CONFIG.get('vcs.server')
     enabled = rhodecode.CONFIG.get('vcs.server.enable')
     protocol = rhodecode.CONFIG.get('vcs.server.protocol') or 'http'
     state = STATE_OK_DEFAULT
     version = None
     workers = 0
 
     try:
         data = get_vcsserver_service_data()
         if data and 'version' in data:
             version = data['version']
 
         if data and 'config' in data:
             conf = data['config']
             workers = conf.get('workers', 'NOT AVAILABLE')
 
         connection = 'connected'
     except Exception as e:
         connection = 'failed'
         state = {'message': str(e), 'type': STATE_ERR}
 
     value = dict(
         url=server_url,
         enabled=enabled,
         protocol=protocol,
         connection=connection,
         version=version,
         text='',
     )
 
     human_value = value.copy()
     human_value['text'] = \
         '{url}@ver:{ver} via {mode} mode[workers:{workers}], connection:{conn}'.format(
             url=server_url, ver=version, workers=workers, mode=protocol,
             conn=connection)
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def vcs_server_config():
     from rhodecode.lib.vcs.backends import get_vcsserver_service_data
     state = STATE_OK_DEFAULT
 
     value = {}
     try:
         data = get_vcsserver_service_data()
         value = data['app_config']
     except Exception as e:
         state = {'message': str(e), 'type': STATE_ERR}
 
     human_value = value.copy()
     human_value['text'] = 'VCS Server config'
 
     return SysInfoRes(value=value, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def rhodecode_app_info():
     import rhodecode
     edition = rhodecode.CONFIG.get('rhodecode.edition')
 
     value = dict(
         rhodecode_version=rhodecode.__version__,
         rhodecode_lib_path=os.path.abspath(rhodecode.__file__),
         text=''
     )
     human_value = value.copy()
     human_value['text'] = 'RhodeCode {edition}, version {ver}'.format(
         edition=edition, ver=value['rhodecode_version']
     )
     return SysInfoRes(value=value, human_value=human_value)
 
 
 @register_sysinfo
 def rhodecode_config():
     import rhodecode
     path = rhodecode.CONFIG.get('__file__')
     rhodecode_ini_safe = rhodecode.CONFIG.copy()
     cert_path = get_cert_path(path)
 
     try:
         config = configparser.ConfigParser()
         config.read(path)
         parsed_ini = config
         if parsed_ini.has_section('server:main'):
             parsed_ini = dict(parsed_ini.items('server:main'))
     except Exception:
         log.exception('Failed to read .ini file for display')
         parsed_ini = {}
 
     rhodecode_ini_safe['server:main'] = parsed_ini
 
     blacklist = [
         f'rhodecode_{LicenseModel.LICENSE_DB_KEY}',
         'routes.map',
         'sqlalchemy.db1.url',
         'channelstream.secret',
         'beaker.session.secret',
         'rhodecode.encrypted_values.secret',
         'rhodecode_auth_github_consumer_key',
         'rhodecode_auth_github_consumer_secret',
         'rhodecode_auth_google_consumer_key',
         'rhodecode_auth_google_consumer_secret',
         'rhodecode_auth_bitbucket_consumer_secret',
         'rhodecode_auth_bitbucket_consumer_key',
         'rhodecode_auth_twitter_consumer_secret',
         'rhodecode_auth_twitter_consumer_key',
 
         'rhodecode_auth_twitter_secret',
         'rhodecode_auth_github_secret',
         'rhodecode_auth_google_secret',
         'rhodecode_auth_bitbucket_secret',
 
         'appenlight.api_key',
         ('app_conf', 'sqlalchemy.db1.url')
     ]
     for k in blacklist:
         if isinstance(k, tuple):
             section, key = k
             if section in rhodecode_ini_safe:
                 rhodecode_ini_safe[section] = '**OBFUSCATED**'
         else:
             rhodecode_ini_safe.pop(k, None)
 
     # TODO: maybe put some CONFIG checks here ?
     return SysInfoRes(value={'config': rhodecode_ini_safe,
                              'path': path, 'cert_path': cert_path})
 
 
 @register_sysinfo
 def database_info():
     import rhodecode
     from sqlalchemy.engine import url as engine_url
     from rhodecode.model import meta
     from rhodecode.model.meta import Session
     from rhodecode.model.db import DbMigrateVersion
 
     state = STATE_OK_DEFAULT
 
     db_migrate = DbMigrateVersion.query().filter(
         DbMigrateVersion.repository_id == 'rhodecode_db_migrations').one()
 
     db_url_obj = engine_url.make_url(rhodecode.CONFIG['sqlalchemy.db1.url'])
 
     try:
         engine = meta.get_engine()
         db_server_info = engine.dialect._get_server_version_info(
             Session.connection(bind=engine))
         db_version = '.'.join(map(str, db_server_info))
     except Exception:
         log.exception('failed to fetch db version')
         db_version = 'UNKNOWN'
 
     db_info = dict(
         migrate_version=db_migrate.version,
         type=db_url_obj.get_backend_name(),
         version=db_version,
         url=repr(db_url_obj)
     )
     current_version = db_migrate.version
     expected_version = rhodecode.__dbversion__
     if state['type'] == STATE_OK and current_version != expected_version:
         msg = 'Critical: database schema mismatch, ' \
               'expected version {}, got {}. ' \
               'Please run migrations on your database.'.format(
             expected_version, current_version)
         state = {'message': msg, 'type': STATE_ERR}
 
     human_value = db_info.copy()
     human_value['url'] = "{} @ migration version: {}".format(
         db_info['url'], db_info['migrate_version'])
     human_value['version'] = "{} {}".format(db_info['type'], db_info['version'])
     return SysInfoRes(value=db_info, state=state, human_value=human_value)
 
 
 @register_sysinfo
 def server_info(environ):
     import rhodecode
     from rhodecode.lib.base import get_server_ip_addr, get_server_port
 
     value = {
         'server_ip': '{}:{}'.format(
             get_server_ip_addr(environ, log_errors=False),
             get_server_port(environ)
         ),
         'server_id': rhodecode.CONFIG.get('instance_id'),
     }
     return SysInfoRes(value=value)
 
 
 @register_sysinfo
 def usage_info():
-    from rhodecode.model.db import User, Repository
+    from rhodecode.model.db import User, Repository, true
     value = {
         'users': User.query().count(),
-        'users_active': User.query().filter(User.active == True).count(),
+        'users_active': User.query().filter(User.active == true()).count(),
         'repositories': Repository.query().count(),
         'repository_types': {
             'hg': Repository.query().filter(
                 Repository.repo_type == 'hg').count(),
             'git': Repository.query().filter(
                 Repository.repo_type == 'git').count(),
             'svn': Repository.query().filter(
                 Repository.repo_type == 'svn').count(),
         },
     }
     return SysInfoRes(value=value)
 
 
 def get_system_info(environ):
     environ = environ or {}
     return {
         'rhodecode_app': SysInfo(rhodecode_app_info)(),
         'rhodecode_config': SysInfo(rhodecode_config)(),
         'rhodecode_usage': SysInfo(usage_info)(),
         'python': SysInfo(python_info)(),
         'py_modules': SysInfo(py_modules)(),
 
         'platform': SysInfo(platform_type)(),
         'locale': SysInfo(locale_info)(),
         'server': SysInfo(server_info, environ=environ)(),
         'database': SysInfo(database_info)(),
         'ulimit': SysInfo(ulimit_info)(),
         'storage': SysInfo(storage)(),
         'storage_inodes': SysInfo(storage_inodes)(),
         'storage_archive': SysInfo(storage_archives)(),
         'storage_gist': SysInfo(storage_gist)(),
         'storage_temp': SysInfo(storage_temp)(),
 
         'search': SysInfo(search_info)(),
 
         'uptime': SysInfo(uptime)(),
         'load': SysInfo(machine_load)(),
         'cpu': SysInfo(cpu)(),
         'memory': SysInfo(memory)(),
 
         'vcs_backends': SysInfo(vcs_backends)(),
         'vcs_server': SysInfo(vcs_server)(),
 
         'vcs_server_config': SysInfo(vcs_server_config)(),
 
         'git': SysInfo(git_info)(),
         'hg': SysInfo(hg_info)(),
         'svn': SysInfo(svn_info)(),
     }
 
 
 def load_system_info(key):
     """
     get_sys_info('vcs_server')
     get_sys_info('database')
     """
     return SysInfo(registered_helpers[key])()

rhodecode/lib/utils2.py

@@ -1,987 +1,987 b''
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 """
 Some simple helper functions
 """
 
 import collections
 import datetime
 import dateutil.relativedelta
 import logging
 import re
 import sys
 import time
 import urllib.request
 import urllib.parse
 import urllib.error
 import urlobject
 import uuid
 import getpass
 import socket
 import errno
 import random
 import functools
 from contextlib import closing
 
 import pygments.lexers
 import sqlalchemy
 import sqlalchemy.event
 import sqlalchemy.engine.url
 import sqlalchemy.exc
 import sqlalchemy.sql
 import webob
 from pyramid.settings import asbool
 
 import rhodecode
 from rhodecode.translation import _, _pluralize
 from rhodecode.lib.str_utils import safe_str, safe_int, safe_bytes
 from rhodecode.lib.hash_utils import md5, md5_safe, sha1, sha1_safe
 from rhodecode.lib.type_utils import aslist, str2bool, StrictAttributeDict, AttributeDict
 
 
 def __get_lem(extra_mapping=None):
     """
     Get language extension map based on what's inside pygments lexers
     """
     d = collections.defaultdict(lambda: [])
 
     def __clean(s):
         s = s.lstrip('*')
         s = s.lstrip('.')
 
         if s.find('[') != -1:
             exts = []
             start, stop = s.find('['), s.find(']')
 
             for suffix in s[start + 1:stop]:
                 exts.append(s[:s.find('[')] + suffix)
             return [e.lower() for e in exts]
         else:
             return [s.lower()]
 
     for lx, t in sorted(pygments.lexers.LEXERS.items()):
         m = list(map(__clean, t[-2]))
         if m:
             m = functools.reduce(lambda x, y: x + y, m)
             for ext in m:
                 desc = lx.replace('Lexer', '')
                 d[ext].append(desc)
 
     data = dict(d)
 
     extra_mapping = extra_mapping or {}
     if extra_mapping:
         for k, v in list(extra_mapping.items()):
             if k not in data:
                 # register new mapping2lexer
                 data[k] = [v]
 
     return data
 
 
 def convert_line_endings(line: str, mode) -> str:
     """
     Converts a given line  "line end" accordingly to given mode
 
     Available modes are::
         0 - Unix
         1 - Mac
         2 - DOS
 
     :param line: given line to convert
     :param mode: mode to convert to
     :return: converted line according to mode
     """
     if mode == 0:
         line = line.replace('\r\n', '\n')
         line = line.replace('\r', '\n')
     elif mode == 1:
         line = line.replace('\r\n', '\r')
         line = line.replace('\n', '\r')
     elif mode == 2:
         line = re.sub('\r(?!\n)|(?<!\r)\n', '\r\n', line)
     return line
 
 
 def detect_mode(line: str, default) -> int:
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
 
     :param line: str line
     :param default: default
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 
 
 def remove_suffix(s, suffix):
     if s.endswith(suffix):
         s = s[:-1 * len(suffix)]
     return s
 
 
 def remove_prefix(s, prefix):
     if s.startswith(prefix):
         s = s[len(prefix):]
     return s
 
 
 def find_calling_context(ignore_modules=None, depth=4, output_writer=None, indent=True):
     """
     Look through the calling stack and return the frame which called
     this function and is part of core module ( ie. rhodecode.* )
 
     :param ignore_modules: list of modules to ignore eg. ['rhodecode.lib']
     :param depth:
     :param output_writer:
     :param indent:
 
     usage::
 
         from rhodecode.lib.utils2 import find_calling_context
 
         calling_context = find_calling_context(ignore_modules=[
             'rhodecode.lib.caching_query',
             'rhodecode.model.settings',
         ])
 
     """
     import inspect
     if not output_writer:
         try:
             from rich import print as pprint
         except ImportError:
             pprint = print
         output_writer = pprint
 
     frame = inspect.currentframe()
     cc = []
     try:
         for i in range(depth):  # current frame + 3 callers
             frame = frame.f_back
             if not frame:
                 break
 
             info = inspect.getframeinfo(frame)
             name = frame.f_globals.get('__name__')
             if name not in ignore_modules:
                 cc.insert(0, f'CALL_CONTEXT:{i}: file {info.filename}:{info.lineno} -> {info.function}')
     finally:
         # Avoids a reference cycle
         del frame
 
     output_writer('* INFO: This code was called from: *')
     for cnt, frm_info in enumerate(cc):
         if not indent:
             cnt = 1
         output_writer(' ' * cnt + frm_info)
 
 
 def ping_connection(connection, branch):
     if branch:
         # "branch" refers to a sub-connection of a connection,
         # we don't want to bother pinging on these.
         return
 
     # turn off "close with result".  This flag is only used with
     # "connectionless" execution, otherwise will be False in any case
     save_should_close_with_result = connection.should_close_with_result
     connection.should_close_with_result = False
 
     try:
         # run a SELECT 1.   use a core select() so that
         # the SELECT of a scalar value without a table is
         # appropriately formatted for the backend
         connection.scalar(sqlalchemy.sql.select([1]))
     except sqlalchemy.exc.DBAPIError as err:
         # catch SQLAlchemy's DBAPIError, which is a wrapper
         # for the DBAPI's exception.  It includes a .connection_invalidated
         # attribute which specifies if this connection is a "disconnect"
         # condition, which is based on inspection of the original exception
         # by the dialect in use.
         if err.connection_invalidated:
             # run the same SELECT again - the connection will re-validate
             # itself and establish a new connection.  The disconnect detection
             # here also causes the whole connection pool to be invalidated
             # so that all stale connections are discarded.
             connection.scalar(sqlalchemy.sql.select([1]))
         else:
             raise
     finally:
         # restore "close with result"
         connection.should_close_with_result = save_should_close_with_result
 
 
 def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
     """Custom engine_from_config functions."""
     log = logging.getLogger('sqlalchemy.engine')
     use_ping_connection = asbool(configuration.pop('sqlalchemy.db1.ping_connection', None))
     debug = asbool(configuration.pop('sqlalchemy.db1.debug_query', None))
 
     engine = sqlalchemy.engine_from_config(configuration, prefix, **kwargs)
 
     def color_sql(sql):
         color_seq = '\033[1;33m'  # This is yellow: code 33
         normal = '\x1b[0m'
         return ''.join([color_seq, sql, normal])
 
     if use_ping_connection:
         log.debug('Adding ping_connection on the engine config.')
         sqlalchemy.event.listen(engine, "engine_connect", ping_connection)
 
     if debug:
         # attach events only for debug configuration
         def before_cursor_execute(conn, cursor, statement,
                                   parameters, context, executemany):
             setattr(conn, 'query_start_time', time.time())
             log.info(color_sql(">>>>> STARTING QUERY >>>>>"))
             find_calling_context(ignore_modules=[
                 'rhodecode.lib.caching_query',
                 'rhodecode.model.settings',
             ], output_writer=log.info)
 
         def after_cursor_execute(conn, cursor, statement,
                                  parameters, context, executemany):
             delattr(conn, 'query_start_time')
 
         sqlalchemy.event.listen(engine, "before_cursor_execute", before_cursor_execute)
         sqlalchemy.event.listen(engine, "after_cursor_execute", after_cursor_execute)
 
     return engine
 
 
 def get_encryption_key(config) -> bytes:
     secret = config.get('rhodecode.encrypted_values.secret')
     default = config['beaker.session.secret']
     enc_key = secret or default
 
     return safe_bytes(enc_key)
 
 
 def age(prevdate, now=None, show_short_version=False, show_suffix=True, short_format=False):
     """
     Turns a datetime into an age string.
     If show_short_version is True, this generates a shorter string with
     an approximate age; ex. '1 day ago', rather than '1 day and 23 hours ago'.
 
     * IMPORTANT*
     Code of this function is written in special way so it's easier to
     backport it to javascript. If you mean to update it, please also update
     `jquery.timeago-extension.js` file
 
     :param prevdate: datetime object
     :param now: get current time, if not define we use
         `datetime.datetime.now()`
     :param show_short_version: if it should approximate the date and
         return a shorter string
     :param show_suffix:
     :param short_format: show short format, eg 2D instead of 2 days
     :rtype: unicode
     :returns: unicode words describing age
     """
 
     def _get_relative_delta(now, prevdate):
         base = dateutil.relativedelta.relativedelta(now, prevdate)
         return {
             'year': base.years,
             'month': base.months,
             'day': base.days,
             'hour': base.hours,
             'minute': base.minutes,
             'second': base.seconds,
         }
 
     def _is_leap_year(year):
         return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0)
 
     def get_month(prevdate):
         return prevdate.month
 
     def get_year(prevdate):
         return prevdate.year
 
     now = now or datetime.datetime.now()
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     future = False
 
     if prevdate > now:
         now_old = now
         now = prevdate
         prevdate = now_old
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     for part in order:
         rel_delta = _get_relative_delta(now, prevdate)
         deltas[part] = rel_delta[part]
 
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     offsets = [[5, 60], [4, 60], [3, 24]]
     for element in offsets:  # seconds, minutes, hours
         num = element[0]
         length = element[1]
 
         part = order[num]
         carry_part = order[num - 1]
 
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
 
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if get_month(prevdate) == 2 and _is_leap_year(get_year(prevdate)):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[get_month(prevdate) - 1]
 
         deltas['month'] -= 1
 
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
 
     # Format the result
     if short_format:
         fmt_funcs = {
             'year': lambda d: '%dy' % d,
             'month': lambda d: '%dm' % d,
             'day': lambda d: '%dd' % d,
             'hour': lambda d: '%dh' % d,
             'minute': lambda d: '%dmin' % d,
             'second': lambda d: '%dsec' % d,
         }
     else:
         fmt_funcs = {
             'year': lambda d: _pluralize('${num} year', '${num} years', d, mapping={'num': d}).interpolate(),
             'month': lambda d: _pluralize('${num} month', '${num} months', d, mapping={'num': d}).interpolate(),
             'day': lambda d: _pluralize('${num} day', '${num} days', d, mapping={'num': d}).interpolate(),
             'hour': lambda d: _pluralize('${num} hour', '${num} hours', d, mapping={'num': d}).interpolate(),
             'minute': lambda d: _pluralize('${num} minute', '${num} minutes', d, mapping={'num': d}).interpolate(),
             'second': lambda d: _pluralize('${num} second', '${num} seconds', d, mapping={'num': d}).interpolate(),
         }
 
     i = 0
     for part in order:
         value = deltas[part]
         if value != 0:
 
             if i < 5:
                 sub_part = order[i + 1]
                 sub_value = deltas[sub_part]
             else:
                 sub_value = 0
 
             if sub_value == 0 or show_short_version:
                 _val = fmt_funcs[part](value)
                 if future:
                     if show_suffix:
                         return _('in ${ago}', mapping={'ago': _val})
                     else:
                         return _(_val)
 
                 else:
                     if show_suffix:
                         return _('${ago} ago', mapping={'ago': _val})
                     else:
                         return _(_val)
 
             val = fmt_funcs[part](value)
             val_detail = fmt_funcs[sub_part](sub_value)
             mapping = {'val': val, 'detail': val_detail}
 
             if short_format:
                 datetime_tmpl = _('${val}, ${detail}', mapping=mapping)
                 if show_suffix:
                     datetime_tmpl = _('${val}, ${detail} ago', mapping=mapping)
                     if future:
                         datetime_tmpl = _('in ${val}, ${detail}', mapping=mapping)
             else:
                 datetime_tmpl = _('${val} and ${detail}', mapping=mapping)
                 if show_suffix:
                     datetime_tmpl = _('${val} and ${detail} ago', mapping=mapping)
                     if future:
                         datetime_tmpl = _('in ${val} and ${detail}', mapping=mapping)
 
             return datetime_tmpl
         i += 1
     return _('just now')
 
 
 def age_from_seconds(seconds):
     seconds = safe_int(seconds) or 0
     prevdate = time_to_datetime(time.time() + seconds)
     return age(prevdate, show_suffix=False, show_short_version=True)
 
 
 def cleaned_uri(uri):
     """
     Quotes '[' and ']' from uri if there is only one of them.
         according to RFC3986 we cannot use such chars in uri
     :param uri:
     :return: uri without this chars
     """
     return urllib.parse.quote(uri, safe='@$:/')
 
 
 def credentials_filter(uri):
     """
     Returns a url with removed credentials
 
     :param uri:
     """
     import urlobject
     if isinstance(uri, rhodecode.lib.encrypt.InvalidDecryptedValue):
         return 'InvalidDecryptionKey'
 
     url_obj = urlobject.URLObject(cleaned_uri(uri))
     url_obj = url_obj.without_password().without_username()
 
     return url_obj
 
 
 def get_host_info(request):
     """
     Generate host info, to obtain full url e.g https://server.com
     use this
     `{scheme}://{netloc}`
     """
     if not request:
         return {}
 
     qualified_home_url = request.route_url('home')
     parsed_url = urlobject.URLObject(qualified_home_url)
     decoded_path = safe_str(urllib.parse.unquote(parsed_url.path.rstrip('/')))
 
     return {
         'scheme': parsed_url.scheme,
         'netloc': parsed_url.netloc+decoded_path,
         'hostname': parsed_url.hostname,
     }
 
 
 def get_clone_url(request, uri_tmpl, repo_name, repo_id, repo_type, **override):
     qualified_home_url = request.route_url('home')
     parsed_url = urlobject.URLObject(qualified_home_url)
     decoded_path = safe_str(urllib.parse.unquote(parsed_url.path.rstrip('/')))
 
     args = {
         'scheme': parsed_url.scheme,
         'user': '',
         'sys_user': getpass.getuser(),
         # path if we use proxy-prefix
         'netloc': parsed_url.netloc+decoded_path,
         'hostname': parsed_url.hostname,
         'prefix': decoded_path,
         'repo': repo_name,
         'repoid': str(repo_id),
         'repo_type': repo_type
     }
     args.update(override)
     args['user'] = urllib.parse.quote(safe_str(args['user']))
 
     for k, v in list(args.items()):
         tmpl_key = '{%s}' % k
         uri_tmpl = uri_tmpl.replace(tmpl_key, v)
 
     # special case for SVN clone url
     if repo_type == 'svn':
         uri_tmpl = uri_tmpl.replace('ssh://', 'svn+ssh://')
 
     # remove leading @ sign if it's present. Case of empty user
     url_obj = urlobject.URLObject(uri_tmpl)
     url = url_obj.with_netloc(url_obj.netloc.lstrip('@'))
 
     return safe_str(url)
 
 
 def get_commit_safe(repo, commit_id=None, commit_idx=None, pre_load=None,
                     maybe_unreachable=False, reference_obj=None):
     """
     Safe version of get_commit if this commit doesn't exists for a
     repository it returns a Dummy one instead
 
     :param repo: repository instance
     :param commit_id: commit id as str
     :param commit_idx: numeric commit index
     :param pre_load: optional list of commit attributes to load
     :param maybe_unreachable: translate unreachable commits on git repos
     :param reference_obj: explicitly search via a reference obj in git. E.g "branch:123" would mean branch "123"
     """
     # TODO(skreft): remove these circular imports
     from rhodecode.lib.vcs.backends.base import BaseRepository, EmptyCommit
     from rhodecode.lib.vcs.exceptions import RepositoryError
     if not isinstance(repo, BaseRepository):
         raise Exception('You must pass an Repository '
                         'object as first argument got %s', type(repo))
 
     try:
         commit = repo.get_commit(
             commit_id=commit_id, commit_idx=commit_idx, pre_load=pre_load,
             maybe_unreachable=maybe_unreachable, reference_obj=reference_obj)
     except (RepositoryError, LookupError):
         commit = EmptyCommit()
     return commit
 
 
 def datetime_to_time(dt):
     if dt:
         return time.mktime(dt.timetuple())
 
 
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, str):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
 
 
 def time_to_utcdatetime(tm):
     if tm:
         if isinstance(tm, str):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.utcfromtimestamp(tm)
 
 
 MENTIONS_REGEX = re.compile(
     # ^@ or @ without any special chars in front
     r'(?:^@|[^a-zA-Z0-9\-\_\.]@)'
     # main body starts with letter, then can be . - _
     r'([a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+)',
     re.VERBOSE | re.MULTILINE)
 
 
 def extract_mentioned_users(s):
     """
     Returns unique usernames from given string s that have @mention
 
     :param s: string to get mentions
     """
     usrs = set()
     for username in MENTIONS_REGEX.findall(s):
         usrs.add(username)
 
     return sorted(list(usrs), key=lambda k: k.lower())
 
 
 def fix_PATH(os_=None):
     """
     Get current active python path, and append it to PATH variable to fix
     issues of subprocess calls and different python versions
     """
     if os_ is None:
         import os
     else:
         os = os_
 
     cur_path = os.path.split(sys.executable)[0]
     os_path = os.environ['PATH']
     if not os.environ['PATH'].startswith(cur_path):
         os.environ['PATH'] = f'{cur_path}:{os_path}'
 
 
 def obfuscate_url_pw(engine):
     _url = engine or ''
     try:
         _url = sqlalchemy.engine.url.make_url(engine)
     except Exception:
         pass
     return repr(_url)
 
 
 def get_server_url(environ):
     req = webob.Request(environ)
     return req.host_url + req.script_name
 
 
 def unique_id(hexlen=32):
     alphabet = "23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz"
     return suuid(truncate_to=hexlen, alphabet=alphabet)
 
 
 def suuid(url=None, truncate_to=22, alphabet=None):
     """
     Generate and return a short URL safe UUID.
 
     If the url parameter is provided, set the namespace to the provided
     URL and generate a UUID.
 
     :param url to get the uuid for
     :truncate_to: truncate the basic 22 UUID to shorter version
 
     The IDs won't be universally unique any longer, but the probability of
     a collision will still be very low.
     """
     # Define our alphabet.
     _ALPHABET = alphabet or "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
 
     # If no URL is given, generate a random UUID.
     if url is None:
         unique_id = uuid.uuid4().int
     else:
         unique_id = uuid.uuid3(uuid.NAMESPACE_URL, url).int
 
     alphabet_length = len(_ALPHABET)
     output = []
     while unique_id > 0:
         digit = unique_id % alphabet_length
         output.append(_ALPHABET[digit])
         unique_id = int(unique_id / alphabet_length)
     return "".join(output)[:truncate_to]
 
 
 def get_current_rhodecode_user(request=None):
     """
     Gets rhodecode user from request
     """
     import pyramid.threadlocal
     pyramid_request = request or pyramid.threadlocal.get_current_request()
 
     # web case
     if pyramid_request and hasattr(pyramid_request, 'user'):
         return pyramid_request.user
 
     # api case
     if pyramid_request and hasattr(pyramid_request, 'rpc_user'):
         return pyramid_request.rpc_user
 
     return None
 
 
 def action_logger_generic(action, namespace=''):
     """
     A generic logger for actions useful to the system overview, tries to find
     an acting user for the context of the call otherwise reports unknown user
 
     :param action: logging message eg 'comment 5 deleted'
     :param type: string
 
     :param namespace: namespace of the logging message eg. 'repo.comments'
     :param type: string
 
     """
 
     logger_name = 'rhodecode.actions'
 
     if namespace:
         logger_name += '.' + namespace
 
     log = logging.getLogger(logger_name)
 
     # get a user if we can
     user = get_current_rhodecode_user()
 
     logfunc = log.info
 
     if not user:
         user = '<unknown user>'
         logfunc = log.warning
 
     logfunc(f'Logging action by {user}: {action}')
 
 
 def escape_split(text, sep=',', maxsplit=-1):
     r"""
     Allows for escaping of the separator: e.g. arg='foo\, bar'
 
     It should be noted that the way bash et. al. do command line parsing, those
     single quotes are required.
     """
     escaped_sep = r'\%s' % sep
 
     if escaped_sep not in text:
         return text.split(sep, maxsplit)
 
     before, _mid, after = text.partition(escaped_sep)
     startlist = before.split(sep, maxsplit)  # a regular split is fine here
     unfinished = startlist[-1]
     startlist = startlist[:-1]
 
     # recurse because there may be more escaped separators
     endlist = escape_split(after, sep, maxsplit)
 
     # finish building the escaped value. we use endlist[0] becaue the first
     # part of the string sent in recursion is the rest of the escaped value.
     unfinished += sep + endlist[0]
 
     return startlist + [unfinished] + endlist[1:]  # put together all the parts
 
 
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute. Example::
 
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid)
             # calls
 
     """
 
     def __init__(self, attr_name):
         self.attr_name = attr_name
 
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
 
     def __call__(self):
         return self
 
 
 # alias
 OAttr = OptionalAttr
 
 
 class Optional(object):
     """
     Defines an optional parameter::
 
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
 
     is equivalent of::
 
         param = Optional.extract(param)
 
     """
 
     def __init__(self, type_):
         self.type_ = type_
 
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
 
     def __call__(self):
         return self.getval()
 
     def getval(self):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             # use params name
             return self.type_.attr_name
         return self.type_
 
     @classmethod
     def extract(cls, val):
         """
         Extracts value from Optional() instance
 
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             return val.getval()
         return val
 
 
 def glob2re(pat):
     import fnmatch
     return fnmatch.translate(pat)
 
 
 def parse_byte_string(size_str):
     match = re.match(r'(\d+)(MB|KB)', size_str, re.IGNORECASE)
     if not match:
         raise ValueError(f'Given size:{size_str} is invalid, please make sure '
                          f'to use format of <num>(MB|KB)')
 
     _parts = match.groups()
     num, type_ = _parts
     return int(num) * {'mb': 1024*1024, 'kb': 1024}[type_.lower()]
 
 
 class CachedProperty(object):
     """
     Lazy Attributes. With option to invalidate the cache by running a method
 
     >>> class Foo(object):
     ...
     ...    @CachedProperty
     ...    def heavy_func(self):
     ...        return 'super-calculation'
     ...
     ... foo = Foo()
     ... foo.heavy_func() # first computation
     ... foo.heavy_func() # fetch from cache
     ... foo._invalidate_prop_cache('heavy_func')
 
     # at this point calling foo.heavy_func() will be re-computed
     """
 
     def __init__(self, func, func_name=None):
 
         if func_name is None:
             func_name = func.__name__
         self.data = (func, func_name)
         functools.update_wrapper(self, func)
 
     def __get__(self, inst, class_):
         if inst is None:
             return self
 
         func, func_name = self.data
         value = func(inst)
         inst.__dict__[func_name] = value
         if '_invalidate_prop_cache' not in inst.__dict__:
             inst.__dict__['_invalidate_prop_cache'] = functools.partial(
                 self._invalidate_prop_cache, inst)
         return value
 
     def _invalidate_prop_cache(self, inst, name):
         inst.__dict__.pop(name, None)
 
 
 def retry(func=None, exception=Exception, n_tries=5, delay=5, backoff=1, logger=True):
     """
     Retry decorator with exponential backoff.
 
     Parameters
     ----------
     func : typing.Callable, optional
         Callable on which the decorator is applied, by default None
     exception : Exception or tuple of Exceptions, optional
         Exception(s) that invoke retry, by default Exception
     n_tries : int, optional
         Number of tries before giving up, by default 5
     delay : int, optional
         Initial delay between retries in seconds, by default 5
     backoff : int, optional
         Backoff multiplier e.g. value of 2 will double the delay, by default 1
     logger : bool, optional
         Option to log or print, by default False
 
     Returns
     -------
     typing.Callable
         Decorated callable that calls itself when exception(s) occur.
 
     Examples
     --------
     >>> import random
     >>> @retry(exception=Exception, n_tries=3)
     ... def test_random(text):
     ...    x = random.random()
     ...    if x < 0.5:
     ...        raise Exception("Fail")
     ...    else:
     ...        print("Success: ", text)
     >>> test_random("It works!")
     """
 
     if func is None:
         return functools.partial(
             retry,
             exception=exception,
             n_tries=n_tries,
             delay=delay,
             backoff=backoff,
             logger=logger,
         )
 
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
         _n_tries, n_delay = n_tries, delay
         log = logging.getLogger('rhodecode.retry')
 
         while _n_tries > 1:
             try:
                 return func(*args, **kwargs)
             except exception as e:
                 e_details = repr(e)
                 msg = "Exception on calling func {func}: {e}, " \
                       "Retrying in {n_delay} seconds..."\
                     .format(func=func, e=e_details, n_delay=n_delay)
                 if logger:
                     log.warning(msg)
                 else:
                     print(msg)
                 time.sleep(n_delay)
                 _n_tries -= 1
                 n_delay *= backoff
 
         return func(*args, **kwargs)
 
     return wrapper
 
 
 def user_agent_normalizer(user_agent_raw, safe=True):
     log = logging.getLogger('rhodecode.user_agent_normalizer')
     ua = (user_agent_raw or '').strip().lower()
     ua = ua.replace('"', '')
 
     try:
         if 'mercurial/proto-1.0' in ua:
             ua = ua.replace('mercurial/proto-1.0', '')
             ua = ua.replace('(', '').replace(')', '').strip()
             ua = ua.replace('mercurial ', 'mercurial/')
         elif ua.startswith('git'):
             parts = ua.split(' ')
             if parts:
                 ua = parts[0]
             ua = re.sub(r'\.windows\.\d', '', ua).strip()
 
         return ua
     except Exception:
         log.exception('Failed to parse scm user-agent')
         if not safe:
             raise
 
     return ua
 
 
 def get_available_port(min_port=40000, max_port=55555, use_range=False):
     hostname = ''
-    for _ in range(min_port, max_port):
+    for _check_port in range(min_port, max_port):
         pick_port = 0
         if use_range:
             pick_port = random.randint(min_port, max_port)
 
         with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
             try:
                 s.bind((hostname, pick_port))
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                 return s.getsockname()[1]
-            except OSError:
-                continue
             except socket.error as e:
                 if e.args[0] in [errno.EADDRINUSE, errno.ECONNREFUSED]:
                     continue
                 raise
+            except OSError:
+                continue

rhodecode/lib/vcs/backends/base.py

@@ -1,1984 +1,1984 b''
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Base module for all VCS systems
 """
 import os
 import re
 import time
 import shutil
 import datetime
 import fnmatch
 import itertools
 import logging
 import dataclasses
 import warnings
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 
 import rhodecode
 from rhodecode.translation import lazy_ugettext
 from rhodecode.lib.utils2 import safe_str, CachedProperty
 from rhodecode.lib.vcs.utils import author_name, author_email
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.exceptions import (
     CommitError, EmptyRepositoryError, NodeAlreadyAddedError,
     NodeAlreadyChangedError, NodeAlreadyExistsError, NodeAlreadyRemovedError,
     NodeDoesNotExistError, NodeNotChangedError, VCSError,
     ImproperArchiveTypeError, BranchDoesNotExistError, CommitDoesNotExistError,
     RepositoryError)
 
 
 log = logging.getLogger(__name__)
 
 
 FILEMODE_DEFAULT = 0o100644
 FILEMODE_EXECUTABLE = 0o100755
 EMPTY_COMMIT_ID = '0' * 40
 
 
 @dataclasses.dataclass
 class Reference:
     type: str
     name: str
     commit_id: str
 
     def __iter__(self):
         yield self.type
         yield self.name
         yield self.commit_id
 
     @property
     def branch(self):
         if self.type == 'branch':
             return self.name
 
     @property
     def bookmark(self):
         if self.type == 'book':
             return self.name
 
     @property
     def to_str(self):
         return reference_to_unicode(self)
 
     def asdict(self):
         return dict(
             type=self.type,
             name=self.name,
             commit_id=self.commit_id
         )
 
 
 def unicode_to_reference(raw: str):
     """
     Convert a unicode (or string) to a reference object.
     If unicode evaluates to False it returns None.
     """
     if raw:
         refs = raw.split(':')
         return Reference(*refs)
     else:
         return None
 
 
 def reference_to_unicode(ref: Reference):
     """
     Convert a reference object to unicode.
     If reference is None it returns None.
     """
     if ref:
         return ':'.join(ref)
     else:
         return None
 
 
 class MergeFailureReason(object):
     """
     Enumeration with all the reasons why the server side merge could fail.
 
     DO NOT change the number of the reasons, as they may be stored in the
     database.
 
     Changing the name of a reason is acceptable and encouraged to deprecate old
     reasons.
     """
 
     # Everything went well.
     NONE = 0
 
     # An unexpected exception was raised. Check the logs for more details.
     UNKNOWN = 1
 
     # The merge was not successful, there are conflicts.
     MERGE_FAILED = 2
 
     # The merge succeeded but we could not push it to the target repository.
     PUSH_FAILED = 3
 
     # The specified target is not a head in the target repository.
     TARGET_IS_NOT_HEAD = 4
 
     # The source repository contains more branches than the target. Pushing
     # the merge will create additional branches in the target.
     HG_SOURCE_HAS_MORE_BRANCHES = 5
 
     # The target reference has multiple heads. That does not allow to correctly
     # identify the target location. This could only happen for mercurial
     # branches.
     HG_TARGET_HAS_MULTIPLE_HEADS = 6
 
     # The target repository is locked
     TARGET_IS_LOCKED = 7
 
     # Deprecated, use MISSING_TARGET_REF or MISSING_SOURCE_REF instead.
     # A involved commit could not be found.
     _DEPRECATED_MISSING_COMMIT = 8
 
     # The target repo reference is missing.
     MISSING_TARGET_REF = 9
 
     # The source repo reference is missing.
     MISSING_SOURCE_REF = 10
 
     # The merge was not successful, there are conflicts related to sub
     # repositories.
     SUBREPO_MERGE_FAILED = 11
 
 
 class UpdateFailureReason(object):
     """
     Enumeration with all the reasons why the pull request update could fail.
 
     DO NOT change the number of the reasons, as they may be stored in the
     database.
 
     Changing the name of a reason is acceptable and encouraged to deprecate old
     reasons.
     """
 
     # Everything went well.
     NONE = 0
 
     # An unexpected exception was raised. Check the logs for more details.
     UNKNOWN = 1
 
     # The pull request is up to date.
     NO_CHANGE = 2
 
     # The pull request has a reference type that is not supported for update.
     WRONG_REF_TYPE = 3
 
     # Update failed because the target reference is missing.
     MISSING_TARGET_REF = 4
 
     # Update failed because the source reference is missing.
     MISSING_SOURCE_REF = 5
 
 
 class MergeResponse(object):
 
     # uses .format(**metadata) for variables
     MERGE_STATUS_MESSAGES = {
         MergeFailureReason.NONE: lazy_ugettext(
             'This pull request can be automatically merged.'),
         MergeFailureReason.UNKNOWN: lazy_ugettext(
             'This pull request cannot be merged because of an unhandled exception. '
             '{exception}'),
         MergeFailureReason.MERGE_FAILED: lazy_ugettext(
             'This pull request cannot be merged because of merge conflicts. {unresolved_files}'),
         MergeFailureReason.PUSH_FAILED: lazy_ugettext(
             'This pull request could not be merged because push to '
             'target:`{target}@{merge_commit}` failed.'),
         MergeFailureReason.TARGET_IS_NOT_HEAD: lazy_ugettext(
             'This pull request cannot be merged because the target '
             '`{target_ref.name}` is not a head.'),
         MergeFailureReason.HG_SOURCE_HAS_MORE_BRANCHES: lazy_ugettext(
             'This pull request cannot be merged because the source contains '
             'more branches than the target.'),
         MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS: lazy_ugettext(
             'This pull request cannot be merged because the target `{target_ref.name}` '
             'has multiple heads: `{heads}`.'),
         MergeFailureReason.TARGET_IS_LOCKED: lazy_ugettext(
             'This pull request cannot be merged because the target repository is '
             'locked by {locked_by}.'),
 
         MergeFailureReason.MISSING_TARGET_REF: lazy_ugettext(
             'This pull request cannot be merged because the target '
             'reference `{target_ref.name}` is missing.'),
         MergeFailureReason.MISSING_SOURCE_REF: lazy_ugettext(
             'This pull request cannot be merged because the source '
             'reference `{source_ref.name}` is missing.'),
         MergeFailureReason.SUBREPO_MERGE_FAILED: lazy_ugettext(
             'This pull request cannot be merged because of conflicts related '
             'to sub repositories.'),
 
         # Deprecations
         MergeFailureReason._DEPRECATED_MISSING_COMMIT: lazy_ugettext(
             'This pull request cannot be merged because the target or the '
             'source reference is missing.'),
 
     }
 
-    def __init__(self, possible, executed, merge_ref, failure_reason, metadata=None):
+    def __init__(self, possible, executed, merge_ref: Reference, failure_reason, metadata=None):
         self.possible = possible
         self.executed = executed
         self.merge_ref = merge_ref
         self.failure_reason = failure_reason
         self.metadata = metadata or {}
 
     def __repr__(self):
         return f'<MergeResponse:{self.label} {self.failure_reason}>'
 
     def __eq__(self, other):
         same_instance = isinstance(other, self.__class__)
         return same_instance \
                and self.possible == other.possible \
                and self.executed == other.executed \
                and self.failure_reason == other.failure_reason
 
     @property
     def label(self):
         label_dict = dict((v, k) for k, v in MergeFailureReason.__dict__.items() if
                           not k.startswith('_'))
         return label_dict.get(self.failure_reason)
 
     @property
     def merge_status_message(self):
         """
         Return a human friendly error message for the given merge status code.
         """
         msg = safe_str(self.MERGE_STATUS_MESSAGES[self.failure_reason])
 
         try:
             return msg.format(**self.metadata)
         except Exception:
             log.exception('Failed to format %s message', self)
             return msg
 
     def asdict(self):
         data = {}
         for k in ['possible', 'executed', 'merge_ref', 'failure_reason',
                   'merge_status_message']:
             data[k] = getattr(self, k)
         return data
 
 
 class TargetRefMissing(ValueError):
     pass
 
 
 class SourceRefMissing(ValueError):
     pass
 
 
 class BaseRepository(object):
     """
     Base Repository for final backends
 
     .. attribute:: DEFAULT_BRANCH_NAME
 
        name of default branch (i.e. "trunk" for svn, "master" for git etc.
 
     .. attribute:: commit_ids
 
        list of all available commit ids, in ascending order
 
     .. attribute:: path
 
        absolute path to the repository
 
     .. attribute:: bookmarks
 
        Mapping from name to :term:`Commit ID` of the bookmark. Empty in case
        there are no bookmarks or the backend implementation does not support
        bookmarks.
 
     .. attribute:: tags
 
        Mapping from name to :term:`Commit ID` of the tag.
 
     """
 
     DEFAULT_BRANCH_NAME = None
     DEFAULT_CONTACT = "Unknown"
     DEFAULT_DESCRIPTION = "unknown"
     EMPTY_COMMIT_ID = '0' * 40
     COMMIT_ID_PAT = re.compile(r'[0-9a-fA-F]{40}')
 
     path = None
 
     _is_empty = None
     _commit_ids = {}
 
     def __init__(self, repo_path, config=None, create=False, **kwargs):
         """
         Initializes repository. Raises RepositoryError if repository could
         not be find at the given ``repo_path`` or directory at ``repo_path``
         exists and ``create`` is set to True.
 
         :param repo_path: local path of the repository
         :param config: repository configuration
         :param create=False: if set to True, would try to create repository.
         :param src_url=None: if set, should be proper url from which repository
           would be cloned; requires ``create`` parameter to be set to True -
           raises RepositoryError if src_url is set and create evaluates to
           False
         """
         raise NotImplementedError
 
     def __repr__(self):
         return f'<{self.__class__.__name__} at {self.path}>'
 
     def __len__(self):
         return self.count()
 
     def __eq__(self, other):
         same_instance = isinstance(other, self.__class__)
         return same_instance and other.path == self.path
 
     def __ne__(self, other):
         return not self.__eq__(other)
 
     def get_create_shadow_cache_pr_path(self, db_repo):
         path = db_repo.cached_diffs_dir
         if not os.path.exists(path):
             os.makedirs(path, 0o755)
         return path
 
     @classmethod
     def get_default_config(cls, default=None):
         config = Config()
         if default and isinstance(default, list):
             for section, key, val in default:
                 config.set(section, key, val)
         return config
 
     @LazyProperty
     def _remote(self):
         raise NotImplementedError
 
     def _heads(self, branch=None):
         return []
 
     @LazyProperty
     def EMPTY_COMMIT(self):
         return EmptyCommit(self.EMPTY_COMMIT_ID)
 
     @LazyProperty
     def alias(self):
         for k, v in settings.BACKENDS.items():
             if v.split('.')[-1] == str(self.__class__.__name__):
                 return k
 
     @LazyProperty
     def name(self):
         return safe_str(os.path.basename(self.path))
 
     @LazyProperty
     def description(self):
         raise NotImplementedError
 
     def refs(self):
         """
         returns a `dict` with branches, bookmarks, tags, and closed_branches
         for this repository
         """
         return dict(
             branches=self.branches,
             branches_closed=self.branches_closed,
             tags=self.tags,
             bookmarks=self.bookmarks
         )
 
     @LazyProperty
     def branches(self):
         """
         A `dict` which maps branch names to commit ids.
         """
         raise NotImplementedError
 
     @LazyProperty
     def branches_closed(self):
         """
         A `dict` which maps tags names to commit ids.
         """
         raise NotImplementedError
 
     @LazyProperty
     def bookmarks(self):
         """
         A `dict` which maps tags names to commit ids.
         """
         raise NotImplementedError
 
     @LazyProperty
     def tags(self):
         """
         A `dict` which maps tags names to commit ids.
         """
         raise NotImplementedError
 
     @LazyProperty
     def size(self):
         """
         Returns combined size in bytes for all repository files
         """
         tip = self.get_commit()
         return tip.size
 
     def size_at_commit(self, commit_id):
         commit = self.get_commit(commit_id)
         return commit.size
 
     def _check_for_empty(self):
         no_commits = len(self._commit_ids) == 0
         if no_commits:
             # check on remote to be sure
             return self._remote.is_empty()
         else:
             return False
 
     def is_empty(self):
         if rhodecode.is_test:
             return self._check_for_empty()
 
         if self._is_empty is None:
             # cache empty for production, but not tests
             self._is_empty = self._check_for_empty()
 
         return self._is_empty
 
     @staticmethod
     def check_url(url, config):
         """
         Function will check given url and try to verify if it's a valid
         link.
         """
         raise NotImplementedError
 
     @staticmethod
     def is_valid_repository(path):
         """
         Check if given `path` contains a valid repository of this backend
         """
         raise NotImplementedError
 
     # ==========================================================================
     # COMMITS
     # ==========================================================================
 
     @CachedProperty
     def commit_ids(self):
         raise NotImplementedError
 
     def append_commit_id(self, commit_id):
         if commit_id not in self.commit_ids:
             self._rebuild_cache(self.commit_ids + [commit_id])
 
         # clear cache
         self._invalidate_prop_cache('commit_ids')
         self._is_empty = False
 
     def get_commit(self, commit_id=None, commit_idx=None, pre_load=None,
                    translate_tag=None, maybe_unreachable=False, reference_obj=None):
         """
         Returns instance of `BaseCommit` class. If `commit_id` and `commit_idx`
         are both None, most recent commit is returned.
 
         :param pre_load: Optional. List of commit attributes to load.
 
         :raises ``EmptyRepositoryError``: if there are no commits
         """
         raise NotImplementedError
 
     def __iter__(self):
         for commit_id in self.commit_ids:
             yield self.get_commit(commit_id=commit_id)
 
     def get_commits(
             self, start_id=None, end_id=None, start_date=None, end_date=None,
             branch_name=None, show_hidden=False, pre_load=None, translate_tags=None):
         """
         Returns iterator of `BaseCommit` objects from start to end
         not inclusive. This should behave just like a list, ie. end is not
         inclusive.
 
         :param start_id: None or str, must be a valid commit id
         :param end_id: None or str, must be a valid commit id
         :param start_date:
         :param end_date:
         :param branch_name:
         :param show_hidden:
         :param pre_load:
         :param translate_tags:
         """
         raise NotImplementedError
 
     def __getitem__(self, key):
         """
         Allows index based access to the commit objects of this repository.
         """
         pre_load = ["author", "branch", "date", "message", "parents"]
         if isinstance(key, slice):
             return self._get_range(key, pre_load)
         return self.get_commit(commit_idx=key, pre_load=pre_load)
 
     def _get_range(self, slice_obj, pre_load):
         for commit_id in self.commit_ids.__getitem__(slice_obj):
             yield self.get_commit(commit_id=commit_id, pre_load=pre_load)
 
     def count(self):
         return len(self.commit_ids)
 
     def tag(self, name, user, commit_id=None, message=None, date=None, **opts):
         """
         Creates and returns a tag for the given ``commit_id``.
 
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <[e-mail]>"
         :param commit_id: commit id for which new tag would be created
         :param message: message of the tag's commit
         :param date: date of tag's commit
 
         :raises TagAlreadyExistError: if tag with same name already exists
         """
         raise NotImplementedError
 
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
 
         :param name: name of the tag to be removed
         :param user: full username, i.e.: "Joe Doe <[e-mail]>"
         :param message: message of the tag's removal commit
         :param date: date of tag's removal commit
 
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         raise NotImplementedError
 
     def get_diff(
             self, commit1, commit2, path=None, ignore_whitespace=False,
             context=3, path1=None):
         """
         Returns (git like) *diff*, as plain text. Shows changes introduced by
         `commit2` since `commit1`.
 
         :param commit1: Entry point from which diff is shown. Can be
           ``self.EMPTY_COMMIT`` - in this case, patch showing all
           the changes since empty state of the repository until `commit2`
         :param commit2: Until which commit changes should be shown.
         :param path: Can be set to a path of a file to create a diff of that
           file. If `path1` is also set, this value is only associated to
           `commit2`.
         :param ignore_whitespace: If set to ``True``, would not show whitespace
           changes. Defaults to ``False``.
         :param context: How many lines before/after changed lines should be
           shown. Defaults to ``3``.
         :param path1: Can be set to a path to associate with `commit1`. This
           parameter works only for backends which support diff generation for
           different paths. Other backends will raise a `ValueError` if `path1`
           is set and has a different value than `path`.
         :param file_path: filter this diff by given path pattern
         """
         raise NotImplementedError
 
     def strip(self, commit_id, branch=None):
         """
         Strip given commit_id from the repository
         """
         raise NotImplementedError
 
     def get_common_ancestor(self, commit_id1, commit_id2, repo2):
         """
         Return a latest common ancestor commit if one exists for this repo
         `commit_id1` vs `commit_id2` from `repo2`.
 
         :param commit_id1: Commit it from this repository to use as a
            target for the comparison.
         :param commit_id2: Source commit id to use for comparison.
         :param repo2: Source repository to use for comparison.
         """
         raise NotImplementedError
 
     def compare(self, commit_id1, commit_id2, repo2, merge, pre_load=None):
         """
         Compare this repository's revision `commit_id1` with `commit_id2`.
 
         Returns a tuple(commits, ancestor) that would be merged from
         `commit_id2`. Doing a normal compare (``merge=False``), ``None``
         will be returned as ancestor.
 
         :param commit_id1: Commit it from this repository to use as a
            target for the comparison.
         :param commit_id2: Source commit id to use for comparison.
         :param repo2: Source repository to use for comparison.
         :param merge: If set to ``True`` will do a merge compare which also
            returns the common ancestor.
         :param pre_load: Optional. List of commit attributes to load.
         """
         raise NotImplementedError
 
     def merge(self, repo_id, workspace_id, target_ref, source_repo, source_ref,
               user_name='', user_email='', message='', dry_run=False,
               use_rebase=False, close_branch=False):
         """
         Merge the revisions specified in `source_ref` from `source_repo`
         onto the `target_ref` of this repository.
 
         `source_ref` and `target_ref` are named tupls with the following
         fields `type`, `name` and `commit_id`.
 
         Returns a MergeResponse named tuple with the following fields
         'possible', 'executed', 'source_commit', 'target_commit',
         'merge_commit'.
 
         :param repo_id: `repo_id` target repo id.
         :param workspace_id: `workspace_id` unique identifier.
         :param target_ref: `target_ref` points to the commit on top of which
             the `source_ref` should be merged.
         :param source_repo: The repository that contains the commits to be
             merged.
         :param source_ref: `source_ref` points to the topmost commit from
             the `source_repo` which should be merged.
         :param user_name: Merge commit `user_name`.
         :param user_email: Merge commit `user_email`.
         :param message: Merge commit `message`.
         :param dry_run: If `True` the merge will not take place.
         :param use_rebase: If `True` commits from the source will be rebased
             on top of the target instead of being merged.
         :param close_branch: If `True` branch will be close before merging it
         """
         if dry_run:
             message = message or settings.MERGE_DRY_RUN_MESSAGE
             user_email = user_email or settings.MERGE_DRY_RUN_EMAIL
             user_name = user_name or settings.MERGE_DRY_RUN_USER
         else:
             if not user_name:
                 raise ValueError('user_name cannot be empty')
             if not user_email:
                 raise ValueError('user_email cannot be empty')
             if not message:
                 raise ValueError('message cannot be empty')
 
         try:
             return self._merge_repo(
                 repo_id, workspace_id, target_ref, source_repo,
                 source_ref, message, user_name, user_email, dry_run=dry_run,
                 use_rebase=use_rebase, close_branch=close_branch)
         except RepositoryError as exc:
             log.exception('Unexpected failure when running merge, dry-run=%s', dry_run)
             return MergeResponse(
                 False, False, None, MergeFailureReason.UNKNOWN,
                 metadata={'exception': str(exc)})
 
     def _merge_repo(self, repo_id, workspace_id, target_ref,
                     source_repo, source_ref, merge_message,
                     merger_name, merger_email, dry_run=False,
                     use_rebase=False, close_branch=False):
         """Internal implementation of merge."""
         raise NotImplementedError
 
     def _maybe_prepare_merge_workspace(
             self, repo_id, workspace_id, target_ref, source_ref):
         """
         Create the merge workspace.
 
         :param workspace_id: `workspace_id` unique identifier.
         """
         raise NotImplementedError
 
     @classmethod
     def _get_legacy_shadow_repository_path(cls, repo_path, workspace_id):
         """
         Legacy version that was used before. We still need it for
         backward compat
         """
         return os.path.join(
             os.path.dirname(repo_path),
             f'.__shadow_{os.path.basename(repo_path)}_{workspace_id}')
 
     @classmethod
     def _get_shadow_repository_path(cls, repo_path, repo_id, workspace_id):
         # The name of the shadow repository must start with '.', so it is
         # skipped by 'rhodecode.lib.utils.get_filesystem_repos'.
         legacy_repository_path = cls._get_legacy_shadow_repository_path(repo_path, workspace_id)
         if os.path.exists(legacy_repository_path):
             return legacy_repository_path
         else:
             return os.path.join(
                 os.path.dirname(repo_path),
                 f'.__shadow_repo_{repo_id}_{workspace_id}')
 
     def cleanup_merge_workspace(self, repo_id, workspace_id):
         """
         Remove merge workspace.
 
         This function MUST not fail in case there is no workspace associated to
         the given `workspace_id`.
 
         :param workspace_id: `workspace_id` unique identifier.
         """
         shadow_repository_path = self._get_shadow_repository_path(
             self.path, repo_id, workspace_id)
         shadow_repository_path_del = '{}.{}.delete'.format(
             shadow_repository_path, time.time())
 
         # move the shadow repo, so it never conflicts with the one used.
         # we use this method because shutil.rmtree had some edge case problems
         # removing symlinked repositories
         if not os.path.isdir(shadow_repository_path):
             return
 
         shutil.move(shadow_repository_path, shadow_repository_path_del)
         try:
             shutil.rmtree(shadow_repository_path_del, ignore_errors=False)
         except Exception:
             log.exception('Failed to gracefully remove shadow repo under %s',
                           shadow_repository_path_del)
             shutil.rmtree(shadow_repository_path_del, ignore_errors=True)
 
     # ========== #
     # COMMIT API #
     # ========== #
 
     @LazyProperty
     def in_memory_commit(self):
         """
         Returns :class:`InMemoryCommit` object for this repository.
         """
         raise NotImplementedError
 
     # ======================== #
     # UTILITIES FOR SUBCLASSES #
     # ======================== #
 
     def _validate_diff_commits(self, commit1, commit2):
         """
         Validates that the given commits are related to this repository.
 
         Intended as a utility for sub classes to have a consistent validation
         of input parameters in methods like :meth:`get_diff`.
         """
         self._validate_commit(commit1)
         self._validate_commit(commit2)
         if (isinstance(commit1, EmptyCommit) and
                 isinstance(commit2, EmptyCommit)):
             raise ValueError("Cannot compare two empty commits")
 
     def _validate_commit(self, commit):
         if not isinstance(commit, BaseCommit):
             raise TypeError(
                 "%s is not of type BaseCommit" % repr(commit))
         if commit.repository != self and not isinstance(commit, EmptyCommit):
             raise ValueError(
                 "Commit %s must be a valid commit from this repository %s, "
                 "related to this repository instead %s." %
                 (commit, self, commit.repository))
 
     def _validate_commit_id(self, commit_id):
         if not isinstance(commit_id, str):
             raise TypeError(f"commit_id must be a string value got {type(commit_id)} instead")
 
     def _validate_commit_idx(self, commit_idx):
         if not isinstance(commit_idx, int):
             raise TypeError(f"commit_idx must be a numeric value, got {type(commit_idx)}")
 
     def _validate_branch_name(self, branch_name):
         if branch_name and branch_name not in self.branches_all:
             msg = (f"Branch {branch_name} not found in {self}")
             raise BranchDoesNotExistError(msg)
 
     #
     # Supporting deprecated API parts
     # TODO: johbo: consider to move this into a mixin
     #
 
     @property
     def EMPTY_CHANGESET(self):
         warnings.warn(
             "Use EMPTY_COMMIT or EMPTY_COMMIT_ID instead", DeprecationWarning)
         return self.EMPTY_COMMIT_ID
 
     @property
     def revisions(self):
         warnings.warn("Use commits attribute instead", DeprecationWarning)
         return self.commit_ids
 
     @revisions.setter
     def revisions(self, value):
         warnings.warn("Use commits attribute instead", DeprecationWarning)
         self.commit_ids = value
 
     def get_changeset(self, revision=None, pre_load=None):
         warnings.warn("Use get_commit instead", DeprecationWarning)
         commit_id = None
         commit_idx = None
         if isinstance(revision, str):
             commit_id = revision
         else:
             commit_idx = revision
         return self.get_commit(
             commit_id=commit_id, commit_idx=commit_idx, pre_load=pre_load)
 
     def get_changesets(
             self, start=None, end=None, start_date=None, end_date=None,
             branch_name=None, pre_load=None):
         warnings.warn("Use get_commits instead", DeprecationWarning)
         start_id = self._revision_to_commit(start)
         end_id = self._revision_to_commit(end)
         return self.get_commits(
             start_id=start_id, end_id=end_id, start_date=start_date,
             end_date=end_date, branch_name=branch_name, pre_load=pre_load)
 
     def _revision_to_commit(self, revision):
         """
         Translates a revision to a commit_id
 
         Helps to support the old changeset based API which allows to use
         commit ids and commit indices interchangeable.
         """
         if revision is None:
             return revision
 
         if isinstance(revision, str):
             commit_id = revision
         else:
             commit_id = self.commit_ids[revision]
         return commit_id
 
     @property
     def in_memory_changeset(self):
         warnings.warn("Use in_memory_commit instead", DeprecationWarning)
         return self.in_memory_commit
 
     def get_path_permissions(self, username):
         """
         Returns a path permission checker or None if not supported
 
         :param username: session user name
         :return: an instance of BasePathPermissionChecker or None
         """
         return None
 
     def install_hooks(self, force=False):
         return self._remote.install_hooks(force)
 
     def get_hooks_info(self):
         return self._remote.get_hooks_info()
 
     def vcsserver_invalidate_cache(self, delete=False):
         return self._remote.vcsserver_invalidate_cache(delete)
 
 
 class BaseCommit(object):
     """
     Each backend should implement it's commit representation.
 
     **Attributes**
 
         ``repository``
             repository object within which commit exists
 
         ``id``
             The commit id, may be ``raw_id`` or i.e. for mercurial's tip
             just ``tip``.
 
         ``raw_id``
             raw commit representation (i.e. full 40 length sha for git
             backend)
 
         ``short_id``
             shortened (if apply) version of ``raw_id``; it would be simple
             shortcut for ``raw_id[:12]`` for git/mercurial backends or same
             as ``raw_id`` for subversion
 
         ``idx``
             commit index
 
         ``files``
             list of ``FileNode`` (``Node`` with NodeKind.FILE) objects
 
         ``dirs``
             list of ``DirNode`` (``Node`` with NodeKind.DIR) objects
 
         ``nodes``
             combined list of ``Node`` objects
 
         ``author``
             author of the commit, as unicode
 
         ``message``
             message of the commit, as unicode
 
         ``parents``
             list of parent commits
 
     """
     repository = None
     branch = None
 
     """
     Depending on the backend this should be set to the branch name of the
     commit. Backends not supporting branches on commits should leave this
     value as ``None``.
     """
 
     _ARCHIVE_PREFIX_TEMPLATE = '{repo_name}-{short_id}'
     """
     This template is used to generate a default prefix for repository archives
     if no prefix has been specified.
     """
 
     def __repr__(self):
         return self.__str__()
 
     def __str__(self):
         return f'<{self.__class__.__name__} at {self.idx}:{self.short_id}>'
 
     def __eq__(self, other):
         same_instance = isinstance(other, self.__class__)
         return same_instance and self.raw_id == other.raw_id
 
     def __json__(self):
         parents = []
         try:
             for parent in self.parents:
                 parents.append({'raw_id': parent.raw_id})
         except NotImplementedError:
             # empty commit doesn't have parents implemented
             pass
 
         return {
             'short_id': self.short_id,
             'raw_id': self.raw_id,
             'revision': self.idx,
             'message': self.message,
             'date': self.date,
             'author': self.author,
             'parents': parents,
             'branch': self.branch
         }
 
     def __getstate__(self):
         d = self.__dict__.copy()
         d.pop('_remote', None)
         d.pop('repository', None)
         return d
 
     def get_remote(self):
         return self._remote
 
     def serialize(self):
         return self.__json__()
 
     def _get_refs(self):
         return {
             'branches': [self.branch] if self.branch else [],
             'bookmarks': getattr(self, 'bookmarks', []),
             'tags': self.tags
         }
 
     @LazyProperty
     def last(self):
         """
         ``True`` if this is last commit in repository, ``False``
         otherwise; trying to access this attribute while there is no
         commits would raise `EmptyRepositoryError`
         """
         if self.repository is None:
             raise CommitError("Cannot check if it's most recent commit")
         return self.raw_id == self.repository.commit_ids[-1]
 
     @LazyProperty
     def parents(self):
         """
         Returns list of parent commits.
         """
         raise NotImplementedError
 
     @LazyProperty
     def first_parent(self):
         """
         Returns list of parent commits.
         """
         return self.parents[0] if self.parents else EmptyCommit()
 
     @property
     def merge(self):
         """
         Returns boolean if commit is a merge.
         """
         return len(self.parents) > 1
 
     @LazyProperty
     def children(self):
         """
         Returns list of child commits.
         """
         raise NotImplementedError
 
     @LazyProperty
     def id(self):
         """
         Returns string identifying this commit.
         """
         raise NotImplementedError
 
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this commit.
         """
         raise NotImplementedError
 
     @LazyProperty
     def short_id(self):
         """
         Returns shortened version of ``raw_id`` attribute, as string,
         identifying this commit, useful for presentation to users.
         """
         raise NotImplementedError
 
     @LazyProperty
     def idx(self):
         """
         Returns integer identifying this commit.
         """
         raise NotImplementedError
 
     @LazyProperty
     def committer(self):
         """
         Returns committer for this commit
         """
         raise NotImplementedError
 
     @LazyProperty
     def committer_name(self):
         """
         Returns committer name for this commit
         """
 
         return author_name(self.committer)
 
     @LazyProperty
     def committer_email(self):
         """
         Returns committer email address for this commit
         """
 
         return author_email(self.committer)
 
     @LazyProperty
     def author(self):
         """
         Returns author for this commit
         """
 
         raise NotImplementedError
 
     @LazyProperty
     def author_name(self):
         """
         Returns author name for this commit
         """
 
         return author_name(self.author)
 
     @LazyProperty
     def author_email(self):
         """
         Returns author email address for this commit
         """
 
         return author_email(self.author)
 
     def get_file_mode(self, path: bytes):
         """
         Returns stat mode of the file at `path`.
         """
         raise NotImplementedError
 
     def is_link(self, path):
         """
         Returns ``True`` if given `path` is a symlink
         """
         raise NotImplementedError
 
     def is_node_binary(self, path):
         """
         Returns ``True`` is given path is a binary file
         """
         raise NotImplementedError
 
     def node_md5_hash(self, path):
         """
         Returns md5 hash of a node data
         """
         raise NotImplementedError
 
     def get_file_content(self, path) -> bytes:
         """
         Returns content of the file at the given `path`.
         """
         raise NotImplementedError
 
     def get_file_content_streamed(self, path):
         """
         returns a streaming response from vcsserver with file content
         """
         raise NotImplementedError
 
     def get_file_size(self, path):
         """
         Returns size of the file at the given `path`.
         """
         raise NotImplementedError
 
     def get_path_commit(self, path, pre_load=None):
         """
         Returns last commit of the file at the given `path`.
 
         :param pre_load: Optional. List of commit attributes to load.
         """
         commits = self.get_path_history(path, limit=1, pre_load=pre_load)
         if not commits:
             raise RepositoryError(
                 'Failed to fetch history for path {}. '
                 'Please check if such path exists in your repository'.format(
                     path))
         return commits[0]
 
     def get_path_history(self, path, limit=None, pre_load=None):
         """
         Returns history of file as reversed list of :class:`BaseCommit`
         objects for which file at given `path` has been modified.
 
         :param limit: Optional. Allows to limit the size of the returned
            history. This is intended as a hint to the underlying backend, so
            that it can apply optimizations depending on the limit.
         :param pre_load: Optional. List of commit attributes to load.
         """
         raise NotImplementedError
 
     def get_file_annotate(self, path, pre_load=None):
         """
         Returns a generator of four element tuples with
         lineno, sha, commit lazy loader and line
 
         :param pre_load: Optional. List of commit attributes to load.
         """
         raise NotImplementedError
 
     def get_nodes(self, path, pre_load=None):
         """
         Returns combined ``DirNode`` and ``FileNode`` objects list representing
         state of commit at the given ``path``.
 
         :raises ``CommitError``: if node at the given ``path`` is not
           instance of ``DirNode``
         """
         raise NotImplementedError
 
     def get_node(self, path):
         """
         Returns ``Node`` object from the given ``path``.
 
         :raises ``NodeDoesNotExistError``: if there is no node at the given
           ``path``
         """
         raise NotImplementedError
 
     def get_largefile_node(self, path):
         """
         Returns the path to largefile from Mercurial/Git-lfs storage.
         or None if it's not a largefile node
         """
         return None
 
     def archive_repo(self, archive_name_key, kind='tgz', subrepos=None,
                      archive_dir_name=None, write_metadata=False, mtime=None,
                      archive_at_path='/', cache_config=None):
         """
         Creates an archive containing the contents of the repository.
 
         :param archive_name_key: unique key under this archive should be generated
         :param kind: one of the following: ``"tbz2"``, ``"tgz"``, ``"zip"``.
         :param archive_dir_name: name of root directory in archive.
             Default is repository name and commit's short_id joined with dash:
             ``"{repo_name}-{short_id}"``.
         :param write_metadata: write a metadata file into archive.
         :param mtime: custom modification time for archive creation, defaults
             to time.time() if not given.
         :param archive_at_path: pack files at this path (default '/')
         :param cache_config: config spec to send to vcsserver to configure the backend to store files
 
         :raise VCSError: If prefix has a problem.
         """
         cache_config = cache_config or {}
         allowed_kinds = [x[0] for x in settings.ARCHIVE_SPECS]
         if kind not in allowed_kinds:
             raise ImproperArchiveTypeError(
                 f'Archive kind ({kind}) not supported use one of {allowed_kinds}')
 
         archive_dir_name = self._validate_archive_prefix(archive_dir_name)
         mtime = mtime is not None or time.mktime(self.date.timetuple())
         commit_id = self.raw_id
 
         return self.repository._remote.archive_repo(
             archive_name_key, kind, mtime, archive_at_path,
             archive_dir_name, commit_id, cache_config)
 
     def _validate_archive_prefix(self, archive_dir_name):
         if archive_dir_name is None:
             archive_dir_name = self._ARCHIVE_PREFIX_TEMPLATE.format(
                 repo_name=safe_str(self.repository.name),
                 short_id=self.short_id)
         elif not isinstance(archive_dir_name, str):
             raise ValueError(f"archive_dir_name is not str object but: {type(archive_dir_name)}")
         elif archive_dir_name.startswith('/'):
             raise VCSError("Prefix cannot start with leading slash")
         elif archive_dir_name.strip() == '':
             raise VCSError("Prefix cannot be empty")
         elif not archive_dir_name.isascii():
             raise VCSError("Prefix cannot contain non ascii characters")
         return archive_dir_name
 
     @LazyProperty
     def root(self):
         """
         Returns ``RootNode`` object for this commit.
         """
         return self.get_node('')
 
     def next(self, branch=None):
         """
         Returns next commit from current, if branch is gives it will return
         next commit belonging to this branch
 
         :param branch: show commits within the given named branch
         """
         indexes = range(self.idx + 1, self.repository.count())
         return self._find_next(indexes, branch)
 
     def prev(self, branch=None):
         """
         Returns previous commit from current, if branch is gives it will
         return previous commit belonging to this branch
 
         :param branch: show commit within the given named branch
         """
         indexes = range(self.idx - 1, -1, -1)
         return self._find_next(indexes, branch)
 
     def _find_next(self, indexes, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on commit not belonging '
                            'to that branch')
 
         for next_idx in indexes:
             commit = self.repository.get_commit(commit_idx=next_idx)
             if branch and branch != commit.branch:
                 continue
             return commit
         raise CommitDoesNotExistError
 
     def diff(self, ignore_whitespace=True, context=3):
         """
         Returns a `Diff` object representing the change made by this commit.
         """
         parent = self.first_parent
         diff = self.repository.get_diff(
             parent, self,
             ignore_whitespace=ignore_whitespace,
             context=context)
         return diff
 
     @LazyProperty
     def added(self):
         """
         Returns list of added ``FileNode`` objects.
         """
         raise NotImplementedError
 
     @LazyProperty
     def changed(self):
         """
         Returns list of modified ``FileNode`` objects.
         """
         raise NotImplementedError
 
     @LazyProperty
     def removed(self):
         """
         Returns list of removed ``FileNode`` objects.
         """
         raise NotImplementedError
 
     @LazyProperty
     def size(self):
         """
         Returns total number of bytes from contents of all filenodes.
         """
         return sum(node.size for node in self.get_filenodes_generator())
 
     def walk(self, topurl=''):
         """
         Similar to os.walk method. Insted of filesystem it walks through
         commit starting at given ``topurl``.  Returns generator of tuples
         (top_node, dirnodes, filenodes).
         """
         from rhodecode.lib.vcs.nodes import DirNode
 
         if isinstance(topurl, DirNode):
             top_node = topurl
         else:
             top_node = self.get_node(topurl)
 
         has_default_pre_load = False
         if isinstance(top_node, DirNode):
             # used to inject as we walk same defaults as given top_node
             default_pre_load = top_node.default_pre_load
             has_default_pre_load = True
 
         if not top_node.is_dir():
             return
         yield top_node, top_node.dirs, top_node.files
         for dir_node in top_node.dirs:
             if has_default_pre_load:
                 dir_node.default_pre_load = default_pre_load
             yield from self.walk(dir_node)
 
     def get_filenodes_generator(self):
         """
         Returns generator that yields *all* file nodes.
         """
         for topnode, dirs, files in self.walk():
             yield from files
 
     #
     # Utilities for sub classes to support consistent behavior
     #
 
     def no_node_at_path(self, path):
         return NodeDoesNotExistError(
             f"There is no file nor directory at the given path: "
             f"`{safe_str(path)}` at commit {self.short_id}")
 
     def _fix_path(self, path: str) -> str:
         """
         Paths are stored without trailing slash so we need to get rid off it if
         needed.
         """
         return safe_str(path).rstrip('/')
 
     #
     # Deprecated API based on changesets
     #
 
     @property
     def revision(self):
         warnings.warn("Use idx instead", DeprecationWarning)
         return self.idx
 
     @revision.setter
     def revision(self, value):
         warnings.warn("Use idx instead", DeprecationWarning)
         self.idx = value
 
     def get_file_changeset(self, path):
         warnings.warn("Use get_path_commit instead", DeprecationWarning)
         return self.get_path_commit(path)
 
 
 class BaseChangesetClass(type):
 
     def __instancecheck__(self, instance):
         return isinstance(instance, BaseCommit)
 
 
 class BaseChangeset(BaseCommit, metaclass=BaseChangesetClass):
 
     def __new__(cls, *args, **kwargs):
         warnings.warn(
             "Use BaseCommit instead of BaseChangeset", DeprecationWarning)
         return super().__new__(cls, *args, **kwargs)
 
 
 class BaseInMemoryCommit(object):
     """
     Represents differences between repository's state (most recent head) and
     changes made *in place*.
 
     **Attributes**
 
         ``repository``
             repository object for this in-memory-commit
 
         ``added``
             list of ``FileNode`` objects marked as *added*
 
         ``changed``
             list of ``FileNode`` objects marked as *changed*
 
         ``removed``
             list of ``FileNode`` or ``RemovedFileNode`` objects marked to be
             *removed*
 
         ``parents``
             list of :class:`BaseCommit` instances representing parents of
             in-memory commit. Should always be 2-element sequence.
 
     """
 
     def __init__(self, repository):
         self.repository = repository
         self.added = []
         self.changed = []
         self.removed = []
         self.parents = []
 
     def add(self, *filenodes):
         """
         Marks given ``FileNode`` objects as *to be committed*.
 
         :raises ``NodeAlreadyExistsError``: if node with same path exists at
            latest commit
         :raises ``NodeAlreadyAddedError``: if node with same path is already
            marked as *added*
         """
         # Check if not already marked as *added* first
         for node in filenodes:
             if node.path in (n.path for n in self.added):
                 raise NodeAlreadyAddedError(
                     "Such FileNode %s is already marked for addition"
                     % node.path)
         for node in filenodes:
             self.added.append(node)
 
     def change(self, *filenodes):
         """
         Marks given ``FileNode`` objects to be *changed* in next commit.
 
         :raises ``EmptyRepositoryError``: if there are no commits yet
         :raises ``NodeAlreadyExistsError``: if node with same path is already
            marked to be *changed*
         :raises ``NodeAlreadyRemovedError``: if node with same path is already
            marked to be *removed*
         :raises ``NodeDoesNotExistError``: if node doesn't exist in latest
            commit
         :raises ``NodeNotChangedError``: if node hasn't really be changed
         """
         for node in filenodes:
             if node.path in (n.path for n in self.removed):
                 raise NodeAlreadyRemovedError(
                     "Node at %s is already marked as removed" % node.path)
         try:
             self.repository.get_commit()
         except EmptyRepositoryError:
             raise EmptyRepositoryError(
                 "Nothing to change - try to *add* new nodes rather than "
                 "changing them")
         for node in filenodes:
             if node.path in (n.path for n in self.changed):
                 raise NodeAlreadyChangedError(
                     "Node at '%s' is already marked as changed" % node.path)
             self.changed.append(node)
 
     def remove(self, *filenodes):
         """
         Marks given ``FileNode`` (or ``RemovedFileNode``) objects to be
         *removed* in next commit.
 
         :raises ``NodeAlreadyRemovedError``: if node has been already marked to
            be *removed*
         :raises ``NodeAlreadyChangedError``: if node has been already marked to
            be *changed*
         """
         for node in filenodes:
             if node.path in (n.path for n in self.removed):
                 raise NodeAlreadyRemovedError(
                     "Node is already marked to for removal at %s" % node.path)
             if node.path in (n.path for n in self.changed):
                 raise NodeAlreadyChangedError(
                     "Node is already marked to be changed at %s" % node.path)
             # We only mark node as *removed* - real removal is done by
             # commit method
             self.removed.append(node)
 
     def reset(self):
         """
         Resets this instance to initial state (cleans ``added``, ``changed``
         and ``removed`` lists).
         """
         self.added = []
         self.changed = []
         self.removed = []
         self.parents = []
 
     def get_ipaths(self):
         """
         Returns generator of paths from nodes marked as added, changed or
         removed.
         """
         for node in itertools.chain(self.added, self.changed, self.removed):
             yield node.path
 
     def get_paths(self):
         """
         Returns list of paths from nodes marked as added, changed or removed.
         """
         return list(self.get_ipaths())
 
     def check_integrity(self, parents=None):
         """
         Checks in-memory commit's integrity. Also, sets parents if not
         already set.
 
         :raises CommitError: if any error occurs (i.e.
           ``NodeDoesNotExistError``).
         """
         if not self.parents:
             parents = parents or []
             if len(parents) == 0:
                 try:
                     parents = [self.repository.get_commit(), None]
                 except EmptyRepositoryError:
                     parents = [None, None]
             elif len(parents) == 1:
                 parents += [None]
             self.parents = parents
 
         # Local parents, only if not None
         parents = [p for p in self.parents if p]
 
         # Check nodes marked as added
         for p in parents:
             for node in self.added:
                 try:
                     p.get_node(node.path)
                 except NodeDoesNotExistError:
                     pass
                 else:
                     raise NodeAlreadyExistsError(
                         f"Node `{node.path}` already exists at {p}")
 
         # Check nodes marked as changed
         missing = set(self.changed)
         not_changed = set(self.changed)
         if self.changed and not parents:
             raise NodeDoesNotExistError(str(self.changed[0].path))
         for p in parents:
             for node in self.changed:
                 try:
                     old = p.get_node(node.path)
                     missing.remove(node)
                     # if content actually changed, remove node from not_changed
                     if old.content != node.content:
                         not_changed.remove(node)
                 except NodeDoesNotExistError:
                     pass
         if self.changed and missing:
             raise NodeDoesNotExistError(
                 "Node `%s` marked as modified but missing in parents: %s"
                 % (node.path, parents))
 
         if self.changed and not_changed:
             raise NodeNotChangedError(
                 "Node `%s` wasn't actually changed (parents: %s)"
                 % (not_changed.pop().path, parents))
 
         # Check nodes marked as removed
         if self.removed and not parents:
             raise NodeDoesNotExistError(
                 "Cannot remove node at %s as there "
                 "were no parents specified" % self.removed[0].path)
         really_removed = set()
         for p in parents:
             for node in self.removed:
                 try:
                     p.get_node(node.path)
                     really_removed.add(node)
                 except CommitError:
                     pass
         not_removed = set(self.removed) - really_removed
         if not_removed:
             # TODO: johbo: This code branch does not seem to be covered
             raise NodeDoesNotExistError(
                 "Cannot remove node at %s from "
                 "following parents: %s" % (not_removed, parents))
 
     def commit(self, message, author, parents=None, branch=None, date=None, **kwargs):
         """
         Performs in-memory commit (doesn't check workdir in any way) and
         returns newly created :class:`BaseCommit`. Updates repository's
         attribute `commits`.
 
         .. note::
 
            While overriding this method each backend's should call
            ``self.check_integrity(parents)`` in the first place.
 
         :param message: message of the commit
         :param author: full username, i.e. "Joe Doe <[e-mail]>"
         :param parents: single parent or sequence of parents from which commit
            would be derived
         :param date: ``datetime.datetime`` instance. Defaults to
            ``datetime.datetime.now()``.
         :param branch: branch name, as string. If none given, default backend's
            branch would be used.
 
         :raises ``CommitError``: if any error occurs while committing
         """
         raise NotImplementedError
 
 
 class BaseInMemoryChangesetClass(type):
 
     def __instancecheck__(self, instance):
         return isinstance(instance, BaseInMemoryCommit)
 
 
 class BaseInMemoryChangeset(BaseInMemoryCommit, metaclass=BaseInMemoryChangesetClass):
 
     def __new__(cls, *args, **kwargs):
         warnings.warn(
             "Use BaseCommit instead of BaseInMemoryCommit", DeprecationWarning)
         return super().__new__(cls, *args, **kwargs)
 
 
 class EmptyCommit(BaseCommit):
     """
     An dummy empty commit. It's possible to pass hash when creating
     an EmptyCommit
     """
 
     def __init__(
             self, commit_id=EMPTY_COMMIT_ID, repo=None, alias=None, idx=-1,
             message='', author='', date=None):
         self._empty_commit_id = commit_id
         # TODO: johbo: Solve idx parameter, default value does not make
         # too much sense
         self.idx = idx
         self.message = message
         self.author = author
         self.date = date or datetime.datetime.fromtimestamp(0)
         self.repository = repo
         self.alias = alias
 
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this commit, useful for web
         representation.
         """
 
         return self._empty_commit_id
 
     @LazyProperty
     def branch(self):
         if self.alias:
             from rhodecode.lib.vcs.backends import get_backend
             return get_backend(self.alias).DEFAULT_BRANCH_NAME
 
     @LazyProperty
     def short_id(self):
         return self.raw_id[:12]
 
     @LazyProperty
     def id(self):
         return self.raw_id
 
     def get_path_commit(self, path, pre_load=None):
         return self
 
     def get_file_content(self, path) -> bytes:
         return b''
 
     def get_file_content_streamed(self, path):
         yield self.get_file_content(path)
 
     def get_file_size(self, path):
         return 0
 
 
 class EmptyChangesetClass(type):
 
     def __instancecheck__(self, instance):
         return isinstance(instance, EmptyCommit)
 
 
 class EmptyChangeset(EmptyCommit, metaclass=EmptyChangesetClass):
 
     def __new__(cls, *args, **kwargs):
         warnings.warn(
             "Use EmptyCommit instead of EmptyChangeset", DeprecationWarning)
         return super(EmptyCommit, cls).__new__(cls, *args, **kwargs)
 
     def __init__(self, cs=EMPTY_COMMIT_ID, repo=None, requested_revision=None,
                  alias=None, revision=-1, message='', author='', date=None):
         if requested_revision is not None:
             warnings.warn(
                 "Parameter requested_revision not supported anymore",
                 DeprecationWarning)
         super().__init__(
             commit_id=cs, repo=repo, alias=alias, idx=revision,
             message=message, author=author, date=date)
 
         @property
         def revision(self):
             warnings.warn("Use idx instead", DeprecationWarning)
             return self.idx
 
         @revision.setter
         def revision(self, value):
             warnings.warn("Use idx instead", DeprecationWarning)
             self.idx = value
 
 
 class EmptyRepository(BaseRepository):
     def __init__(self, repo_path=None, config=None, create=False, **kwargs):
         pass
 
     def get_diff(self, *args, **kwargs):
         from rhodecode.lib.vcs.backends.git.diff import GitDiff
         return GitDiff(b'')
 
 
 class CollectionGenerator(object):
 
     def __init__(self, repo, commit_ids, collection_size=None, pre_load=None, translate_tag=None):
         self.repo = repo
         self.commit_ids = commit_ids
         self.collection_size = collection_size
         self.pre_load = pre_load
         self.translate_tag = translate_tag
 
     def __len__(self):
         if self.collection_size is not None:
             return self.collection_size
         return self.commit_ids.__len__()
 
     def __iter__(self):
         for commit_id in self.commit_ids:
             # TODO: johbo: Mercurial passes in commit indices or commit ids
             yield self._commit_factory(commit_id)
 
     def _commit_factory(self, commit_id):
         """
         Allows backends to override the way commits are generated.
         """
         return self.repo.get_commit(
             commit_id=commit_id, pre_load=self.pre_load,
             translate_tag=self.translate_tag)
 
     def __getitem__(self, key):
         """Return either a single element by index, or a sliced collection."""
 
         if isinstance(key, slice):
             commit_ids = self.commit_ids[key.start:key.stop]
 
         else:
             # single item
             commit_ids = self.commit_ids[key]
 
         return self.__class__(
             self.repo, commit_ids, pre_load=self.pre_load,
             translate_tag=self.translate_tag)
 
     def __repr__(self):
         return '<CollectionGenerator[len:%s]>' % (self.__len__())
 
 
 class Config(object):
     """
     Represents the configuration for a repository.
 
     The API is inspired by :class:`ConfigParser.ConfigParser` from the
     standard library. It implements only the needed subset.
     """
 
     def __init__(self):
         self._values = {}
 
     def copy(self):
         clone = Config()
         for section, values in self._values.items():
             clone._values[section] = values.copy()
         return clone
 
     def __repr__(self):
         return '<Config({} sections) at {}>'.format(
             len(self._values), hex(id(self)))
 
     def items(self, section):
         return self._values.get(section, {}).items()
 
     def get(self, section, option):
         return self._values.get(section, {}).get(option)
 
     def set(self, section, option, value):
         section_values = self._values.setdefault(section, {})
         section_values[option] = value
 
     def clear_section(self, section):
         self._values[section] = {}
 
     def serialize(self):
         """
         Creates a list of three tuples (section, key, value) representing
         this config object.
         """
         items = []
         for section in self._values:
             for option, value in self._values[section].items():
                 items.append(
                     (safe_str(section), safe_str(option), safe_str(value)))
         return items
 
 
 class Diff(object):
     """
     Represents a diff result from a repository backend.
 
     Subclasses have to provide a backend specific value for
     :attr:`_header_re` and :attr:`_meta_re`.
     """
     _meta_re = None
     _header_re: bytes = re.compile(br"")
 
     def __init__(self, raw_diff: bytes):
         if not isinstance(raw_diff, bytes):
             raise Exception(f'raw_diff must be bytes - got {type(raw_diff)}')
 
         self.raw = memoryview(raw_diff)
 
     def get_header_re(self):
         return self._header_re
 
     def chunks(self):
         """
         split the diff in chunks of separate --git a/file b/file chunks
         to make diffs consistent we must prepend with \n, and make sure
         we can detect last chunk as this was also has special rule
         """
 
         diff_parts = (b'\n' + bytes(self.raw)).split(b'\ndiff --git')
 
         chunks = diff_parts[1:]
         total_chunks = len(chunks)
 
         def diff_iter(_chunks):
             for cur_chunk, chunk in enumerate(_chunks, start=1):
                 yield DiffChunk(chunk, self, cur_chunk == total_chunks)
         return diff_iter(chunks)
 
 
 class DiffChunk(object):
 
     def __init__(self, chunk: bytes, diff_obj: Diff, is_last_chunk: bool):
         self.diff_obj = diff_obj
 
         # since we split by \ndiff --git that part is lost from original diff
         # we need to re-apply it at the end, EXCEPT ! if it's last chunk
         if not is_last_chunk:
             chunk += b'\n'
         header_re = self.diff_obj.get_header_re()
         match = header_re.match(chunk)
         self.header = match.groupdict()
         self.diff = chunk[match.end():]
         self.raw = chunk
 
     @property
     def header_as_str(self):
         if self.header:
             def safe_str_on_bytes(val):
                 if isinstance(val, bytes):
                     return safe_str(val)
                 return val
             return {safe_str(k): safe_str_on_bytes(v) for k, v in self.header.items()}
 
     def __repr__(self):
         return f'DiffChunk({self.header_as_str})'
 
 
 class BasePathPermissionChecker(object):
 
     @staticmethod
     def create_from_patterns(includes, excludes):
         if includes and '*' in includes and not excludes:
             return AllPathPermissionChecker()
         elif excludes and '*' in excludes:
             return NonePathPermissionChecker()
         else:
             return PatternPathPermissionChecker(includes, excludes)
 
     @property
     def has_full_access(self):
         raise NotImplementedError()
 
     def has_access(self, path):
         raise NotImplementedError()
 
 
 class AllPathPermissionChecker(BasePathPermissionChecker):
 
     @property
     def has_full_access(self):
         return True
 
     def has_access(self, path):
         return True
 
 
 class NonePathPermissionChecker(BasePathPermissionChecker):
 
     @property
     def has_full_access(self):
         return False
 
     def has_access(self, path):
         return False
 
 
 class PatternPathPermissionChecker(BasePathPermissionChecker):
 
     def __init__(self, includes, excludes):
         self.includes = includes
         self.excludes = excludes
         self.includes_re = [] if not includes else [
             re.compile(fnmatch.translate(pattern)) for pattern in includes]
         self.excludes_re = [] if not excludes else [
             re.compile(fnmatch.translate(pattern)) for pattern in excludes]
 
     @property
     def has_full_access(self):
         return '*' in self.includes and not self.excludes
 
     def has_access(self, path):
         for regex in self.excludes_re:
             if regex.match(path):
                 return False
         for regex in self.includes_re:
             if regex.match(path):
                 return True
         return False

rhodecode/lib/vcs/backends/git/repository.py

@@ -1,1053 +1,1054 b''
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 GIT repository module
 """
 
 import logging
 import os
 import re
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 from collections import OrderedDict
 from rhodecode.lib.datelib import (
     utcdate_fromtimestamp, makedate, date_astimestamp)
 from rhodecode.lib.hash_utils import safe_str
 from rhodecode.lib.utils2 import CachedProperty
 from rhodecode.lib.vcs import connection, path as vcspath
 from rhodecode.lib.vcs.backends.base import (
     BaseRepository, CollectionGenerator, Config, MergeResponse,
     MergeFailureReason, Reference)
 from rhodecode.lib.vcs.backends.git.commit import GitCommit
 from rhodecode.lib.vcs.backends.git.diff import GitDiff
 from rhodecode.lib.vcs.backends.git.inmemory import GitInMemoryCommit
 from rhodecode.lib.vcs.exceptions import (
     CommitDoesNotExistError, EmptyRepositoryError,
     RepositoryError, TagAlreadyExistError, TagDoesNotExistError, VCSError, UnresolvedFilesInRepo)
 
 
 SHA_PATTERN = re.compile(r'^([0-9a-fA-F]{12}|[0-9a-fA-F]{40})$')
 
 log = logging.getLogger(__name__)
 
 
 class GitRepository(BaseRepository):
     """
     Git repository backend.
     """
     DEFAULT_BRANCH_NAME = os.environ.get('GIT_DEFAULT_BRANCH_NAME') or 'master'
     DEFAULT_REF = f'branch:{DEFAULT_BRANCH_NAME}'
 
     contact = BaseRepository.DEFAULT_CONTACT
 
     def __init__(self, repo_path, config=None, create=False, src_url=None,
                  do_workspace_checkout=False, with_wire=None, bare=False):
 
         self.path = safe_str(os.path.abspath(repo_path))
         self.config = config if config else self.get_default_config()
         self.with_wire = with_wire or {"cache": False}  # default should not use cache
 
         self._init_repo(create, src_url, do_workspace_checkout, bare)
 
         # caches
         self._commit_ids = {}
 
     @LazyProperty
     def _remote(self):
         repo_id = self.path
         return connection.Git(self.path, repo_id, self.config, with_wire=self.with_wire)
 
     @LazyProperty
     def bare(self):
         return self._remote.bare()
 
     @LazyProperty
     def head(self):
         return self._remote.head()
 
     @CachedProperty
     def commit_ids(self):
         """
         Returns list of commit ids, in ascending order.  Being lazy
         attribute allows external tools to inject commit ids from cache.
         """
         commit_ids = self._get_all_commit_ids()
         self._rebuild_cache(commit_ids)
         return commit_ids
 
     def _rebuild_cache(self, commit_ids):
         self._commit_ids = {commit_id: index
                                 for index, commit_id in enumerate(commit_ids)}
 
     def run_git_command(self, cmd, **opts):
         """
         Runs given ``cmd`` as git command and returns tuple
         (stdout, stderr).
 
         :param cmd: git command to be executed
         :param opts: env options to pass into Subprocess command
         """
         if not isinstance(cmd, list):
             raise ValueError(f'cmd must be a list, got {type(cmd)} instead')
 
         skip_stderr_log = opts.pop('skip_stderr_log', False)
         out, err = self._remote.run_git_command(cmd, **opts)
         if err and not skip_stderr_log:
             log.debug('Stderr output of git command "%s":\n%s', cmd, err)
         return out, err
 
     @staticmethod
     def check_url(url, config):
         """
         Function will check given url and try to verify if it's a valid
         link. Sometimes it may happened that git will issue basic
         auth request that can cause whole API to hang when used from python
         or other external calls.
 
         On failures it'll raise urllib2.HTTPError, exception is also thrown
         when the return code is non 200
         """
         # check first if it's not an url
         if os.path.isdir(url) or url.startswith('file:'):
             return True
 
         if '+' in url.split('://', 1)[0]:
             url = url.split('+', 1)[1]
 
         # Request the _remote to verify the url
         return connection.Git.check_url(url, config.serialize())
 
     @staticmethod
     def is_valid_repository(path):
         if os.path.isdir(os.path.join(path, '.git')):
             return True
         # check case of bare repository
         try:
             GitRepository(path)
             return True
         except VCSError:
             pass
         return False
 
     def _init_repo(self, create, src_url=None, do_workspace_checkout=False,
                    bare=False):
         if create and os.path.exists(self.path):
             raise RepositoryError(
                 "Cannot create repository at %s, location already exist"
                 % self.path)
 
         if bare and do_workspace_checkout:
             raise RepositoryError("Cannot update a bare repository")
         try:
 
             if src_url:
                 # check URL before any actions
                 GitRepository.check_url(src_url, self.config)
 
             if create:
                 os.makedirs(self.path, mode=0o755)
 
                 if bare:
                     self._remote.init_bare()
                 else:
                     self._remote.init()
 
                 if src_url and bare:
                     # bare repository only allows a fetch and checkout is not allowed
                     self.fetch(src_url, commit_ids=None)
                 elif src_url:
                     self.pull(src_url, commit_ids=None,
                               update_after=do_workspace_checkout)
 
             else:
                 if not self._remote.assert_correct_path():
                     raise RepositoryError(
                         'Path "%s" does not contain a Git repository' %
                         (self.path,))
 
         # TODO: johbo: check if we have to translate the OSError here
         except OSError as err:
             raise RepositoryError(err)
 
     def _get_all_commit_ids(self):
         return self._remote.get_all_commit_ids()
 
     def _get_commit_ids(self, filters=None):
         # we must check if this repo is not empty, since later command
         # fails if it is. And it's cheaper to ask than throw the subprocess
         # errors
 
         head = self._remote.head(show_exc=False)
 
         if not head:
             return []
 
         rev_filter = ['--branches', '--tags']
         extra_filter = []
 
         if filters:
             if filters.get('since'):
                 extra_filter.append('--since=%s' % (filters['since']))
             if filters.get('until'):
                 extra_filter.append('--until=%s' % (filters['until']))
             if filters.get('branch_name'):
                 rev_filter = []
                 extra_filter.append(filters['branch_name'])
         rev_filter.extend(extra_filter)
 
             # if filters.get('start') or filters.get('end'):
             #     # skip is offset, max-count is limit
             #     if filters.get('start'):
             #         extra_filter += ' --skip=%s' % filters['start']
             #     if filters.get('end'):
             #         extra_filter += ' --max-count=%s' % (filters['end'] - (filters['start'] or 0))
 
         cmd = ['rev-list', '--reverse', '--date-order'] + rev_filter
         try:
             output, __ = self.run_git_command(cmd)
         except RepositoryError:
             # Can be raised for empty repositories
             return []
         return output.splitlines()
 
     def _lookup_commit(self, commit_id_or_idx, translate_tag=True, maybe_unreachable=False, reference_obj=None):
 
         def is_null(value):
             return len(value) == commit_id_or_idx.count('0')
 
         if commit_id_or_idx in (None, '', 'tip', 'HEAD', 'head', -1):
             return self.commit_ids[-1]
 
         commit_missing_err = "Commit {} does not exist for `{}`".format(
             *map(safe_str, [commit_id_or_idx, self.name]))
 
         is_bstr = isinstance(commit_id_or_idx, str)
         is_branch = reference_obj and reference_obj.branch
 
         lookup_ok = False
         if is_bstr:
             # Need to call remote to translate id for tagging scenarios,
             # or branch that are numeric
             try:
                 remote_data = self._remote.get_object(commit_id_or_idx,
                                                       maybe_unreachable=maybe_unreachable)
                 commit_id_or_idx = remote_data["commit_id"]
                 lookup_ok = True
             except (CommitDoesNotExistError,):
                 lookup_ok = False
 
         if lookup_ok is False:
             is_numeric_idx = \
                 (is_bstr and commit_id_or_idx.isdigit() and len(commit_id_or_idx) < 12) \
                 or isinstance(commit_id_or_idx, int)
             if not is_branch and (is_numeric_idx or is_null(commit_id_or_idx)):
                 try:
                     commit_id_or_idx = self.commit_ids[int(commit_id_or_idx)]
                     lookup_ok = True
                 except Exception:
                     raise CommitDoesNotExistError(commit_missing_err)
 
         # we failed regular lookup, and by integer number lookup
         if lookup_ok is False:
             raise CommitDoesNotExistError(commit_missing_err)
 
         # Ensure we return full id
         if not SHA_PATTERN.match(str(commit_id_or_idx)):
             raise CommitDoesNotExistError(
                 "Given commit id %s not recognized" % commit_id_or_idx)
         return commit_id_or_idx
 
     def get_hook_location(self):
         """
         returns absolute path to location where hooks are stored
         """
         loc = os.path.join(self.path, 'hooks')
         if not self.bare:
             loc = os.path.join(self.path, '.git', 'hooks')
         return loc
 
     @LazyProperty
     def last_change(self):
         """
         Returns last change made on this repository as
         `datetime.datetime` object.
         """
         try:
             return self.get_commit().date
         except RepositoryError:
             tzoffset = makedate()[1]
             return utcdate_fromtimestamp(self._get_fs_mtime(), tzoffset)
 
     def _get_fs_mtime(self):
         idx_loc = '' if self.bare else '.git'
         # fallback to filesystem
         in_path = os.path.join(self.path, idx_loc, "index")
         he_path = os.path.join(self.path, idx_loc, "HEAD")
         if os.path.exists(in_path):
             return os.stat(in_path).st_mtime
         else:
             return os.stat(he_path).st_mtime
 
     @LazyProperty
     def description(self):
         description = self._remote.get_description()
         return safe_str(description or self.DEFAULT_DESCRIPTION)
 
     def _get_refs_entries(self, prefix='', reverse=False, strip_prefix=True):
         if self.is_empty():
             return OrderedDict()
 
         result = []
         for ref, sha in self._refs.items():
             if ref.startswith(prefix):
                 ref_name = ref
                 if strip_prefix:
                     ref_name = ref[len(prefix):]
                 result.append((safe_str(ref_name), sha))
 
         def get_name(entry):
             return entry[0]
 
         return OrderedDict(sorted(result, key=get_name, reverse=reverse))
 
     def _get_branches(self):
         return self._get_refs_entries(prefix='refs/heads/', strip_prefix=True)
 
     @CachedProperty
     def branches(self):
         return self._get_branches()
 
     @CachedProperty
     def branches_closed(self):
         return {}
 
     @CachedProperty
     def bookmarks(self):
         return {}
 
     @CachedProperty
     def branches_all(self):
         all_branches = {}
         all_branches.update(self.branches)
         all_branches.update(self.branches_closed)
         return all_branches
 
     @CachedProperty
     def tags(self):
         return self._get_tags()
 
     def _get_tags(self):
         return self._get_refs_entries(prefix='refs/tags/', strip_prefix=True, reverse=True)
 
     def tag(self, name, user, commit_id=None, message=None, date=None,
             **kwargs):
         # TODO: fix this method to apply annotated tags correct with message
         """
         Creates and returns a tag for the given ``commit_id``.
 
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <[e-mail]>"
         :param commit_id: commit id for which new tag would be created
         :param message: message of the tag's commit
         :param date: date of tag's commit
 
         :raises TagAlreadyExistError: if tag with same name already exists
         """
         if name in self.tags:
             raise TagAlreadyExistError("Tag %s already exists" % name)
         commit = self.get_commit(commit_id=commit_id)
         message = message or f"Added tag {name} for commit {commit.raw_id}"
 
         self._remote.set_refs('refs/tags/%s' % name, commit.raw_id)
 
         self._invalidate_prop_cache('tags')
         self._invalidate_prop_cache('_refs')
 
         return commit
 
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
 
         :param name: name of the tag to be removed
         :param user: full username, i.e.: "Joe Doe <[e-mail]>"
         :param message: message of the tag's removal commit
         :param date: date of tag's removal commit
 
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         if name not in self.tags:
             raise TagDoesNotExistError("Tag %s does not exist" % name)
 
         self._remote.tag_remove(name)
         self._invalidate_prop_cache('tags')
         self._invalidate_prop_cache('_refs')
 
     def _get_refs(self):
         return self._remote.get_refs()
 
     @CachedProperty
     def _refs(self):
         return self._get_refs()
 
     @property
     def _ref_tree(self):
         node = tree = {}
         for ref, sha in self._refs.items():
             path = ref.split('/')
             for bit in path[:-1]:
                 node = node.setdefault(bit, {})
             node[path[-1]] = sha
             node = tree
         return tree
 
     def get_remote_ref(self, ref_name):
         ref_key = f'refs/remotes/origin/{safe_str(ref_name)}'
         try:
             return self._refs[ref_key]
         except Exception:
             return
 
     def get_commit(self, commit_id=None, commit_idx=None, pre_load=None,
                    translate_tag=True, maybe_unreachable=False, reference_obj=None):
         """
         Returns `GitCommit` object representing commit from git repository
         at the given `commit_id` or head (most recent commit) if None given.
         """
 
         if self.is_empty():
             raise EmptyRepositoryError("There are no commits yet")
 
         if commit_id is not None:
             self._validate_commit_id(commit_id)
             try:
                 # we have cached idx, use it without contacting the remote
                 idx = self._commit_ids[commit_id]
                 return GitCommit(self, commit_id, idx, pre_load=pre_load)
             except KeyError:
                 pass
 
         elif commit_idx is not None:
             self._validate_commit_idx(commit_idx)
             try:
                 _commit_id = self.commit_ids[commit_idx]
                 if commit_idx < 0:
                     commit_idx = self.commit_ids.index(_commit_id)
                 return GitCommit(self, _commit_id, commit_idx, pre_load=pre_load)
             except IndexError:
                 commit_id = commit_idx
         else:
             commit_id = "tip"
 
         if translate_tag:
             commit_id = self._lookup_commit(
                 commit_id, maybe_unreachable=maybe_unreachable,
                 reference_obj=reference_obj)
 
         try:
             idx = self._commit_ids[commit_id]
         except KeyError:
             idx = -1
 
         return GitCommit(self, commit_id, idx, pre_load=pre_load)
 
     def get_commits(
             self, start_id=None, end_id=None, start_date=None, end_date=None,
             branch_name=None, show_hidden=False, pre_load=None, translate_tags=True):
         """
         Returns generator of `GitCommit` objects from start to end (both
         are inclusive), in ascending date order.
 
         :param start_id: None, str(commit_id)
         :param end_id: None, str(commit_id)
         :param start_date: if specified, commits with commit date less than
           ``start_date`` would be filtered out from returned set
         :param end_date: if specified, commits with commit date greater than
           ``end_date`` would be filtered out from returned set
         :param branch_name: if specified, commits not reachable from given
           branch would be filtered out from returned set
         :param show_hidden: Show hidden commits such as obsolete or hidden from
             Mercurial evolve
         :raise BranchDoesNotExistError: If given `branch_name` does not
             exist.
         :raise CommitDoesNotExistError: If commits for given `start` or
           `end` could not be found.
 
         """
         if self.is_empty():
             raise EmptyRepositoryError("There are no commits yet")
 
         self._validate_branch_name(branch_name)
 
         if start_id is not None:
             self._validate_commit_id(start_id)
         if end_id is not None:
             self._validate_commit_id(end_id)
 
         start_raw_id = self._lookup_commit(start_id)
         start_pos = self._commit_ids[start_raw_id] if start_id else None
         end_raw_id = self._lookup_commit(end_id)
         end_pos = max(0, self._commit_ids[end_raw_id]) if end_id else None
 
         if None not in [start_id, end_id] and start_pos > end_pos:
             raise RepositoryError(
                 "Start commit '%s' cannot be after end commit '%s'" %
                 (start_id, end_id))
 
         if end_pos is not None:
             end_pos += 1
 
         filter_ = []
         if branch_name:
             filter_.append({'branch_name': branch_name})
         if start_date and not end_date:
             filter_.append({'since': start_date})
         if end_date and not start_date:
             filter_.append({'until': end_date})
         if start_date and end_date:
             filter_.append({'since': start_date})
             filter_.append({'until': end_date})
 
         # if start_pos or end_pos:
         #     filter_.append({'start': start_pos})
         #     filter_.append({'end': end_pos})
 
         if filter_:
             revfilters = {
                 'branch_name': branch_name,
                 'since': start_date.strftime('%m/%d/%y %H:%M:%S') if start_date else None,
                 'until': end_date.strftime('%m/%d/%y %H:%M:%S') if end_date else None,
                 'start': start_pos,
                 'end': end_pos,
             }
             commit_ids = self._get_commit_ids(filters=revfilters)
 
         else:
             commit_ids = self.commit_ids
 
         if start_pos or end_pos:
             commit_ids = commit_ids[start_pos: end_pos]
 
         return CollectionGenerator(self, commit_ids, pre_load=pre_load,
                                    translate_tag=translate_tags)
 
     def get_diff(
             self, commit1, commit2, path='', ignore_whitespace=False,
             context=3, path1=None):
         """
         Returns (git like) *diff*, as plain text. Shows changes introduced by
         ``commit2`` since ``commit1``.
 
         :param commit1: Entry point from which diff is shown. Can be
           ``self.EMPTY_COMMIT`` - in this case, patch showing all
           the changes since empty state of the repository until ``commit2``
         :param commit2: Until which commits changes should be shown.
         :param path:
         :param ignore_whitespace: If set to ``True``, would not show whitespace
           changes. Defaults to ``False``.
         :param context: How many lines before/after changed lines should be
           shown. Defaults to ``3``.
         :param path1:
         """
         self._validate_diff_commits(commit1, commit2)
         if path1 is not None and path1 != path:
             raise ValueError("Diff of two different paths not supported.")
 
         if path:
             file_filter = path
         else:
             file_filter = None
 
         diff = self._remote.diff(
             commit1.raw_id, commit2.raw_id, file_filter=file_filter,
             opt_ignorews=ignore_whitespace,
             context=context)
 
         return GitDiff(diff)
 
     def strip(self, commit_id, branch_name):
         commit = self.get_commit(commit_id=commit_id)
         if commit.merge:
             raise Exception('Cannot reset to merge commit')
 
         # parent is going to be the new head now
         commit = commit.parents[0]
         self._remote.set_refs('refs/heads/%s' % branch_name, commit.raw_id)
 
         # clear cached properties
         self._invalidate_prop_cache('commit_ids')
         self._invalidate_prop_cache('_refs')
         self._invalidate_prop_cache('branches')
 
         return len(self.commit_ids)
 
     def get_common_ancestor(self, commit_id1, commit_id2, repo2):
         log.debug('Calculating common ancestor between %sc1:%s and %sc2:%s',
                   self, commit_id1, repo2, commit_id2)
 
         if commit_id1 == commit_id2:
             return commit_id1
 
         if self != repo2:
             commits = self._remote.get_missing_revs(
                 commit_id1, commit_id2, repo2.path)
             if commits:
                 commit = repo2.get_commit(commits[-1])
                 if commit.parents:
                     ancestor_id = commit.parents[0].raw_id
                 else:
                     ancestor_id = None
             else:
                 # no commits from other repo, ancestor_id is the commit_id2
                 ancestor_id = commit_id2
         else:
             output, __ = self.run_git_command(
                 ['merge-base', commit_id1, commit_id2])
             ancestor_id = self.COMMIT_ID_PAT.findall(output)[0]
 
         log.debug('Found common ancestor with sha: %s', ancestor_id)
 
         return ancestor_id
 
     def compare(self, commit_id1, commit_id2, repo2, merge, pre_load=None):
         repo1 = self
         ancestor_id = None
 
         if commit_id1 == commit_id2:
             commits = []
         elif repo1 != repo2:
             missing_ids = self._remote.get_missing_revs(commit_id1, commit_id2,
                                                         repo2.path)
             commits = [
                 repo2.get_commit(commit_id=commit_id, pre_load=pre_load)
                 for commit_id in reversed(missing_ids)]
         else:
             output, __ = repo1.run_git_command(
                 ['log', '--reverse', '--pretty=format: %H', '-s',
                  f'{commit_id1}..{commit_id2}'])
             commits = [
                 repo1.get_commit(commit_id=commit_id, pre_load=pre_load)
                 for commit_id in self.COMMIT_ID_PAT.findall(output)]
 
         return commits
 
     @LazyProperty
     def in_memory_commit(self):
         """
         Returns ``GitInMemoryCommit`` object for this repository.
         """
         return GitInMemoryCommit(self)
 
     def pull(self, url, commit_ids=None, update_after=False):
         """
         Pull changes from external location. Pull is different in GIT
         that fetch since it's doing a checkout
 
         :param commit_ids: Optional. Can be set to a list of commit ids
            which shall be pulled from the other repository.
         """
         refs = None
         if commit_ids is not None:
             remote_refs = self._remote.get_remote_refs(url)
             refs = [ref for ref in remote_refs if remote_refs[ref] in commit_ids]
         self._remote.pull(url, refs=refs, update_after=update_after)
         self._remote.invalidate_vcs_cache()
 
     def fetch(self, url, commit_ids=None):
         """
         Fetch all git objects from external location.
         """
         self._remote.sync_fetch(url, refs=commit_ids)
         self._remote.invalidate_vcs_cache()
 
     def push(self, url):
         refs = None
         self._remote.sync_push(url, refs=refs)
 
     def set_refs(self, ref_name, commit_id):
         self._remote.set_refs(ref_name, commit_id)
         self._invalidate_prop_cache('_refs')
 
     def remove_ref(self, ref_name):
         self._remote.remove_ref(ref_name)
         self._invalidate_prop_cache('_refs')
 
     def run_gc(self, prune=True):
         cmd = ['gc', '--aggressive']
         if prune:
             cmd += ['--prune=now']
         _stdout, stderr = self.run_git_command(cmd, fail_on_stderr=False)
         return stderr
 
     def _update_server_info(self):
         """
         runs gits update-server-info command in this repo instance
         """
         self._remote.update_server_info()
 
     def _current_branch(self):
         """
         Return the name of the current branch.
 
         It only works for non bare repositories (i.e. repositories with a
         working copy)
         """
         if self.bare:
             raise RepositoryError('Bare git repos do not have active branches')
 
         if self.is_empty():
             return None
 
         stdout, _ = self.run_git_command(['rev-parse', '--abbrev-ref', 'HEAD'])
         return stdout.strip()
 
     def _checkout(self, branch_name, create=False, force=False):
         """
         Checkout a branch in the working directory.
 
         It tries to create the branch if create is True, failing if the branch
         already exists.
 
         It only works for non bare repositories (i.e. repositories with a
         working copy)
         """
         if self.bare:
             raise RepositoryError('Cannot checkout branches in a bare git repo')
 
         cmd = ['checkout']
         if force:
             cmd.append('-f')
         if create:
             cmd.append('-b')
         cmd.append(branch_name)
         self.run_git_command(cmd, fail_on_stderr=False)
 
     def _create_branch(self, branch_name, commit_id):
         """
         creates a branch in a GIT repo
         """
         self._remote.create_branch(branch_name, commit_id)
 
     def _identify(self):
         """
         Return the current state of the working directory.
         """
         if self.bare:
             raise RepositoryError('Bare git repos do not have active branches')
 
         if self.is_empty():
             return None
 
         stdout, _ = self.run_git_command(['rev-parse', 'HEAD'])
         return stdout.strip()
 
     def _local_clone(self, clone_path, branch_name, source_branch=None):
         """
         Create a local clone of the current repo.
         """
         # N.B.(skreft): the --branch option is required as otherwise the shallow
         # clone will only fetch the active branch.
         cmd = ['clone', '--branch', branch_name,
                self.path, os.path.abspath(clone_path)]
 
         self.run_git_command(cmd, fail_on_stderr=False)
 
         # if we get the different source branch, make sure we also fetch it for
         # merge conditions
         if source_branch and source_branch != branch_name:
             # check if the ref exists.
             shadow_repo = GitRepository(os.path.abspath(clone_path))
             if shadow_repo.get_remote_ref(source_branch):
                 cmd = ['fetch', self.path, source_branch]
                 self.run_git_command(cmd, fail_on_stderr=False)
 
     def _local_fetch(self, repository_path, branch_name, use_origin=False):
         """
         Fetch a branch from a local repository.
         """
         repository_path = os.path.abspath(repository_path)
         if repository_path == self.path:
             raise ValueError('Cannot fetch from the same repository')
 
         if use_origin:
             branch_name = '+{branch}:refs/heads/{branch}'.format(
                 branch=branch_name)
 
         cmd = ['fetch', '--no-tags', '--update-head-ok',
                repository_path, branch_name]
         self.run_git_command(cmd, fail_on_stderr=False)
 
     def _local_reset(self, branch_name):
         branch_name = f'{branch_name}'
         cmd = ['reset', '--hard', branch_name, '--']
         self.run_git_command(cmd, fail_on_stderr=False)
 
     def _last_fetch_heads(self):
         """
         Return the last fetched heads that need merging.
 
         The algorithm is defined at
         https://github.com/git/git/blob/v2.1.3/git-pull.sh#L283
         """
         if not self.bare:
             fetch_heads_path = os.path.join(self.path, '.git', 'FETCH_HEAD')
         else:
             fetch_heads_path = os.path.join(self.path, 'FETCH_HEAD')
 
         heads = []
         with open(fetch_heads_path) as f:
             for line in f:
                 if '    not-for-merge   ' in line:
                     continue
                 line = re.sub('\t.*', '', line, flags=re.DOTALL)
                 heads.append(line)
 
         return heads
 
     def get_shadow_instance(self, shadow_repository_path, enable_hooks=False, cache=False):
         return GitRepository(shadow_repository_path, with_wire={"cache": cache})
 
     def _local_pull(self, repository_path, branch_name, ff_only=True):
         """
         Pull a branch from a local repository.
         """
         if self.bare:
             raise RepositoryError('Cannot pull into a bare git repository')
         # N.B.(skreft): The --ff-only option is to make sure this is a
         # fast-forward (i.e., we are only pulling new changes and there are no
         # conflicts with our current branch)
         # Additionally, that option needs to go before --no-tags, otherwise git
         # pull complains about it being an unknown flag.
         cmd = ['pull']
         if ff_only:
             cmd.append('--ff-only')
         cmd.extend(['--no-tags', repository_path, branch_name])
         self.run_git_command(cmd, fail_on_stderr=False)
 
     def _local_merge(self, merge_message, user_name, user_email, heads):
         """
         Merge the given head into the checked out branch.
 
         It will force a merge commit.
 
         Currently it raises an error if the repo is empty, as it is not possible
         to create a merge commit in an empty repo.
 
         :param merge_message: The message to use for the merge commit.
         :param heads: the heads to merge.
         """
         if self.bare:
             raise RepositoryError('Cannot merge into a bare git repository')
 
         if not heads:
             return
 
         if self.is_empty():
             # TODO(skreft): do something more robust in this case.
             raise RepositoryError('Do not know how to merge into empty repositories yet')
         unresolved = None
 
         # N.B.(skreft): the --no-ff option is used to enforce the creation of a
         # commit message. We also specify the user who is doing the merge.
         cmd = ['-c', f'user.name="{user_name}"',
                '-c', f'user.email={user_email}',
                'merge', '--no-ff', '-m', safe_str(merge_message)]
 
         merge_cmd = cmd + heads
 
         try:
             self.run_git_command(merge_cmd, fail_on_stderr=False)
         except RepositoryError:
             files = self.run_git_command(['diff', '--name-only', '--diff-filter', 'U'],
                                          fail_on_stderr=False)[0].splitlines()
             # NOTE(marcink): we add U notation for consistent with HG backend output
             unresolved = [f'U {f}' for f in files]
 
             # Cleanup any merge leftovers
             self._remote.invalidate_vcs_cache()
             self.run_git_command(['merge', '--abort'], fail_on_stderr=False)
 
             if unresolved:
                 raise UnresolvedFilesInRepo(unresolved)
             else:
                 raise
 
     def _local_push(
             self, source_branch, repository_path, target_branch,
             enable_hooks=False, rc_scm_data=None):
         """
         Push the source_branch to the given repository and target_branch.
 
         Currently it if the target_branch is not master and the target repo is
         empty, the push will work, but then GitRepository won't be able to find
         the pushed branch or the commits. As the HEAD will be corrupted (i.e.,
         pointing to master, which does not exist).
 
         It does not run the hooks in the target repo.
         """
         # TODO(skreft): deal with the case in which the target repo is empty,
         # and the target_branch is not master.
         target_repo = GitRepository(repository_path)
         if (not target_repo.bare and
                 target_repo._current_branch() == target_branch):
             # Git prevents pushing to the checked out branch, so simulate it by
             # pulling into the target repository.
             target_repo._local_pull(self.path, source_branch)
         else:
             cmd = ['push', os.path.abspath(repository_path),
                    f'{source_branch}:{target_branch}']
             gitenv = {}
             if rc_scm_data:
                 gitenv.update({'RC_SCM_DATA': rc_scm_data})
 
             if not enable_hooks:
                 gitenv['RC_SKIP_HOOKS'] = '1'
             self.run_git_command(cmd, fail_on_stderr=False, extra_env=gitenv)
 
     def _get_new_pr_branch(self, source_branch, target_branch):
         prefix = f'pr_{source_branch}-{target_branch}_'
         pr_branches = []
         for branch in self.branches:
             if branch.startswith(prefix):
                 pr_branches.append(int(branch[len(prefix):]))
 
         if not pr_branches:
             branch_id = 0
         else:
             branch_id = max(pr_branches) + 1
 
         return '%s%d' % (prefix, branch_id)
 
     def _maybe_prepare_merge_workspace(
             self, repo_id, workspace_id, target_ref, source_ref):
         shadow_repository_path = self._get_shadow_repository_path(
             self.path, repo_id, workspace_id)
         if not os.path.exists(shadow_repository_path):
             self._local_clone(
                 shadow_repository_path, target_ref.name, source_ref.name)
             log.debug('Prepared %s shadow repository in %s',
                       self.alias, shadow_repository_path)
 
         return shadow_repository_path
 
     def _merge_repo(self, repo_id, workspace_id, target_ref,
                     source_repo, source_ref, merge_message,
                     merger_name, merger_email, dry_run=False,
                     use_rebase=False, close_branch=False):
 
         log.debug('Executing merge_repo with %s strategy, dry_run mode:%s',
                   'rebase' if use_rebase else 'merge', dry_run)
+
         if target_ref.commit_id != self.branches[target_ref.name]:
             log.warning('Target ref %s commit mismatch %s vs %s', target_ref,
                         target_ref.commit_id, self.branches[target_ref.name])
             return MergeResponse(
                 False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD,
                 metadata={'target_ref': target_ref})
 
         shadow_repository_path = self._maybe_prepare_merge_workspace(
             repo_id, workspace_id, target_ref, source_ref)
         shadow_repo = self.get_shadow_instance(shadow_repository_path)
 
         # checkout source, if it's different. Otherwise we could not
         # fetch proper commits for merge testing
         if source_ref.name != target_ref.name:
             if shadow_repo.get_remote_ref(source_ref.name):
                 shadow_repo._checkout(source_ref.name, force=True)
 
         # checkout target, and fetch changes
         shadow_repo._checkout(target_ref.name, force=True)
 
         # fetch/reset pull the target, in case it is changed
         # this handles even force changes
         shadow_repo._local_fetch(self.path, target_ref.name, use_origin=True)
         shadow_repo._local_reset(target_ref.name)
 
         # Need to reload repo to invalidate the cache, or otherwise we cannot
         # retrieve the last target commit.
         shadow_repo = self.get_shadow_instance(shadow_repository_path)
         if target_ref.commit_id != shadow_repo.branches[target_ref.name]:
             log.warning('Shadow Target ref %s commit mismatch %s vs %s',
                         target_ref, target_ref.commit_id,
                         shadow_repo.branches[target_ref.name])
             return MergeResponse(
                 False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD,
                 metadata={'target_ref': target_ref})
 
         # calculate new branch
         pr_branch = shadow_repo._get_new_pr_branch(
             source_ref.name, target_ref.name)
         log.debug('using pull-request merge branch: `%s`', pr_branch)
         # checkout to temp branch, and fetch changes
         shadow_repo._checkout(pr_branch, create=True)
         try:
             shadow_repo._local_fetch(source_repo.path, source_ref.name)
         except RepositoryError:
             log.exception('Failure when doing local fetch on '
                           'shadow repo: %s', shadow_repo)
             return MergeResponse(
                 False, False, None, MergeFailureReason.MISSING_SOURCE_REF,
                 metadata={'source_ref': source_ref})
 
         merge_ref = None
         merge_failure_reason = MergeFailureReason.NONE
         metadata = {}
         try:
             shadow_repo._local_merge(merge_message, merger_name, merger_email,
                                      [source_ref.commit_id])
             merge_possible = True
 
             # Need to invalidate the cache, or otherwise we
             # cannot retrieve the merge commit.
             shadow_repo = shadow_repo.get_shadow_instance(shadow_repository_path)
             merge_commit_id = shadow_repo.branches[pr_branch]
 
             # Set a reference pointing to the merge commit. This reference may
             # be used to easily identify the last successful merge commit in
             # the shadow repository.
             shadow_repo.set_refs('refs/heads/pr-merge', merge_commit_id)
             merge_ref = Reference('branch', 'pr-merge', merge_commit_id)
         except RepositoryError as e:
             log.exception('Failure when doing local merge on git shadow repo')
             if isinstance(e, UnresolvedFilesInRepo):
                 metadata['unresolved_files'] = '\n* conflict: ' + ('\n * conflict: '.join(e.args[0]))
 
             merge_possible = False
             merge_failure_reason = MergeFailureReason.MERGE_FAILED
 
         if merge_possible and not dry_run:
             try:
                 shadow_repo._local_push(
                     pr_branch, self.path, target_ref.name, enable_hooks=True,
                     rc_scm_data=self.config.get('rhodecode', 'RC_SCM_DATA'))
                 merge_succeeded = True
             except RepositoryError:
                 log.exception(
                     'Failure when doing local push from the shadow '
                     'repository to the target repository at %s.', self.path)
                 merge_succeeded = False
                 merge_failure_reason = MergeFailureReason.PUSH_FAILED
                 metadata['target'] = 'git shadow repo'
                 metadata['merge_commit'] = pr_branch
         else:
             merge_succeeded = False
 
         return MergeResponse(
             merge_possible, merge_succeeded, merge_ref, merge_failure_reason,
             metadata=metadata)

rhodecode/tests/fixture.py

@@ -1,427 +1,428 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Helpers for fixture generation
 """
 
 import os
 import time
 import tempfile
 import shutil
 import configparser
 
 from rhodecode.model.settings import SettingsModel
-from rhodecode.tests import *
 from rhodecode.model.db import Repository, User, RepoGroup, UserGroup, Gist, UserEmailMap
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.gist import GistModel
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.authentication.plugins.auth_rhodecode import \
     RhodeCodeAuthPlugin
 
+from rhodecode.tests import TEST_USER_ADMIN_LOGIN
+
 dn = os.path.dirname
 FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'tests', 'fixtures')
 
 
 def error_function(*args, **kwargs):
     raise Exception('Total Crash !')
 
 
 class TestINI(object):
     """
     Allows to create a new test.ini file as a copy of existing one with edited
     data. Example usage::
 
         with TestINI('test.ini', [{'section':{'key':val'}]) as new_test_ini_path:
             print('paster server %s' % new_test_ini)
     """
 
     def __init__(self, ini_file_path, ini_params, new_file_prefix='DEFAULT',
                  destroy=True, dir=None):
         self.ini_file_path = ini_file_path
         self.ini_params = ini_params
         self.new_path = None
         self.new_path_prefix = new_file_prefix
         self._destroy = destroy
         self._dir = dir
 
     def __enter__(self):
         return self.create()
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.destroy()
 
     def create(self):
         parser = configparser.ConfigParser()
         parser.read(self.ini_file_path)
 
         for data in self.ini_params:
             section, ini_params = list(data.items())[0]
 
             for key, val in ini_params.items():
                 parser[section][key] = str(val)
 
         with tempfile.NamedTemporaryFile(
                 mode='w',
                 prefix=self.new_path_prefix, suffix='.ini', dir=self._dir,
                 delete=False) as new_ini_file:
             parser.write(new_ini_file)
             self.new_path = new_ini_file.name
 
         return self.new_path
 
     def destroy(self):
         if self._destroy:
             os.remove(self.new_path)
 
 
 class Fixture(object):
 
     def anon_access(self, status):
         """
         Context process for disabling anonymous access. use like:
         fixture = Fixture()
         with fixture.anon_access(False):
             #tests
 
         after this block anon access will be set to `not status`
         """
 
         class context(object):
             def __enter__(self):
                 anon = User.get_default_user()
                 anon.active = status
                 Session().add(anon)
                 Session().commit()
                 time.sleep(1.5)  # must sleep for cache (1s to expire)
 
             def __exit__(self, exc_type, exc_val, exc_tb):
                 anon = User.get_default_user()
                 anon.active = not status
                 Session().add(anon)
                 Session().commit()
 
         return context()
 
     def auth_restriction(self, registry, auth_restriction):
         """
         Context process for changing the builtin rhodecode plugin auth restrictions.
         Use like:
         fixture = Fixture()
         with fixture.auth_restriction('super_admin'):
             #tests
 
         after this block auth restriction will be taken off
         """
 
         class context(object):
             def _get_plugin(self):
                 plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
                 plugin = RhodeCodeAuthPlugin(plugin_id)
                 return plugin
 
             def __enter__(self):
 
                 plugin = self._get_plugin()
                 plugin.create_or_update_setting('auth_restriction', auth_restriction)
                 Session().commit()
                 SettingsModel().invalidate_settings_cache(hard=True)
 
             def __exit__(self, exc_type, exc_val, exc_tb):
 
                 plugin = self._get_plugin()
                 plugin.create_or_update_setting(
                     'auth_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_NONE)
                 Session().commit()
                 SettingsModel().invalidate_settings_cache(hard=True)
 
         return context()
 
     def scope_restriction(self, registry, scope_restriction):
         """
         Context process for changing the builtin rhodecode plugin scope restrictions.
         Use like:
         fixture = Fixture()
         with fixture.scope_restriction('scope_http'):
             #tests
 
         after this block scope restriction will be taken off
         """
 
         class context(object):
             def _get_plugin(self):
                 plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
                 plugin = RhodeCodeAuthPlugin(plugin_id)
                 return plugin
 
             def __enter__(self):
                 plugin = self._get_plugin()
                 plugin.create_or_update_setting('scope_restriction', scope_restriction)
                 Session().commit()
                 SettingsModel().invalidate_settings_cache(hard=True)
 
             def __exit__(self, exc_type, exc_val, exc_tb):
                 plugin = self._get_plugin()
                 plugin.create_or_update_setting(
                     'scope_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_ALL)
                 Session().commit()
                 SettingsModel().invalidate_settings_cache(hard=True)
 
         return context()
 
     def _get_repo_create_params(self, **custom):
         repo_type = custom.get('repo_type') or 'hg'
 
         default_landing_ref, landing_ref_lbl = ScmModel.backend_landing_ref(repo_type)
 
         defs = {
             'repo_name': None,
             'repo_type': repo_type,
             'clone_uri': '',
             'push_uri': '',
             'repo_group': '-1',
             'repo_description': 'DESC',
             'repo_private': False,
             'repo_landing_commit_ref': default_landing_ref,
             'repo_copy_permissions': False,
             'repo_state': Repository.STATE_CREATED,
         }
         defs.update(custom)
         if 'repo_name_full' not in custom:
             defs.update({'repo_name_full': defs['repo_name']})
 
         # fix the repo name if passed as repo_name_full
         if defs['repo_name']:
             defs['repo_name'] = defs['repo_name'].split('/')[-1]
 
         return defs
 
     def _get_group_create_params(self, **custom):
         defs = {
             'group_name': None,
             'group_description': 'DESC',
             'perm_updates': [],
             'perm_additions': [],
             'perm_deletions': [],
             'group_parent_id': -1,
             'enable_locking': False,
             'recursive': False,
         }
         defs.update(custom)
 
         return defs
 
     def _get_user_create_params(self, name, **custom):
         defs = {
             'username': name,
             'password': 'qweqwe',
             'email': '[e-mail]' % name,
             'firstname': 'TestUser',
             'lastname': 'Test',
             'description': 'test description',
             'active': True,
             'admin': False,
             'extern_type': 'rhodecode',
             'extern_name': None,
         }
         defs.update(custom)
 
         return defs
 
     def _get_user_group_create_params(self, name, **custom):
         defs = {
             'users_group_name': name,
             'user_group_description': 'DESC',
             'users_group_active': True,
             'user_group_data': {},
         }
         defs.update(custom)
 
         return defs
 
     def create_repo(self, name, **kwargs):
         repo_group = kwargs.get('repo_group')
         if isinstance(repo_group, RepoGroup):
             kwargs['repo_group'] = repo_group.group_id
             name = name.split(Repository.NAME_SEP)[-1]
             name = Repository.NAME_SEP.join((repo_group.group_name, name))
 
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             r = Repository.get_by_repo_name(name)
             if r:
                 return r
 
         form_data = self._get_repo_create_params(repo_name=name, **kwargs)
         cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create(form_data, cur_user)
         Session().commit()
         repo = Repository.get_by_repo_name(name)
         assert repo
         return repo
 
     def create_fork(self, repo_to_fork, fork_name, **kwargs):
         repo_to_fork = Repository.get_by_repo_name(repo_to_fork)
 
         form_data = self._get_repo_create_params(
             repo_name=fork_name,
             fork_parent_id=repo_to_fork.repo_id,
             repo_type=repo_to_fork.repo_type,
             **kwargs)
 
-        #TODO: fix it !!
+        # TODO: fix it !!
         form_data['description'] = form_data['repo_description']
         form_data['private'] = form_data['repo_private']
         form_data['landing_rev'] = form_data['repo_landing_commit_ref']
 
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create_fork(form_data, cur_user=owner)
         Session().commit()
         r = Repository.get_by_repo_name(fork_name)
         assert r
         return r
 
     def destroy_repo(self, repo_name, **kwargs):
         RepoModel().delete(repo_name, pull_requests='delete', **kwargs)
         Session().commit()
 
     def destroy_repo_on_filesystem(self, repo_name):
         rm_path = os.path.join(RepoModel().repos_path, repo_name)
         if os.path.isdir(rm_path):
             shutil.rmtree(rm_path)
 
     def create_repo_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = RepoGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_group_create_params(group_name=name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         gr = RepoGroupModel().create(
             group_name=form_data['group_name'],
             group_description=form_data['group_name'],
             owner=owner)
         Session().commit()
         gr = RepoGroup.get_by_group_name(gr.group_name)
         return gr
 
     def destroy_repo_group(self, repogroupid):
         RepoGroupModel().delete(repogroupid)
         Session().commit()
 
     def create_user(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             user = User.get_by_username(name)
             if user:
                 return user
         form_data = self._get_user_create_params(name, **kwargs)
         user = UserModel().create(form_data)
 
         # create token for user
         AuthTokenModel().create(
                 user=user, description=u'TEST_USER_TOKEN')
 
         Session().commit()
         user = User.get_by_username(user.username)
         return user
 
     def destroy_user(self, userid):
         UserModel().delete(userid)
         Session().commit()
 
     def create_additional_user_email(self, user, email):
         uem = UserEmailMap()
         uem.user = user
         uem.email = email
         Session().add(uem)
         return uem
 
     def destroy_users(self, userid_iter):
         for user_id in userid_iter:
             if User.get_by_username(user_id):
                 UserModel().delete(user_id)
                 Session().commit()
 
     def create_user_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = UserGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         # map active flag to the real attribute. For API consistency of fixtures
         if 'active' in kwargs:
             kwargs['users_group_active'] = kwargs['active']
             del kwargs['active']
         form_data = self._get_user_group_create_params(name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         user_group = UserGroupModel().create(
             name=form_data['users_group_name'],
             description=form_data['user_group_description'],
             owner=owner, active=form_data['users_group_active'],
             group_data=form_data['user_group_data'])
         Session().commit()
         user_group = UserGroup.get_by_group_name(user_group.users_group_name)
         return user_group
 
     def destroy_user_group(self, usergroupid):
         UserGroupModel().delete(user_group=usergroupid, force=True)
         Session().commit()
 
     def create_gist(self, **kwargs):
         form_data = {
             'description': 'new-gist',
             'owner': TEST_USER_ADMIN_LOGIN,
             'gist_type': GistModel.cls.GIST_PUBLIC,
             'lifetime': -1,
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'gist_mapping': {b'filename1.txt': {'content': b'hello world'},}
         }
         form_data.update(kwargs)
         gist = GistModel().create(
             description=form_data['description'], owner=form_data['owner'],
             gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
             lifetime=form_data['lifetime'], gist_acl_level=form_data['acl_level']
         )
         Session().commit()
         return gist
 
     def destroy_gists(self, gistid=None):
         for g in GistModel.cls.get_all():
             if gistid:
                 if gistid == g.gist_access_id:
                     GistModel().delete(g)
             else:
                 GistModel().delete(g)
         Session().commit()
 
     def load_resource(self, resource_name, strip=False):
         with open(os.path.join(FIXTURES, resource_name), 'rb') as f:
             source = f.read()
             if strip:
                 source = source.strip()
 
         return source

rhodecode/tests/lib/test_diffs.py

@@ -1,825 +1,826 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import textwrap
 
 import mock
 import pytest
 
 from rhodecode.lib.codeblocks import DiffSet
 from rhodecode.lib.diffs import (
     DiffProcessor,
     NEW_FILENODE, DEL_FILENODE, MOD_FILENODE, RENAMED_FILENODE,
     CHMOD_FILENODE, BIN_FILENODE, COPIED_FILENODE)
-from rhodecode.lib.str_utils import safe_bytes
+
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.lib.vcs.backends.git import GitCommit
-from rhodecode.tests.fixture import Fixture, no_newline_id_generator
+from rhodecode.tests.fixture import Fixture
+from rhodecode.tests import no_newline_id_generator
 from rhodecode.lib.vcs.backends.git.repository import GitDiff
 from rhodecode.lib.vcs.backends.hg.repository import MercurialDiff
 from rhodecode.lib.vcs.backends.svn.repository import SubversionDiff
 
 fixture = Fixture()
 
 
 class TestMixedFilenameEncodings(object):
 
     @pytest.fixture(scope="class")
     def raw_diff(self):
         return fixture.load_resource(
             'hg_diff_mixed_filename_encodings.diff')
 
     @pytest.fixture()
     def processor(self, raw_diff):
         diff = MercurialDiff(raw_diff)
         processor = DiffProcessor(diff, diff_format='newdiff')
         return processor
 
     def test_filenames_are_decoded_to_unicode(self, processor):
         diff_data = processor.prepare()
         filenames = [item['filename'] for item in diff_data]
         assert filenames == [
             u'späcial-utf8.txt', u'sp�cial-cp1252.txt', u'sp�cial-latin1.txt']
 
     def test_raw_diff_is_decoded_to_unicode(self, processor):
         diff_data = processor.prepare()
         raw_diffs = [item['raw_diff'] for item in diff_data]
         new_file_message = u'\nnew file mode 100644\n'
         expected_raw_diffs = [
             u' a/späcial-utf8.txt b/späcial-utf8.txt' + new_file_message,
             u' a/sp�cial-cp1252.txt b/sp�cial-cp1252.txt' + new_file_message,
             u' a/sp�cial-latin1.txt b/sp�cial-latin1.txt' + new_file_message]
         assert raw_diffs == expected_raw_diffs
 
     def test_as_raw_preserves_the_encoding(self, processor, raw_diff):
         assert processor.as_raw() == raw_diff
 
 
 # TODO: mikhail: format the following data structure properly
 DIFF_FIXTURES = [
     ('hg',
      'hg_diff_add_single_binary_file.diff',
      [('US Warszawa.jpg', 'A',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '100755',
         'ops': {NEW_FILENODE: 'new file 100755',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ]),
     ('hg',
      'hg_diff_mod_single_binary_file.diff',
      [('US Warszawa.jpg', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ]),
     ('hg',
      'hg_diff_mod_single_file_and_rename_and_chmod.diff',
      [('README', 'M',
        {'added': 3,
         'deleted': 0,
         'binary': False,
         'old_mode': '100755',
         'new_mode': '100644',
         'renamed': ('README.rst', 'README'),
         'ops': {MOD_FILENODE: 'modified file',
                 RENAMED_FILENODE: 'file renamed from README.rst to README',
                 CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
       ]),
     ('hg',
      'hg_diff_no_newline.diff',
      [('server.properties', 'M',
        {'added': 2,
         'deleted': 1,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ]),
     ('hg',
      'hg_diff_mod_file_and_rename.diff',
      [('README.rst', 'M',
        {'added': 3,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'renamed': ('README', 'README.rst'),
         'ops': {MOD_FILENODE: 'modified file',
                 RENAMED_FILENODE: 'file renamed from README to README.rst'}}),
       ]),
     ('hg',
      'hg_diff_del_single_binary_file.diff',
      [('US Warszawa.jpg', 'D',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'ops': {DEL_FILENODE: 'deleted file',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ]),
     ('hg',
      'hg_diff_chmod_and_mod_single_binary_file.diff',
      [('gravatar.png', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '100644',
         'new_mode': '100755',
         'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ]),
     ('hg',
      'hg_diff_chmod.diff',
      [('file', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '100755',
         'new_mode': '100644',
         'ops': {CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
       ]),
     ('hg',
      'hg_diff_rename_file.diff',
      [('file_renamed', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'renamed': ('file', 'file_renamed'),
         'ops': {RENAMED_FILENODE: 'file renamed from file to file_renamed'}}),
       ]),
     ('hg',
      'hg_diff_rename_and_chmod_file.diff',
      [('README', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '100644',
         'new_mode': '100755',
         'renamed': ('README.rst', 'README'),
         'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                 RENAMED_FILENODE: 'file renamed from README.rst to README'}}),
       ]),
     ('hg',
      'hg_diff_binary_and_normal.diff',
      [('img/baseline-10px.png', 'A',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'new_mode': '100644',
         'old_mode': '',
         'ops': {NEW_FILENODE: 'new file 100644',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ('js/jquery/hashgrid.js', 'A',
        {'added': 340,
         'deleted': 0,
         'binary': False,
         'new_mode': '100755',
         'old_mode': '',
         'ops': {NEW_FILENODE: 'new file 100755'}}),
       ('index.html', 'M',
        {'added': 3,
         'deleted': 2,
         'binary': False,
         'new_mode': '',
         'old_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('less/docs.less', 'M',
        {'added': 34,
         'deleted': 0,
         'binary': False,
         'new_mode': '',
         'old_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('less/scaffolding.less', 'M',
        {'added': 1,
         'deleted': 3,
         'binary': False,
         'new_mode': '',
         'old_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('readme.markdown', 'M',
        {'added': 1,
         'deleted': 10,
         'binary': False,
         'new_mode': '',
         'old_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('img/baseline-20px.png', 'D',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'new_mode': '',
         'old_mode': '',
         'ops': {DEL_FILENODE: 'deleted file',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ('js/global.js', 'D',
        {'added': 0,
         'deleted': 75,
         'binary': False,
         'new_mode': '',
         'old_mode': '',
         'ops': {DEL_FILENODE: 'deleted file'}})
       ]),
     ('git',
      'git_diff_chmod.diff',
      [('work-horus.xls', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '100644',
         'new_mode': '100755',
         'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}})
       ]),
     ('git',
      'git_diff_js_chars.diff',
      [('\\"><img src=x onerror=prompt(0)>/\\"><img src=x onerror=prompt(1)>.txt', 'M',
        {'added': 1,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}})
       ]),
     ('git',
      'git_diff_rename_file.diff',
      [('file.xls', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'renamed': ('work-horus.xls', 'file.xls'),
         'ops': {
             RENAMED_FILENODE: 'file renamed from work-horus.xls to file.xls'}})
       ]),
     ('git',
      'git_diff_mod_single_binary_file.diff',
      [('US Warszawa.jpg', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file',
                 BIN_FILENODE: 'binary diff hidden'}})
       ]),
     ('git',
      'git_diff_binary_and_normal.diff',
      [('img/baseline-10px.png', 'A',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {NEW_FILENODE: 'new file 100644',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ('js/jquery/hashgrid.js', 'A',
        {'added': 340,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '100755',
         'ops': {NEW_FILENODE: 'new file 100755'}}),
       ('index.html', 'M',
        {'added': 3,
         'deleted': 2,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('less/docs.less', 'M',
        {'added': 34,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('less/scaffolding.less', 'M',
        {'added': 1,
         'deleted': 3,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('readme.markdown', 'M',
        {'added': 1,
         'deleted': 10,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('img/baseline-20px.png', 'D',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'ops': {DEL_FILENODE: 'deleted file',
                 BIN_FILENODE: 'binary diff hidden'}}),
       ('js/global.js', 'D',
        {'added': 0,
         'deleted': 75,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {DEL_FILENODE: 'deleted file'}}),
       ]),
     ('hg',
      'diff_with_diff_data.diff',
      [('vcs/backends/base.py', 'M',
        {'added': 18,
         'deleted': 2,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('vcs/backends/git/repository.py', 'M',
        {'added': 46,
         'deleted': 15,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('vcs/backends/hg.py', 'M',
        {'added': 22,
         'deleted': 3,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('vcs/tests/test_git.py', 'M',
        {'added': 5,
         'deleted': 5,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ('vcs/tests/test_repository.py', 'M',
        {'added': 174,
         'deleted': 2,
         'binary': False,
         'old_mode': '',
         'new_mode': '100644',
         'ops': {MOD_FILENODE: 'modified file'}}),
       ]),
     ('hg',
      'hg_diff_copy_file.diff',
      [('file2', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '',
         'new_mode': '',
         'copied': ('file1', 'file2'),
         'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),
       ]),
     ('hg',
      'hg_diff_copy_and_modify_file.diff',
      [('file3', 'M',
        {'added': 1,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'copied': ('file2', 'file3'),
         'ops': {COPIED_FILENODE: 'file copied from file2 to file3',
                 MOD_FILENODE: 'modified file'}}),
       ]),
     ('hg',
      'hg_diff_copy_and_chmod_file.diff',
      [('file4', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': True,
         'old_mode': '100644',
         'new_mode': '100755',
         'copied': ('file3', 'file4'),
         'ops': {COPIED_FILENODE: 'file copied from file3 to file4',
                 CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}}),
       ]),
     ('hg',
      'hg_diff_copy_chmod_and_edit_file.diff',
      [('file5', 'M',
        {'added': 2,
         'deleted': 1,
         'binary': False,
         'old_mode': '100755',
         'new_mode': '100644',
         'copied': ('file4', 'file5'),
         'ops': {COPIED_FILENODE: 'file copied from file4 to file5',
                 CHMOD_FILENODE: 'modified file chmod 100755 => 100644',
                 MOD_FILENODE: 'modified file'}})]),
 
     # Diffs to validate rename and copy file with space in its name
     ('git',
      'git_diff_rename_file_with_spaces.diff',
      [('file_with_  two spaces.txt', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'old_mode': '',
           'new_mode': '',
           'renamed': ('file_with_ spaces.txt', 'file_with_  two spaces.txt'),
           'ops': {
               RENAMED_FILENODE: (
                   'file renamed from file_with_ spaces.txt to file_with_ '
                   ' two spaces.txt')}
           }), ]),
     ('hg',
      'hg_diff_rename_file_with_spaces.diff',
      [('file_changed _.txt', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'old_mode': '',
           'new_mode': '',
           'renamed': ('file_ with update.txt', 'file_changed _.txt'),
           'ops': {
               RENAMED_FILENODE: (
                   'file renamed from file_ with update.txt to file_changed _.txt')}
           }), ]),
     ('hg',
      'hg_diff_copy_file_with_spaces.diff',
      [('file_copied_ with  spaces.txt', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'old_mode': '',
           'new_mode': '',
           'copied': ('file_changed_without_spaces.txt', 'file_copied_ with  spaces.txt'),
           'ops': {
               COPIED_FILENODE: (
                   'file copied from file_changed_without_spaces.txt to'
                   ' file_copied_ with  spaces.txt')}
           }),
       ]),
 
     # special signs from git
     ('git',
      'git_diff_binary_special_files.diff',
      [('css/_Icon\\r', 'A',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'old_mode': '',
           'new_mode': '100644',
           'ops': {NEW_FILENODE: 'new file 100644',
                   BIN_FILENODE: 'binary diff hidden'}
           }),
       ]),
     ('git',
      'git_diff_binary_special_files_2.diff',
      [('css/Icon\\r', 'A',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'old_mode': '',
           'new_mode': '100644',
           'ops': {NEW_FILENODE: 'new file 100644', }
           }),
       ]),
 
     ('svn',
      'svn_diff_binary_add_file.diff',
      [('intl.dll', 'A',
        {'added': 0,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '10644',
         'ops': {NEW_FILENODE: 'new file 10644',
                 #TODO(Marcink): depends on binary detection on svn patches
                 # BIN_FILENODE: 'binary diff hidden'
                 }
         }),
       ]),
 
     ('svn',
      'svn_diff_multiple_changes.diff',
      [('trunk/doc/images/SettingsOverlay.png', 'M',
        {'added': 0,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file',
                 #TODO(Marcink): depends on binary detection on svn patches
                 # BIN_FILENODE: 'binary diff hidden'
                 }
         }),
       ('trunk/doc/source/de/tsvn_ch04.xml', 'M',
        {'added': 89,
         'deleted': 34,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/doc/source/en/tsvn_ch04.xml', 'M',
        {'added': 66,
         'deleted': 21,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/Changelog.txt', 'M',
        {'added': 2,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/Resources/TortoiseProcENG.rc', 'M',
        {'added': 19,
         'deleted': 13,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/TortoiseProc/SetOverlayPage.cpp', 'M',
        {'added': 16,
         'deleted': 1,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/TortoiseProc/SetOverlayPage.h', 'M',
        {'added': 3,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/TortoiseProc/resource.h', 'M',
        {'added': 2,
         'deleted': 0,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ('trunk/src/TortoiseShell/ShellCache.h', 'M',
        {'added': 50,
         'deleted': 1,
         'binary': False,
         'old_mode': '',
         'new_mode': '',
         'ops': {MOD_FILENODE: 'modified file'}
         }),
       ]),
 
 ]
 
 DIFF_FIXTURES_WITH_CONTENT = [
     (
         'hg', 'hg_diff_single_file_change_newline.diff',
         [
             (
                 'file_b',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1 @@\n+test_content b\n'  # diff
             ),
         ],
     ),
     (
         'hg', 'hg_diff_double_file_change_newline.diff',
         [
             (
                 'file_b',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1 @@\n+test_content b\n'  # diff
             ),
             (
                 'file_c',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1 @@\n+test_content c\n'  # diff
             ),
         ],
     ),
     (
         'hg', 'hg_diff_double_file_change_double_newline.diff',
         [
             (
                 'file_b',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1 @@\n+test_content b\n\n'  # diff
             ),
             (
                 'file_c',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1 @@\n+test_content c\n'  # diff
             ),
         ],
     ),
     (
         'hg', 'hg_diff_four_file_change_newline.diff',
         [
             (
                 'file',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1,1 @@\n+file\n'  # diff
             ),
             (
                 'file2',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1,1 @@\n+another line\n'  # diff
             ),
             (
                 'file3',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                    'old_mode': '',
                    'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1,1 @@\n+newline\n'  # diff
             ),
             (
                 'file4',  # filename
                 'A',  # change
                 {  # stats
                    'added': 1,
                    'deleted': 0,
                    'binary': False,
                     'old_mode': '',
                     'new_mode': '100644',
                    'ops': {NEW_FILENODE: 'new file 100644', }
                 },
                 '@@ -0,0 +1,1 @@\n+fil4\n\\ No newline at end of file'  # diff
             ),
         ],
     ),
 
 ]
 
 
 diff_class = {
     'git': GitDiff,
     'hg': MercurialDiff,
     'svn': SubversionDiff,
 }
 
 
 @pytest.mark.parametrize('vcs_type, diff_file, expected_data', DIFF_FIXTURES)
 def test_diff_lib(vcs_type, diff_file, expected_data):
     diff_txt = fixture.load_resource(diff_file)
     diff = diff_class[vcs_type](diff_txt)
 
     diff_proc = DiffProcessor(diff, diff_format='newdiff')
     diff_proc_d = diff_proc.prepare()
     data = [(x['filename'], x['operation'], x['stats'])
             for x in diff_proc_d]
     assert expected_data == data
 
 
 @pytest.mark.parametrize('vcs_type, diff_file, expected_data', DIFF_FIXTURES_WITH_CONTENT)
 def test_diff_lib_newlines(vcs_type, diff_file, expected_data):
     diff_txt = fixture.load_resource(diff_file)
     diff = diff_class[vcs_type](diff_txt)
 
     diff_proc = DiffProcessor(diff, diff_format='newdiff')
     diff_proc_d = diff_proc.prepare()
     data = [(x['filename'], x['operation'], x['stats'], x['raw_diff'])
             for x in diff_proc_d]
     assert expected_data == data
 
 
 @pytest.mark.parametrize('input_str', [
     b'',
     b'\n',
     b'\n\n',
     b'First\n+second',
     b'First\n+second\n',
 
     b'\n\n\n Multi \n\n\n',
     b'\n\n\n Multi beginning',
     b'Multi end \n\n\n',
     b'Multi end',
     b'@@ -0,0 +1 @@\n+test_content \n\n b\n'
 ], ids=no_newline_id_generator)
 def test_splitlines(input_str):
     result = DiffProcessor.diff_splitter(input_str)
     assert list(result) == input_str.splitlines(True)
 
 
 def test_diff_over_limit(request):
 
     diff_limit = 1024
     file_limit = 1024
 
     raw_diff = fixture.load_resource('large_diff.diff')
     vcs_diff = GitDiff(raw_diff)
     diff_processor = DiffProcessor(vcs_diff, diff_format='newdiff',
                                    diff_limit=diff_limit, file_limit=file_limit,
                                    show_full_diff=False)
 
     _parsed = diff_processor.prepare()
 
     commit1 = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
     commit2 = GitCommit(repository=mock.Mock(), raw_id='abcdef34', idx=2)
 
     diffset = DiffSet(
         repo_name='repo_name',
         source_node_getter=lambda *a, **kw: AttributeDict({'commit': commit1}),
         target_node_getter=lambda *a, **kw: AttributeDict({'commit': commit2})
     )
 
     diffset = diffset.render_patchset(_parsed, commit1, commit2)
 
     assert len(diffset.files) == 2
     assert diffset.limited_diff is True
     assert diffset.files[0].patch['filename'] == 'example.go'
     assert diffset.files[0].limited_diff is True
 
     assert diffset.files[1].patch['filename'] == 'README.md'
     assert diffset.files[1].limited_diff is False