chore(mercurial): Fixed usage of str in generation of mercurial configs, and fixed largefiles call

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

@@ -1,1013 +1,1017 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/
 
 """
 HG repository module
 """
 import os
 import logging
 import binascii
 import configparser
 import urllib.request
 import urllib.parse
 import urllib.error
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 from collections import OrderedDict
 from rhodecode.lib.datelib import (
     date_to_timestamp_plus_offset, utcdate_fromtimestamp, makedate)
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.utils2 import CachedProperty
 from rhodecode.lib.vcs import connection, exceptions
 from rhodecode.lib.vcs.backends.base import (
     BaseRepository, CollectionGenerator, Config, MergeResponse,
     MergeFailureReason, Reference, BasePathPermissionChecker)
 from rhodecode.lib.vcs.backends.hg.commit import MercurialCommit
 from rhodecode.lib.vcs.backends.hg.diff import MercurialDiff
 from rhodecode.lib.vcs.backends.hg.inmemory import MercurialInMemoryCommit
 from rhodecode.lib.vcs.exceptions import (
     EmptyRepositoryError, RepositoryError, TagAlreadyExistError,
     TagDoesNotExistError, CommitDoesNotExistError, SubrepoMergeError, UnresolvedFilesInRepo)
 
 hexlify = binascii.hexlify
 nullid = "\0" * 20
 
 log = logging.getLogger(__name__)
 
 
 class MercurialRepository(BaseRepository):
     """
     Mercurial repository backend
     """
     DEFAULT_BRANCH_NAME = 'default'
 
     def __init__(self, repo_path, config=None, create=False, src_url=None,
                  do_workspace_checkout=False, with_wire=None, bare=False):
         """
         Raises RepositoryError if repository could not be find at the given
         ``repo_path``.
 
         :param repo_path: local path of the repository
         :param config: config object containing the repo configuration
         :param create=False: if set to True, would try to create repository if
            it does not exist rather than raising exception
         :param src_url=None: would try to clone repository from given location
         :param do_workspace_checkout=False: sets update of working copy after
            making a clone
         :param bare: not used, compatible with other VCS
         """
 
         self.path = safe_str(os.path.abspath(repo_path))
         # mercurial since 4.4.X requires certain configuration to be present
         # because sometimes we init the repos with config we need to meet
         # special requirements
         self.config = config if config else self.get_default_config(
-            default=[('extensions', 'largefiles', '1')])
+            default=[('extensions', 'largefiles', '')])
         self.with_wire = with_wire or {"cache": False}  # default should not use cache
 
         self._init_repo(create, src_url, do_workspace_checkout)
 
         # caches
         self._commit_ids = {}
 
     @LazyProperty
     def _remote(self):
         repo_id = self.path
         return connection.Hg(self.path, repo_id, self.config, with_wire=self.with_wire)
 
     @CachedProperty
     def commit_ids(self):
         """
         Returns list of commit ids, in ascending order.  Being lazy
         attribute allows external tools to inject shas 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)}
 
     @CachedProperty
     def branches(self):
         return self._get_branches()
 
     @CachedProperty
     def branches_closed(self):
         return self._get_branches(active=False, closed=True)
 
     @CachedProperty
     def branches_all(self):
         all_branches = {}
         all_branches.update(self.branches)
         all_branches.update(self.branches_closed)
         return all_branches
 
     def _get_branches(self, active=True, closed=False):
         """
         Gets branches for this repository
         Returns only not closed active branches by default
 
         :param active: return also active branches
         :param closed: return also closed branches
 
         """
         if self.is_empty():
             return {}
 
         def get_name(ctx):
             return ctx[0]
 
         _branches = [(n, h,) for n, h in
                      self._remote.branches(active, closed).items()]
 
         return OrderedDict(sorted(_branches, key=get_name, reverse=False))
 
     @CachedProperty
     def tags(self):
         """
         Gets tags for this repository
         """
         return self._get_tags()
 
     def _get_tags(self):
         if self.is_empty():
             return {}
 
         def get_name(ctx):
             return ctx[0]
 
         _tags = [(n, h,) for n, h in
                  self._remote.tags().items()]
 
         return OrderedDict(sorted(_tags, key=get_name, reverse=True))
 
     def tag(self, name, user, commit_id=None, message=None, date=None, **kwargs):
         """
         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)
         local = kwargs.setdefault('local', False)
 
         if message is None:
             message = f"Added tag {name} for commit {commit.short_id}"
 
         date, tz = date_to_timestamp_plus_offset(date)
 
         self._remote.tag(name, commit.raw_id, message, local, user, date, tz)
         self._remote.invalidate_vcs_cache()
 
         # Reinitialize tags
         self._invalidate_prop_cache('tags')
         tag_id = self.tags[name]
 
         return self.get_commit(commit_id=tag_id)
 
     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)
 
         if message is None:
             message = "Removed tag %s" % name
         local = False
 
         date, tz = date_to_timestamp_plus_offset(date)
 
         self._remote.tag(name, nullid, message, local, user, date, tz)
         self._remote.invalidate_vcs_cache()
         self._invalidate_prop_cache('tags')
 
     @LazyProperty
     def bookmarks(self):
         """
         Gets bookmarks for this repository
         """
         return self._get_bookmarks()
 
     def _get_bookmarks(self):
         if self.is_empty():
             return {}
 
         def get_name(ctx):
             return ctx[0]
 
         _bookmarks = [
             (n, h) for n, h in
             self._remote.bookmarks().items()]
 
         return OrderedDict(sorted(_bookmarks, key=get_name))
 
     def _get_all_commit_ids(self):
         return self._remote.get_all_commit_ids('visible')
 
     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 commit changes should be shown.
         :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``.
         """
         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 = [self.path, path]
         else:
             file_filter = None
 
         diff = self._remote.diff(
             commit1.raw_id, commit2.raw_id, file_filter=file_filter,
             opt_git=True, opt_ignorews=ignore_whitespace,
             context=context)
         return MercurialDiff(diff)
 
     def strip(self, commit_id, branch=None):
         self._remote.strip(commit_id, update=False, backup=False)
 
         self._remote.invalidate_vcs_cache()
         # clear cache
         self._invalidate_prop_cache('commit_ids')
 
         return len(self.commit_ids)
 
     def verify(self):
         verify = self._remote.verify()
 
         self._remote.invalidate_vcs_cache()
         return verify
 
     def hg_update_cache(self):
         update_cache = self._remote.hg_update_cache()
 
         self._remote.invalidate_vcs_cache()
         return update_cache
 
     def hg_rebuild_fn_cache(self):
         update_cache = self._remote.hg_rebuild_fn_cache()
 
         self._remote.invalidate_vcs_cache()
         return update_cache
 
     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
 
         ancestors = self._remote.revs_from_revspec(
             "ancestor(id(%s), id(%s))", commit_id1, commit_id2,
             other_path=repo2.path)
 
         ancestor_id = repo2[ancestors[0]].raw_id if ancestors else None
 
         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):
         if commit_id1 == commit_id2:
             commits = []
         else:
             if merge:
                 indexes = self._remote.revs_from_revspec(
                     "ancestors(id(%s)) - ancestors(id(%s)) - id(%s)",
                     commit_id2, commit_id1, commit_id1, other_path=repo2.path)
             else:
                 indexes = self._remote.revs_from_revspec(
                     "id(%s)..id(%s) - id(%s)", commit_id1, commit_id2,
                     commit_id1, other_path=repo2.path)
 
             commits = [repo2.get_commit(commit_idx=idx, pre_load=pre_load)
                        for idx in indexes]
 
         return commits
 
     @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 mercurial 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 local url
         if os.path.isdir(url) or url.startswith('file:'):
             return True
 
         # Request the _remote to verify the url
         return connection.Hg.check_url(url, config.serialize())
 
     @staticmethod
     def is_valid_repository(path):
         return os.path.isdir(os.path.join(path, '.hg'))
 
     def _init_repo(self, create, src_url=None, do_workspace_checkout=False):
         """
         Function will check for mercurial repository in given path. If there
         is no repository in that path it will raise an exception unless
         `create` parameter is set to True - in that case repository would
         be created.
 
         If `src_url` is given, would try to clone repository from the
         location at given clone_point. Additionally it'll make update to
         working copy accordingly to `do_workspace_checkout` flag.
         """
         if create and os.path.exists(self.path):
             raise RepositoryError(
                 f"Cannot create repository at {self.path}, location already exist")
 
         if src_url:
             url = str(self._get_url(src_url))
             MercurialRepository.check_url(url, self.config)
 
             self._remote.clone(url, self.path, do_workspace_checkout)
 
             # Don't try to create if we've already cloned repo
             create = False
 
         if create:
             os.makedirs(self.path, mode=0o755)
 
         self._remote.localrepository(create)
 
     @LazyProperty
     def in_memory_commit(self):
         return MercurialInMemoryCommit(self)
 
     @LazyProperty
     def description(self):
         description = self._remote.get_config_value(
             'web', 'description', untrusted=True)
         return safe_str(description or self.DEFAULT_DESCRIPTION)
 
     @LazyProperty
     def contact(self):
         contact = (
             self._remote.get_config_value("web", "contact") or
             self._remote.get_config_value("ui", "username"))
         return safe_str(contact or self.DEFAULT_CONTACT)
 
     @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):
         # fallback to filesystem
         cl_path = os.path.join(self.path, '.hg', "00changelog.i")
         st_path = os.path.join(self.path, '.hg', "store")
         if os.path.exists(cl_path):
             return os.stat(cl_path).st_mtime
         else:
             return os.stat(st_path).st_mtime
 
     def _get_url(self, url):
         """
         Returns normalized url. If schema is not given, would fall
         to filesystem
         (``file:///``) schema.
         """
         if url != 'default' and '://' not in url:
             url = "file:" + urllib.request.pathname2url(url)
         return url
 
     def get_hook_location(self):
         """
         returns absolute path to location where hooks are stored
         """
         return os.path.join(self.path, '.hg', '.hgrc')
 
     def get_commit(self, commit_id=None, commit_idx=None, pre_load=None,
                    translate_tag=None, maybe_unreachable=False, reference_obj=None):
         """
         Returns ``MercurialCommit`` object representing repository's
         commit at the given `commit_id` or `commit_idx`.
         """
         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 MercurialCommit(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 MercurialCommit(self, _commit_id, commit_idx, pre_load=pre_load)
             except IndexError:
                 commit_id = commit_idx
         else:
             commit_id = "tip"
 
         # case here is no cached version, do an actual lookup instead
         try:
             raw_id, idx = self._remote.lookup(commit_id, both=True)
         except CommitDoesNotExistError:
             msg = "Commit {} does not exist for `{}`".format(
                 *map(safe_str, [commit_id, self.name]))
             raise CommitDoesNotExistError(msg)
 
         return MercurialCommit(self, raw_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=None):
         """
         Returns generator of ``MercurialCommit`` objects from start to end
         (both are inclusive)
 
         :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 commit for given ``start`` or
           ``end`` could not be found.
         """
         # actually we should check now if it's not an empty repo
         if self.is_empty():
             raise EmptyRepositoryError("There are no commits yet")
         self._validate_branch_name(branch_name)
 
         branch_ancestors = False
         if start_id is not None:
             self._validate_commit_id(start_id)
             c_start = self.get_commit(commit_id=start_id)
             start_pos = self._commit_ids[c_start.raw_id]
         else:
             start_pos = None
 
         if end_id is not None:
             self._validate_commit_id(end_id)
             c_end = self.get_commit(commit_id=end_id)
             end_pos = max(0, self._commit_ids[c_end.raw_id])
         else:
             end_pos = 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
 
         commit_filter = []
 
         if branch_name and not branch_ancestors:
             commit_filter.append(f'branch("{branch_name}")')
         elif branch_name and branch_ancestors:
             commit_filter.append(f'ancestors(branch("{branch_name}"))')
 
         if start_date and not end_date:
             commit_filter.append(f'date(">{start_date}")')
         if end_date and not start_date:
             commit_filter.append(f'date("<{end_date}")')
         if start_date and end_date:
             commit_filter.append(
                 f'date(">{start_date}") and date("<{end_date}")')
 
         if not show_hidden:
             commit_filter.append('not obsolete()')
             commit_filter.append('not hidden()')
 
         # TODO: johbo: Figure out a simpler way for this solution
         collection_generator = CollectionGenerator
         if commit_filter:
             commit_filter = ' and '.join(map(safe_str, commit_filter))
             revisions = self._remote.rev_range([commit_filter])
             collection_generator = MercurialIndexBasedCollectionGenerator
         else:
             revisions = self.commit_ids
 
         if start_pos or end_pos:
             revisions = revisions[start_pos:end_pos]
 
         return collection_generator(self, revisions, pre_load=pre_load)
 
     def pull(self, url, commit_ids=None):
         """
         Pull changes from external location.
 
         :param commit_ids: Optional. Can be set to a list of commit ids
            which shall be pulled from the other repository.
         """
         url = self._get_url(url)
         self._remote.pull(url, commit_ids=commit_ids)
         self._remote.invalidate_vcs_cache()
 
     def fetch(self, url, commit_ids=None):
         """
         Backward compatibility with GIT fetch==pull
         """
         return self.pull(url, commit_ids=commit_ids)
 
     def push(self, url):
         url = self._get_url(url)
         self._remote.sync_push(url)
 
     def _local_clone(self, clone_path):
         """
         Create a local clone of the current repo.
         """
         self._remote.clone(self.path, clone_path, update_after_clone=True,
                            hooks=False)
 
     def _update(self, revision, clean=False):
         """
         Update the working copy to the specified revision.
         """
         log.debug('Doing checkout to commit: `%s` for %s', revision, self)
         self._remote.update(revision, clean=clean)
 
     def _identify(self):
         """
         Return the current state of the working directory.
         """
         return self._remote.identify().strip().rstrip('+')
 
     def _heads(self, branch=None):
         """
         Return the commit ids of the repository heads.
         """
         return self._remote.heads(branch=branch).strip().split(' ')
 
     def _ancestor(self, revision1, revision2):
         """
         Return the common ancestor of the two revisions.
         """
         return self._remote.ancestor(revision1, revision2)
 
     def _local_push(
             self, revision, repository_path, push_branches=False,
             enable_hooks=False):
         """
         Push the given revision to the specified repository.
 
         :param push_branches: allow to create branches in the target repo.
         """
         self._remote.push(
             [revision], repository_path, hooks=enable_hooks,
             push_branches=push_branches)
 
     def _local_merge(self, target_ref, merge_message, user_name, user_email,
                      source_ref, use_rebase=False, close_commit_id=None, dry_run=False):
         """
         Merge the given source_revision into the checked out revision.
 
         Returns the commit id of the merge and a boolean indicating if the
         commit needs to be pushed.
         """
+
         source_ref_commit_id = source_ref.commit_id
         target_ref_commit_id = target_ref.commit_id
 
         # update our workdir to target ref, for proper merge
         self._update(target_ref_commit_id, clean=True)
 
         ancestor = self._ancestor(target_ref_commit_id, source_ref_commit_id)
         is_the_same_branch = self._is_the_same_branch(target_ref, source_ref)
 
         if close_commit_id:
             # NOTE(marcink): if we get the close commit, this is our new source
             # which will include the close commit itself.
             source_ref_commit_id = close_commit_id
 
         if ancestor == source_ref_commit_id:
             # Nothing to do, the changes were already integrated
             return target_ref_commit_id, False
 
         elif ancestor == target_ref_commit_id and is_the_same_branch:
             # In this case we should force a commit message
             return source_ref_commit_id, True
 
         unresolved = None
         if use_rebase:
             try:
                 bookmark_name = f'rcbook{source_ref_commit_id}{target_ref_commit_id}'
                 self.bookmark(bookmark_name, revision=source_ref.commit_id)
                 self._remote.rebase(
                     source=source_ref_commit_id, dest=target_ref_commit_id)
                 self._remote.invalidate_vcs_cache()
                 self._update(bookmark_name, clean=True)
                 return self._identify(), True
             except RepositoryError as e:
                 # The rebase-abort may raise another exception which 'hides'
                 # the original one, therefore we log it here.
                 log.exception('Error while rebasing shadow repo during merge.')
                 if 'unresolved conflicts' in safe_str(e):
                     unresolved = self._remote.get_unresolved_files()
                     log.debug('unresolved files: %s', unresolved)
 
                 # Cleanup any rebase leftovers
                 self._remote.invalidate_vcs_cache()
                 self._remote.rebase(abort=True)
                 self._remote.invalidate_vcs_cache()
                 self._remote.update(clean=True)
                 if unresolved:
                     raise UnresolvedFilesInRepo(unresolved)
                 else:
                     raise
         else:
             try:
                 self._remote.merge(source_ref_commit_id)
                 self._remote.invalidate_vcs_cache()
                 self._remote.commit(
                     message=safe_str(merge_message),
                     username=safe_str(f'{user_name} <{user_email}>'))
                 self._remote.invalidate_vcs_cache()
                 return self._identify(), True
             except RepositoryError as e:
                 # The merge-abort may raise another exception which 'hides'
                 # the original one, therefore we log it here.
                 log.exception('Error while merging shadow repo during merge.')
                 if 'unresolved merge conflicts' in safe_str(e):
                     unresolved = self._remote.get_unresolved_files()
                     log.debug('unresolved files: %s', unresolved)
 
                 # Cleanup any merge leftovers
                 self._remote.update(clean=True)
                 if unresolved:
                     raise UnresolvedFilesInRepo(unresolved)
                 else:
                     raise
 
     def _local_close(self, target_ref, user_name, user_email,
                      source_ref, close_message=''):
         """
         Close the branch of the given source_revision
 
         Returns the commit id of the close and a boolean indicating if the
         commit needs to be pushed.
         """
         self._update(source_ref.commit_id)
         message = close_message or f"Closing branch: `{source_ref.name}`"
         try:
             self._remote.commit(
                 message=safe_str(message),
                 username=safe_str(f'{user_name} <{user_email}>'),
                 close_branch=True)
             self._remote.invalidate_vcs_cache()
             return self._identify(), True
         except RepositoryError:
             # Cleanup any commit leftovers
             self._remote.update(clean=True)
             raise
 
     def _is_the_same_branch(self, target_ref, source_ref):
         return (
             self._get_branch_name(target_ref) ==
             self._get_branch_name(source_ref))
 
     def _get_branch_name(self, ref):
         if ref.type == 'branch':
             return ref.name
         return self._remote.ctx_branch(ref.commit_id)
 
     def _maybe_prepare_merge_workspace(
             self, repo_id, workspace_id, unused_target_ref, unused_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)
             log.debug(
                 'Prepared shadow repository in %s', 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 not in self._heads():
             return MergeResponse(
                 False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD,
                 metadata={'target_ref': target_ref})
 
         try:
             if target_ref.type == 'branch' and len(self._heads(target_ref.name)) != 1:
                 heads_all = self._heads(target_ref.name)
                 max_heads = 10
                 if len(heads_all) > max_heads:
                     heads = '\n,'.join(
                         heads_all[:max_heads] +
                         [f'and {len(heads_all)-max_heads} more.'])
                 else:
                     heads = '\n,'.join(heads_all)
                 metadata = {
                     'target_ref': target_ref,
                     'source_ref': source_ref,
                     'heads': heads
                 }
                 return MergeResponse(
                     False, False, None,
                     MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS,
                     metadata=metadata)
         except CommitDoesNotExistError:
             log.exception('Failure when looking up branch heads on hg target')
             return MergeResponse(
                 False, False, None, MergeFailureReason.MISSING_TARGET_REF,
                 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)
 
         log.debug('Pulling in target reference %s', target_ref)
         self._validate_pull_reference(target_ref)
         shadow_repo._local_pull(self.path, target_ref)
 
         try:
             log.debug('Pulling in source reference %s', source_ref)
             source_repo._validate_pull_reference(source_ref)
             shadow_repo._local_pull(source_repo.path, source_ref)
         except CommitDoesNotExistError:
             log.exception('Failure when doing local pull on hg shadow repo')
             return MergeResponse(
                 False, False, None, MergeFailureReason.MISSING_SOURCE_REF,
                 metadata={'source_ref': source_ref})
 
         merge_ref = None
         merge_commit_id = None
         close_commit_id = None
         merge_failure_reason = MergeFailureReason.NONE
         metadata = {}
 
         # enforce that close branch should be used only in case we source from
         # an actual Branch
         close_branch = close_branch and source_ref.type == 'branch'
 
         # don't allow to close branch if source and target are the same
         close_branch = close_branch and source_ref.name != target_ref.name
 
         needs_push_on_close = False
         if close_branch and not use_rebase and not dry_run:
             try:
                 close_commit_id, needs_push_on_close = shadow_repo._local_close(
                     target_ref, merger_name, merger_email, source_ref)
                 merge_possible = True
             except RepositoryError:
                 log.exception('Failure when doing close branch on '
                               'shadow repo: %s', shadow_repo)
                 merge_possible = False
                 merge_failure_reason = MergeFailureReason.MERGE_FAILED
         else:
             merge_possible = True
 
         needs_push = False
         if merge_possible:
 
             try:
                 merge_commit_id, needs_push = shadow_repo._local_merge(
                     target_ref, merge_message, merger_name, merger_email,
                     source_ref, use_rebase=use_rebase,
                     close_commit_id=close_commit_id, dry_run=dry_run)
                 merge_possible = True
 
                 # read the state of the close action, if it
                 # maybe required a push
                 needs_push = needs_push or needs_push_on_close
 
                 # Set a bookmark pointing to the merge commit. This bookmark
                 # may be used to easily identify the last successful merge
                 # commit in the shadow repository.
                 shadow_repo.bookmark('pr-merge', revision=merge_commit_id)
                 merge_ref = Reference('book', 'pr-merge', merge_commit_id)
             except SubrepoMergeError:
                 log.exception(
                     'Subrepo merge error during local merge on hg shadow repo.')
                 merge_possible = False
                 merge_failure_reason = MergeFailureReason.SUBREPO_MERGE_FAILED
                 needs_push = False
             except RepositoryError as e:
                 log.exception('Failure when doing local merge on hg shadow repo')
+                metadata['unresolved_files'] = 'no unresolved files found'
+
                 if isinstance(e, UnresolvedFilesInRepo):
                     all_conflicts = list(e.args[0])
                     max_conflicts = 20
                     if len(all_conflicts) > max_conflicts:
                         conflicts = all_conflicts[:max_conflicts] \
                                     + [f'and {len(all_conflicts)-max_conflicts} more.']
                     else:
                         conflicts = all_conflicts
                     metadata['unresolved_files'] = \
                         '\n* conflict: ' + \
                         ('\n * conflict: '.join(conflicts))
 
                 merge_possible = False
                 merge_failure_reason = MergeFailureReason.MERGE_FAILED
                 needs_push = False
 
         if merge_possible and not dry_run:
             if needs_push:
                 # In case the target is a bookmark, update it, so after pushing
                 # the bookmarks is also updated in the target.
                 if target_ref.type == 'book':
                     shadow_repo.bookmark(
                         target_ref.name, revision=merge_commit_id)
                 try:
                     shadow_repo_with_hooks = self.get_shadow_instance(
                         shadow_repository_path,
                         enable_hooks=True)
                     # This is the actual merge action, we push from shadow
                     # into origin.
                     # Note: the push_branches option will push any new branch
                     # defined in the source repository to the target. This may
                     # be dangerous as branches are permanent in Mercurial.
                     # This feature was requested in issue #441.
                     shadow_repo_with_hooks._local_push(
                         merge_commit_id, self.path, push_branches=True,
                         enable_hooks=True)
 
                     # maybe we also need to push the close_commit_id
                     if close_commit_id:
                         shadow_repo_with_hooks._local_push(
                             close_commit_id, self.path, push_branches=True,
                             enable_hooks=True)
                     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'] = 'hg shadow repo'
                     metadata['merge_commit'] = merge_commit_id
             else:
                 merge_succeeded = True
         else:
             merge_succeeded = False
 
         return MergeResponse(
             merge_possible, merge_succeeded, merge_ref, merge_failure_reason,
             metadata=metadata)
 
     def get_shadow_instance(self, shadow_repository_path, enable_hooks=False, cache=False):
         config = self.config.copy()
         if not enable_hooks:
             config.clear_section('hooks')
         return MercurialRepository(shadow_repository_path, config, with_wire={"cache": cache})
 
     def _validate_pull_reference(self, reference):
         if not (reference.name in self.bookmarks or
                 reference.name in self.branches or
                 self.get_commit(reference.commit_id)):
             raise CommitDoesNotExistError(
                 'Unknown branch, bookmark or commit id')
 
     def _local_pull(self, repository_path, reference):
         """
         Fetch a branch, bookmark or commit from a local repository.
         """
         repository_path = os.path.abspath(repository_path)
         if repository_path == self.path:
             raise ValueError('Cannot pull from the same repository')
 
         reference_type_to_option_name = {
             'book': 'bookmark',
             'branch': 'branch',
         }
         option_name = reference_type_to_option_name.get(
             reference.type, 'revision')
 
         if option_name == 'revision':
             ref = reference.commit_id
         else:
             ref = reference.name
 
         options = {option_name: [ref]}
         self._remote.pull_cmd(repository_path, hooks=False, **options)
         self._remote.invalidate_vcs_cache()
 
     def bookmark(self, bookmark, revision=None):
         if isinstance(bookmark, str):
             bookmark = safe_str(bookmark)
         self._remote.bookmark(bookmark, revision=revision)
         self._remote.invalidate_vcs_cache()
 
     def get_path_permissions(self, username):
         hgacl_file = os.path.join(self.path, '.hg/hgacl')
 
         def read_patterns(suffix):
             svalue = None
             for section, option in [
                 ('narrowacl', username + suffix),
                 ('narrowacl', 'default' + suffix),
                 ('narrowhgacl', username + suffix),
                 ('narrowhgacl', 'default' + suffix)
             ]:
                 try:
                     svalue = hgacl.get(section, option)
                     break  # stop at the first value we find
                 except configparser.NoOptionError:
                     pass
             if not svalue:
                 return None
             result = ['/']
             for pattern in svalue.split():
                 result.append(pattern)
                 if '*' not in pattern and '?' not in pattern:
                     result.append(pattern + '/*')
             return result
 
         if os.path.exists(hgacl_file):
             try:
                 hgacl = configparser.RawConfigParser()
                 hgacl.read(hgacl_file)
 
                 includes = read_patterns('.includes')
                 excludes = read_patterns('.excludes')
                 return BasePathPermissionChecker.create_from_patterns(
                     includes, excludes)
             except BaseException as e:
                 msg = 'Cannot read ACL settings from {} on {}: {}'.format(
                     hgacl_file, self.name, e)
                 raise exceptions.RepositoryRequirementError(msg)
         else:
             return None
 
 
 class MercurialIndexBasedCollectionGenerator(CollectionGenerator):
 
     def _commit_factory(self, commit_id):
         if isinstance(commit_id, int):
             return self.repo.get_commit(
                 commit_idx=commit_id, pre_load=self.pre_load)
         else:
             return self.repo.get_commit(
                 commit_id=commit_id, pre_load=self.pre_load)

rhodecode/tests/vcs/test_hg.py

@@ -1,1181 +1,1181 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 os
 
 import mock
 import pytest
 
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.utils import make_db_config
 from rhodecode.lib.vcs import backends
 from rhodecode.lib.vcs.backends.base import (
     Reference, MergeResponse, MergeFailureReason)
 from rhodecode.lib.vcs.backends.hg import MercurialRepository, MercurialCommit
 from rhodecode.lib.vcs.exceptions import (
     RepositoryError, VCSError, NodeDoesNotExistError, CommitDoesNotExistError)
 from rhodecode.lib.vcs.nodes import FileNode, NodeKind, NodeState
 from rhodecode.tests import TEST_HG_REPO, TEST_HG_REPO_CLONE, repo_id_generator
 
 
 pytestmark = pytest.mark.backends("hg")
 
 
 def repo_path_generator():
     """
     Return a different path to be used for cloning repos.
     """
     i = 0
     while True:
         i += 1
         yield '%s-%d' % (TEST_HG_REPO_CLONE, i)
 
 REPO_PATH_GENERATOR = repo_path_generator()
 
 
 @pytest.fixture(scope='class', autouse=True)
 def repo(request, baseapp):
     repo = MercurialRepository(TEST_HG_REPO)
     if request.cls:
         request.cls.repo = repo
     return repo
 
 
 class TestMercurialRepository(object):
 
     # pylint: disable=protected-access
 
     def get_clone_repo(self):
         """
         Return a clone of the base repo.
         """
         clone_path = next(REPO_PATH_GENERATOR)
         repo_clone = MercurialRepository(
             clone_path, create=True, src_url=self.repo.path)
 
         return repo_clone
 
     def get_empty_repo(self):
         """
         Return an empty repo.
         """
         return MercurialRepository(next(REPO_PATH_GENERATOR), create=True)
 
     def test_wrong_repo_path(self):
         wrong_repo_path = '/tmp/errorrepo_hg'
         with pytest.raises(RepositoryError):
             MercurialRepository(wrong_repo_path)
 
     def test_unicode_path_repo(self):
         with pytest.raises(VCSError):
             MercurialRepository('iShouldFail')
 
     def test_unicode_commit_id(self):
         with pytest.raises(CommitDoesNotExistError):
             self.repo.get_commit('unicode-commit-id')
         with pytest.raises(CommitDoesNotExistError):
             self.repo.get_commit('unícøde-spéçial-chäråcter-commit-id')
 
     def test_unicode_bookmark(self):
         self.repo.bookmark('unicode-bookmark')
         self.repo.bookmark('unícøde-spéçial-chäråcter-bookmark')
 
     def test_unicode_branch(self):
         with pytest.raises(KeyError):
             assert self.repo.branches['unicode-branch']
         with pytest.raises(KeyError):
             assert self.repo.branches['unícøde-spéçial-chäråcter-branch']
 
     def test_repo_clone(self):
         if os.path.exists(TEST_HG_REPO_CLONE):
             self.fail(
                 'Cannot test mercurial clone repo as location %s already '
                 'exists. You should manually remove it first.'
                 % TEST_HG_REPO_CLONE)
 
         repo = MercurialRepository(TEST_HG_REPO)
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE,
                                          src_url=TEST_HG_REPO)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # Checking hashes of commits should be enough
         for commit in repo.get_commits():
             raw_id = commit.raw_id
             assert raw_id == repo_clone.get_commit(raw_id).raw_id
 
     def test_repo_clone_with_update(self):
         repo = MercurialRepository(TEST_HG_REPO)
         repo_clone = MercurialRepository(
             TEST_HG_REPO_CLONE + '_w_update',
             src_url=TEST_HG_REPO, do_workspace_checkout=True)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
 
         # check if current workdir was updated
         assert os.path.isfile(
             os.path.join(TEST_HG_REPO_CLONE + '_w_update', 'MANIFEST.in'))
 
     def test_repo_clone_without_update(self):
         repo = MercurialRepository(TEST_HG_REPO)
         repo_clone = MercurialRepository(
             TEST_HG_REPO_CLONE + '_wo_update',
             src_url=TEST_HG_REPO, do_workspace_checkout=False)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         assert not os.path.isfile(
             os.path.join(TEST_HG_REPO_CLONE + '_wo_update', 'MANIFEST.in'))
 
     def test_commit_ids(self):
         # there are 21 commits at bitbucket now
         # so we can assume they would be available from now on
         subset = {'b986218ba1c9b0d6a259fac9b050b1724ed8e545', '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
                   '6cba7170863a2411822803fa77a0a264f1310b35', '56349e29c2af3ac913b28bde9a2c6154436e615b',
                   '2dda4e345facb0ccff1a191052dd1606dba6781d', '6fff84722075f1607a30f436523403845f84cd9e',
                   '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7', '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
                   'dc5d2c0661b61928834a785d3e64a3f80d3aad9c', 'be90031137367893f1c406e0a8683010fd115b79',
                   'db8e58be770518cbb2b1cdfa69146e47cd481481', '84478366594b424af694a6c784cb991a16b87c21',
                   '17f8e105dddb9f339600389c6dc7175d395a535c', '20a662e756499bde3095ffc9bc0643d1def2d0eb',
                   '2e319b85e70a707bba0beff866d9f9de032aa4f9', '786facd2c61deb9cf91e9534735124fb8fc11842',
                   '94593d2128d38210a2fcd1aabff6dda0d6d9edf8', 'aa6a0de05b7612707db567078e130a6cd114a9a7',
                   'eada5a770da98ab0dd7325e29d00e0714f228d09'
                   }
         assert subset.issubset(set(self.repo.commit_ids))
 
         # check if we have the proper order of commits
         org = [
             'b986218ba1c9b0d6a259fac9b050b1724ed8e545',
             '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
             '6cba7170863a2411822803fa77a0a264f1310b35',
             '56349e29c2af3ac913b28bde9a2c6154436e615b',
             '2dda4e345facb0ccff1a191052dd1606dba6781d',
             '6fff84722075f1607a30f436523403845f84cd9e',
             '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',
             '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
             'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',
             'be90031137367893f1c406e0a8683010fd115b79',
             'db8e58be770518cbb2b1cdfa69146e47cd481481',
             '84478366594b424af694a6c784cb991a16b87c21',
             '17f8e105dddb9f339600389c6dc7175d395a535c',
             '20a662e756499bde3095ffc9bc0643d1def2d0eb',
             '2e319b85e70a707bba0beff866d9f9de032aa4f9',
             '786facd2c61deb9cf91e9534735124fb8fc11842',
             '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',
             'aa6a0de05b7612707db567078e130a6cd114a9a7',
             'eada5a770da98ab0dd7325e29d00e0714f228d09',
             '2c1885c735575ca478bf9e17b0029dca68824458',
             'd9bcd465040bf869799b09ad732c04e0eea99fe9',
             '469e9c847fe1f6f7a697b8b25b4bc5b48780c1a7',
             '4fb8326d78e5120da2c7468dcf7098997be385da',
             '62b4a097164940bd66030c4db51687f3ec035eed',
             '536c1a19428381cfea92ac44985304f6a8049569',
             '965e8ab3c44b070cdaa5bf727ddef0ada980ecc4',
             '9bb326a04ae5d98d437dece54be04f830cf1edd9',
             'f8940bcb890a98c4702319fbe36db75ea309b475',
             'ff5ab059786ebc7411e559a2cc309dfae3625a3b',
             '6b6ad5f82ad5bb6190037671bd254bd4e1f4bf08',
             'ee87846a61c12153b51543bf860e1026c6d3dcba',
         ]
         assert org == self.repo.commit_ids[:31]
 
     def test_iter_slice(self):
         sliced = list(self.repo[:10])
         itered = list(self.repo)[:10]
         assert sliced == itered
 
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             indexes = list(self.repo[sfrom:sto])
             assert len(indexes) == size
             assert indexes[0] == self.repo.get_commit(commit_idx=sfrom)
             assert indexes[-1] == self.repo.get_commit(commit_idx=sto - 1)
 
     def test_branches(self):
         # TODO: Need more tests here
 
         # active branches
         assert 'default' in self.repo.branches
         assert 'stable' in self.repo.branches
 
         # closed
         assert 'git' in self.repo._get_branches(closed=True)
         assert 'web' in self.repo._get_branches(closed=True)
 
         for name, id in self.repo.branches.items():
             assert isinstance(self.repo.get_commit(id), MercurialCommit)
 
     def test_tip_in_tags(self):
         # tip is always a tag
         assert 'tip' in self.repo.tags
 
     def test_tip_commit_in_tags(self):
         tip = self.repo.get_commit()
         assert self.repo.tags['tip'] == tip.raw_id
 
     def test_initial_commit(self):
         init_commit = self.repo.get_commit(commit_idx=0)
         init_author = init_commit.author
 
         assert init_commit.message == 'initial import'
         assert init_author == 'Marcin Kuzminski <[e-mail]>'
         assert init_author == init_commit.committer
         assert sorted(init_commit._file_paths) == sorted([
             'vcs/__init__.py',
             'vcs/backends/BaseRepository.py',
             'vcs/backends/__init__.py',
         ])
         assert sorted(init_commit._dir_paths) == sorted(
             ['', 'vcs', 'vcs/backends'])
 
         assert init_commit._dir_paths + init_commit._file_paths == \
             init_commit._paths
 
         with pytest.raises(NodeDoesNotExistError):
             init_commit.get_node(path='foobar')
 
         node = init_commit.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
 
     def test_not_existing_commit(self):
         # rawid
         with pytest.raises(RepositoryError):
             self.repo.get_commit('abcd' * 10)
         # shortid
         with pytest.raises(RepositoryError):
             self.repo.get_commit('erro' * 4)
         # numeric
         with pytest.raises(RepositoryError):
             self.repo.get_commit(commit_idx=self.repo.count() + 1)
 
         # Small chance we ever get to this one
         idx = pow(2, 30)
         with pytest.raises(RepositoryError):
             self.repo.get_commit(commit_idx=idx)
 
     def test_commit10(self):
         commit10 = self.repo.get_commit(commit_idx=10)
         README = """===
 VCS
 ===
 
 Various Version Control System management abstraction layer for Python.
 
 Introduction
 ------------
 
 TODO: To be written...
 
 """
         node = commit10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.str_content == README
 
     def test_local_clone(self):
         clone_path = next(REPO_PATH_GENERATOR)
         self.repo._local_clone(clone_path)
         repo_clone = MercurialRepository(clone_path)
 
         assert self.repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_fails_if_target_exists(self):
         with pytest.raises(RepositoryError):
             self.repo._local_clone(self.repo.path)
 
     def test_update(self):
         repo_clone = self.get_clone_repo()
         branches = repo_clone.branches
 
         repo_clone._update('default')
         assert branches['default'] == repo_clone._identify()
         repo_clone._update('stable')
         assert branches['stable'] == repo_clone._identify()
 
     def test_local_pull_branch(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         default = Reference(
             'branch', 'default', source_repo.branches['default'])
         target_repo._local_pull(source_repo.path, default)
         target_repo = MercurialRepository(target_repo.path)
         assert (target_repo.branches['default'] ==
                 source_repo.branches['default'])
 
         stable = Reference('branch', 'stable', source_repo.branches['stable'])
         target_repo._local_pull(source_repo.path, stable)
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.branches['stable'] == source_repo.branches['stable']
 
     def test_local_pull_bookmark(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         commits = list(source_repo.get_commits(branch_name='default'))
         foo1_id = commits[-5].raw_id
         foo1 = Reference('book', 'foo1', foo1_id)
         source_repo._update(foo1_id)
         source_repo.bookmark('foo1')
 
         foo2_id = commits[-3].raw_id
         foo2 = Reference('book', 'foo2', foo2_id)
         source_repo._update(foo2_id)
         source_repo.bookmark('foo2')
 
         target_repo._local_pull(source_repo.path, foo1)
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.branches['default'] == commits[-5].raw_id
 
         target_repo._local_pull(source_repo.path, foo2)
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.branches['default'] == commits[-3].raw_id
 
     def test_local_pull_commit(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         commits = list(source_repo.get_commits(branch_name='default'))
         commit_id = commits[-5].raw_id
         commit = Reference('rev', commit_id, commit_id)
         target_repo._local_pull(source_repo.path, commit)
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.branches['default'] == commit_id
 
         commit_id = commits[-3].raw_id
         commit = Reference('rev', commit_id, commit_id)
         target_repo._local_pull(source_repo.path, commit)
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.branches['default'] == commit_id
 
     def test_local_pull_from_same_repo(self):
         reference = Reference('branch', 'default', None)
         with pytest.raises(ValueError):
             self.repo._local_pull(self.repo.path, reference)
 
     def test_validate_pull_reference_raises_on_missing_reference(
             self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         reference = Reference(
             'book', 'invalid_reference', 'a' * 40)
 
         with pytest.raises(CommitDoesNotExistError):
             target_repo._validate_pull_reference(reference)
 
     def test_heads(self):
         assert set(self.repo._heads()) == set(self.repo.branches.values())
 
     def test_ancestor(self):
         commits = [
             c.raw_id for c in self.repo.get_commits(branch_name='default')]
         assert self.repo._ancestor(commits[-3], commits[-5]) == commits[-5]
         assert self.repo._ancestor(commits[-5], commits[-3]) == commits[-5]
 
     def test_local_push(self):
         target_repo = self.get_empty_repo()
 
         revisions = list(self.repo.get_commits(branch_name='default'))
         revision = revisions[-5].raw_id
         self.repo._local_push(revision, target_repo.path)
 
         target_repo = MercurialRepository(target_repo.path)
 
         assert target_repo.branches['default'] == revision
 
     def test_hooks_can_be_enabled_for_local_push(self):
         revision = 'deadbeef'
         repo_path = 'test_group/test_repo'
         with mock.patch.object(self.repo, '_remote') as remote_mock:
             self.repo._local_push(revision, repo_path, enable_hooks=True)
         remote_mock.push.assert_called_once_with(
             [revision], repo_path, hooks=True, push_branches=False)
 
     def test_local_merge(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
-        vcsbackend_hg.add_file(target_repo, 'README_MERGE1', 'Version 1')
+        vcsbackend_hg.add_file(target_repo, b'README_MERGE1', b'Version 1')
         target_repo = MercurialRepository(target_repo.path)
         target_rev = target_repo.branches['default']
         target_ref = Reference(
             type='branch', name='default', commit_id=target_rev)
-        vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
+        vcsbackend_hg.add_file(source_repo, b'README_MERGE2', b'Version 2')
         source_repo = MercurialRepository(source_repo.path)
         source_rev = source_repo.branches['default']
         source_ref = Reference(
             type='branch', name='default', commit_id=source_rev)
 
         target_repo._local_pull(source_repo.path, source_ref)
 
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         merge_commit_id, needs_push = target_repo._local_merge(
             target_ref, merge_message, user_name, user_email, source_ref)
         assert needs_push
 
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.commit_ids[-3] == target_rev
         assert target_repo.commit_ids[-2] == source_rev
         last_commit = target_repo.get_commit(merge_commit_id)
         assert last_commit.message.strip() == merge_message
         assert last_commit.author == '%s <%s>' % (user_name, user_email)
 
         assert not os.path.exists(
             os.path.join(target_repo.path, '.hg', 'merge', 'state'))
 
     def test_local_merge_source_is_fast_forward(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
         target_rev = target_repo.branches['default']
         target_ref = Reference(
             type='branch', name='default', commit_id=target_rev)
         vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
         source_repo = MercurialRepository(source_repo.path)
         source_rev = source_repo.branches['default']
         source_ref = Reference(
             type='branch', name='default', commit_id=source_rev)
 
         target_repo._local_pull(source_repo.path, source_ref)
 
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         merge_commit_id, needs_push = target_repo._local_merge(
             target_ref, merge_message, user_name, user_email, source_ref)
         assert merge_commit_id == source_rev
         assert needs_push
 
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.commit_ids[-2] == target_rev
         assert target_repo.commit_ids[-1] == source_rev
 
         assert not os.path.exists(
             os.path.join(target_repo.path, '.hg', 'merge', 'state'))
 
     def test_local_merge_source_is_integrated(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         target_rev = target_repo.branches['default']
         target_ref = Reference(
             type='branch', name='default', commit_id=target_rev)
 
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         merge_commit_id, needs_push = target_repo._local_merge(
             target_ref, merge_message, user_name, user_email, target_ref)
         assert merge_commit_id == target_rev
         assert not needs_push
 
         target_repo = MercurialRepository(target_repo.path)
         assert target_repo.commit_ids[-1] == target_rev
 
         assert not os.path.exists(
             os.path.join(target_repo.path, '.hg', 'merge', 'state'))
 
     def test_local_merge_raises_exception_on_conflict(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
         vcsbackend_hg.add_file(target_repo, 'README_MERGE', 'Version 1')
         target_repo = MercurialRepository(target_repo.path)
         target_rev = target_repo.branches['default']
         target_ref = Reference(
             type='branch', name='default', commit_id=target_rev)
         vcsbackend_hg.add_file(source_repo, 'README_MERGE', 'Version 2')
         source_repo = MercurialRepository(source_repo.path)
         source_rev = source_repo.branches['default']
         source_ref = Reference(
             type='branch', name='default', commit_id=source_rev)
 
         target_repo._local_pull(source_repo.path, source_ref)
         with pytest.raises(RepositoryError):
             target_repo._local_merge(
                 target_ref, 'merge_message', 'user name', '[e-mail]',
                 source_ref)
 
         # Check we are not left in an intermediate merge state
         assert not os.path.exists(
             os.path.join(target_repo.path, '.hg', 'merge', 'state'))
 
     def test_local_merge_of_two_branches_of_the_same_repo(self, backend_hg):
         commits = [
             {'message': 'a'},
             {'message': 'b', 'branch': 'b'},
         ]
         repo = backend_hg.create_repo(commits)
         commit_ids = backend_hg.commit_ids
         target_ref = Reference(
             type='branch', name='default', commit_id=commit_ids['a'])
         source_ref = Reference(
             type='branch', name='b', commit_id=commit_ids['b'])
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         vcs_repo = repo.scm_instance()
         merge_commit_id, needs_push = vcs_repo._local_merge(
             target_ref, merge_message, user_name, user_email, source_ref)
         assert merge_commit_id != source_ref.commit_id
         assert needs_push is True
         commit = vcs_repo.get_commit(merge_commit_id)
         assert commit.merge is True
         assert commit.message == merge_message
 
     def test_maybe_prepare_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             1, 'pr2', 'unused', 'unused2')
 
         assert os.path.isdir(workspace)
         workspace_repo = MercurialRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             1, 'pr2', 'unused', 'unused2')
         assert os.path.isdir(workspace)
 
     def test_cleanup_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             1, 'pr3', 'unused', 'unused2')
 
         assert os.path.isdir(workspace)
         self.repo.cleanup_merge_workspace(1, 'pr3')
 
         assert not os.path.exists(workspace)
 
     def test_cleanup_merge_workspace_invalid_workspace_id(self):
         # No assert: because in case of an inexistent workspace this function
         # should still succeed.
         self.repo.cleanup_merge_workspace(1, 'pr4')
 
     def test_merge_target_is_bookmark(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
         vcsbackend_hg.add_file(target_repo, 'README_MERGE1', 'Version 1')
         vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
         imc = source_repo.in_memory_commit
         imc.add(FileNode(b'file_x', content=source_repo.name))
         imc.commit(
             message='Automatic commit from repo merge test',
             author='Automatic <[e-mail]>')
         target_commit = target_repo.get_commit()
         source_commit = source_repo.get_commit()
         default_branch = target_repo.DEFAULT_BRANCH_NAME
         bookmark_name = 'bookmark'
         target_repo._update(default_branch)
         target_repo.bookmark(bookmark_name)
         target_ref = Reference('book', bookmark_name, target_commit.raw_id)
         source_ref = Reference('branch', default_branch, source_commit.raw_id)
         workspace_id = 'test-merge'
         repo_id = repo_id_generator(target_repo.path)
         merge_response = target_repo.merge(
             repo_id, workspace_id, target_ref, source_repo, source_ref,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(vcsbackend_hg.alias)(
             target_repo.path)
         target_commits = list(target_repo.get_commits())
         commit_ids = [c.raw_id for c in target_commits[:-1]]
         assert source_ref.commit_id in commit_ids
         assert target_ref.commit_id in commit_ids
 
         merge_commit = target_commits[-1]
         assert merge_commit.raw_id == merge_response.merge_ref.commit_id
         assert merge_commit.message.strip() == 'merge message 1'
         assert merge_commit.author == 'test user <[e-mail]>'
 
         # Check the bookmark was updated in the target repo
         assert (
             target_repo.bookmarks[bookmark_name] ==
             merge_response.merge_ref.commit_id)
 
     def test_merge_source_is_bookmark(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
         imc = source_repo.in_memory_commit
         imc.add(FileNode(b'file_x', content=source_repo.name))
         imc.commit(
             message='Automatic commit from repo merge test',
             author='Automatic <[e-mail]>')
         target_commit = target_repo.get_commit()
         source_commit = source_repo.get_commit()
         default_branch = target_repo.DEFAULT_BRANCH_NAME
         bookmark_name = 'bookmark'
         target_ref = Reference('branch', default_branch, target_commit.raw_id)
         source_repo._update(default_branch)
         source_repo.bookmark(bookmark_name)
         source_ref = Reference('book', bookmark_name, source_commit.raw_id)
         workspace_id = 'test-merge'
         repo_id = repo_id_generator(target_repo.path)
         merge_response = target_repo.merge(
             repo_id, workspace_id, target_ref, source_repo, source_ref,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(vcsbackend_hg.alias)(
             target_repo.path)
         target_commits = list(target_repo.get_commits())
         commit_ids = [c.raw_id for c in target_commits]
         assert source_ref.commit_id == commit_ids[-1]
         assert target_ref.commit_id == commit_ids[-2]
 
     def test_merge_target_has_multiple_heads(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=2)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
         vcsbackend_hg.add_file(target_repo, 'README_MERGE1', 'Version 1')
         vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
 
         # add an extra head to the target repo
         imc = target_repo.in_memory_commit
         imc.add(FileNode(b'file_x', content='foo'))
         commits = list(target_repo.get_commits())
         imc.commit(
             message='Automatic commit from repo merge test',
             author='Automatic <[e-mail]>', parents=commits[0:1])
 
         target_commit = target_repo.get_commit()
         source_commit = source_repo.get_commit()
         default_branch = target_repo.DEFAULT_BRANCH_NAME
         target_repo._update(default_branch)
 
         target_ref = Reference('branch', default_branch, target_commit.raw_id)
         source_ref = Reference('branch', default_branch, source_commit.raw_id)
         workspace_id = 'test-merge'
 
         assert len(target_repo._heads(branch='default')) == 2
         heads = target_repo._heads(branch='default')
         expected_merge_response = MergeResponse(
             False, False, None,
             MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS,
             metadata={'heads': heads})
         repo_id = repo_id_generator(target_repo.path)
         merge_response = target_repo.merge(
             repo_id, workspace_id, target_ref, source_repo, source_ref,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False)
         assert merge_response == expected_merge_response
 
     def test_merge_rebase_source_is_updated_bookmark(self, vcsbackend_hg):
         target_repo = vcsbackend_hg.create_repo(number_of_commits=1)
         source_repo = vcsbackend_hg.clone_repo(target_repo)
-        vcsbackend_hg.add_file(target_repo, 'README_MERGE1', 'Version 1')
-        vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
+        vcsbackend_hg.add_file(target_repo, b'README_MERGE1', b'Version 1')
+        vcsbackend_hg.add_file(source_repo, b'README_MERGE2', b'Version 2')
 
         imc = source_repo.in_memory_commit
         imc.add(FileNode(b'file_x', content=safe_bytes(source_repo.name)))
         imc.commit(
             message='Automatic commit from repo merge test',
             author='Automatic <[e-mail]>')
 
         target_commit = target_repo.get_commit()
         source_commit = source_repo.get_commit()
 
-        vcsbackend_hg.add_file(source_repo, 'LICENSE', 'LICENSE Info')
+        vcsbackend_hg.add_file(source_repo, b'LICENSE', b'LICENSE Info')
 
         default_branch = target_repo.DEFAULT_BRANCH_NAME
         bookmark_name = 'bookmark'
         source_repo._update(default_branch)
         source_repo.bookmark(bookmark_name)
 
         target_ref = Reference('branch', default_branch, target_commit.raw_id)
         source_ref = Reference('book', bookmark_name, source_commit.raw_id)
         repo_id = repo_id_generator(target_repo.path)
         workspace_id = 'test-merge'
 
         merge_response = target_repo.merge(
             repo_id, workspace_id, target_ref, source_repo, source_ref,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False, use_rebase=True)
 
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(vcsbackend_hg.alias)(
             target_repo.path)
         last_commit = target_repo.get_commit()
         assert last_commit.message == source_commit.message
         assert last_commit.author == source_commit.author
         # This checks that we effectively did a rebase
         assert last_commit.raw_id != source_commit.raw_id
 
         # Check the target has only 4 commits: 2 were already in target and
         # only two should have been added
         assert len(target_repo.commit_ids) == 2 + 2
 
 
 class TestGetShadowInstance(object):
 
     @pytest.fixture()
     def repo(self, vcsbackend_hg, monkeypatch):
         repo = vcsbackend_hg.repo
         monkeypatch.setattr(repo, 'config', mock.Mock())
         monkeypatch.setattr('rhodecode.lib.vcs.connection.Hg', mock.Mock())
         return repo
 
     def test_passes_config(self, repo):
         shadow = repo.get_shadow_instance(repo.path)
         assert shadow.config == repo.config.copy()
 
     def test_disables_hooks(self, repo):
         shadow = repo.get_shadow_instance(repo.path)
         shadow.config.clear_section.assert_called_once_with('hooks')
 
     def test_allows_to_keep_hooks(self, repo):
         shadow = repo.get_shadow_instance(repo.path, enable_hooks=True)
         assert not shadow.config.clear_section.called
 
 
 class TestMercurialCommit(object):
 
     def _test_equality(self, commit):
         idx = commit.idx
         assert commit == self.repo.get_commit(commit_idx=idx)
 
     def test_equality(self):
         indexes = [0, 10, 20]
         commits = [self.repo.get_commit(commit_idx=idx) for idx in indexes]
         for commit in commits:
             self._test_equality(commit)
 
     def test_default_commit(self):
         tip = self.repo.get_commit('tip')
         assert tip == self.repo.get_commit()
         assert tip == self.repo.get_commit(commit_id=None)
         assert tip == self.repo.get_commit(commit_idx=None)
         assert tip == list(self.repo[-1:])[0]
 
     def test_root_node(self):
         tip = self.repo.get_commit('tip')
         assert tip.root is tip.get_node('')
 
     def test_lazy_fetch(self):
         """
         Test if commit's nodes expands and are cached as we walk through
         the commit. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit = self.repo.get_commit(commit_idx=45)
         assert len(commit.nodes) == 0
         root = commit.root
         assert len(commit.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates commit.nodes
         assert len(commit.nodes) == 9
 
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(commit.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates commit.nodes
         assert len(commit.nodes) == 17
 
         assert docs is commit.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is commit.get_node('docs')
 
     def test_nodes_with_commit(self):
         commit = self.repo.get_commit(commit_idx=45)
         root = commit.root
         docs = root.get_node('docs')
         assert docs is commit.get_node('docs')
         api = docs.get_node('api')
         assert api is commit.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is commit.get_node('docs/api/index.rst')
         assert index is commit.get_node(
             'docs').get_node('api').get_node('index.rst')
 
     def test_branch_and_tags(self):
         commit0 = self.repo.get_commit(commit_idx=0)
         assert commit0.branch == 'default'
         assert commit0.tags == []
 
         commit10 = self.repo.get_commit(commit_idx=10)
         assert commit10.branch == 'default'
         assert commit10.tags == []
 
         commit44 = self.repo.get_commit(commit_idx=44)
         assert commit44.branch == 'web'
 
         tip = self.repo.get_commit('tip')
         assert 'tip' in tip.tags
 
     def test_bookmarks(self):
         commit0 = self.repo.get_commit(commit_idx=0)
         assert commit0.bookmarks == []
 
     def _test_file_size(self, idx, path, size):
         node = self.repo.get_commit(commit_idx=idx).get_node(path)
         assert node.is_file()
         assert node.size == size
 
     def test_file_size(self):
         to_check = (
             (10, 'setup.py', 1068),
             (20, 'setup.py', 1106),
             (60, 'setup.py', 1074),
 
             (10, 'vcs/backends/base.py', 2921),
             (20, 'vcs/backends/base.py', 3936),
             (60, 'vcs/backends/base.py', 6189),
         )
         for idx, path, size in to_check:
             self._test_file_size(idx, path, size)
 
     def test_file_history_from_commits(self):
         node = self.repo[10].get_node('setup.py')
         commit_ids = [commit.raw_id for commit in node.history]
         assert ['3803844fdbd3b711175fc3da9bdacfcd6d29a6fb'] == commit_ids
 
         node = self.repo[20].get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['eada5a770da98ab0dd7325e29d00e0714f228d09',
                 '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb'] == node_ids
 
         # special case we check history from commit that has this particular
         # file changed this means we check if it's included as well
         node = self.repo.get_commit('eada5a770da98ab0dd7325e29d00e0714f228d09')\
             .get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['eada5a770da98ab0dd7325e29d00e0714f228d09',
                 '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb'] == node_ids
 
     def test_file_history(self):
         # we can only check if those commits are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [7, 18, 45, 46, 47, 69, 77],
             'vcs/nodes.py': [
                 7, 8, 24, 26, 30, 45, 47, 49, 56, 57, 58, 59, 60, 61, 73, 76],
             'vcs/backends/hg.py': [
                 4, 5, 6, 11, 12, 13, 14, 15, 16, 21, 22, 23, 26, 27, 28, 30,
                 31, 33, 35, 36, 37, 38, 39, 40, 41, 44, 45, 47, 48, 49, 53, 54,
                 55, 58, 60, 61, 67, 68, 69, 70, 73, 77, 78, 79, 82],
         }
         for path, indexes in files.items():
             tip = self.repo.get_commit(commit_idx=indexes[-1])
             node = tip.get_node(path)
             node_indexes = [commit.idx for commit in node.history]
             assert set(indexes).issubset(set(node_indexes)), (
                 "We assumed that %s is subset of commits for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (indexes, path, node_indexes))
 
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 89: {
                     'lines_no': 31,
                     'commits': [
                         32, 32, 61, 32, 32, 37, 32, 32, 32, 44,
                         37, 37, 37, 37, 45, 37, 44, 37, 37, 37,
                         32, 32, 32, 32, 37, 32, 37, 37, 32,
                         32, 32
                     ]
                 },
                 20: {
                     'lines_no': 1,
                     'commits': [4]
                 },
                 55: {
                     'lines_no': 31,
                     'commits': [
                         32, 32, 45, 32, 32, 37, 32, 32, 32, 44,
                         37, 37, 37, 37, 45, 37, 44, 37, 37, 37,
                         32, 32, 32, 32, 37, 32, 37, 37, 32,
                         32, 32
                     ]
                 }
             },
             'vcs/exceptions.py': {
                 89: {
                     'lines_no': 18,
                     'commits': [
                         16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
                         16, 16, 17, 16, 16, 18, 18, 18
                     ]
                 },
                 20: {
                     'lines_no': 18,
                     'commits': [
                         16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
                         16, 16, 17, 16, 16, 18, 18, 18
                     ]
                 },
                 55: {
                     'lines_no': 18,
                     'commits': [
                         16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
                         17, 16, 16, 18, 18, 18
                     ]
                 }
             },
             'MANIFEST.in': {
                 89: {
                     'lines_no': 5,
                     'commits': [7, 7, 7, 71, 71]
                 },
                 20: {
                     'lines_no': 3,
                     'commits': [7, 7, 7]
                 },
                 55: {
                     'lines_no': 3,
                     'commits': [7, 7, 7]
                 }
             }
         }
 
         for fname, commit_dict in files.items():
             for idx, __ in commit_dict.items():
                 commit = self.repo.get_commit(commit_idx=idx)
                 l1_1 = [x[1] for x in commit.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in commit.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_2 = [
                     x[2]().idx for x in commit.get_file_annotate(fname)]
                 l2 = files[fname][idx]['commits']
                 assert l1 == l2, (
                     "The lists of commit for %s@commit_id%s"
                     "from annotation list should match each other,"
                     "got \n%s \nvs \n%s " % (fname, idx, l1, l2))
 
     def test_commit_state(self):
         """
         Tests which files have been added/changed/removed at particular commit
         """
 
         # commit_id 46ad32a4f974:
         # hg st --rev 46ad32a4f974
         #    changed: 13
         #    added:   20
         #    removed: 1
         changed = set([
             '.hgignore', 'README.rst', 'docs/conf.py', 'docs/index.rst',
             'setup.py', 'tests/test_hg.py', 'tests/test_nodes.py',
             'vcs/__init__.py', 'vcs/backends/__init__.py',
             'vcs/backends/base.py', 'vcs/backends/hg.py', 'vcs/nodes.py',
             'vcs/utils/__init__.py'])
 
         added = set([
             'docs/api/backends/hg.rst', 'docs/api/backends/index.rst',
             'docs/api/index.rst', 'docs/api/nodes.rst',
             'docs/api/web/index.rst', 'docs/api/web/simplevcs.rst',
             'docs/installation.rst', 'docs/quickstart.rst', 'setup.cfg',
             'vcs/utils/baseui_config.py', 'vcs/utils/web.py',
             'vcs/web/__init__.py', 'vcs/web/exceptions.py',
             'vcs/web/simplevcs/__init__.py', 'vcs/web/simplevcs/exceptions.py',
             'vcs/web/simplevcs/middleware.py', 'vcs/web/simplevcs/models.py',
             'vcs/web/simplevcs/settings.py', 'vcs/web/simplevcs/utils.py',
             'vcs/web/simplevcs/views.py'])
 
         removed = set(['docs/api.rst'])
 
         commit64 = self.repo.get_commit('46ad32a4f974')
         assert set((node.path for node in commit64.added)) == added
         assert set((node.path for node in commit64.changed)) == changed
         assert set((node.path for node in commit64.removed)) == removed
 
         # commit_id b090f22d27d6:
         # hg st --rev b090f22d27d6
         #    changed: 13
         #    added:   20
         #    removed: 1
         commit88 = self.repo.get_commit('b090f22d27d6')
         assert set((node.path for node in commit88.added)) == set()
         assert set((node.path for node in commit88.changed)) == \
             set(['.hgignore'])
         assert set((node.path for node in commit88.removed)) == set()
 
         #
         # 85:
         #    added:   2 [
         #       'vcs/utils/diffs.py', 'vcs/web/simplevcs/views/diffs.py']
         #    changed: 4 ['vcs/web/simplevcs/models.py', ...]
         #    removed: 1 ['vcs/utils/web.py']
         commit85 = self.repo.get_commit(commit_idx=85)
         assert set((node.path for node in commit85.added)) == set([
             'vcs/utils/diffs.py',
             'vcs/web/simplevcs/views/diffs.py'])
         assert set((node.path for node in commit85.changed)) == set([
             'vcs/web/simplevcs/models.py',
             'vcs/web/simplevcs/utils.py',
             'vcs/web/simplevcs/views/__init__.py',
             'vcs/web/simplevcs/views/repository.py',
             ])
         assert set((node.path for node in commit85.removed)) == \
             set(['vcs/utils/web.py'])
 
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         commit = self.repo.get_commit(commit_idx=85)
         node = commit.get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
 
         commit = self.repo.get_commit(commit_idx=88)
         node = commit.get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
 
         commit = self.repo.get_commit(commit_idx=85)
         node = commit.get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
 
         # If node has REMOVED state then trying to fetch it would raise
         # CommitError exception
         commit = self.repo.get_commit(commit_idx=2)
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             commit.get_node(path)
         # but it would be one of ``removed`` (commit's attribute)
         assert path in [rf.path for rf in commit.removed]
 
     def test_commit_message_is_unicode(self):
         for cm in self.repo:
             assert type(cm.message) == str
 
     def test_commit_author_is_unicode(self):
         for cm in self.repo:
             assert type(cm.author) == str
 
     def test_repo_files_content_type(self):
         test_commit = self.repo.get_commit(commit_idx=100)
         for node in test_commit.get_node('/'):
             if node.is_file():
                 assert type(node.content) == bytes
                 assert type(node.str_content) == str
 
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         with pytest.raises(VCSError):
             self.repo.get_commit().get_node(path)
 
     def test_author_email(self):
         assert '[e-mail]' == \
             self.repo.get_commit('b986218ba1c9').author_email
         assert '[e-mail]' == \
             self.repo.get_commit('3803844fdbd3').author_email
         assert '' == self.repo.get_commit('84478366594b').author_email
 
     def test_author_username(self):
         assert 'Marcin Kuzminski' == \
             self.repo.get_commit('b986218ba1c9').author_name
         assert 'Lukasz Balcerzak' == \
             self.repo.get_commit('3803844fdbd3').author_name
         assert 'marcink' == \
             self.repo.get_commit('84478366594b').author_name
 
 
 class TestLargeFileRepo(object):
 
     def test_large_file(self, backend_hg):
         repo = backend_hg.create_test_repo('largefiles', make_db_config())
 
         tip = repo.scm_instance().get_commit()
         node = tip.get_node('.hglf/thisfileislarge')
 
         lf_node = node.get_largefile_node()
 
         assert lf_node.is_largefile() is True
         assert lf_node.size == 1024000
         assert lf_node.name == '.hglf/thisfileislarge'
 
 
 class TestGetBranchName(object):
     def test_returns_ref_name_when_type_is_branch(self):
         ref = self._create_ref('branch', 'fake-name')
         result = self.repo._get_branch_name(ref)
         assert result == ref.name
 
     @pytest.mark.parametrize("type_", ("book", "tag"))
     def test_queries_remote_when_type_is_not_branch(self, type_):
         ref = self._create_ref(type_, 'wrong-fake-name')
         with mock.patch.object(self.repo, "_remote") as remote_mock:
             remote_mock.ctx_branch.return_value = "fake-name"
             result = self.repo._get_branch_name(ref)
         assert result == "fake-name"
         remote_mock.ctx_branch.assert_called_once_with(ref.commit_id)
 
     def _create_ref(self, type_, name):
         ref = mock.Mock()
         ref.type = type_
         ref.name = 'wrong-fake-name'
         ref.commit_id = "deadbeef"
         return ref
 
 
 class TestIsTheSameBranch(object):
     def test_returns_true_when_branches_are_equal(self):
         source_ref = mock.Mock(name="source-ref")
         target_ref = mock.Mock(name="target-ref")
         branch_name_patcher = mock.patch.object(
             self.repo, "_get_branch_name", return_value="default")
         with branch_name_patcher as branch_name_mock:
             result = self.repo._is_the_same_branch(source_ref, target_ref)
 
         expected_calls = [mock.call(source_ref), mock.call(target_ref)]
         assert branch_name_mock.call_args_list == expected_calls
         assert result is True
 
     def test_returns_false_when_branches_are_not_equal(self):
         source_ref = mock.Mock(name="source-ref")
         source_ref.name = "source-branch"
         target_ref = mock.Mock(name="target-ref")
         source_ref.name = "target-branch"
 
         def side_effect(ref):
             return ref.name
 
         branch_name_patcher = mock.patch.object(
             self.repo, "_get_branch_name", side_effect=side_effect)
         with branch_name_patcher as branch_name_mock:
             result = self.repo._is_the_same_branch(source_ref, target_ref)
 
         expected_calls = [mock.call(source_ref), mock.call(target_ref)]
         assert branch_name_mock.call_args_list == expected_calls
         assert result is False