comparison hgext/convert/gnuarch.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

    dateutil,
    procutil,
)
from . import common

class gnuarch_source(common.converter_source, common.commandline):

    class gnuarch_rev(object):
        def __init__(self, rev):
            self.rev = rev
            self.summary = ''
            self.date = None

    def __init__(self, ui, repotype, path, revs=None):
        super(gnuarch_source, self).__init__(ui, repotype, path, revs=revs)

        if not os.path.exists(os.path.join(path, '{arch}')):
            raise common.NoRepo(_("%s does not look like a GNU Arch repository")
                         % path)

        # Could use checktool, but we want to check for baz or tla.
        self.execmd = None
        if procutil.findexe('baz'):
            self.execmd = 'baz'

        self.encoding = encoding.encoding
        self.archives = []

    def before(self):
        # Get registered archives
        self.archives = [i.rstrip('\n')
                         for i in self.runlines0('archives', '-n')]

        if self.execmd == 'tla':
            output = self.run0('tree-version', self.path)
        else:
            output = self.run0('tree-version', '-d', self.path)
        self.treeversion = output.strip()

        # Get name of temporary directory
        version = self.treeversion.split('/')
        self.tmppath = os.path.join(pycompat.fsencode(tempfile.gettempdir()),
                                    'hg-%s' % version[1])

        # Generate parents dictionary
        self.parents[None] = []
        treeversion = self.treeversion
        child = None
        while treeversion:
            self.ui.status(_('analyzing tree version %s...\n') % treeversion)

            archive = treeversion.split('/')[0]
            if archive not in self.archives:
                self.ui.status(_('tree analysis stopped because it points to '
                                 'an unregistered archive %s...\n') % archive)
                break

            # Get the complete list of revisions for that tree version
            output, status = self.runlines('revisions', '-r', '-f', treeversion)
            self.checkexit(status, 'failed retrieving revisions for %s'
                           % treeversion)

            # No new iteration unless a revision has a continuation-of header
            treeversion = None

            for l in output:

                self.parents[rev] = []

                # Read author, date and summary
                catlog, status = self.run('cat-log', '-d', self.path, rev)
                if status:
                    catlog  = self.run0('cat-archive-log', rev)
                self._parsecatlog(catlog, rev)

                # Populate the parents map
                self.parents[child].append(rev)

                # Check if we have to follow the usual incremental history
                # or if we have to 'jump' to a different treeversion given
                # by the continuation-of header.
                if self.changes[rev].continuationof:
                    treeversion = '--'.join(
                        self.changes[rev].continuationof.split('--')[:-1])
                    break

                # If we reached a base-0 revision w/o any continuation-of
                # header, it means the tree history ends here.
                if rev[-6:] == 'base-0':

        self.lastrev = rev
        return sorted(set(changes)), copies, set()

    def getcommit(self, rev):
        changes = self.changes[rev]
        return common.commit(author=changes.author, date=changes.date,
                             desc=changes.summary, parents=self.parents[rev],
                             rev=rev)

    def gettags(self):
        return self.tags

    def _execute(self, cmd, *args, **kwargs):

        self.ui.debug(cmdline, '\n')
        return os.system(pycompat.rapply(procutil.tonativestr, cmdline))

    def _update(self, rev):
        self.ui.debug('applying revision %s...\n' % rev)
        changeset, status = self.runlines('replay', '-d', self.tmppath,
                                              rev)
        if status:
            # Something went wrong while merging (baz or tla
            # issue?), get latest revision and try from there
            shutil.rmtree(self.tmppath, ignore_errors=True)
            self._obtainrevision(rev)
        else:
            old_rev = self.parents[rev][0]
            self.ui.debug('computing changeset between %s and %s...\n'
                          % (old_rev, rev))
            self._parsechangeset(changeset, rev)

    def _getfile(self, name, rev):
        mode = os.lstat(os.path.join(self.tmppath, name)).st_mode
        if stat.S_ISLNK(mode):

        try:
            catlog = self.catlogparser.parsestr(data)

            # Commit date
            self.changes[rev].date = dateutil.datestr(
                dateutil.strdate(catlog['Standard-date'],
                             '%Y-%m-%d %H:%M:%S'))

            # Commit author
            self.changes[rev].author = self.recode(catlog['Creator'])

            # Commit description
            self.changes[rev].summary = '\n\n'.join((catlog['Summary'],
                                                    catlog.get_payload()))
            self.changes[rev].summary = self.recode(self.changes[rev].summary)

            # Commit revision origin when dealing with a branch or tag
            if 'Continuation-of' in catlog:
                self.changes[rev].continuationof = self.recode(
                    catlog['Continuation-of'])
        except Exception:
            raise error.Abort(_('could not parse cat-log of %s') % rev)

    def _parsechangeset(self, data, rev):
        for l in data:

43076:2372284d9457

    dateutil,
    procutil,
)
from . import common


class gnuarch_source(common.converter_source, common.commandline):
    class gnuarch_rev(object):
        def __init__(self, rev):
            self.rev = rev
            self.summary = ''
            self.date = None

    def __init__(self, ui, repotype, path, revs=None):
        super(gnuarch_source, self).__init__(ui, repotype, path, revs=revs)

        if not os.path.exists(os.path.join(path, '{arch}')):
            raise common.NoRepo(
                _("%s does not look like a GNU Arch repository") % path
            )

        # Could use checktool, but we want to check for baz or tla.
        self.execmd = None
        if procutil.findexe('baz'):
            self.execmd = 'baz'

        self.encoding = encoding.encoding
        self.archives = []

    def before(self):
        # Get registered archives
        self.archives = [
            i.rstrip('\n') for i in self.runlines0('archives', '-n')
        ]

        if self.execmd == 'tla':
            output = self.run0('tree-version', self.path)
        else:
            output = self.run0('tree-version', '-d', self.path)
        self.treeversion = output.strip()

        # Get name of temporary directory
        version = self.treeversion.split('/')
        self.tmppath = os.path.join(
            pycompat.fsencode(tempfile.gettempdir()), 'hg-%s' % version[1]
        )

        # Generate parents dictionary
        self.parents[None] = []
        treeversion = self.treeversion
        child = None
        while treeversion:
            self.ui.status(_('analyzing tree version %s...\n') % treeversion)

            archive = treeversion.split('/')[0]
            if archive not in self.archives:
                self.ui.status(
                    _(
                        'tree analysis stopped because it points to '
                        'an unregistered archive %s...\n'
                    )
                    % archive
                )
                break

            # Get the complete list of revisions for that tree version
            output, status = self.runlines('revisions', '-r', '-f', treeversion)
            self.checkexit(
                status, 'failed retrieving revisions for %s' % treeversion
            )

            # No new iteration unless a revision has a continuation-of header
            treeversion = None

            for l in output:

                self.parents[rev] = []

                # Read author, date and summary
                catlog, status = self.run('cat-log', '-d', self.path, rev)
                if status:
                    catlog = self.run0('cat-archive-log', rev)
                self._parsecatlog(catlog, rev)

                # Populate the parents map
                self.parents[child].append(rev)

                # Check if we have to follow the usual incremental history
                # or if we have to 'jump' to a different treeversion given
                # by the continuation-of header.
                if self.changes[rev].continuationof:
                    treeversion = '--'.join(
                        self.changes[rev].continuationof.split('--')[:-1]
                    )
                    break

                # If we reached a base-0 revision w/o any continuation-of
                # header, it means the tree history ends here.
                if rev[-6:] == 'base-0':

        self.lastrev = rev
        return sorted(set(changes)), copies, set()

    def getcommit(self, rev):
        changes = self.changes[rev]
        return common.commit(
            author=changes.author,
            date=changes.date,
            desc=changes.summary,
            parents=self.parents[rev],
            rev=rev,
        )

    def gettags(self):
        return self.tags

    def _execute(self, cmd, *args, **kwargs):

        self.ui.debug(cmdline, '\n')
        return os.system(pycompat.rapply(procutil.tonativestr, cmdline))

    def _update(self, rev):
        self.ui.debug('applying revision %s...\n' % rev)
        changeset, status = self.runlines('replay', '-d', self.tmppath, rev)
        if status:
            # Something went wrong while merging (baz or tla
            # issue?), get latest revision and try from there
            shutil.rmtree(self.tmppath, ignore_errors=True)
            self._obtainrevision(rev)
        else:
            old_rev = self.parents[rev][0]
            self.ui.debug(
                'computing changeset between %s and %s...\n' % (old_rev, rev)
            )
            self._parsechangeset(changeset, rev)

    def _getfile(self, name, rev):
        mode = os.lstat(os.path.join(self.tmppath, name)).st_mode
        if stat.S_ISLNK(mode):

        try:
            catlog = self.catlogparser.parsestr(data)

            # Commit date
            self.changes[rev].date = dateutil.datestr(
                dateutil.strdate(catlog['Standard-date'], '%Y-%m-%d %H:%M:%S')
            )

            # Commit author
            self.changes[rev].author = self.recode(catlog['Creator'])

            # Commit description
            self.changes[rev].summary = '\n\n'.join(
                (catlog['Summary'], catlog.get_payload())
            )
            self.changes[rev].summary = self.recode(self.changes[rev].summary)

            # Commit revision origin when dealing with a branch or tag
            if 'Continuation-of' in catlog:
                self.changes[rev].continuationof = self.recode(
                    catlog['Continuation-of']
                )
        except Exception:
            raise error.Abort(_('could not parse cat-log of %s') % rev)

    def _parsechangeset(self, data, rev):
        for l in data: