mercurial-scm/hg: mercurial/bundle2.py comparison

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

author	Augie Fackler <augie@google.com>
date	Sun, 06 Oct 2019 09:45:02 -0400
parents	181ee2118a96
children	687b865b95ad

comparison

equal deleted inserted replaced

-:57875cf423c9
+:2372284d9457
 streamclone,
 tags,
 url,
 util,
 )
-from .utils import (
+from .utils import stringutil
-stringutil,
-)
 urlerr = util.urlerr
 urlreq = util.urlreq
 _pack = struct.pack
 preferedchunksize = 32768
 _parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')
 def outdebug(ui, message):
 """debug regarding output stream (bundling)"""
 if ui.configbool('devel', 'bundle2.debug'):
 ui.debug('bundle2-output: %s\n' % message)
 def indebug(ui, message):
 """debug on input stream (unbundling)"""
 if ui.configbool('devel', 'bundle2.debug'):
 ui.debug('bundle2-input: %s\n' % message)
 def validateparttype(parttype):
 """raise ValueError if a parttype contains invalid character"""
 if _parttypeforbidden.search(parttype):
 raise ValueError(parttype)
 def _makefpartparamsizes(nbparams):
 """return a struct format to read part parameter sizes
 The number parameters is variable so we need to build that format
 dynamically.
 """
-return '>'+('BB'*nbparams)
+return '>' + ('BB' * nbparams)
 parthandlermapping = {}
 def parthandler(parttype, params=()):
 """decorator that register a function as a bundle2 part handler
 eg::
 def myparttypehandler(...):
 '''process a part of type "my part".'''
 ...
 """
 validateparttype(parttype)
 def _decorator(func):
-lparttype = parttype.lower() # enforce lower case matching.
+lparttype = parttype.lower()  # enforce lower case matching.
 assert lparttype not in parthandlermapping
 parthandlermapping[lparttype] = func
 func.params = frozenset(params)
 return func
 return _decorator
 class unbundlerecords(object):
 """keep record of what happens during and unbundle
 New records are added using `records.add('cat', obj)`. Where 'cat' is a
 def __nonzero__(self):
 return bool(self._sequences)
 __bool__ = __nonzero__
 class bundleoperation(object):
 """an object that represents a single bundling process
 Its purpose is to carry unbundle-related objects and states.
 return transaction
 def addhookargs(self, hookargs):
 if self.hookargs is None:
-raise error.ProgrammingError('attempted to add hookargs to '
+raise error.ProgrammingError(
-'operation after transaction started')
+'attempted to add hookargs to '
+'operation after transaction started'
+)
 self.hookargs.update(hookargs)
 class TransactionUnavailable(RuntimeError):
 pass
 def _notransaction():
 """default method to get a transaction while processing a bundle
 Raise an exception to highlight the fact that no transaction was expected
 to be created"""
 raise TransactionUnavailable()
 def applybundle(repo, unbundler, tr, source, url=None, **kwargs):
 # transform me into unbundler.apply() as soon as the freeze is lifted
 if isinstance(unbundler, unbundle20):
 tr.hookargs['bundle2'] = '1'
 # the transactiongetter won't be used, but we might as well set it
 op = bundleoperation(repo, lambda: tr, source=source)
 _processchangegroup(op, unbundler, tr, source, url, **kwargs)
 return op
 class partiterator(object):
 def __init__(self, repo, op, unbundler):
 self.repo = repo
 self.op = op
 self.unbundler = unbundler
 self.count = count
 self.current = p
 yield p
 p.consume()
 self.current = None
 self.iterator = func()
 return self.iterator
 def __exit__(self, type, exc, tb):
 if not self.iterator:
 # Re-raising from a variable loses the original stack. So only use
 # that form if we need to.
 if seekerror:
 raise exc
-self.repo.ui.debug('bundle2-input-bundle: %i parts total\n' %
+self.repo.ui.debug(
-self.count)
+'bundle2-input-bundle: %i parts total\n' % self.count
+)
 def processbundle(repo, unbundler, transactiongetter=None, op=None, source=''):
 """This function process a bundle, apply effect to/from a repo
 It iterates over each part then searches for and uses the proper handling
 processparts(repo, op, unbundler)
 return op
 def processparts(repo, op, unbundler):
 with partiterator(repo, op, unbundler) as parts:
 for part in parts:
 _processpart(op, part)
 def _processchangegroup(op, cg, tr, source, url, **kwargs):
 ret = cg.apply(op.repo, tr, source, url, **kwargs)
-op.records.add('changegroup', {
+op.records.add('changegroup', {'return': ret,})
-'return': ret,
-})
 return ret
 def _gethandler(op, part):
-status = 'unknown' # used by debug output
+status = 'unknown'  # used by debug output
 try:
 handler = parthandlermapping.get(part.type)
 if handler is None:
 status = 'unsupported-type'
 raise error.BundleUnknownFeatureError(parttype=part.type)
 unknownparams = part.mandatorykeys - handler.params
 if unknownparams:
 unknownparams = list(unknownparams)
 unknownparams.sort()
 status = 'unsupported-params (%s)' % ', '.join(unknownparams)
-raise error.BundleUnknownFeatureError(parttype=part.type,
+raise error.BundleUnknownFeatureError(
-params=unknownparams)
+parttype=part.type, params=unknownparams
+)
 status = 'supported'
 except error.BundleUnknownFeatureError as exc:
-if part.mandatory: # mandatory parts
+if part.mandatory:  # mandatory parts
 raise
 indebug(op.ui, 'ignoring unsupported advisory part %s' % exc)
-return # skip to part processing
+return  # skip to part processing
 finally:
 if op.ui.debugflag:
 msg = ['bundle2-input-part: "%s"' % part.type]
 if not part.mandatory:
 msg.append(' (advisory)')
 msg.append(' %s\n' % status)
 op.ui.debug(''.join(msg))
 return handler
 def _processpart(op, part):
 """process a single part from a bundle
 The part is guaranteed to have been fully consumed when the function exits
 (even if an exception is raised)."""
 handler(op, part)
 finally:
 if output is not None:
 output = op.ui.popbuffer()
 if output:
-outpart = op.reply.newpart('output', data=output,
+outpart = op.reply.newpart('output', data=output, mandatory=False)
-mandatory=False)
 outpart.addparam(
-'in-reply-to', pycompat.bytestr(part.id), mandatory=False)
+'in-reply-to', pycompat.bytestr(part.id), mandatory=False
+)
 def decodecaps(blob):
 """decode a bundle2 caps bytes blob into a dictionary
 The blob is a list of capabilities (one per line)
 key = urlreq.unquote(key)
 vals = [urlreq.unquote(v) for v in vals]
 caps[key] = vals
 return caps
 def encodecaps(caps):
 """encode a bundle2 caps dictionary into a bytes blob"""
 chunks = []
 for ca in sorted(caps):
 vals = caps[ca]
 if vals:
 ca = "%s=%s" % (ca, ','.join(vals))
 chunks.append(ca)
 return '\n'.join(chunks)
 bundletypes = {
-"": ("", 'UN'),       # only when using unbundle on ssh and old http servers
+"": ("", 'UN'),  # only when using unbundle on ssh and old http servers
 # since the unification ssh accepts a header but there
 # is no capability signaling it.
-"HG20": (), # special-cased below
+"HG20": (),  # special-cased below
 "HG10UN": ("HG10UN", 'UN'),
 "HG10BZ": ("HG10", 'BZ'),
 "HG10GZ": ("HG10GZ", 'GZ'),
 }
 # hgweb uses this list to communicate its preferred type
 bundlepriority = ['HG10GZ', 'HG10BZ', 'HG10UN']
 class bundle20(object):
 """represent an outgoing bundle2 container
 Use the `addparam` method to add stream level parameter. and `newpart` to
 def addparam(self, name, value=None):
 """add a stream level parameter"""
 if not name:
 raise error.ProgrammingError(b'empty parameter name')
 if name[0:1] not in pycompat.bytestr(string.ascii_letters):
-raise error.ProgrammingError(b'non letter first character: %s'
+raise error.ProgrammingError(
-% name)
+b'non letter first character: %s' % name
+)
 self._params.append((name, value))
 def addpart(self, part):
 """add a new part to the bundle2 container
 Parts contains the actual applicative payload."""
 assert part.id is None
-part.id = len(self._parts) # very cheap counter
+part.id = len(self._parts)  # very cheap counter
 self._parts.append(part)
 def newpart(self, typeid, *args, **kwargs):
 """create a new part and add it to the containers
 param = self._paramchunk()
 outdebug(self.ui, 'bundle parameter: %s' % param)
 yield _pack(_fstreamparamsize, len(param))
 if param:
 yield param
-for chunk in self._compengine.compressstream(self._getcorechunk(),
+for chunk in self._compengine.compressstream(
-self._compopts):
+self._getcorechunk(), self._compopts
+):
 yield chunk
 def _paramchunk(self):
 """return a encoded version of all stream parameters"""
 blocks = []
 for chunk in part.getchunks(ui=self.ui):
 yield chunk
 outdebug(self.ui, 'end of bundle')
 yield _pack(_fpartheadersize, 0)
 def salvageoutput(self):
 """return a list with a copy of all output parts in the bundle
 This is meant to be used during error handling to make sure we preserve
 server output"""
 They directly manipulate the low level stream including bundle2 level
 instruction.
 Do not use it to implement higher-level logic or methods."""
 return changegroup.readexactly(self._fp, size)
 def getunbundler(ui, fp, magicstring=None):
 """return a valid unbundler object for a given magicstring"""
 if magicstring is None:
 magicstring = changegroup.readexactly(fp, 4)
 magic, version = magicstring[0:2], magicstring[2:4]
 if magic != 'HG':
 ui.debug(
 "error: invalid magic: %r (version %r), should be 'HG'\n"
-% (magic, version))
+% (magic, version)
+)
 raise error.Abort(_('not a Mercurial bundle'))
 unbundlerclass = formatmap.get(version)
 if unbundlerclass is None:
 raise error.Abort(_('unknown bundle version %s') % version)
 unbundler = unbundlerclass(ui, fp)
 indebug(ui, 'start processing of %s stream' % magicstring)
 return unbundler
 class unbundle20(unpackermixin):
 """interpret a bundle2 stream
 This class is fed with a binary stream and yields parts through its
 `iterparts` methods."""
 """dictionary of stream level parameters"""
 indebug(self.ui, 'reading bundle2 stream parameters')
 params = {}
 paramssize = self._unpack(_fstreamparamsize)[0]
 if paramssize < 0:
-raise error.BundleValueError('negative bundle param size: %i'
+raise error.BundleValueError(
-% paramssize)
+'negative bundle param size: %i' % paramssize
+)
 if paramssize:
 params = self._readexact(paramssize)
 params = self._processallparams(params)
 return params
 if len(p) < 2:
 p.append(None)
 self._processparam(*p)
 params[p[0]] = p[1]
 return params
 def _processparam(self, name, value):
 """process a parameter, applying its effect if needed
 Parameter starting with a lower case letter are advisory and will be
 """
 yield self._magicstring
 assert 'params' not in vars(self)
 paramssize = self._unpack(_fstreamparamsize)[0]
 if paramssize < 0:
-raise error.BundleValueError('negative bundle param size: %i'
+raise error.BundleValueError(
-% paramssize)
+'negative bundle param size: %i' % paramssize
+)
 if paramssize:
 params = self._readexact(paramssize)
 self._processallparams(params)
 # The payload itself is decompressed below, so drop
 # the compression parameter passed down to compensate.
 continue
 elif size < 0:
 raise error.BundleValueError('negative chunk size: %i')
 yield self._readexact(size)
 def iterparts(self, seekable=False):
 """yield all parts contained in the stream"""
 cls = seekableunbundlepart if seekable else unbundlepart
 # make sure param have been loaded
 self.params
 """reads a part header size and return the bytes blob
 returns None if empty"""
 headersize = self._unpack(_fpartheadersize)[0]
 if headersize < 0:
-raise error.BundleValueError('negative part header size: %i'
+raise error.BundleValueError(
-% headersize)
+'negative part header size: %i' % headersize
+)
 indebug(self.ui, 'part header size: %i' % headersize)
 if headersize:
 return self._readexact(headersize)
 return None
 def compressed(self):
-self.params # load params
+self.params  # load params
 return self._compressed
 def close(self):
 """close underlying file"""
 if util.safehasattr(self._fp, 'close'):
 return self._fp.close()
 formatmap = {'20': unbundle20}
 b2streamparamsmap = {}
 def b2streamparamhandler(name):
 """register a handler for a stream level parameter"""
 def decorator(func):
 assert name not in formatmap
 b2streamparamsmap[name] = func
 return func
 return decorator
 @b2streamparamhandler('compression')
 def processcompression(unbundler, param, value):
 """read compression parameter and install payload decompression"""
 if value not in util.compengines.supportedbundletypes:
-raise error.BundleUnknownFeatureError(params=(param,),
+raise error.BundleUnknownFeatureError(params=(param,), values=(value,))
-values=(value,))
 unbundler._compengine = util.compengines.forbundletype(value)
 if value is not None:
 unbundler._compressed = True
 class bundlepart(object):
 """A bundle2 part contains application level payload
 The part `type` is used to route the part to the application level
 handler.
 should be able to safely ignore the advisory ones.
 Both data and parameters cannot be modified after the generation has begun.
 """
-def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
+def __init__(
-data='', mandatory=True):
+self,
+parttype,
+mandatoryparams=(),
+advisoryparams=(),
+data='',
+mandatory=True,
+):
 validateparttype(parttype)
 self.id = None
 self.type = parttype
 self._data = data
 self._mandatoryparams = list(mandatoryparams)
 self._generated = None
 self.mandatory = mandatory
 def __repr__(self):
 cls = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)
-return ('<%s object at %x; id: %s; type: %s; mandatory: %s>'
+return '<%s object at %x; id: %s; type: %s; mandatory: %s>' % (
-% (cls, id(self), self.id, self.type, self.mandatory))
+cls,
+id(self),
+self.id,
+self.type,
+self.mandatory,
+)
 def copy(self):
 """return a copy of the part
 The new part have the very same content but no partid assigned yet.
 Parts with generated data cannot be copied."""
 assert not util.safehasattr(self.data, 'next')
-return self.__class__(self.type, self._mandatoryparams,
+return self.__class__(
-self._advisoryparams, self._data, self.mandatory)
+self.type,
+self._mandatoryparams,
+self._advisoryparams,
+self._data,
+self.mandatory,
+)
 # methods used to defines the part content
 @property
 def data(self):
 return self._data
 if nbap:
 msg.append(' %i advisory' % nbmp)
 msg.append(')')
 if not self.data:
 msg.append(' empty payload')
-elif (util.safehasattr(self.data, 'next')
+elif util.safehasattr(self.data, 'next') or util.safehasattr(
-or util.safehasattr(self.data, '__next__')):
+self.data, '__next__'
+):
 msg.append(' streamed payload')
 else:
 msg.append(' %i bytes payload' % len(self.data))
 msg.append('\n')
 ui.debug(''.join(msg))
 parttype = self.type.upper()
 else:
 parttype = self.type.lower()
 outdebug(ui, 'part %s: "%s"' % (pycompat.bytestr(self.id), parttype))
 ## parttype
-header = [_pack(_fparttypesize, len(parttype)),
+header = [
-parttype, _pack(_fpartid, self.id),
+_pack(_fparttypesize, len(parttype)),
-]
+parttype,
+_pack(_fpartid, self.id),
+]
 ## parameters
 # count
 manpar = self.mandatoryparams
 advpar = self.advisoryparams
 header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
 header.append(value)
 ## finalize header
 try:
 headerchunk = ''.join(header)
 except TypeError:
-raise TypeError(r'Found a non-bytes trying to '
+raise TypeError(
-r'build bundle part header: %r' % header)
+r'Found a non-bytes trying to '
+r'build bundle part header: %r' % header
+)
 outdebug(ui, 'header chunk size: %i' % len(headerchunk))
 yield _pack(_fpartheadersize, len(headerchunk))
 yield headerchunk
 ## payload
 try:
 ui.debug('bundle2-generatorexit\n')
 raise
 except BaseException as exc:
 bexc = stringutil.forcebytestr(exc)
 # backup exception data for later
-ui.debug('bundle2-input-stream-interrupt: encoding exception %s'
+ui.debug(
-% bexc)
+'bundle2-input-stream-interrupt: encoding exception %s' % bexc
+)
 tb = sys.exc_info()[2]
 msg = 'unexpected error: %s' % bexc
-interpart = bundlepart('error:abort', [('message', msg)],
+interpart = bundlepart(
-mandatory=False)
+'error:abort', [('message', msg)], mandatory=False
+)
 interpart.id = 0
 yield _pack(_fpayloadsize, -1)
 for chunk in interpart.getchunks(ui=ui):
 yield chunk
 outdebug(ui, 'closing payload chunk')
 """yield chunks of a the part payload
 Exists to handle the different methods to provide data to a part."""
 # we only support fixed size data now.
 # This will be improved in the future.
-if (util.safehasattr(self.data, 'next')
+if util.safehasattr(self.data, 'next') or util.safehasattr(
-or util.safehasattr(self.data, '__next__')):
+self.data, '__next__'
+):
 buff = util.chunkbuffer(self.data)
 chunk = buff.read(preferedchunksize)
 while chunk:
 yield chunk
 chunk = buff.read(preferedchunksize)
 yield self.data
 flaginterrupt = -1
 class interrupthandler(unpackermixin):
 """read one part and process it with restricted capability
 This allows to transmit exception raised on the producer size during part
 iteration while the consumer is reading a part.
 """reads a part header size and return the bytes blob
 returns None if empty"""
 headersize = self._unpack(_fpartheadersize)[0]
 if headersize < 0:
-raise error.BundleValueError('negative part header size: %i'
+raise error.BundleValueError(
-% headersize)
+'negative part header size: %i' % headersize
+)
 indebug(self.ui, 'part header size: %i\n' % headersize)
 if headersize:
 return self._readexact(headersize)
 return None
 def __call__(self):
-self.ui.debug('bundle2-input-stream-interrupt:'
+self.ui.debug(
-' opening out of band context\n')
+'bundle2-input-stream-interrupt:' ' opening out of band context\n'
+)
 indebug(self.ui, 'bundle2 stream interruption, looking for a part.')
 headerblock = self._readpartheader()
 if headerblock is None:
 indebug(self.ui, 'no part found during interruption.')
 return
 hardabort = True
 raise
 finally:
 if not hardabort:
 part.consume()
-self.ui.debug('bundle2-input-stream-interrupt:'
+self.ui.debug(
-' closing out of band context\n')
+'bundle2-input-stream-interrupt:' ' closing out of band context\n'
+)
 class interruptoperation(object):
 """A limited operation to be use by part handler during interruption
 It only have access to an ui object.
 def repo(self):
 raise error.ProgrammingError('no repo access from stream interruption')
 def gettransaction(self):
 raise TransactionUnavailable('no repo access from stream interruption')
 def decodepayloadchunks(ui, fh):
 """Reads bundle2 part payload data into chunks.
 Part payload data consists of framed chunks. This function takes
 # changegroup.readexactly() is inlined below for performance.
 while chunksize:
 if chunksize >= 0:
 s = read(chunksize)
 if len(s) < chunksize:
-raise error.Abort(_('stream ended unexpectedly '
+raise error.Abort(
-' (got %d bytes, expected %d)') %
+_(
-(len(s), chunksize))
+'stream ended unexpectedly '
+' (got %d bytes, expected %d)'
+)
+% (len(s), chunksize)
+)
 yield s
 elif chunksize == flaginterrupt:
 # Interrupt "signal" detected. The regular stream is interrupted
 # and a bundle2 part follows. Consume it.
 interrupthandler(ui, fh)()
 else:
 raise error.BundleValueError(
-'negative payload chunk size: %s' % chunksize)
+'negative payload chunk size: %s' % chunksize
+)
 s = read(headersize)
 if len(s) < headersize:
-raise error.Abort(_('stream ended unexpectedly '
+raise error.Abort(
-' (got %d bytes, expected %d)') %
+_('stream ended unexpectedly ' ' (got %d bytes, expected %d)')
-(len(s), chunksize))
+% (len(s), chunksize)
+)
 chunksize = unpack(s)[0]
 # indebug() inlined for performance.
 if dolog:
 debug('bundle2-input: payload chunk size: %i\n' % chunksize)
 class unbundlepart(unpackermixin):
 """a bundle part read from a bundle"""
 def __init__(self, ui, header, fp):
 super(unbundlepart, self).__init__(fp)
-self._seekable = (util.safehasattr(fp, 'seek') and
+self._seekable = util.safehasattr(fp, 'seek') and util.safehasattr(
-util.safehasattr(fp, 'tell'))
+fp, 'tell'
+)
 self.ui = ui
 # unbundle state attr
 self._headerdata = header
 self._headeroffset = 0
 self._initialized = False
 self._pos = 0
 def _fromheader(self, size):
 """return the next <size> byte from the header"""
 offset = self._headeroffset
-data = self._headerdata[offset:(offset + size)]
+data = self._headerdata[offset : (offset + size)]
 self._headeroffset = offset + size
 return data
 def _unpackheader(self, format):
 """read given format from header
 def _initparams(self, mandatoryparams, advisoryparams):
 """internal function to setup all logic related parameters"""
 # make it read only to prevent people touching it by mistake.
 self.mandatoryparams = tuple(mandatoryparams)
-self.advisoryparams  = tuple(advisoryparams)
+self.advisoryparams = tuple(advisoryparams)
 # user friendly UI
 self.params = util.sortdict(self.mandatoryparams)
 self.params.update(self.advisoryparams)
 self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
 self.type = self._fromheader(typesize)
 indebug(self.ui, 'part type: "%s"' % self.type)
 self.id = self._unpackheader(_fpartid)[0]
 indebug(self.ui, 'part id: "%s"' % pycompat.bytestr(self.id))
 # extract mandatory bit from type
-self.mandatory = (self.type != self.type.lower())
+self.mandatory = self.type != self.type.lower()
 self.type = self.type.lower()
 ## reading parameters
 # param count
 mancount, advcount = self._unpackheader(_fpartparamcount)
 indebug(self.ui, 'part parameters: %i' % (mancount + advcount))
 else:
 data = self._payloadstream.read(size)
 self._pos += len(data)
 if size is None or len(data) < size:
 if not self.consumed and self._pos:
-self.ui.debug('bundle2-input-part: total payload size %i\n'
+self.ui.debug(
-% self._pos)
+'bundle2-input-part: total payload size %i\n' % self._pos
+)
 self.consumed = True
 return data
 class seekableunbundlepart(unbundlepart):
 """A bundle2 part in a bundle that is seekable.
 Regular ``unbundlepart`` instances can only be read once. This class
 class maintain a mapping between offsets in the underlying stream and
 the decoded payload. This mapping will consume memory in proportion
 to the number of chunks within the payload (which almost certainly
 increases in proportion with the size of the part).
 """
 def __init__(self, ui, header, fp):
 # (payload, file) offsets for chunk starts.
 self._chunkindex = []
 super(seekableunbundlepart, self).__init__(ui, header, fp)
 if len(self._chunkindex) == 0:
 assert chunknum == 0, 'Must start with chunk 0'
 self._chunkindex.append((0, self._tellfp()))
 else:
 assert chunknum < len(self._chunkindex), (
-'Unknown chunk %d' % chunknum)
+'Unknown chunk %d' % chunknum
+)
 self._seekfp(self._chunkindex[chunknum][1])
 pos = self._chunkindex[chunknum][0]
 for chunk in decodepayloadchunks(self.ui, self._fp):
 self._seekable = False
 else:
 raise
 return None
 # These are only the static capabilities.
 # Check the 'getrepocaps' function for the rest.
-capabilities = {'HG20': (),
+capabilities = {
-'bookmarks': (),
+'HG20': (),
-'error': ('abort', 'unsupportedcontent', 'pushraced',
+'bookmarks': (),
-'pushkey'),
+'error': ('abort', 'unsupportedcontent', 'pushraced', 'pushkey'),
 'listkeys': (),
 'pushkey': (),
 'digests': tuple(sorted(util.DIGESTS.keys())),
 'remote-changegroup': ('http', 'https'),
 'hgtagsfnodes': (),
 'rev-branch-cache': (),
 'phases': ('heads',),
 'stream': ('v2',),
 }
 def getrepocaps(repo, allowpushback=False, role=None):
 """return the bundle2 capabilities for a given repo
 Exists to allow extensions (like evolution) to mutate the capabilities.
 """
 if role not in ('client', 'server'):
 raise error.ProgrammingError('role argument must be client or server')
 caps = capabilities.copy()
-caps['changegroup'] = tuple(sorted(
+caps['changegroup'] = tuple(
-changegroup.supportedincomingversions(repo)))
+sorted(changegroup.supportedincomingversions(repo))
+)
 if obsolete.isenabled(repo, obsolete.exchangeopt):
 supportedformat = tuple('V%i' % v for v in obsolete.formats)
 caps['obsmarkers'] = supportedformat
 if allowpushback:
 caps['pushback'] = ()
 if 'phases' in repo.ui.configlist('devel', 'legacy.exchange'):
 caps.pop('phases')
 # Don't advertise stream clone support in server mode if not configured.
 if role == 'server':
-streamsupported = repo.ui.configbool('server', 'uncompressed',
+streamsupported = repo.ui.configbool(
-untrusted=True)
+'server', 'uncompressed', untrusted=True
+)
 featuresupported = repo.ui.configbool('server', 'bundle2.stream')
 if not streamsupported or not featuresupported:
 caps.pop('stream')
 # Else always advertise support on client, because payload support
 # should always be advertised.
 return caps
 def bundle2caps(remote):
 """return the bundle capabilities of a peer as dict"""
 raw = remote.capable('bundle2')
 if not raw and raw != '':
 return {}
 capsblob = urlreq.unquote(remote.capable('bundle2'))
 return decodecaps(capsblob)
 def obsmarkersversion(caps):
 """extract the list of supported obsmarkers versions from a bundle2caps dict
 """
 obscaps = caps.get('obsmarkers', ())
 return [int(c[1:]) for c in obscaps if c.startswith('V')]
-def writenewbundle(ui, repo, source, filename, bundletype, outgoing, opts,
-vfs=None, compression=None, compopts=None):
+def writenewbundle(
+ui,
+repo,
+source,
+filename,
+bundletype,
+outgoing,
+opts,
+vfs=None,
+compression=None,
+compopts=None,
+):
 if bundletype.startswith('HG10'):
 cg = changegroup.makechangegroup(repo, outgoing, '01', source)
-return writebundle(ui, cg, filename, bundletype, vfs=vfs,
+return writebundle(
-compression=compression, compopts=compopts)
+ui,
+cg,
+filename,
+bundletype,
+vfs=vfs,
+compression=compression,
+compopts=compopts,
+)
 elif not bundletype.startswith('HG20'):
 raise error.ProgrammingError('unknown bundle type: %s' % bundletype)
 caps = {}
 if 'obsolescence' in opts:
 bundle.setcompression(compression, compopts)
 _addpartsfromopts(ui, repo, bundle, source, outgoing, opts)
 chunkiter = bundle.getchunks()
 return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)
 def _addpartsfromopts(ui, repo, bundler, source, outgoing, opts):
 # We should eventually reconcile this logic with the one behind
 # 'exchange.getbundle2partsgenerator'.
 #
 cgversion = changegroup.safeversion(repo)
 cg = changegroup.makechangegroup(repo, outgoing, cgversion, source)
 part = bundler.newpart('changegroup', data=cg.getchunks())
 part.addparam('version', cg.version)
 if 'clcount' in cg.extras:
-part.addparam('nbchanges', '%d' % cg.extras['clcount'],
+part.addparam(
-mandatory=False)
+'nbchanges', '%d' % cg.extras['clcount'], mandatory=False
-if opts.get('phases') and repo.revs('%ln and secret()',
+)
-outgoing.missingheads):
+if opts.get('phases') and repo.revs(
+'%ln and secret()', outgoing.missingheads
+):
 part.addparam('targetphase', '%d' % phases.secret, mandatory=False)
 if opts.get('streamv2', False):
 addpartbundlestream2(bundler, repo, stream=True)
 if opts.get('phases', False):
 headsbyphase = phases.subsetphaseheads(repo, outgoing.missing)
 phasedata = phases.binaryencode(headsbyphase)
 bundler.newpart('phase-heads', data=phasedata)
 def addparttagsfnodescache(repo, bundler, outgoing):
 # we include the tags fnode cache for the bundle changeset
 # (as an optional parts)
 cache = tags.hgtagsfnodescache(repo.unfiltered())
 chunks.extend([node, fnode])
 if chunks:
 bundler.newpart('hgtagsfnodes', data=''.join(chunks))
 def addpartrevbranchcache(repo, bundler, outgoing):
 # we include the rev branch cache for the bundle changeset
 # (as an optional parts)
 cache = repo.revbranchcache()
 cl = repo.unfiltered().changelog
 for n in sorted(nodes):
 yield n
 for n in sorted(closed):
 yield n
-bundler.newpart('cache:rev-branch-cache', data=generate(),
+bundler.newpart('cache:rev-branch-cache', data=generate(), mandatory=False)
-mandatory=False)
 def _formatrequirementsspec(requirements):
 requirements = [req for req in requirements if req != "shared"]
 return urlreq.quote(','.join(sorted(requirements)))
 def _formatrequirementsparams(requirements):
 requirements = _formatrequirementsspec(requirements)
 params = "%s%s" % (urlreq.quote("requirements="), requirements)
 return params
 def addpartbundlestream2(bundler, repo, **kwargs):
 if not kwargs.get(r'stream', False):
 return
 if not streamclone.allowservergeneration(repo):
-raise error.Abort(_('stream data requested but server does not allow '
+raise error.Abort(
-'this feature'),
+_(
-hint=_('well-behaved clients should not be '
+'stream data requested but server does not allow '
-'requesting stream data from servers not '
+'this feature'
-'advertising it; the client may be buggy'))
+),
+hint=_(
+'well-behaved clients should not be '
+'requesting stream data from servers not '
+'advertising it; the client may be buggy'
+),
+)
 # Stream clones don't compress well. And compression undermines a
 # goal of stream clones, which is to be fast. Communicate the desire
 # to avoid compression to consumers of the bundle.
 bundler.prefercompressed = False
 # get the includes and excludes
 includepats = kwargs.get(r'includepats')
 excludepats = kwargs.get(r'excludepats')
-narrowstream = repo.ui.configbool('experimental',
+narrowstream = repo.ui.configbool(
-'server.stream-narrow-clones')
+'experimental', 'server.stream-narrow-clones'
+)
 if (includepats or excludepats) and not narrowstream:
 raise error.Abort(_('server does not support narrow stream clones'))
 includeobsmarkers = False
 if repo.obsstore:
 remoteversions = obsmarkersversion(bundler.capabilities)
 if not remoteversions:
-raise error.Abort(_('server has obsolescence markers, but client '
+raise error.Abort(
-'cannot receive them via stream clone'))
+_(
+'server has obsolescence markers, but client '
+'cannot receive them via stream clone'
+)
+)
 elif repo.obsstore._version in remoteversions:
 includeobsmarkers = True
-filecount, bytecount, it = streamclone.generatev2(repo, includepats,
+filecount, bytecount, it = streamclone.generatev2(
-excludepats,
+repo, includepats, excludepats, includeobsmarkers
-includeobsmarkers)
+)
 requirements = _formatrequirementsspec(repo.requirements)
 part = bundler.newpart('stream2', data=it)
 part.addparam('bytecount', '%d' % bytecount, mandatory=True)
 part.addparam('filecount', '%d' % filecount, mandatory=True)
 part.addparam('requirements', requirements, mandatory=True)
 def buildobsmarkerspart(bundler, markers):
 """add an obsmarker part to the bundler with <markers>
 No part is created if markers is empty.
 if version is None:
 raise ValueError('bundler does not support common obsmarker format')
 stream = obsolete.encodemarkers(markers, True, version=version)
 return bundler.newpart('obsmarkers', data=stream)
-def writebundle(ui, cg, filename, bundletype, vfs=None, compression=None,
-compopts=None):
+def writebundle(
+ui, cg, filename, bundletype, vfs=None, compression=None, compopts=None
+):
 """Write a bundle file and return its filename.
 Existing files will not be overwritten.
 If no filename is specified, a temporary file is created.
 bz2 compression can be turned off.
 bundle = bundle20(ui)
 bundle.setcompression(compression, compopts)
 part = bundle.newpart('changegroup', data=cg.getchunks())
 part.addparam('version', cg.version)
 if 'clcount' in cg.extras:
-part.addparam('nbchanges', '%d' % cg.extras['clcount'],
+part.addparam(
-mandatory=False)
+'nbchanges', '%d' % cg.extras['clcount'], mandatory=False
+)
 chunkiter = bundle.getchunks()
 else:
 # compression argument is only for the bundle2 case
 assert compression is None
 if cg.version != '01':
-raise error.Abort(_('old bundle types only supports v1 '
+raise error.Abort(
-'changegroups'))
+_('old bundle types only supports v1 ' 'changegroups')
+)
 header, comp = bundletypes[bundletype]
 if comp not in util.compengines.supportedbundletypes:
-raise error.Abort(_('unknown stream compression type: %s')
+raise error.Abort(_('unknown stream compression type: %s') % comp)
-% comp)
 compengine = util.compengines.forbundletype(comp)
 def chunkiter():
 yield header
 for chunk in compengine.compressstream(cg.getchunks(), compopts):
 yield chunk
 chunkiter = chunkiter()
 # parse the changegroup data, otherwise we will block
 # in case of sshrepo because we don't know the end of the stream
 return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)
 def combinechangegroupresults(op):
 """logic to combine 0 or more addchangegroup results into one"""
-results = [r.get('return', 0)
+results = [r.get('return', 0) for r in op.records['changegroup']]
-for r in op.records['changegroup']]
 changedheads = 0
 result = 1
 for ret in results:
 # If any changegroup result is 0, return 0
 if ret == 0:
 result = 1 + changedheads
 elif changedheads < 0:
 result = -1 + changedheads
 return result
-@parthandler('changegroup', ('version', 'nbchanges', 'treemanifest',
-'targetphase'))
+@parthandler(
+'changegroup', ('version', 'nbchanges', 'treemanifest', 'targetphase')
+)
 def handlechangegroup(op, inpart):
 """apply a changegroup part on the repo
 This is a very early implementation that will massive rework before being
 inflicted to any end-user.
 # the source and url passed here are overwritten by the one contained in
 # the transaction.hookargs argument. So 'bundle2' is a placeholder
 nbchangesets = None
 if 'nbchanges' in inpart.params:
 nbchangesets = int(inpart.params.get('nbchanges'))
-if ('treemanifest' in inpart.params and
+if (
-'treemanifest' not in op.repo.requirements):
+'treemanifest' in inpart.params
+and 'treemanifest' not in op.repo.requirements
+):
 if len(op.repo.changelog) != 0:
-raise error.Abort(_(
+raise error.Abort(
-"bundle contains tree manifests, but local repo is "
+_(
-"non-empty and does not use tree manifests"))
+"bundle contains tree manifests, but local repo is "
+"non-empty and does not use tree manifests"
+)
+)
 op.repo.requirements.add('treemanifest')
 op.repo.svfs.options = localrepo.resolvestorevfsoptions(
-op.repo.ui, op.repo.requirements, op.repo.features)
+op.repo.ui, op.repo.requirements, op.repo.features
+)
 op.repo._writerequirements()
 extrakwargs = {}
 targetphase = inpart.params.get('targetphase')
 if targetphase is not None:
 extrakwargs[r'targetphase'] = int(targetphase)
-ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2',
+ret = _processchangegroup(
-expectedtotal=nbchangesets, **extrakwargs)
+op,
+cg,
+tr,
+'bundle2',
+'bundle2',
+expectedtotal=nbchangesets,
+**extrakwargs
+)
 if op.reply is not None:
 # This is definitely not the final form of this
 # return. But one need to start somewhere.
 part = op.reply.newpart('reply:changegroup', mandatory=False)
 part.addparam(
-'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
+'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False
+)
 part.addparam('return', '%i' % ret, mandatory=False)
 assert not inpart.read()
-_remotechangegroupparams = tuple(['url', 'size', 'digests'] +
-['digest:%s' % k for k in util.DIGESTS.keys()])
+_remotechangegroupparams = tuple(
+['url', 'size', 'digests'] + ['digest:%s' % k for k in util.DIGESTS.keys()]
+)
 @parthandler('remote-changegroup', _remotechangegroupparams)
 def handleremotechangegroup(op, inpart):
 """apply a bundle10 on the repo, given an url and validation information
 All the information about the remote bundle to import are given as
 raw_url = inpart.params['url']
 except KeyError:
 raise error.Abort(_('remote-changegroup: missing "%s" param') % 'url')
 parsed_url = util.url(raw_url)
 if parsed_url.scheme not in capabilities['remote-changegroup']:
-raise error.Abort(_('remote-changegroup does not support %s urls') %
+raise error.Abort(
-parsed_url.scheme)
+_('remote-changegroup does not support %s urls') % parsed_url.scheme
+)
 try:
 size = int(inpart.params['size'])
 except ValueError:
-raise error.Abort(_('remote-changegroup: invalid value for param "%s"')
+raise error.Abort(
-% 'size')
+_('remote-changegroup: invalid value for param "%s"') % 'size'
+)
 except KeyError:
 raise error.Abort(_('remote-changegroup: missing "%s" param') % 'size')
 digests = {}
 for typ in inpart.params.get('digests', '').split():
 param = 'digest:%s' % typ
 try:
 value = inpart.params[param]
 except KeyError:
-raise error.Abort(_('remote-changegroup: missing "%s" param') %
+raise error.Abort(
-param)
+_('remote-changegroup: missing "%s" param') % param
+)
 digests[typ] = value
 real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)
 tr = op.gettransaction()
 from . import exchange
 cg = exchange.readbundle(op.repo.ui, real_part, raw_url)
 if not isinstance(cg, changegroup.cg1unpacker):
-raise error.Abort(_('%s: not a bundle version 1.0') %
+raise error.Abort(
-util.hidepassword(raw_url))
+_('%s: not a bundle version 1.0') % util.hidepassword(raw_url)
+)
 ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2')
 if op.reply is not None:
 # This is definitely not the final form of this
 # return. But one need to start somewhere.
 part = op.reply.newpart('reply:changegroup')
 part.addparam(
-'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
+'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False
+)
 part.addparam('return', '%i' % ret, mandatory=False)
 try:
 real_part.validate()
 except error.Abort as e:
-raise error.Abort(_('bundle at %s is corrupted:\n%s') %
+raise error.Abort(
-(util.hidepassword(raw_url), bytes(e)))
+_('bundle at %s is corrupted:\n%s')
+% (util.hidepassword(raw_url), bytes(e))
+)
 assert not inpart.read()
 @parthandler('reply:changegroup', ('return', 'in-reply-to'))
 def handlereplychangegroup(op, inpart):
 ret = int(inpart.params['return'])
 replyto = int(inpart.params['in-reply-to'])
 op.records.add('changegroup', {'return': ret}, replyto)
 @parthandler('check:bookmarks')
 def handlecheckbookmarks(op, inpart):
 """check location of bookmarks
 This part is to be used to detect push race regarding bookmark, it
 contains binary encoded (bookmark, node) tuple. If the local state does
 not marks the one in the part, a PushRaced exception is raised
 """
 bookdata = bookmarks.binarydecode(inpart)
-msgstandard = ('remote repository changed while pushing - please try again '
+msgstandard = (
-'(bookmark "%s" move from %s to %s)')
+'remote repository changed while pushing - please try again '
-msgmissing = ('remote repository changed while pushing - please try again '
+'(bookmark "%s" move from %s to %s)'
-'(bookmark "%s" is missing, expected %s)')
+)
-msgexist = ('remote repository changed while pushing - please try again '
+msgmissing = (
-'(bookmark "%s" set on %s, expected missing)')
+'remote repository changed while pushing - please try again '
+'(bookmark "%s" is missing, expected %s)'
+)
+msgexist = (
+'remote repository changed while pushing - please try again '
+'(bookmark "%s" set on %s, expected missing)'
+)
 for book, node in bookdata:
 currentnode = op.repo._bookmarks.get(book)
 if currentnode != node:
 if node is None:
 finalmsg = msgexist % (book, nodemod.short(currentnode))
 elif currentnode is None:
 finalmsg = msgmissing % (book, nodemod.short(node))
 else:
-finalmsg = msgstandard % (book, nodemod.short(node),
+finalmsg = msgstandard % (
-nodemod.short(currentnode))
+book,
+nodemod.short(node),
+nodemod.short(currentnode),
+)
 raise error.PushRaced(finalmsg)
 @parthandler('check:heads')
 def handlecheckheads(op, inpart):
 """check that head of the repo did not change
 assert not h
 # Trigger a transaction so that we are guaranteed to have the lock now.
 if op.ui.configbool('experimental', 'bundle2lazylocking'):
 op.gettransaction()
 if sorted(heads) != sorted(op.repo.heads()):
-raise error.PushRaced('remote repository changed while pushing - '
+raise error.PushRaced(
-'please try again')
+'remote repository changed while pushing - ' 'please try again'
+)
 @parthandler('check:updated-heads')
 def handlecheckupdatedheads(op, inpart):
 """check for race on the heads touched by a push
 for ls in op.repo.branchmap().iterheads():
 currentheads.update(ls)
 for h in heads:
 if h not in currentheads:
-raise error.PushRaced('remote repository changed while pushing - '
+raise error.PushRaced(
-'please try again')
+'remote repository changed while pushing - ' 'please try again'
+)
 @parthandler('check:phases')
 def handlecheckphases(op, inpart):
 """check that phase boundaries of the repository did not change
 """
 phasetonodes = phases.binarydecode(inpart)
 unfi = op.repo.unfiltered()
 cl = unfi.changelog
 phasecache = unfi._phasecache
-msg = ('remote repository changed while pushing - please try again '
+msg = (
-'(%s is %s expected %s)')
+'remote repository changed while pushing - please try again '
+'(%s is %s expected %s)'
+)
 for expectedphase, nodes in enumerate(phasetonodes):
 for n in nodes:
 actualphase = phasecache.phase(unfi, cl.rev(n))
 if actualphase != expectedphase:
-finalmsg = msg % (nodemod.short(n),
+finalmsg = msg % (
-phases.phasenames[actualphase],
+nodemod.short(n),
-phases.phasenames[expectedphase])
+phases.phasenames[actualphase],
+phases.phasenames[expectedphase],
+)
 raise error.PushRaced(finalmsg)
 @parthandler('output')
 def handleoutput(op, inpart):
 """forward output captured on the server to the client"""
 for line in inpart.read().splitlines():
 op.ui.status(_('remote: %s\n') % line)
 @parthandler('replycaps')
 def handlereplycaps(op, inpart):
 """Notify that a reply bundle should be created
 The payload contains the capabilities information for the reply"""
 caps = decodecaps(inpart.read())
 if op.reply is None:
 op.reply = bundle20(op.ui, caps)
 class AbortFromPart(error.Abort):
 """Sub-class of Abort that denotes an error from a bundle2 part."""
 @parthandler('error:abort', ('message', 'hint'))
 def handleerrorabort(op, inpart):
 """Used to transmit abort error over the wire"""
-raise AbortFromPart(inpart.params['message'],
+raise AbortFromPart(
-hint=inpart.params.get('hint'))
+inpart.params['message'], hint=inpart.params.get('hint')
+)
-@parthandler('error:pushkey', ('namespace', 'key', 'new', 'old', 'ret',
-'in-reply-to'))
+@parthandler(
+'error:pushkey', ('namespace', 'key', 'new', 'old', 'ret', 'in-reply-to')
+)
 def handleerrorpushkey(op, inpart):
 """Used to transmit failure of a mandatory pushkey over the wire"""
 kwargs = {}
 for name in ('namespace', 'key', 'new', 'old', 'ret'):
 value = inpart.params.get(name)
 if value is not None:
 kwargs[name] = value
-raise error.PushkeyFailed(inpart.params['in-reply-to'],
+raise error.PushkeyFailed(
-**pycompat.strkwargs(kwargs))
+inpart.params['in-reply-to'], **pycompat.strkwargs(kwargs)
+)
 @parthandler('error:unsupportedcontent', ('parttype', 'params'))
 def handleerrorunsupportedcontent(op, inpart):
 """Used to transmit unknown content error over the wire"""
 kwargs = {}
 if params is not None:
 kwargs['params'] = params.split('\0')
 raise error.BundleUnknownFeatureError(**pycompat.strkwargs(kwargs))
 @parthandler('error:pushraced', ('message',))
 def handleerrorpushraced(op, inpart):
 """Used to transmit push race error over the wire"""
 raise error.ResponseError(_('push failed:'), inpart.params['message'])
 @parthandler('listkeys', ('namespace',))
 def handlelistkeys(op, inpart):
 """retrieve pushkey namespace content stored in a bundle2"""
 namespace = inpart.params['namespace']
 r = pushkey.decodekeys(inpart.read())
 op.records.add('listkeys', (namespace, r))
 @parthandler('pushkey', ('namespace', 'key', 'old', 'new'))
 def handlepushkey(op, inpart):
 """process a pushkey request"""
 dec = pushkey.decode
 # Grab the transaction to ensure that we have the lock before performing the
 # pushkey.
 if op.ui.configbool('experimental', 'bundle2lazylocking'):
 op.gettransaction()
 ret = op.repo.pushkey(namespace, key, old, new)
-record = {'namespace': namespace,
+record = {'namespace': namespace, 'key': key, 'old': old, 'new': new}
-'key': key,
-'old': old,
-'new': new}
 op.records.add('pushkey', record)
 if op.reply is not None:
 rpart = op.reply.newpart('reply:pushkey')
 rpart.addparam(
-'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
+'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False
+)
 rpart.addparam('return', '%i' % ret, mandatory=False)
 if inpart.mandatory and not ret:
 kwargs = {}
 for key in ('namespace', 'key', 'new', 'old', 'ret'):
 if key in inpart.params:
 kwargs[key] = inpart.params[key]
-raise error.PushkeyFailed(partid='%d' % inpart.id,
+raise error.PushkeyFailed(
-**pycompat.strkwargs(kwargs))
+partid='%d' % inpart.id, **pycompat.strkwargs(kwargs)
+)
 @parthandler('bookmarks')
 def handlebookmark(op, inpart):
 """transmit bookmark information
 hookargs['old'] = nodemod.hex(bookstore.get(book, ''))
 hookargs['new'] = nodemod.hex(node if node is not None else '')
 allhooks.append(hookargs)
 for hookargs in allhooks:
-op.repo.hook('prepushkey', throw=True,
+op.repo.hook(
-**pycompat.strkwargs(hookargs))
+'prepushkey', throw=True, **pycompat.strkwargs(hookargs)
+)
 bookstore.applychanges(op.repo, op.gettransaction(), changes)
 if pushkeycompat:
 def runhook():
 for hookargs in allhooks:
 op.repo.hook('pushkey', **pycompat.strkwargs(hookargs))
 op.repo._afterlock(runhook)
 elif bookmarksmode == 'records':
 for book, node in changes:
 record = {'bookmark': book, 'node': node}
 op.records.add('bookmarks', record)
 else:
 raise error.ProgrammingError('unkown bookmark mode: %s' % bookmarksmode)
 @parthandler('phase-heads')
 def handlephases(op, inpart):
 """apply phases from bundle part to repo"""
 headsbyphase = phases.binarydecode(inpart)
 phases.updatephases(op.repo.unfiltered(), op.gettransaction, headsbyphase)
 @parthandler('reply:pushkey', ('return', 'in-reply-to'))
 def handlepushkeyreply(op, inpart):
 """retrieve the result of a pushkey request"""
 ret = int(inpart.params['return'])
 partid = int(inpart.params['in-reply-to'])
 op.records.add('pushkey', {'return': ret}, partid)
 @parthandler('obsmarkers')
 def handleobsmarker(op, inpart):
 """add a stream of obsmarkers to the repo"""
 tr = op.gettransaction()
 markerdata = inpart.read()
 if op.ui.config('experimental', 'obsmarkers-exchange-debug'):
-op.ui.write(('obsmarker-exchange: %i bytes received\n')
+op.ui.write('obsmarker-exchange: %i bytes received\n' % len(markerdata))
-% len(markerdata))
 # The mergemarkers call will crash if marker creation is not enabled.
 # we want to avoid this if the part is advisory.
 if not inpart.mandatory and op.repo.obsstore.readonly:
 op.repo.ui.debug('ignoring obsolescence markers, feature not enabled\n')
 return
 op.repo.invalidatevolatilesets()
 op.records.add('obsmarkers', {'new': new})
 if op.reply is not None:
 rpart = op.reply.newpart('reply:obsmarkers')
 rpart.addparam(
-'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
+'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False
+)
 rpart.addparam('new', '%i' % new, mandatory=False)
 @parthandler('reply:obsmarkers', ('new', 'in-reply-to'))
 def handleobsmarkerreply(op, inpart):
 """retrieve the result of a pushkey request"""
 ret = int(inpart.params['new'])
 partid = int(inpart.params['in-reply-to'])
 op.records.add('obsmarkers', {'new': ret}, partid)
 @parthandler('hgtagsfnodes')
 def handlehgtagsfnodes(op, inpart):
 """Applies .hgtags fnodes cache entries to the local repo.
 count += 1
 cache.write()
 op.ui.debug('applied %i hgtags fnodes cache entries\n' % count)
 rbcstruct = struct.Struct('>III')
 @parthandler('cache:rev-branch-cache')
 def handlerbc(op, inpart):
 """receive a rev-branch-cache payload and update the local cache
 rev = cl.rev(node)
 cache.setdata(branch, rev, node, True)
 rawheader = inpart.read(rbcstruct.size)
 cache.write()
 @parthandler('pushvars')
 def bundle2getvars(op, part):
 '''unbundle a bundle2 containing shellvars on the server'''
 # An option to disable unbundling on server-side for security reasons
 if op.ui.configbool('push', 'pushvars.server'):
 # they came from the --pushvar flag.
 key = "USERVAR_" + key
 hookargs[key] = value
 op.addhookargs(hookargs)
 @parthandler('stream2', ('requirements', 'filecount', 'bytecount'))
 def handlestreamv2bundle(op, part):
 requirements = urlreq.unquote(part.params['requirements']).split(',')
 filecount = int(part.params['filecount'])
 if len(repo):
 msg = _('cannot apply stream clone to non empty repository')
 raise error.Abort(msg)
 repo.ui.debug('applying stream bundle\n')
-streamclone.applybundlev2(repo, part, filecount, bytecount,
+streamclone.applybundlev2(repo, part, filecount, bytecount, requirements)
-requirements)
-def widen_bundle(bundler, repo, oldmatcher, newmatcher, common,
+def widen_bundle(
-known, cgversion, ellipses):
+bundler, repo, oldmatcher, newmatcher, common, known, cgversion, ellipses
+):
 """generates bundle2 for widening a narrow clone
 bundler is the bundle to which data should be added
 repo is the localrepository instance
 oldmatcher matches what the client already has
 for r in repo.revs("::%ln", common):
 commonnodes.add(cl.node(r))
 if commonnodes:
 # XXX: we should only send the filelogs (and treemanifest). user
 # already has the changelog and manifest
-packer = changegroup.getbundler(cgversion, repo,
+packer = changegroup.getbundler(
-oldmatcher=oldmatcher,
+cgversion,
-matcher=newmatcher,
+repo,
-fullnodes=commonnodes)
+oldmatcher=oldmatcher,
-cgdata = packer.generate({nodemod.nullid}, list(commonnodes),
+matcher=newmatcher,
-False, 'narrow_widen', changelog=False)
+fullnodes=commonnodes,
+)
+cgdata = packer.generate(
+{nodemod.nullid},
+list(commonnodes),
+False,
+'narrow_widen',
+changelog=False,
+)
 part = bundler.newpart('changegroup', data=cgdata)
 part.addparam('version', cgversion)
 if 'treemanifest' in repo.requirements:
 part.addparam('treemanifest', '1')

Mercurial > public > mercurial-scm > hg

comparison mercurial/bundle2.py @ 43076:2372284d9457