Changeset - e885a8efef34
[Not reviewed]
default
0 1 0
Mads Kiilerich - 7 years ago 2018-09-30 18:31:37
mads@kiilerich.com
tests: fix vcs Hg helper function get_config_value to not modify global state when using a custom config file

This leak was what made test_get_user_name and test_get_user_email pass when
test_get_config_value had been run first.
1 file changed with 4 insertions and 2 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/vcs/backends/hg/repository.py
Show inline comments
 
@@ -413,219 +413,221 @@ class MercurialRepository(BaseRepository
 
        """
 
        Gets an ID revision given as str. This will always return a full
 
        40 char revision number
 

	
 
        :param revision: str or int or None
 
        """
 
        if isinstance(revision, unicode):
 
            revision = safe_str(revision)
 

	
 
        if self._empty:
 
            raise EmptyRepositoryError("There are no changesets yet")
 

	
 
        if revision in [-1, None]:
 
            revision = 'tip'
 

	
 
        try:
 
            if isinstance(revision, int):
 
                return self._repo[revision].hex()
 
            try:
 
                return scmutil.revsymbol(self._repo, revision).hex()
 
            except AttributeError: # revsymbol was introduced in Mercurial 4.6
 
                return self._repo[revision].hex()
 
        except (IndexError, ValueError, RepoLookupError, TypeError):
 
            msg = ("Revision %s does not exist for %s" % (revision, self))
 
            raise ChangesetDoesNotExistError(msg)
 
        except (LookupError, ):
 
            msg = ("Ambiguous identifier `%s` for %s" % (revision, self))
 
            raise ChangesetDoesNotExistError(msg)
 

	
 
    def get_ref_revision(self, ref_type, ref_name):
 
        """
 
        Returns revision number for the given reference.
 
        """
 
        ref_name = safe_str(ref_name)
 
        if ref_type == 'rev' and not ref_name.strip('0'):
 
            return self.EMPTY_CHANGESET
 
        # lookup up the exact node id
 
        _revset_predicates = {
 
                'branch': 'branch',
 
                'book': 'bookmark',
 
                'tag': 'tag',
 
                'rev': 'id',
 
            }
 
        # avoid expensive branch(x) iteration over whole repo
 
        rev_spec = "%%s & %s(%%s)" % _revset_predicates[ref_type]
 
        try:
 
            revs = self._repo.revs(rev_spec, ref_name, ref_name)
 
        except LookupError:
 
            msg = ("Ambiguous identifier %s:%s for %s" % (ref_type, ref_name, self.name))
 
            raise ChangesetDoesNotExistError(msg)
 
        except RepoLookupError:
 
            msg = ("Revision %s:%s does not exist for %s" % (ref_type, ref_name, self.name))
 
            raise ChangesetDoesNotExistError(msg)
 
        if revs:
 
            try:
 
                revision = revs.last()
 
            except AttributeError:
 
                # removed in hg 3.2
 
                revision = revs[-1]
 
        else:
 
            # TODO: just report 'not found'?
 
            revision = ref_name
 

	
 
        return self._get_revision(revision)
 

	
 
    def _get_archives(self, archive_name='tip'):
 
        allowed = self.baseui.configlist("web", "allow_archive",
 
                                         untrusted=True)
 
        for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:
 
            if i[0] in allowed or self._repo.ui.configbool("web",
 
                                                           "allow" + i[0],
 
                                                           untrusted=True):
 
                yield {"type": i[0], "extension": i[1], "node": archive_name}
 

	
 
    def _get_url(self, url):
 
        """
 
        Returns normalized url. If schema is not given, would fall
 
        to filesystem
 
        (``file:///``) schema.
 
        """
 
        url = safe_str(url)
 
        if url != 'default' and '://' not in url:
 
            url = "file:" + urllib.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_changeset(self, revision=None):
 
        """
 
        Returns ``MercurialChangeset`` object representing repository's
 
        changeset at the given ``revision``.
 
        """
 
        revision = self._get_revision(revision)
 
        changeset = MercurialChangeset(repository=self, revision=revision)
 
        return changeset
 

	
 
    def get_changesets(self, start=None, end=None, start_date=None,
 
                       end_date=None, branch_name=None, reverse=False, max_revisions=None):
 
        """
 
        Returns iterator of ``MercurialChangeset`` objects from start to end
 
        (both are inclusive)
 

	
 
        :param start: None, str, int or mercurial lookup format
 
        :param end:  None, str, int or mercurial lookup format
 
        :param start_date:
 
        :param end_date:
 
        :param branch_name:
 
        :param reversed: return changesets in reversed order
 
        """
 
        start_raw_id = self._get_revision(start)
 
        start_pos = self.revisions.index(start_raw_id) if start else None
 
        end_raw_id = self._get_revision(end)
 
        end_pos = self.revisions.index(end_raw_id) if end else None
 

	
 
        if None not in [start, end] and start_pos > end_pos:
 
            raise RepositoryError("Start revision '%s' cannot be "
 
                                  "after end revision '%s'" % (start, end))
 

	
 
        if branch_name and branch_name not in self.allbranches.keys():
 
            msg = ("Branch %s not found in %s" % (branch_name, self))
 
            raise BranchDoesNotExistError(msg)
 
        if end_pos is not None:
 
            end_pos += 1
 
        # filter branches
 
        filter_ = []
 
        if branch_name:
 
            filter_.append('branch("%s")' % safe_str(branch_name))
 
        if start_date:
 
            filter_.append('date(">%s")' % start_date)
 
        if end_date:
 
            filter_.append('date("<%s")' % end_date)
 
        if filter_ or max_revisions:
 
            if filter_:
 
                revspec = ' and '.join(filter_)
 
            else:
 
                revspec = 'all()'
 
            if max_revisions:
 
                revspec = 'limit(%s, %s)' % (revspec, max_revisions)
 
            revisions = scmutil.revrange(self._repo, [revspec])
 
        else:
 
            revisions = self.revisions
 

	
 
        # this is very much a hack to turn this into a list; a better solution
 
        # would be to get rid of this function entirely and use revsets
 
        revs = list(revisions)[start_pos:end_pos]
 
        if reverse:
 
            revs = reversed(revs)
 

	
 
        return CollectionGenerator(self, revs)
 

	
 
    def pull(self, url):
 
        """
 
        Tries to pull changes from external location.
 
        """
 
        url = self._get_url(url)
 
        other = peer(self._repo, {}, url)
 
        try:
 
            # hg 3.2 moved push / pull to exchange module
 
            from mercurial import exchange
 
            exchange.pull(self._repo, other, heads=None, force=None)
 
        except ImportError:
 
            self._repo.pull(other, heads=None, force=None)
 
        except Abort as err:
 
            # Propagate error but with vcs's type
 
            raise RepositoryError(str(err))
 

	
 
    @LazyProperty
 
    def workdir(self):
 
        """
 
        Returns ``Workdir`` instance for this repository.
 
        """
 
        return MercurialWorkdir(self)
 

	
 
    def get_config_value(self, section, name=None, config_file=None):
 
        """
 
        Returns configuration value for a given [``section``] and ``name``.
 

	
 
        :param section: Section we want to retrieve value from
 
        :param name: Name of configuration we want to retrieve
 
        :param config_file: A path to file which should be used to retrieve
 
          configuration from (might also be a list of file paths)
 
        """
 
        if config_file is None:
 
            config_file = []
 
        elif isinstance(config_file, basestring):
 
            config_file = [config_file]
 

	
 
        config = self._repo.ui
 
        for path in config_file:
 
            config.readconfig(path)
 
        if config_file:
 
            config = ui.ui()
 
            for path in config_file:
 
                config.readconfig(path)
 
        return config.config(section, name)
 

	
 
    def get_user_name(self, config_file=None):
 
        """
 
        Returns user's name from global configuration file.
 

	
 
        :param config_file: A path to file which should be used to retrieve
 
          configuration from (might also be a list of file paths)
 
        """
 
        username = self.get_config_value('ui', 'username', config_file=config_file)
 
        if username:
 
            return author_name(username)
 
        return None
 

	
 
    def get_user_email(self, config_file=None):
 
        """
 
        Returns user's email from global configuration file.
 

	
 
        :param config_file: A path to file which should be used to retrieve
 
          configuration from (might also be a list of file paths)
 
        """
 
        username = self.get_config_value('ui', 'username', config_file=config_file)
 
        if username:
 
            return author_email(username)
 
        return None
0 comments (0 inline, 0 general)