comparison mercurial/archival.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

import time
import zipfile
import zlib

from .i18n import _
from .node import (
    nullrev,
)

from . import (
    error,
    formatter,
    match as matchmod,
    pycompat,
    scmutil,
    util,
    vfs as vfsmod,
)
stringio = util.stringio

# from unzip source code:
_UNX_IFREG = 0x8000
_UNX_IFLNK = 0xa000

def tidyprefix(dest, kind, prefix):
    '''choose prefix to use for names in archive.  make sure prefix is
    safe for consumers.'''

            raise ValueError('dest must be string if no prefix')
        prefix = os.path.basename(dest)
        lower = prefix.lower()
        for sfx in exts.get(kind, []):
            if lower.endswith(sfx):
                prefix = prefix[:-len(sfx)]
                break
    lpfx = os.path.normpath(util.localpath(prefix))
    prefix = util.pconvert(lpfx)
    if not prefix.endswith('/'):
        prefix += '/'

        prefix = prefix[2:]
    if prefix.startswith('../') or os.path.isabs(lpfx) or '/../' in prefix:
        raise error.Abort(_('archive prefix contains illegal components'))
    return prefix

exts = {
    'tar': ['.tar'],
    'tbz2': ['.tbz2', '.tar.bz2'],
    'tgz': ['.tgz', '.tar.gz'],
    'zip': ['.zip'],
    'txz': ['.txz', '.tar.xz']
    }

def guesskind(dest):
    for kind, extensions in exts.iteritems():
        if any(dest.endswith(ext) for ext in extensions):
            return kind
    return None

def _rootctx(repo):
    # repo[0] may be hidden
    for rev in repo:
        return repo[rev]
    return repo[nullrev]

# {tags} on ctx includes local tags and 'tip', with no current way to limit
# that to global tags.  Therefore, use {latesttag} as a substitute when
# the distance is 0, since that will be the list of global tags on ctx.
_defaultmetatemplate = br'''

{ifeq(latesttagdistance, 0, join(latesttag % "tag: {tag}", "\n"),
      separate("\n",
               join(latesttag % "latesttag: {tag}", "\n"),
               "latesttagdistance: {latesttagdistance}",
               "changessincelatesttag: {changessincelatesttag}"))}
'''[1:]  # drop leading '\n'

def buildmetadata(ctx):
    '''build content of .hg_archival.txt'''
    repo = ctx.repo()

    opts = {
        'template': repo.ui.config('experimental', 'archivemetatemplate',
                                   _defaultmetatemplate)
    }

    out = util.stringio()

    fm = formatter.formatter(repo.ui, out, 'archive', opts)

        fm.data(dirty=dirty)
    fm.end()

    return out.getvalue()

class tarit(object):
    '''write archive to tar file or stream.  can write uncompressed,
    or compress with gzip or bzip2.'''

    class GzipFileWithTime(gzip.GzipFile):

        def __init__(self, *args, **kw):
            timestamp = None
            if r'timestamp' in kw:
                timestamp = kw.pop(r'timestamp')
            if timestamp is None:

            else:
                self.timestamp = timestamp
            gzip.GzipFile.__init__(self, *args, **kw)

        def _write_gzip_header(self):
            self.fileobj.write('\037\213')             # magic header
            self.fileobj.write('\010')                 # compression method
            fname = self.name
            if fname and fname.endswith('.gz'):
                fname = fname[:-3]
            flags = 0
            if fname:

        def taropen(mode, name='', fileobj=None):
            if kind == 'gz':
                mode = mode[0:1]
                if not fileobj:
                    fileobj = open(name, mode + 'b')
                gzfileobj = self.GzipFileWithTime(name,
                                                  pycompat.sysstr(mode + 'b'),
                                                  zlib.Z_BEST_COMPRESSION,
                                                  fileobj, timestamp=mtime)
                self.fileobj = gzfileobj
                return tarfile.TarFile.taropen(
                    name, pycompat.sysstr(mode), gzfileobj)
            else:
                return tarfile.open(
                    name, pycompat.sysstr(mode + kind), fileobj)

        if isinstance(dest, bytes):
            self.z = taropen('w:', name=dest)
        else:
            self.z = taropen('w|', fileobj=dest)

    def done(self):
        self.z.close()
        if self.fileobj:
            self.fileobj.close()

class zipit(object):
    '''write archive to zip file or stream.  can write uncompressed,
    or compressed with deflate.'''

    def __init__(self, dest, mtime, compress=True):
        if isinstance(dest, bytes):
            dest = pycompat.fsdecode(dest)
        self.z = zipfile.ZipFile(dest, r'w',
                                 compress and zipfile.ZIP_DEFLATED or
                                 zipfile.ZIP_STORED)

        # Python's zipfile module emits deprecation warnings if we try
        # to store files with a date before 1980.
        epoch = 315532800 # calendar.timegm((1980, 1, 1, 0, 0, 0, 1, 1, 0))
        if mtime < epoch:
            mtime = epoch

        self.mtime = mtime
        self.date_time = time.gmtime(mtime)[:6]

            ftype = _UNX_IFLNK
        i.external_attr = (mode | ftype) << 16
        # add "extended-timestamp" extra block, because zip archives
        # without this will be extracted with unexpected timestamp,
        # if TZ is not configured as GMT
        i.extra += struct.pack('<hhBl',
                               0x5455,     # block type: "extended-timestamp"
                               1 + 4,      # size of this block
                               1,          # "modification time is present"
                               int(self.mtime)) # last modification (UTC)
        self.z.writestr(i, data)

    def done(self):
        self.z.close()

class fileit(object):
    '''write archive as files in directory.'''

    def __init__(self, name, mtime):

            os.utime(destfile, (self.mtime, self.mtime))

    def done(self):
        pass

archivers = {
    'files': fileit,
    'tar': tarit,
    'tbz2': lambda name, mtime: tarit(name, mtime, 'bz2'),
    'tgz': lambda name, mtime: tarit(name, mtime, 'gz'),
    'txz': lambda name, mtime: tarit(name, mtime, 'xz'),
    'uzip': lambda name, mtime: zipit(name, mtime, False),
    'zip': zipit,
    }

def archive(repo, dest, node, kind, decode=True, match=None,
            prefix='', mtime=None, subrepos=False):
    '''create archive of repo as it was at node.

    dest can be name of directory, name of archive file, or file
    object to write archive to.

    files = [f for f in ctx.manifest().matches(match)]
    total = len(files)
    if total:
        files.sort()
        scmutil.prefetchfiles(repo, [ctx.rev()],
                              scmutil.matchfiles(repo, files))
        progress = repo.ui.makeprogress(_('archiving'), unit=_('files'),
                                        total=total)
        progress.update(0)
        for f in files:
            ff = ctx.flags(f)
            write(f, 'x' in ff and 0o755 or 0o644, 'l' in ff, ctx[f].data)
            progress.increment(item=f)

43076:2372284d9457

import time
import zipfile
import zlib

from .i18n import _
from .node import nullrev

from . import (
    error,
    formatter,
    match as matchmod,
    pycompat,
    scmutil,
    util,
    vfs as vfsmod,
)

stringio = util.stringio

# from unzip source code:
_UNX_IFREG = 0x8000
_UNX_IFLNK = 0xA000


def tidyprefix(dest, kind, prefix):
    '''choose prefix to use for names in archive.  make sure prefix is
    safe for consumers.'''

            raise ValueError('dest must be string if no prefix')
        prefix = os.path.basename(dest)
        lower = prefix.lower()
        for sfx in exts.get(kind, []):
            if lower.endswith(sfx):
                prefix = prefix[: -len(sfx)]
                break
    lpfx = os.path.normpath(util.localpath(prefix))
    prefix = util.pconvert(lpfx)
    if not prefix.endswith('/'):
        prefix += '/'

        prefix = prefix[2:]
    if prefix.startswith('../') or os.path.isabs(lpfx) or '/../' in prefix:
        raise error.Abort(_('archive prefix contains illegal components'))
    return prefix


exts = {
    'tar': ['.tar'],
    'tbz2': ['.tbz2', '.tar.bz2'],
    'tgz': ['.tgz', '.tar.gz'],
    'zip': ['.zip'],
    'txz': ['.txz', '.tar.xz'],
}


def guesskind(dest):
    for kind, extensions in exts.iteritems():
        if any(dest.endswith(ext) for ext in extensions):
            return kind
    return None


def _rootctx(repo):
    # repo[0] may be hidden
    for rev in repo:
        return repo[rev]
    return repo[nullrev]


# {tags} on ctx includes local tags and 'tip', with no current way to limit
# that to global tags.  Therefore, use {latesttag} as a substitute when
# the distance is 0, since that will be the list of global tags on ctx.
_defaultmetatemplate = br'''

{ifeq(latesttagdistance, 0, join(latesttag % "tag: {tag}", "\n"),
      separate("\n",
               join(latesttag % "latesttag: {tag}", "\n"),
               "latesttagdistance: {latesttagdistance}",
               "changessincelatesttag: {changessincelatesttag}"))}
'''[
    1:
]  # drop leading '\n'


def buildmetadata(ctx):
    '''build content of .hg_archival.txt'''
    repo = ctx.repo()

    opts = {
        'template': repo.ui.config(
            'experimental', 'archivemetatemplate', _defaultmetatemplate
        )
    }

    out = util.stringio()

    fm = formatter.formatter(repo.ui, out, 'archive', opts)

        fm.data(dirty=dirty)
    fm.end()

    return out.getvalue()


class tarit(object):
    '''write archive to tar file or stream.  can write uncompressed,
    or compress with gzip or bzip2.'''

    class GzipFileWithTime(gzip.GzipFile):
        def __init__(self, *args, **kw):
            timestamp = None
            if r'timestamp' in kw:
                timestamp = kw.pop(r'timestamp')
            if timestamp is None:

            else:
                self.timestamp = timestamp
            gzip.GzipFile.__init__(self, *args, **kw)

        def _write_gzip_header(self):
            self.fileobj.write('\037\213')  # magic header
            self.fileobj.write('\010')  # compression method
            fname = self.name
            if fname and fname.endswith('.gz'):
                fname = fname[:-3]
            flags = 0
            if fname:

        def taropen(mode, name='', fileobj=None):
            if kind == 'gz':
                mode = mode[0:1]
                if not fileobj:
                    fileobj = open(name, mode + 'b')
                gzfileobj = self.GzipFileWithTime(
                    name,
                    pycompat.sysstr(mode + 'b'),
                    zlib.Z_BEST_COMPRESSION,
                    fileobj,
                    timestamp=mtime,
                )
                self.fileobj = gzfileobj
                return tarfile.TarFile.taropen(
                    name, pycompat.sysstr(mode), gzfileobj
                )
            else:
                return tarfile.open(name, pycompat.sysstr(mode + kind), fileobj)

        if isinstance(dest, bytes):
            self.z = taropen('w:', name=dest)
        else:
            self.z = taropen('w|', fileobj=dest)

    def done(self):
        self.z.close()
        if self.fileobj:
            self.fileobj.close()


class zipit(object):
    '''write archive to zip file or stream.  can write uncompressed,
    or compressed with deflate.'''

    def __init__(self, dest, mtime, compress=True):
        if isinstance(dest, bytes):
            dest = pycompat.fsdecode(dest)
        self.z = zipfile.ZipFile(
            dest, r'w', compress and zipfile.ZIP_DEFLATED or zipfile.ZIP_STORED
        )

        # Python's zipfile module emits deprecation warnings if we try
        # to store files with a date before 1980.
        epoch = 315532800  # calendar.timegm((1980, 1, 1, 0, 0, 0, 1, 1, 0))
        if mtime < epoch:
            mtime = epoch

        self.mtime = mtime
        self.date_time = time.gmtime(mtime)[:6]

            ftype = _UNX_IFLNK
        i.external_attr = (mode | ftype) << 16
        # add "extended-timestamp" extra block, because zip archives
        # without this will be extracted with unexpected timestamp,
        # if TZ is not configured as GMT
        i.extra += struct.pack(
            '<hhBl',
            0x5455,  # block type: "extended-timestamp"
            1 + 4,  # size of this block
            1,  # "modification time is present"
            int(self.mtime),
        )  # last modification (UTC)
        self.z.writestr(i, data)

    def done(self):
        self.z.close()


class fileit(object):
    '''write archive as files in directory.'''

    def __init__(self, name, mtime):

            os.utime(destfile, (self.mtime, self.mtime))

    def done(self):
        pass


archivers = {
    'files': fileit,
    'tar': tarit,
    'tbz2': lambda name, mtime: tarit(name, mtime, 'bz2'),
    'tgz': lambda name, mtime: tarit(name, mtime, 'gz'),
    'txz': lambda name, mtime: tarit(name, mtime, 'xz'),
    'uzip': lambda name, mtime: zipit(name, mtime, False),
    'zip': zipit,
}


def archive(
    repo,
    dest,
    node,
    kind,
    decode=True,
    match=None,
    prefix='',
    mtime=None,
    subrepos=False,
):
    '''create archive of repo as it was at node.

    dest can be name of directory, name of archive file, or file
    object to write archive to.

    files = [f for f in ctx.manifest().matches(match)]
    total = len(files)
    if total:
        files.sort()
        scmutil.prefetchfiles(
            repo, [ctx.rev()], scmutil.matchfiles(repo, files)
        )
        progress = repo.ui.makeprogress(
            _('archiving'), unit=_('files'), total=total
        )
        progress.update(0)
        for f in files:
            ff = ctx.flags(f)
            write(f, 'x' in ff and 0o755 or 0o644, 'l' in ff, ctx[f].data)
            progress.increment(item=f)