( )⚙ D7515 pytype: [WIP] suppress various warnings inline to get a clean run

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D7515

pytype: [WIP] suppress various warnings inline to get a clean run
Changes PlannedPublic

Authored by mharbison72 on Nov 23 2019, 7:47 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Summary

This depends on D7295 and D7384, which in turn was applied to 4222b9d5d4fb. I
made no attempt to keep it under 80 columns or blacken it. But it seems useful
to share now, because it is so time consuming to get this far, and presumably
much of this will entail actual fixes. These things were flagged by pytype
2019.12.06 under WSL.

The following bundlerepo warnings were fixed with assertions, but those
assertions fire when running the testsuite for some reason:

line 256, in __init__: No attribute 'ui' on bundlerepository [attribute-error]
line 260, in __init__: No attribute 'ui' on bundlerepository [attribute-error]
line 292, in __init__: No attribute 'ui' on bundlerepository [attribute-error]
line 292, in __init__: No attribute 'vfs' on bundlerepository [attribute-error]
line 330, in _writetempbundle: No attribute 'vfs' on bundlerepository [attribute-error]
  Called from (traceback):
    line 290, in __init__
line 341, in _writetempbundle: No attribute 'vfs' on bundlerepository [attribute-error]
  Called from (traceback):
    line 290, in __init__
line 378, in _consumemanifest: No attribute '_cgunpacker' on bundlerepository [attribute-error]
line 379, in _consumemanifest: No attribute '_cgunpacker' on bundlerepository [attribute-error]
line 380, in _consumemanifest: No attribute '_cgunpacker' on bundlerepository [attribute-error]
line 382, in _consumemanifest: No attribute '_cgunpacker' on bundlerepository [attribute-error]
line 406, in file: No attribute '_cgfilespos' on bundlerepository [attribute-error]
line 412, in file: No attribute 'unfiltered' on bundlerepository [attribute-error]
line 413, in file: No attribute 'svfs' on bundlerepository [attribute-error]
line 413, in file: No attribute 'file' on super [attribute-error]
line 420, in close: No attribute '_bundlefile' on bundlerepository [attribute-error]
line 422, in close: No attribute 'vfs' on bundlerepository [attribute-error]
line 441, in setparents: No attribute 'setparents' on super [attribute-error]
line 441, in setparents: No attribute 'ui' on bundlerepository [attribute-error]
line 443, in setparents: No attribute 'ui' on bundlerepository [attribute-error]

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

mharbison72 created this revision.Nov 23 2019, 7:47 PM
mharbison72 planned changes to this revision.Nov 23 2019, 7:54 PM

Adjusting this to stay out of the review queue. While most of the files only had a handful of errors (and a few could be removed from the blacklist), the few added had a bunch of errors. I didn't see any obvious fixes, and ran out of time Friday to individually suppress them.

I'm assuming we will fix the problems, and then apply the suppressions in narrow categories rather than a monolithic commit like this.

mharbison72 edited the summary of this revision. (Show Details)Dec 17 2019, 12:43 PM
mharbison72 updated this revision to Diff 18818.
mharbison72 planned changes to this revision.Dec 17 2019, 12:52 PM

Still not a queue-able thing. But this adds in the errors that are being suppressed so others can try to fix some of this without the ~10-15minute wait for failures. It looks like there is a recurring issue with converting exceptions to bytes for printing. Also, a few things seem to have gotten fixed in logcmdutil and patch since last time, though I'm slightly concerned that it missed issues that it previously flagged in hgweb/server.py. The only difference I'm aware of is running a new version of pytype, and running under WSL instead of on a Mac.

Revision Contents
Changeset List

Diff 18368

mercurial/bundlerepo.py

Instances constructed directly are not usable as repository objects. Instances constructed directly are not usable as repository objects.
Use instance() or makebundlerepository() to create instances. Use instance() or makebundlerepository() to create instances.
""" """
def __init__(self, bundlepath, url, tempparent): def __init__(self, bundlepath, url, tempparent):
self._tempparent = tempparent self._tempparent = tempparent
self._url = url self._url = url
self.ui.setconfig(b'phases', b'publish', False, b'bundlerepo') # These types are OK because the class is a dynamically generated type.
# It would be better to somehow mark that `self` is interfaces.repository.ilocalrepositorymain
self.ui.setconfig(b'phases', b'publish', False, b'bundlerepo') # pytype: disable=attribute-error
self.tempfile = None self.tempfile = None
f = util.posixfile(bundlepath, b"rb") f = util.posixfile(bundlepath, b"rb")
bundle = exchange.readbundle(self.ui, f, bundlepath) bundle = exchange.readbundle(self.ui, f, bundlepath) # pytype: disable=attribute-error
if isinstance(bundle, bundle2.unbundle20): if isinstance(bundle, bundle2.unbundle20):
self._bundlefile = bundle self._bundlefile = bundle
self._cgunpacker = None self._cgunpacker = None
cgpart = None cgpart = None
for part in bundle.iterparts(seekable=True): for part in bundle.iterparts(seekable=True):
if part.type == b'changegroup': if part.type == b'changegroup':
# generator returns control to iterparts(). # generator returns control to iterparts().
cgpart.seek(0, os.SEEK_SET) cgpart.seek(0, os.SEEK_SET)
elif isinstance(bundle, changegroup.cg1unpacker): elif isinstance(bundle, changegroup.cg1unpacker):
if bundle.compressed(): if bundle.compressed():
f = self._writetempbundle( f = self._writetempbundle(
bundle.read, b'.hg10un', header=b'HG10UN' bundle.read, b'.hg10un', header=b'HG10UN'
) )
bundle = exchange.readbundle(self.ui, f, bundlepath, self.vfs) bundle = exchange.readbundle(self.ui, f, bundlepath, self.vfs) # pytype: disable=attribute-error
self._bundlefile = bundle self._bundlefile = bundle
self._cgunpacker = bundle self._cgunpacker = bundle
else: else:
raise error.Abort( raise error.Abort(
_(b'bundle type %s cannot be read') % type(bundle) _(b'bundle type %s cannot be read') % type(bundle)
) )
if bundle.compressed(): if bundle.compressed():
cgstream = self._writetempbundle(part.read, b'.cg%sun' % version) cgstream = self._writetempbundle(part.read, b'.cg%sun' % version)
self._cgunpacker = changegroup.getunbundler(version, cgstream, b'UN') self._cgunpacker = changegroup.getunbundler(version, cgstream, b'UN')
def _writetempbundle(self, readfn, suffix, header=b''): def _writetempbundle(self, readfn, suffix, header=b''):
"""Write a temporary file to disk """Write a temporary file to disk
""" """
fdtemp, temp = self.vfs.mkstemp(prefix=b"hg-bundle-", suffix=suffix) fdtemp, temp = self.vfs.mkstemp(prefix=b"hg-bundle-", suffix=suffix) # pytype: disable=attribute-error
self.tempfile = temp self.tempfile = temp
with os.fdopen(fdtemp, 'wb') as fptemp: with os.fdopen(fdtemp, 'wb') as fptemp:
fptemp.write(header) fptemp.write(header)
while True: while True:
chunk = readfn(2 ** 18) chunk = readfn(2 ** 18)
if not chunk: if not chunk:
break break
fptemp.write(chunk) fptemp.write(chunk)
return self.vfs.open(self.tempfile, mode=b"rb") return self.vfs.open(self.tempfile, mode=b"rb") # pytype: disable=attribute-error
@localrepo.unfilteredpropertycache @localrepo.unfilteredpropertycache
def _phasecache(self): def _phasecache(self):
return bundlephasecache(self, self._phasedefaults) return bundlephasecache(self, self._phasedefaults)
@localrepo.unfilteredpropertycache @localrepo.unfilteredpropertycache
def changelog(self): def changelog(self):
# consume the header if it exists # consume the header if it exists
self._cgunpacker.seek(self.filestart) self._cgunpacker.seek(self.filestart)
self._cgfilespos = _getfilestarts(self._cgunpacker) self._cgfilespos = _getfilestarts(self._cgunpacker)
if f in self._cgfilespos: if f in self._cgfilespos:
self._cgunpacker.seek(self._cgfilespos[f]) self._cgunpacker.seek(self._cgfilespos[f])
linkmapper = self.unfiltered().changelog.rev linkmapper = self.unfiltered().changelog.rev
return bundlefilelog(self.svfs, f, self._cgunpacker, linkmapper) return bundlefilelog(self.svfs, f, self._cgunpacker, linkmapper)
else: else:
return super(bundlerepository, self).file(f) return super(bundlerepository, self).file(f) # pytype: disable=attribute-error
def close(self): def close(self):
"""Close assigned bundle file immediately.""" """Close assigned bundle file immediately."""
self._bundlefile.close() self._bundlefile.close()
if self.tempfile is not None: if self.tempfile is not None:
self.vfs.unlink(self.tempfile) self.vfs.unlink(self.tempfile)
if self._tempparent: if self._tempparent:
shutil.rmtree(self._tempparent, True) shutil.rmtree(self._tempparent, True)
def setparents(self, p1, p2=nullid): def setparents(self, p1, p2=nullid):
p1rev = self.changelog.rev(p1) p1rev = self.changelog.rev(p1)
p2rev = self.changelog.rev(p2) p2rev = self.changelog.rev(p2)
msg = _(b"setting parent to node %s that only exists in the bundle\n") msg = _(b"setting parent to node %s that only exists in the bundle\n")
if self.changelog.repotiprev < p1rev: if self.changelog.repotiprev < p1rev:
self.ui.warn(msg % nodemod.hex(p1)) self.ui.warn(msg % nodemod.hex(p1))
if self.changelog.repotiprev < p2rev: if self.changelog.repotiprev < p2rev:
self.ui.warn(msg % nodemod.hex(p2)) self.ui.warn(msg % nodemod.hex(p2))
return super(bundlerepository, self).setparents(p1, p2) return super(bundlerepository, self).setparents(p1, p2) # pytype: disable=attribute-error
def instance(ui, path, create, intents=None, createopts=None): def instance(ui, path, create, intents=None, createopts=None):
if create: if create:
raise error.Abort(_(b'cannot create new bundle repository')) raise error.Abort(_(b'cannot create new bundle repository'))
# internal config: bundle.mainreporoot # internal config: bundle.mainreporoot
parentpath = ui.config(b"bundle", b"mainreporoot") parentpath = ui.config(b"bundle", b"mainreporoot")
if not parentpath: if not parentpath:

mercurial/commands.py

reflags = re.M reflags = re.M
if opts.get(b'ignore_case'): if opts.get(b'ignore_case'):
reflags |= re.I reflags |= re.I
try: try:
regexp = util.re.compile(pattern, reflags) regexp = util.re.compile(pattern, reflags)
except re.error as inst: except re.error as inst:
ui.warn( ui.warn(
_(b"grep: invalid match pattern: %s\n") % pycompat.bytestr(inst) _(b"grep: invalid match pattern: %s\n") % pycompat.bytestr(inst # pytype: disable=wrong-arg-types
)
) )
return 1 return 1
sep, eol = b':', b'\n' sep, eol = b':', b'\n'
if opts.get(b'print0'): if opts.get(b'print0'):
sep = eol = b'\0' sep = eol = b'\0'
getfile = util.lrucachefunc(repo.file) getfile = util.lrucachefunc(repo.file)

mercurial/dagparser.py

... except Exception as e: print(pycompat.sysstr(bytes(e))) ... except Exception as e: print(pycompat.sysstr(bytes(e)))
invalid character in dag description: bad... invalid character in dag description: bad...
''' '''
if not desc: if not desc:
return return
wordchars = pycompat.bytestr( wordchars = pycompat.bytestr(
string.ascii_letters + string.digits string.ascii_letters + string.digits # pytype: disable=wrong-arg-types
) # pytype: disable=wrong-arg-types )
labels = {} labels = {}
p1 = -1 p1 = -1
r = 0 r = 0
def resolve(ref): def resolve(ref):
if not ref: if not ref:
return p1 return p1
elif ref[0] in pycompat.bytestr( elif ref[0] in pycompat.bytestr(
string.digits string.digits # pytype: disable=wrong-arg-types
): # pytype: disable=wrong-arg-types ):
return r - int(ref) return r - int(ref)
else: else:
return labels[ref] return labels[ref]
chiter = pycompat.iterbytestr(desc) chiter = pycompat.iterbytestr(desc)
def nextch(): def nextch():
return next(chiter, b'\0') return next(chiter, b'\0')
if c == b'"': if c == b'"':
return nextdelimited(nextch(), b'"', b'\\') return nextdelimited(nextch(), b'"', b'\\')
else: else:
return nextrun(c, wordchars) return nextrun(c, wordchars)
c = nextch() c = nextch()
while c != b'\0': while c != b'\0':
while c in pycompat.bytestr( while c in pycompat.bytestr(
string.whitespace string.whitespace # pytype: disable=wrong-arg-types
): # pytype: disable=wrong-arg-types ):
c = nextch() c = nextch()
if c == b'.': if c == b'.':
yield b'n', (r, [p1]) yield b'n', (r, [p1])
p1 = r p1 = r
r += 1 r += 1
c = nextch() c = nextch()
elif c == b'+': elif c == b'+':
c, digs = nextrun( c, digs = nextrun(
nextch(), pycompat.bytestr(string.digits) nextch(), pycompat.bytestr(string.digits) # pytype: disable=wrong-arg-types
) # pytype: disable=wrong-arg-types )
n = int(digs) n = int(digs)
for i in pycompat.xrange(0, n): for i in pycompat.xrange(0, n):
yield b'n', (r, [p1]) yield b'n', (r, [p1])
p1 = r p1 = r
r += 1 r += 1
elif c in b'*/': elif c in b'*/':
if c == b'*': if c == b'*':
c = nextch() c = nextch()

mercurial/debugcommands.py

ui.write(b'%s %s\n' % (hex(head), phases.phasenames[phase])) ui.write(b'%s %s\n' % (hex(head), phases.phasenames[phase]))
def _quasirepr(thing): def _quasirepr(thing):
if isinstance(thing, (dict, util.sortdict, collections.OrderedDict)): if isinstance(thing, (dict, util.sortdict, collections.OrderedDict)):
return b'{%s}' % ( return b'{%s}' % (
b', '.join(b'%s: %s' % (k, thing[k]) for k in sorted(thing)) b', '.join(b'%s: %s' % (k, thing[k]) for k in sorted(thing))
) )
return pycompat.bytestr(repr(thing)) return pycompat.bytestr(repr(thing) # pytype: disable=wrong-arg-types
)
def _debugbundle2(ui, gen, all=None, **opts): def _debugbundle2(ui, gen, all=None, **opts):
"""lists the contents of a bundle2""" """lists the contents of a bundle2"""
if not isinstance(gen, bundle2.unbundle20): if not isinstance(gen, bundle2.unbundle20):
raise error.Abort(_(b'not a bundle2 file')) raise error.Abort(_(b'not a bundle2 file'))
ui.write((b'Stream params: %s\n' % _quasirepr(gen.params))) ui.write((b'Stream params: %s\n' % _quasirepr(gen.params)))
parttypes = opts.get('part_type', []) parttypes = opts.get('part_type', [])
parents=parents, parents=parents,
date=date, date=date,
metadata=metadata, metadata=metadata,
ui=ui, ui=ui,
) )
tr.close() tr.close()
except ValueError as exc: except ValueError as exc:
raise error.Abort( raise error.Abort(
_(b'bad obsmarker input: %s') % pycompat.bytestr(exc) _(b'bad obsmarker input: %s') % pycompat.bytestr(exc # pytype: disable=wrong-arg-types
)
) )
finally: finally:
tr.release() tr.release()
finally: finally:
l.release() l.release()
else: else:
if opts[b'rev']: if opts[b'rev']:
revs = scmutil.revrange(repo, opts[b'rev']) revs = scmutil.revrange(repo, opts[b'rev'])
(b' lvl-%-3d : ' % depth) (b' lvl-%-3d : ' % depth)
+ fmt % pcfmt(snaptotal[depth], totalsize) + fmt % pcfmt(snaptotal[depth], totalsize)
) )
ui.writenoi18n(b' deltas : ' + fmt % pcfmt(deltatotal, totalsize)) ui.writenoi18n(b' deltas : ' + fmt % pcfmt(deltatotal, totalsize))
def fmtchunktype(chunktype): def fmtchunktype(chunktype):
if chunktype == b'empty': if chunktype == b'empty':
return b' %s : ' % chunktype return b' %s : ' % chunktype
elif chunktype in pycompat.bytestr(string.ascii_letters): elif chunktype in pycompat.bytestr(string.ascii_letters # pytype: disable=wrong-arg-types
):
return b' 0x%s (%s) : ' % (hex(chunktype), chunktype) return b' 0x%s (%s) : ' % (hex(chunktype), chunktype)
else: else:
return b' 0x%s : ' % hex(chunktype) return b' 0x%s : ' % hex(chunktype)
ui.write(b'\n') ui.write(b'\n')
ui.writenoi18n(b'chunks : ' + fmt2 % numrevs) ui.writenoi18n(b'chunks : ' + fmt2 % numrevs)
for chunktype in sorted(chunktypecounts): for chunktype in sorted(chunktypecounts):
ui.write(fmtchunktype(chunktype)) ui.write(fmtchunktype(chunktype))

mercurial/error.py

"""Mix-in to provide a hint of an error """Mix-in to provide a hint of an error
This should come first in the inheritance list to consume a hint and This should come first in the inheritance list to consume a hint and
pass remaining arguments to the exception class. pass remaining arguments to the exception class.
""" """
def __init__(self, *args, **kw): def __init__(self, *args, **kw):
self.hint = kw.pop('hint', None) self.hint = kw.pop('hint', None)
super(Hint, self).__init__(*args, **kw) super(Hint, self).__init__(*args, **kw) # pytype: disable=wrong-arg-count
class StorageError(Hint, Exception): class StorageError(Hint, Exception):
"""Raised when an error occurs in a storage layer. """Raised when an error occurs in a storage layer.
Usually subclassed by a storage-specific exception. Usually subclassed by a storage-specific exception.
""" """

mercurial/exchange.py

kwargs[b'bookmarks'] = True kwargs[b'bookmarks'] = True
bookmarksrequested = True bookmarksrequested = True
if b'listkeys' in pullop.remotebundle2caps: if b'listkeys' in pullop.remotebundle2caps:
if b'request-bookmarks' not in pullop.stepsdone: if b'request-bookmarks' not in pullop.stepsdone:
# make sure to always includes bookmark data when migrating # make sure to always includes bookmark data when migrating
# `hg incoming --bundle` to using this function. # `hg incoming --bundle` to using this function.
pullop.stepsdone.add(b'request-bookmarks') pullop.stepsdone.add(b'request-bookmarks')
kwargs.setdefault(b'listkeys', []).append(b'bookmarks') kwargs.setdefault(b'listkeys', []).append(b'bookmarks') # pytype: disable=attribute-error
# If this is a full pull / clone and the server supports the clone bundles # If this is a full pull / clone and the server supports the clone bundles
# feature, tell the server whether we attempted a clone bundle. The # feature, tell the server whether we attempted a clone bundle. The
# presence of this flag indicates the client supports clone bundles. This # presence of this flag indicates the client supports clone bundles. This
# will enable the server to treat clients that support clone bundles # will enable the server to treat clients that support clone bundles
# differently from those that don't. # differently from those that don't.
if ( if (
pullop.remote.capable(b'clonebundles') pullop.remote.capable(b'clonebundles')
) )
user_excludes = ui.configlist( user_excludes = ui.configlist(
_NARROWACL_SECTION, _NARROWACL_SECTION,
username + b'.excludes', username + b'.excludes',
ui.configlist(_NARROWACL_SECTION, b'default.excludes'), ui.configlist(_NARROWACL_SECTION, b'default.excludes'),
) )
if not user_includes: if not user_includes:
raise error.Abort( raise error.Abort(
_(b"{} configuration for user {} is empty").format( _(b"%s configuration for user %s is empty") % (
_NARROWACL_SECTION, username _NARROWACL_SECTION, username
) )
) )
user_includes = [ user_includes = [
b'path:.' if p == b'*' else b'path:' + p for p in user_includes b'path:.' if p == b'*' else b'path:' + p for p in user_includes
] ]
user_excludes = [ user_excludes = [
b'path:.' if p == b'*' else b'path:' + p for p in user_excludes b'path:.' if p == b'*' else b'path:' + p for p in user_excludes
] ]
req_includes = set(kwargs.get('includepats', [])) req_includes = set(kwargs.get('includepats', []))
req_excludes = set(kwargs.get('excludepats', [])) req_excludes = set(kwargs.get('excludepats', []))
req_includes, req_excludes, invalid_includes = narrowspec.restrictpatterns( req_includes, req_excludes, invalid_includes = narrowspec.restrictpatterns(
req_includes, req_excludes, user_includes, user_excludes req_includes, req_excludes, user_includes, user_excludes
) )
if invalid_includes: if invalid_includes:
raise error.Abort( raise error.Abort(
_(b"The following includes are not accessible for {}: {}").format( _(b"The following includes are not accessible for %s: %s") % (
username, invalid_includes username, invalid_includes
) )
) )
new_args = {} new_args = {}
new_args.update(kwargs) new_args.update(kwargs)
new_args['narrow'] = True new_args['narrow'] = True
new_args['narrow_acl'] = True new_args['narrow_acl'] = True
source, source,
bundlecaps=None, bundlecaps=None,
b2caps=None, b2caps=None,
heads=None, heads=None,
common=None, common=None,
**kwargs **kwargs
): ):
"""add a changegroup part to the requested bundle""" """add a changegroup part to the requested bundle"""
if not kwargs.get('cg', True): if not kwargs.get('cg', True) or not b2caps:
return return
version = b'01' version = b'01'
cgversions = b2caps.get(b'changegroup') cgversions = b2caps.get(b'changegroup')
if cgversions: # 3.1 and 3.2 ship with an empty value if cgversions: # 3.1 and 3.2 ship with an empty value
cgversions = [ cgversions = [
v v
for v in cgversions for v in cgversions
@getbundle2partsgenerator(b'bookmarks') @getbundle2partsgenerator(b'bookmarks')
def _getbundlebookmarkpart( def _getbundlebookmarkpart(
bundler, repo, source, bundlecaps=None, b2caps=None, **kwargs bundler, repo, source, bundlecaps=None, b2caps=None, **kwargs
): ):
"""add a bookmark part to the requested bundle""" """add a bookmark part to the requested bundle"""
if not kwargs.get('bookmarks', False): if not kwargs.get('bookmarks', False):
return return
if b'bookmarks' not in b2caps: if not b2caps or b'bookmarks' not in b2caps:
raise error.Abort(_(b'no common bookmarks exchange method')) raise error.Abort(_(b'no common bookmarks exchange method'))
books = bookmod.listbinbookmarks(repo) books = bookmod.listbinbookmarks(repo)
data = bookmod.binaryencode(books) data = bookmod.binaryencode(books)
if data: if data:
bundler.newpart(b'bookmarks', data=data) bundler.newpart(b'bookmarks', data=data)
@getbundle2partsgenerator(b'listkeys') @getbundle2partsgenerator(b'listkeys')
@getbundle2partsgenerator(b'phases') @getbundle2partsgenerator(b'phases')
def _getbundlephasespart( def _getbundlephasespart(
bundler, repo, source, bundlecaps=None, b2caps=None, heads=None, **kwargs bundler, repo, source, bundlecaps=None, b2caps=None, heads=None, **kwargs
): ):
"""add phase heads part to the requested bundle""" """add phase heads part to the requested bundle"""
if kwargs.get('phases', False): if kwargs.get('phases', False):
if not b'heads' in b2caps.get(b'phases'): if not b2caps or not b'heads' in b2caps.get(b'phases'):
raise error.Abort(_(b'no common phases exchange method')) raise error.Abort(_(b'no common phases exchange method'))
if heads is None: if heads is None:
heads = repo.heads() heads = repo.heads()
headsbyphase = collections.defaultdict(set) headsbyphase = collections.defaultdict(set)
if repo.publishing(): if repo.publishing():
headsbyphase[phases.public] = heads headsbyphase[phases.public] = heads
else: else:
Only values for heads in this bundle will be transferred. Only values for heads in this bundle will be transferred.
The part data consists of pairs of 20 byte changeset node and .hgtags The part data consists of pairs of 20 byte changeset node and .hgtags
filenodes raw values. filenodes raw values.
""" """
# Don't send unless: # Don't send unless:
# - changeset are being exchanged, # - changeset are being exchanged,
# - the client supports it. # - the client supports it.
if not (kwargs.get('cg', True) and b'hgtagsfnodes' in b2caps): if not b2caps or not (kwargs.get('cg', True) and b'hgtagsfnodes' in b2caps):
return return
outgoing = _computeoutgoing(repo, heads, common) outgoing = _computeoutgoing(repo, heads, common)
bundle2.addparttagsfnodescache(repo, bundler, outgoing) bundle2.addparttagsfnodescache(repo, bundler, outgoing)
@getbundle2partsgenerator(b'cache:rev-branch-cache') @getbundle2partsgenerator(b'cache:rev-branch-cache')
def _getbundlerevbranchcache( def _getbundlerevbranchcache(
5) closed heads nodes 5) closed heads nodes
""" """
# Don't send unless: # Don't send unless:
# - changeset are being exchanged, # - changeset are being exchanged,
# - the client supports it. # - the client supports it.
# - narrow bundle isn't in play (not currently compatible). # - narrow bundle isn't in play (not currently compatible).
if ( if (
not kwargs.get('cg', True) not kwargs.get('cg', True)
or not b2caps
or b'rev-branch-cache' not in b2caps or b'rev-branch-cache' not in b2caps
or kwargs.get('narrow', False) or kwargs.get('narrow', False)
or repo.ui.has_section(_NARROWACL_SECTION) or repo.ui.has_section(_NARROWACL_SECTION)
): ):
return return
outgoing = _computeoutgoing(repo, heads, common) outgoing = _computeoutgoing(repo, heads, common)
bundle2.addpartrevbranchcache(repo, bundler, outgoing) bundle2.addpartrevbranchcache(repo, bundler, outgoing)
try: try:
def gettransaction(): def gettransaction():
if not lockandtr[2]: if not lockandtr[2]:
if not bookmod.bookmarksinstore(repo): if not bookmod.bookmarksinstore(repo):
lockandtr[0] = repo.wlock() lockandtr[0] = repo.wlock()
lockandtr[1] = repo.lock() lockandtr[1] = repo.lock()
lockandtr[2] = repo.transaction(source) lockandtr[2] = repo.transaction(source)
lockandtr[2].hookargs[b'source'] = source lockandtr[2].hookargs[b'source'] = source # pytype: disable=attribute-error
lockandtr[2].hookargs[b'url'] = url lockandtr[2].hookargs[b'url'] = url # pytype: disable=attribute-error
lockandtr[2].hookargs[b'bundle2'] = b'1' lockandtr[2].hookargs[b'bundle2'] = b'1' # pytype: disable=attribute-error
return lockandtr[2] return lockandtr[2]
# Do greedy locking by default until we're satisfied with lazy # Do greedy locking by default until we're satisfied with lazy
# locking. # locking.
if not repo.ui.configbool( if not repo.ui.configbool(
b'experimental', b'bundle2lazylocking' b'experimental', b'bundle2lazylocking'
): ):
gettransaction() gettransaction()

mercurial/filemerge.py

def cmp(self, fctx): def cmp(self, fctx):
"""compare with other file context """compare with other file context
returns True if different from fctx. returns True if different from fctx.
""" """
return not ( return not (
fctx.isabsent() fctx.isabsent()
and fctx.ctx() == self.ctx() and fctx.ctx() == self._ctx
and fctx.path() == self.path() and fctx.path() == self.path()
) )
def flags(self): def flags(self):
return b'' return b''
def changectx(self): def changectx(self):
return self._ctx return self._ctx
# things for us to import cmdutil. # things for us to import cmdutil.
tres = formatter.templateresources(ui, repo) tres = formatter.templateresources(ui, repo)
t = formatter.maketemplater( t = formatter.maketemplater(
ui, tmpl, defaults=templatekw.keywords, resources=tres ui, tmpl, defaults=templatekw.keywords, resources=tres
) )
ui.status(t.renderdefault(props)) ui.status(t.renderdefault(props))
def _xmerge(repo, mynode, orig, fcd, fco, fca, toolconf, files, labels=None): def _xmerge(repo, mynode, orig, fcd, fco, fca, toolconf, files, labels):
tool, toolpath, binary, symlink, scriptfn = toolconf tool, toolpath, binary, symlink, scriptfn = toolconf
uipathfn = scmutil.getuipathfn(repo) uipathfn = scmutil.getuipathfn(repo)
if fcd.isabsent() or fco.isabsent(): if fcd.isabsent() or fco.isabsent():
repo.ui.warn( repo.ui.warn(
_(b'warning: %s cannot merge change/delete conflict for %s\n') _(b'warning: %s cannot merge change/delete conflict for %s\n')
% (tool, uipathfn(fcd.path())) % (tool, uipathfn(fcd.path()))
) )
return False, 1, None return False, 1, None

mercurial/hg.py

def _local(path): def _local(path):
path = util.expandpath(util.urllocalpath(path)) path = util.expandpath(util.urllocalpath(path))
try: try:
isfile = os.path.isfile(path) isfile = os.path.isfile(path)
# Python 2 raises TypeError, Python 3 ValueError. # Python 2 raises TypeError, Python 3 ValueError.
except (TypeError, ValueError) as e: except (TypeError, ValueError) as e:
raise error.Abort( raise error.Abort(
_(b'invalid path %s: %s') % (path, pycompat.bytestr(e)) _(b'invalid path %s: %s') % (path, pycompat.bytestr(e # pytype: disable=wrong-arg-types
))
) )
return isfile and bundlerepo or localrepo return isfile and bundlerepo or localrepo
def addbranchrevs(lrepo, other, branches, revs): def addbranchrevs(lrepo, other, branches, revs):
peer = other.peer() # a courtesy to callers using a localrepo for other peer = other.peer() # a courtesy to callers using a localrepo for other
hashbranch, branches = branches hashbranch, branches = branches

mercurial/hgweb/server.py

sys.argv # unwrap demand-loader so that reload() works sys.argv # unwrap demand-loader so that reload() works
# resurrect sys.setdefaultencoding() # resurrect sys.setdefaultencoding()
try: try:
importlib.reload(sys) importlib.reload(sys)
except AttributeError: except AttributeError:
reload(sys) reload(sys)
oldenc = sys.getdefaultencoding() oldenc = sys.getdefaultencoding()
sys.setdefaultencoding(b"latin1") # or any full 8-bit encoding # or any full 8-bit encoding
sys.setdefaultencoding(b"latin1") # pytype: disable=module-attr
import mimetypes
mimetypes.init() mimetypes.init()
sys.setdefaultencoding(oldenc) sys.setdefaultencoding(oldenc) # pytype: disable=module-attr
address = ui.config(b'web', b'address') address = ui.config(b'web', b'address')
port = util.getport(ui.config(b'web', b'port')) port = util.getport(ui.config(b'web', b'port'))
try: try:
return cls(ui, app, (address, port), handler) return cls(ui, app, (address, port), handler)
except socket.error as inst: except socket.error as inst:
raise error.Abort( raise error.Abort(
_(b"cannot start server at '%s:%d': %s") _(b"cannot start server at '%s:%d': %s")
% (address, port, encoding.strtolocal(inst.args[1])) % (address, port, encoding.strtolocal(inst.args[1]))
) )

mercurial/hgweb/webcommands.py

linerange = None linerange = None
if lrange is not None: if lrange is not None:
linerange = webutil.formatlinerange(*lrange) linerange = webutil.formatlinerange(*lrange)
# deactivate numeric nav links when linerange is specified as this # deactivate numeric nav links when linerange is specified as this
# would required a dedicated "revnav" class # would required a dedicated "revnav" class
nav = templateutil.mappinglist([]) nav = templateutil.mappinglist([])
if descend: if descend:
it = dagop.blockdescendants(fctx, *lrange) # XXX: This seems correct, because lrange is a tuple of 2.
# Why not complain about formatlinerange() below?
it = dagop.blockdescendants(fctx, *lrange) # pytype: disable=missing-parameter
else: else:
it = dagop.blockancestors(fctx, *lrange) it = dagop.blockancestors(fctx, *lrange) # pytype: disable=missing-parameter
for i, (c, lr) in enumerate(it, 1): for i, (c, lr) in enumerate(it, 1):
diffs = None diffs = None
if patch: if patch:
diffs = diff(c, linerange=lr) diffs = diff(c, linerange=lr)
# follow renames accross filtered (not in range) revisions # follow renames accross filtered (not in range) revisions
path = c.path() path = c.path()
lm = webutil.commonentry(repo, c) lm = webutil.commonentry(repo, c)
lm.update( lm.update(

mercurial/hgweb/webutil.py

} }
separator = b'&' separator = b'&'
def join(self, context, mapping, sep): def join(self, context, mapping, sep):
# could be '{separator}{name}={value|urlescape}' # could be '{separator}{name}={value|urlescape}'
raise error.ParseError(_(b'not displayable without template')) raise error.ParseError(_(b'not displayable without template'))
def show(self, context, mapping): def show(self, context, mapping):
return self.join(context, b'') return self.join(context, mapping, b'')
def tobool(self, context, mapping): def tobool(self, context, mapping):
return bool(self._vars) return bool(self._vars)
def tovalue(self, context, mapping): def tovalue(self, context, mapping):
return self._vars return self._vars

mercurial/logcmdutil.py

dateutil, dateutil,
stringutil, stringutil,
) )
if not globals(): if not globals():
from typing import ( from typing import (
Any, Any,
Optional,
Tuple, Tuple,
Union,
) )
for t in (Any, Tuple): for t in (Any, Optional, Tuple, Union):
assert t assert t
def getlimit(opts): def getlimit(opts):
"""get the log limit according to option -l/--limit""" """get the log limit according to option -l/--limit"""
limit = opts.get(b'limit') limit = opts.get(b'limit')
if limit: if limit:
try: try:
revs = repo.revs(b'.') revs = repo.revs(b'.')
else: else:
revs = smartset.spanset(repo) revs = smartset.spanset(repo)
revs.reverse() revs.reverse()
return revs return revs
def getrevs(repo, pats, opts): def getrevs(repo, pats, opts):
# type: (Any, Any, Any) -> Tuple[smartset.BaseSet, changesetdiffer] # type: (Any, Any, Any) -> Tuple[smartset.baseset, Optional[changesetdiffer]]
"""Return (revs, differ) where revs is a smartset """Return (revs, differ) where revs is a smartset
differ is a changesetdiffer with pre-configured file matcher. differ is a changesetdiffer with pre-configured file matcher.
""" """
follow = opts.get(b'follow') or opts.get(b'follow_first') follow = opts.get(b'follow') or opts.get(b'follow_first')
followfirst = opts.get(b'follow_first') followfirst = opts.get(b'follow_first')
limit = getlimit(opts) limit = getlimit(opts)
revs = _initialrevs(repo, opts) revs = _initialrevs(repo, opts)
if expr: if expr:
matcher = revset.match(None, expr) matcher = revset.match(None, expr)
revs = matcher(repo, revs) revs = matcher(repo, revs)
if limit is not None: if limit is not None:
revs = revs.slice(0, limit) revs = revs.slice(0, limit)
differ = changesetdiffer() differ = changesetdiffer()
differ._makefilematcher = filematcher differ._makefilematcher = filematcher
return revs, differ return revs, differ # pytype: disable=bad-return-type
def _parselinerangeopt(repo, opts): def _parselinerangeopt(repo, opts):
"""Parse --line-range log option and return a list of tuples (filename, """Parse --line-range log option and return a list of tuples (filename,
(fromline, toline)). (fromline, toline)).
""" """
linerangebyfname = [] linerangebyfname = []
for pat in opts.get(b'line_range', []): for pat in opts.get(b'line_range', []):
props = props or {} props = props or {}
formatnode = _graphnodeformatter(ui, displayer) formatnode = _graphnodeformatter(ui, displayer)
state = graphmod.asciistate() state = graphmod.asciistate()
styles = state[b'styles'] styles = state[b'styles']
# only set graph styling if HGPLAIN is not set. # only set graph styling if HGPLAIN is not set.
if ui.plain(b'graph'): if ui.plain(b'graph'):
# set all edge styles to |, the default pre-3.8 behaviour # set all edge styles to |, the default pre-3.8 behaviour
styles.update(dict.fromkeys(styles, b'|')) styles.update(dict.fromkeys(styles, b'|')) # pytype: disable=attribute-error
else: else:
edgetypes = { edgetypes = {
b'parent': graphmod.PARENT, b'parent': graphmod.PARENT,
b'grandparent': graphmod.GRANDPARENT, b'grandparent': graphmod.GRANDPARENT,
b'missing': graphmod.MISSINGPARENT, b'missing': graphmod.MISSINGPARENT,
} }
for name, key in edgetypes.items(): for name, key in edgetypes.items():
# experimental config: experimental.graphstyle.* # experimental config: experimental.graphstyle.*
styles[key] = ui.config( styles[key] = ui.config(
b'experimental', b'graphstyle.%s' % name, styles[key] b'experimental', b'graphstyle.%s' % name, styles[key] # pytype: disable=unsupported-operands
) )
if not styles[key]: if not styles[key]:
styles[key] = None styles[key] = None # pytype: disable=unsupported-operands
# experimental config: experimental.graphshorten # experimental config: experimental.graphshorten
state[b'graphshorten'] = ui.configbool(b'experimental', b'graphshorten') state[b'graphshorten'] = ui.configbool(b'experimental', b'graphshorten')
for rev, type, ctx, parents in dag: for rev, type, ctx, parents in dag:
char = formatnode(repo, ctx) char = formatnode(repo, ctx)
copies = getcopies(ctx) if getcopies else None copies = getcopies(ctx) if getcopies else None
edges = edgefn(type, char, state, rev, parents) edges = edgefn(type, char, state, rev, parents)

mercurial/manifest.py

return (data[pos:zeropos], hashval, flags) return (data[pos:zeropos], hashval, flags)
__next__ = next __next__ = next
def unhexlify(data, extra, pos, length): def unhexlify(data, extra, pos, length):
s = bin(data[pos : pos + length]) s = bin(data[pos : pos + length])
if extra: if extra:
s += chr(extra & 0xFF) s += chr(extra & 0xFF) # pytype: disable=unsupported-operands
return s return s
def _cmp(a, b): def _cmp(a, b):
return (a > b) - (a < b) return (a > b) - (a < b)
class _lazymanifest(object): class _lazymanifest(object):
# If our first parent is in the manifest cache, we can # If our first parent is in the manifest cache, we can
# compute a delta here using properties we know about the # compute a delta here using properties we know about the
# manifest up-front, which may save time later for the # manifest up-front, which may save time later for the
# revlog layer. # revlog layer.
_checkforbidden(added) _checkforbidden(added)
# combine the changed lists into one sorted iterator # combine the changed lists into one sorted iterator
work = heapq.merge( work = heapq.merge(
[(x, False) for x in sorted(added)], # XXX: writesubtree() below passes None. Not clear if this
[(x, True) for x in sorted(removed)], # branch can't happen because of `if` test above.
[(x, False) for x in sorted(added)], # pytype: disable=wrong-arg-types
[(x, True) for x in sorted(removed)], # pytype: disable=wrong-arg-types
) )
arraytext, deltatext = m.fastdelta(self.fulltextcache[p1], work) arraytext, deltatext = m.fastdelta(self.fulltextcache[p1], work)
cachedelta = self._revlog.rev(p1), deltatext cachedelta = self._revlog.rev(p1), deltatext
text = util.buffer(arraytext) text = util.buffer(arraytext)
n = self._revlog.addrevision( n = self._revlog.addrevision(
text, transaction, link, p1, p2, cachedelta text, transaction, link, p1, p2, cachedelta
) )

mercurial/obsutil.py

sset = _succs() sset = _succs()
sset.markers.add(mark) sset.markers.add(mark)
fullsuccessorsets.append(sset) fullsuccessorsets.append(sset)
if not foundany: if not foundany:
fullsuccessorsets.append(_succs()) fullsuccessorsets.append(_succs())
values = [] values = []
for sset in fullsuccessorsets: for sset in fullsuccessorsets:
values.append({b'successors': sset, b'markers': sset.markers}) # This might be OK, because successorsets == [] falls into the else
# case above.
values.append({b'successors': sset, b'markers': sset.markers}) # pytype: disable=attribute-error
return values return values
def _getobsfate(successorssets): def _getobsfate(successorssets):
""" Compute a changeset obsolescence fate based on its successorssets. """ Compute a changeset obsolescence fate based on its successorssets.
Successors can be the tipmost ones or the immediate ones. This function Successors can be the tipmost ones or the immediate ones. This function
return values are not meant to be shown directly to users, it is meant to return values are not meant to be shown directly to users, it is meant to

mercurial/patch.py

parents = [] parents = []
nodeid = msg['X-Mercurial-Node'] nodeid = msg['X-Mercurial-Node']
if nodeid: if nodeid:
data[b'nodeid'] = nodeid = mail.headdecode(nodeid) data[b'nodeid'] = nodeid = mail.headdecode(nodeid)
ui.debug(b'Node ID: %s\n' % nodeid) ui.debug(b'Node ID: %s\n' % nodeid)
if subject: if subject:
if subject.startswith(b'[PATCH'): if subject.startswith(b'[PATCH'): # pytype: disable=attribute-error
pend = subject.find(b']') pend = subject.find(b']') # pytype: disable=attribute-error
if pend >= 0: if pend >= 0:
subject = subject[pend + 1 :].lstrip() subject = subject[pend + 1 :].lstrip()
subject = re.sub(br'\n[ \t]+', b' ', subject) subject = re.sub(br'\n[ \t]+', b' ', subject)
ui.debug(b'Subject: %s\n' % subject) ui.debug(b'Subject: %s\n' % subject)
if data[b'user']: if data[b'user']:
ui.debug(b'From: %s\n' % data[b'user']) ui.debug(b'From: %s\n' % data[b'user'])
diffs_seen = 0 diffs_seen = 0
ok_types = (b'text/plain', b'text/x-diff', b'text/x-patch') ok_types = (b'text/plain', b'text/x-diff', b'text/x-patch')
message = b'' message = b''
for part in msg.walk(): for part in msg.walk():
content_type = pycompat.bytestr(part.get_content_type()) content_type = pycompat.bytestr(part.get_content_type()) # pytype: disable=wrong-arg-types
ui.debug(b'Content-Type: %s\n' % content_type) ui.debug(b'Content-Type: %s\n' % content_type)
if content_type not in ok_types: if content_type not in ok_types:
continue continue
payload = part.get_payload(decode=True) payload = part.get_payload(decode=True)
m = diffre.search(payload) m = diffre.search(payload)
if m: if m:
hgpatch = False hgpatch = False
hgpatchheader = False hgpatchheader = False
ignoretext = False ignoretext = False
ui.debug(b'found patch at byte %d\n' % m.start(0)) ui.debug(b'found patch at byte %d\n' % m.start(0))
diffs_seen += 1 diffs_seen += 1
cfp = stringio() cfp = stringio()
for line in payload[: m.start(0)].splitlines(): for line in payload[: m.start(0)].splitlines(): # pytype: disable=attribute-error
if line.startswith(b'# HG changeset patch') and not hgpatch: if line.startswith(b'# HG changeset patch') and not hgpatch:
ui.debug(b'patch generated by hg export\n') ui.debug(b'patch generated by hg export\n')
hgpatch = True hgpatch = True
hgpatchheader = True hgpatchheader = True
# drop earlier commit message content # drop earlier commit message content
cfp.seek(0) cfp.seek(0)
cfp.truncate() cfp.truncate()
subject = None subject = None
elif line == b'---': elif line == b'---':
ignoretext = True ignoretext = True
if not hgpatchheader and not ignoretext: if not hgpatchheader and not ignoretext:
cfp.write(line) cfp.write(line)
cfp.write(b'\n') cfp.write(b'\n')
message = cfp.getvalue() message = cfp.getvalue()
if tmpfp: if tmpfp:
tmpfp.write(payload) tmpfp.write(payload)
if not payload.endswith(b'\n'): if not payload.endswith(b'\n'): # pytype: disable=attribute-error
tmpfp.write(b'\n') tmpfp.write(b'\n')
elif not diffs_seen and message and content_type == b'text/plain': elif not diffs_seen and message and content_type == b'text/plain':
message += b'\n' + payload message += b'\n' + payload
if subject and not message.startswith(subject): if subject and not message.startswith(subject):
message = b'%s\n%s' % (subject, message) message = b'%s\n%s' % (subject, message)
data[b'message'] = message data[b'message'] = message
tmpfp.close() tmpfp.close()
return [fromfile, tofile] return [fromfile, tofile]
else: else:
return self.diff_re.match(self.header[0]).groups() return self.diff_re.match(self.header[0]).groups()
def filename(self): def filename(self):
return self.files()[-1] return self.files()[-1]
def __repr__(self): def __repr__(self):
return b'<header %s>' % (b' '.join(map(repr, self.files()))) # XXX: This is a bug
return b'<header %s>' % (b' '.join(map(repr, self.files()))) # pytype: disable=wrong-arg-types
def isnewfile(self): def isnewfile(self):
return any(self.newfile_re.match(h) for h in self.header) return any(self.newfile_re.match(h) for h in self.header)
def special(self): def special(self):
# Special files are shown only at the header level and not at the hunk # Special files are shown only at the header level and not at the hunk
# level for example a file that has been deleted is a special file. # level for example a file that has been deleted is a special file.
# The user cannot change the content of the operation, in the case of # The user cannot change the content of the operation, in the case of

mercurial/profiling.py

def __exit__(self, exception_type, exception_value, traceback): def __exit__(self, exception_type, exception_value, traceback):
propagate = None propagate = None
if self._profiler is not None: if self._profiler is not None:
propagate = self._profiler.__exit__( propagate = self._profiler.__exit__(
exception_type, exception_value, traceback exception_type, exception_value, traceback
) )
if self._output == b'blackbox': if self._output == b'blackbox':
val = b'Profile:\n%s' % self._fp.getvalue() val = b'Profile:\n%s' % self._fp.getvalue() # pytype: disable=attribute-error
# ui.log treats the input as a format string, # ui.log treats the input as a format string,
# so we need to escape any % signs. # so we need to escape any % signs.
val = val.replace(b'%', b'%%') val = val.replace(b'%', b'%%')
self._ui.log(b'profile', val) self._ui.log(b'profile', val)
self._closefp() self._closefp()
return propagate return propagate
def _closefp(self): def _closefp(self):
if self._fpdoclose and self._fp is not None: if self._fpdoclose and self._fp is not None:
self._fp.close() self._fp.close()

mercurial/pycompat.py

import io import io
import struct import struct
if os.name == r'nt' and sys.version_info >= (3, 6): if os.name == r'nt' and sys.version_info >= (3, 6):
# MBCS (or ANSI) filesystem encoding must be used as before. # MBCS (or ANSI) filesystem encoding must be used as before.
# Otherwise non-ASCII filenames in existing repositories would be # Otherwise non-ASCII filenames in existing repositories would be
# corrupted. # corrupted.
# This must be set once prior to any fsencode/fsdecode calls. # This must be set once prior to any fsencode/fsdecode calls.
sys._enablelegacywindowsfsencoding() sys._enablelegacywindowsfsencoding() # pytype: disable=module-attr
fsencode = os.fsencode fsencode = os.fsencode
fsdecode = os.fsdecode fsdecode = os.fsdecode
oscurdir = os.curdir.encode('ascii') oscurdir = os.curdir.encode('ascii')
oslinesep = os.linesep.encode('ascii') oslinesep = os.linesep.encode('ascii')
osname = os.name.encode('ascii') osname = os.name.encode('ascii')
ospathsep = os.pathsep.encode('ascii') ospathsep = os.pathsep.encode('ascii')
ospardir = os.pardir.encode('ascii') ospardir = os.pardir.encode('ascii')
doc = getattr(obj, '__doc__', None) doc = getattr(obj, '__doc__', None)
if doc is None: if doc is None:
return doc return doc
return sysbytes(doc) return sysbytes(doc)
def _wrapattrfunc(f): def _wrapattrfunc(f):
@functools.wraps(f) @functools.wraps(f)
def w(object, name, *args): def w(object, name, *args):
return f(object, sysstr(name), *args) return f(object, sysstr(name), *args) # pytype: disable=wrong-arg-count
return w return w
# these wrappers are automagically imported by hgloader # these wrappers are automagically imported by hgloader
delattr = _wrapattrfunc(builtins.delattr) delattr = _wrapattrfunc(builtins.delattr)
getattr = _wrapattrfunc(builtins.getattr) getattr = _wrapattrfunc(builtins.getattr)
hasattr = _wrapattrfunc(builtins.hasattr) hasattr = _wrapattrfunc(builtins.hasattr)
setattr = _wrapattrfunc(builtins.setattr) setattr = _wrapattrfunc(builtins.setattr)

mercurial/scmwindows.py

from __future__ import absolute_import from __future__ import absolute_import
import os import os
from . import ( from . import (
encoding, encoding,
pycompat, pycompat,
util, util,
win32, win32,
) )
try: try:
import _winreg as winreg import _winreg as winreg # pytype: disable=import-error
winreg.CloseKey winreg.CloseKey
except ImportError: except ImportError:
import winreg # py2 only
import winreg # pytype: disable=import-error
# MS-DOS 'more' is the only pager available by default on Windows. # MS-DOS 'more' is the only pager available by default on Windows.
fallbackpager = b'more' fallbackpager = b'more'
def systemrcpath(): def systemrcpath():
'''return default os-specific hgrc search path''' '''return default os-specific hgrc search path'''
rcpath = [] rcpath = []

mercurial/shelve.py

try: try:
d[b'originalwctx'] = nodemod.bin(d[b'originalwctx']) d[b'originalwctx'] = nodemod.bin(d[b'originalwctx'])
d[b'pendingctx'] = nodemod.bin(d[b'pendingctx']) d[b'pendingctx'] = nodemod.bin(d[b'pendingctx'])
d[b'parents'] = [nodemod.bin(h) for h in d[b'parents'].split(b' ')] d[b'parents'] = [nodemod.bin(h) for h in d[b'parents'].split(b' ')]
d[b'nodestoremove'] = [ d[b'nodestoremove'] = [
nodemod.bin(h) for h in d[b'nodestoremove'].split(b' ') nodemod.bin(h) for h in d[b'nodestoremove'].split(b' ')
] ]
except (ValueError, TypeError, KeyError) as err: except (ValueError, TypeError, KeyError) as err:
raise error.CorruptedState(pycompat.bytestr(err)) raise error.CorruptedState(pycompat.bytestr(err)) # pytype: disable=wrong-arg-types
@classmethod @classmethod
def _getversion(cls, repo): def _getversion(cls, repo):
"""Read version information from shelvestate file""" """Read version information from shelvestate file"""
fp = repo.vfs(cls._filename) fp = repo.vfs(cls._filename)
try: try:
version = int(fp.readline().strip()) version = int(fp.readline().strip())
except ValueError as err: except ValueError as err:
raise error.CorruptedState(pycompat.bytestr(err)) raise error.CorruptedState(pycompat.bytestr(err)) # pytype: disable=wrong-arg-types
finally: finally:
fp.close() fp.close()
return version return version
@classmethod @classmethod
def _readold(cls, repo): def _readold(cls, repo):
"""Read the old position-based version of a shelvestate file""" """Read the old position-based version of a shelvestate file"""
# Order is important, because old shelvestate file uses it # Order is important, because old shelvestate file uses it

mercurial/subrepo.py

if info.isdir(): if info.isdir():
continue continue
bname = pycompat.fsencode(info.name) bname = pycompat.fsencode(info.name)
if match and not match(bname): if match and not match(bname):
continue continue
if info.issym(): if info.issym():
data = info.linkname data = info.linkname
else: else:
data = tar.extractfile(info).read() data = tar.extractfile(info).read() # pytype: disable=attribute-error
archiver.addfile(prefix + bname, info.mode, info.issym(), data) archiver.addfile(prefix + bname, info.mode, info.issym(), data)
total += 1 total += 1
progress.increment() progress.increment()
progress.complete() progress.complete()
return total return total
@annotatesubrepoerror @annotatesubrepoerror
def cat(self, match, fm, fntemplate, prefix, **opts): def cat(self, match, fm, fntemplate, prefix, **opts):

mercurial/ui.py

""" """
# Sadly, the prompt string may have been built with a filename # Sadly, the prompt string may have been built with a filename
# containing "$$" so let's try to find the first valid-looking # containing "$$" so let's try to find the first valid-looking
# prompt to start parsing. Sadly, we also can't rely on # prompt to start parsing. Sadly, we also can't rely on
# choices containing spaces, ASCII, or basically anything # choices containing spaces, ASCII, or basically anything
# except an ampersand followed by a character. # except an ampersand followed by a character.
m = re.match(br'(?s)(.+?)\$\$([^\$]*&[^ \$].*)', prompt) m = re.match(br'(?s)(.+?)\$\$([^\$]*&[^ \$].*)', prompt)
msg = m.group(1) msg = m.group(1) # pytype: disable=attribute-error
choices = [p.strip(b' ') for p in m.group(2).split(b'$$')] choices = [p.strip(b' ') for p in m.group(2).split(b'$$')] # pytype: disable=attribute-error
def choicetuple(s): def choicetuple(s):
ampidx = s.index(b'&') ampidx = s.index(b'&')
return s[ampidx + 1 : ampidx + 2].lower(), s.replace(b'&', b'', 1) return s[ampidx + 1 : ampidx + 2].lower(), s.replace(b'&', b'', 1)
return (msg, [choicetuple(s) for s in choices]) return (msg, [choicetuple(s) for s in choices])
def promptchoice(self, prompt, default=0): def promptchoice(self, prompt, default=0):
if cause is not None: if cause is not None:
causetb = traceback.format_tb(cause[2]) causetb = traceback.format_tb(cause[2])
exctb = traceback.format_tb(exc[2]) exctb = traceback.format_tb(exc[2])
exconly = traceback.format_exception_only(cause[0], cause[1]) exconly = traceback.format_exception_only(cause[0], cause[1])
# exclude frame where 'exc' was chained and rethrown from exctb # exclude frame where 'exc' was chained and rethrown from exctb
self.write_err( self.write_err(
b'Traceback (most recent call last):\n', b'Traceback (most recent call last):\n',
b''.join(exctb[:-1]), b''.join(exctb[:-1]), # pytype: disable=wrong-arg-types
b''.join(causetb), b''.join(causetb), # pytype: disable=wrong-arg-types
b''.join(exconly), b''.join(exconly), # pytype: disable=wrong-arg-types
) )
else: else:
output = traceback.format_exception(exc[0], exc[1], exc[2]) output = traceback.format_exception(exc[0], exc[1], exc[2])
self.write_err(encoding.strtolocal(''.join(output))) self.write_err(encoding.strtolocal(''.join(output)))
return self.tracebackflag or force return self.tracebackflag or force
def geteditor(self): def geteditor(self):
'''return editor to use''' '''return editor to use'''

mercurial/unionrepo.py

Instances are not usable if constructed directly. Use ``instance()`` Instances are not usable if constructed directly. Use ``instance()``
or ``makeunionrepository()`` to create a usable instance. or ``makeunionrepository()`` to create a usable instance.
""" """
def __init__(self, repo2, url): def __init__(self, repo2, url):
self.repo2 = repo2 self.repo2 = repo2
self._url = url self._url = url
self.ui.setconfig(b'phases', b'publish', False, b'unionrepo') self.ui.setconfig(b'phases', b'publish', False, b'unionrepo') # pytype: disable=attribute-error
@localrepo.unfilteredpropertycache @localrepo.unfilteredpropertycache
def changelog(self): def changelog(self):
return unionchangelog(self.svfs, self.repo2.svfs) return unionchangelog(self.svfs, self.repo2.svfs)
@localrepo.unfilteredpropertycache @localrepo.unfilteredpropertycache
def manifestlog(self): def manifestlog(self):
rootstore = unionmanifest( rootstore = unionmanifest(
self.svfs, self.repo2.svfs, self.unfiltered()._clrev self.svfs, self.repo2.svfs, self.unfiltered()._clrev
) )
return manifest.manifestlog( return manifest.manifestlog(
self.svfs, self, rootstore, self.narrowmatch() self.svfs, self, rootstore, self.narrowmatch()
) )
def _clrev(self, rev2): def _clrev(self, rev2):
"""map from repo2 changelog rev to temporary rev in self.changelog""" """map from repo2 changelog rev to temporary rev in self.changelog"""
node = self.repo2.changelog.node(rev2) node = self.repo2.changelog.node(rev2)
return self.changelog.rev(node) return self.changelog.rev(node)
def url(self): def url(self):
return self._url return self._url
def file(self, f): def file(self, f):
return unionfilelog( return unionfilelog(
self.svfs, f, self.repo2.svfs, self.unfiltered()._clrev, self self.svfs, # pytype: disable=attribute-error
f,
self.repo2.svfs,
self.unfiltered()._clrev, # pytype: disable=attribute-error
self
) )
def close(self): def close(self):
self.repo2.close() self.repo2.close()
def cancopy(self): def cancopy(self):
return False return False

mercurial/upgrade.py

_(b'migrating %s in store; %s tracked data\n') _(b'migrating %s in store; %s tracked data\n')
% ((util.bytecount(srcsize), util.bytecount(srcrawsize))) % ((util.bytecount(srcsize), util.bytecount(srcrawsize)))
) )
# Used to keep track of progress. # Used to keep track of progress.
progress = None progress = None
def oncopiedrevision(rl, rev, node): def oncopiedrevision(rl, rev, node):
progress.increment() # XXX: It looks like there is no `else` where this is set, and it ends
# up calling progress.complete() at the end of the loop (not
# detected).
progress.increment() # pytype: disable=attribute-error
sidedatacompanion = getsidedatacompanion(srcrepo, dstrepo) sidedatacompanion = getsidedatacompanion(srcrepo, dstrepo)
# Do the actual copying. # Do the actual copying.
# FUTURE this operation can be farmed off to worker processes. # FUTURE this operation can be farmed off to worker processes.
seen = set() seen = set()
for unencoded, encoded, size in alldatafiles: for unencoded, encoded, size in alldatafiles:
if unencoded.endswith(b'.d'): if unencoded.endswith(b'.d'):

mercurial/verify.py

msg = b"%s: %s" % (linkrev, msg) msg = b"%s: %s" % (linkrev, msg)
if filename: if filename:
msg = b"%s@%s" % (filename, msg) msg = b"%s@%s" % (filename, msg)
self.ui.warn(b" " + msg + b"\n") self.ui.warn(b" " + msg + b"\n")
self.errors += 1 self.errors += 1
def _exc(self, linkrev, msg, inst, filename=None): def _exc(self, linkrev, msg, inst, filename=None):
"""record exception raised during the verify process""" """record exception raised during the verify process"""
fmsg = pycompat.bytestr(inst) fmsg = pycompat.bytestr(inst) # pytype: disable=wrong-arg-types
if not fmsg: if not fmsg:
fmsg = pycompat.byterepr(inst) fmsg = pycompat.byterepr(inst)
self._err(linkrev, b"%s: %s" % (msg, fmsg), filename) self._err(linkrev, b"%s: %s" % (msg, fmsg), filename)
def _checkrevlog(self, obj, name, linkrev): def _checkrevlog(self, obj, name, linkrev):
"""verify high level property of a revlog """verify high level property of a revlog
- revlog is present, - revlog is present,
for subdir, linkrevs in pycompat.iteritems(subdirnodes): for subdir, linkrevs in pycompat.iteritems(subdirnodes):
subdirfilenodes = self._verifymanifest( subdirfilenodes = self._verifymanifest(
linkrevs, subdir, storefiles, subdirprogress linkrevs, subdir, storefiles, subdirprogress
) )
for f, onefilenodes in pycompat.iteritems(subdirfilenodes): for f, onefilenodes in pycompat.iteritems(subdirfilenodes):
filenodes.setdefault(f, {}).update(onefilenodes) filenodes.setdefault(f, {}).update(onefilenodes)
if not dir and subdirnodes: if not dir and subdirnodes:
subdirprogress.complete() subdirprogress.complete() # pytype: disable=attribute-error
if self.warnorphanstorefiles: if self.warnorphanstorefiles:
for f in sorted(storefiles): for f in sorted(storefiles):
self._warn(_(b"warning: orphan data file '%s'") % f) self._warn(_(b"warning: orphan data file '%s'") % f)
return filenodes return filenodes
def _crosscheckfiles(self, filelinkrevs, filenodes): def _crosscheckfiles(self, filelinkrevs, filenodes):
repo = self.repo repo = self.repo

mercurial/windows.py

encoding, encoding,
error, error,
policy, policy,
pycompat, pycompat,
win32, win32,
) )
try: try:
import _winreg as winreg import _winreg as winreg # pytype: disable=import-error
winreg.CloseKey winreg.CloseKey
except ImportError: except ImportError:
import winreg # py2 only
import winreg # pytype: disable=import-error
osutil = policy.importmod('osutil') osutil = policy.importmod('osutil')
getfsmountpoint = win32.getvolumename getfsmountpoint = win32.getvolumename
getfstype = win32.getfstype getfstype = win32.getfstype
getuser = win32.getuser getuser = win32.getuser
hidewindow = win32.hidewindow hidewindow = win32.hidewindow
makedir = win32.makedir makedir = win32.makedir
return False return False
def setbinary(fd): def setbinary(fd):
# When run without console, pipes may expose invalid # When run without console, pipes may expose invalid
# fileno(), usually set to -1. # fileno(), usually set to -1.
fno = getattr(fd, 'fileno', None) fno = getattr(fd, 'fileno', None)
if fno is not None and fno() >= 0: if fno is not None and fno() >= 0:
msvcrt.setmode(fno(), os.O_BINARY) msvcrt.setmode(fno(), os.O_BINARY) # pytype: disable=module-attr
def pconvert(path): def pconvert(path):
return path.replace(pycompat.ossep, b'/') return path.replace(pycompat.ossep, b'/')
def localpath(path): def localpath(path):
return path.replace(b'/', b'\\') return path.replace(b'/', b'\\')

mercurial/wireprotoframing.py

'size', 'size',
'fullhashes', 'fullhashes',
'fullhashseed', 'fullhashseed',
'serverdercerts', 'serverdercerts',
'servercadercerts', 'servercadercerts',
): ):
value = getattr(location, a) value = getattr(location, a)
if value is not None: if value is not None:
data[b'location'][pycompat.bytestr(a)] = value data[b'location'][pycompat.bytestr(a
)] = value # pytype: disable=unsupported-operands,wrong-arg-types
payload = b''.join(cborutil.streamencode(data)) payload = b''.join(cborutil.streamencode(data))
if stream.streamsettingssent: if stream.streamsettingssent:
payload = stream.encode(payload) payload = stream.encode(payload)
encoded = True encoded = True
else: else:
encoded = False encoded = False
return stream.makeframe( return stream.makeframe(
requestid=requestid, requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE, typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
payload=payload, payload=payload,
encoded=encoded, encoded=encoded,
) )
def createcommanderrorresponse(stream, requestid, message, args=None): def createcommanderrorresponse(stream, requestid, message, args=None):
# TODO should this be using a list of {'msg': ..., 'args': {}} so atom # TODO should this be using a list of {'msg': ..., 'args': {}} so atom
# formatting works consistently? # formatting works consistently?
m = {b'status': b'error', b'error': {b'message': message,}} m = {b'status': b'error', b'error': {b'message': message,}}
if args: if args:
m[b'error'][b'args'] = args m[b'error'][b'args'] = args # pytype: disable=unsupported-operands
overall = b''.join(cborutil.streamencode(m)) overall = b''.join(cborutil.streamencode(m))
yield stream.makeframe( yield stream.makeframe(
requestid=requestid, requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE, typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_EOS, flags=FLAG_COMMAND_RESPONSE_EOS,
payload=overall, payload=overall,
if not pycompat.ispy3 and isinstance(data, bytearray): if not pycompat.ispy3 and isinstance(data, bytearray):
data = bytes(data) data = bytes(data)
return self._decompressor.decompress(data) return self._decompressor.decompress(data)
class zstdbaseencoder(object): class zstdbaseencoder(object):
def __init__(self, level): def __init__(self, level):
from . import zstd from . import zstd # pytype: disable=import-error
self._zstd = zstd self._zstd = zstd
cctx = zstd.ZstdCompressor(level=level) cctx = zstd.ZstdCompressor(level=level)
self._compressor = cctx.compressobj() self._compressor = cctx.compressobj()
def encode(self, data): def encode(self, data):
return self._compressor.compress(data) return self._compressor.compress(data)
class zstd8mbencoder(zstdbaseencoder): class zstd8mbencoder(zstdbaseencoder):
def __init__(self, ui): def __init__(self, ui):
super(zstd8mbencoder, self).__init__(3) super(zstd8mbencoder, self).__init__(3)
class zstdbasedecoder(object): class zstdbasedecoder(object):
def __init__(self, maxwindowsize): def __init__(self, maxwindowsize):
from . import zstd from . import zstd # pytype: disable=import-error
dctx = zstd.ZstdDecompressor(max_window_size=maxwindowsize) dctx = zstd.ZstdDecompressor(max_window_size=maxwindowsize)
self._decompressor = dctx.decompressobj() self._decompressor = dctx.decompressobj()
def decode(self, data): def decode(self, data):
return self._decompressor.decompress(data) return self._decompressor.decompress(data)
STREAM_ENCODERS_ORDER = [] STREAM_ENCODERS_ORDER = []
def populatestreamencoders(): def populatestreamencoders():
if STREAM_ENCODERS: if STREAM_ENCODERS:
return return
try: try:
from . import zstd from . import zstd # pytype: disable=import-error
zstd.__version__ zstd.__version__
except ImportError: except ImportError:
zstd = None zstd = None
# zstandard is fastest and is preferred. # zstandard is fastest and is preferred.
if zstd: if zstd:
STREAM_ENCODERS[b'zstd-8mb'] = (zstd8mbencoder, zstd8mbdecoder) STREAM_ENCODERS[b'zstd-8mb'] = (zstd8mbencoder, zstd8mbdecoder)
del self._receivingcommands[requestid] del self._receivingcommands[requestid]
if self._receivingcommands: if self._receivingcommands:
self._state = b'command-receiving' self._state = b'command-receiving'
else: else:
self._state = b'idle' self._state = b'idle'
# Decode the payloads as CBOR. # Decode the payloads as CBOR.
entry[b'payload'].seek(0) entry[b'payload'].seek(0) # pytype: disable=attribute-error
request = cborutil.decodeall(entry[b'payload'].getvalue())[0] request = cborutil.decodeall(entry[b'payload'].getvalue())[0] # pytype: disable=attribute-error
if b'name' not in request: if b'name' not in request:
self._state = b'errored' self._state = b'errored'
return self._makeerrorresult( return self._makeerrorresult(
_(b'command request missing "name" field') _(b'command request missing "name" field')
) )
if b'args' not in request: if b'args' not in request:
request[b'args'] = {} request[b'args'] = {}
assert requestid not in self._activecommands assert requestid not in self._activecommands
self._activecommands.add(requestid) self._activecommands.add(requestid)
return ( return (
b'runcommand', b'runcommand',
{ {
b'requestid': requestid, b'requestid': requestid,
b'command': request[b'name'], b'command': request[b'name'],
b'args': request[b'args'], b'args': request[b'args'],
b'redirect': request.get(b'redirect'), b'redirect': request.get(b'redirect'),
b'data': entry[b'data'].getvalue() if entry[b'data'] else None, b'data': entry[b'data'].getvalue() if entry[b'data'] else None, # pytype: disable=attribute-error
}, },
) )
def _makewantframeresult(self): def _makewantframeresult(self):
return b'wantframe', {b'state': self._state,} return b'wantframe', {b'state': self._state,}
def _validatecommandrequestframe(self, frame): def _validatecommandrequestframe(self, frame):
new = frame.flags & FLAG_COMMAND_REQUEST_NEW new = frame.flags & FLAG_COMMAND_REQUEST_NEW

mercurial/wireprotoserver.py

# ErrorResponse. It is kind of a layer violation to catch an hgweb # ErrorResponse. It is kind of a layer violation to catch an hgweb
# exception here. So consider refactoring into a exception type that # exception here. So consider refactoring into a exception type that
# is associated with the wire protocol. # is associated with the wire protocol.
try: try:
_callhttp(repo, req, res, proto, cmd) _callhttp(repo, req, res, proto, cmd)
except hgwebcommon.ErrorResponse as e: except hgwebcommon.ErrorResponse as e:
for k, v in e.headers: for k, v in e.headers:
res.headers[k] = v res.headers[k] = v
res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e)) res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e)) # pytype: disable=wrong-arg-types
# TODO This response body assumes the failed command was # TODO This response body assumes the failed command was
# "unbundle." That assumption is not always valid. # "unbundle." That assumption is not always valid.
res.setbodybytes(b'0\n%s\n' % pycompat.bytestr(e)) res.setbodybytes(b'0\n%s\n' % pycompat.bytestr(e)) # pytype: disable=wrong-arg-types
return True return True
def _availableapis(repo): def _availableapis(repo):
apis = set() apis = set()
# Registered APIs are made available via config options of the name of # Registered APIs are made available via config options of the name of
# the protocol. # the protocol.
for k, v in API_HANDLERS.items(): for k, v in API_HANDLERS.items():
section, option = v[b'config'] section, option = v[b'config'] # pytype: disable=attribute-error
if repo.ui.configbool(section, option): if repo.ui.configbool(section, option):
apis.add(k) apis.add(k)
return apis return apis
def handlewsgiapirequest(rctx, req, res, checkperm): def handlewsgiapirequest(rctx, req, res, checkperm):
"""Handle requests to /api/*.""" """Handle requests to /api/*."""

mercurial/wireprotov1peer.py

The decorator returns a function which wraps this coroutine as a plain The decorator returns a function which wraps this coroutine as a plain
method, but adds the original method as an attribute called "batchable", method, but adds the original method as an attribute called "batchable",
which is used by remotebatch to split the call into separate encoding and which is used by remotebatch to split the call into separate encoding and
decoding phases. decoding phases.
''' '''
def plain(*args, **opts): def plain(*args, **opts):
batchable = f(*args, **opts) batchable = f(*args, **opts)
encargsorres, encresref = next(batchable) encargsorres, encresref = next(batchable) # pytype: disable=attribute-error
if not encresref: if not encresref:
return encargsorres # a local result in this case return encargsorres # a local result in this case
self = args[0] self = args[0]
cmd = pycompat.bytesurl(f.__name__) # ensure cmd is ascii bytestr cmd = pycompat.bytesurl(f.__name__) # ensure cmd is ascii bytestr
encresref.set(self._submitone(cmd, encargsorres)) encresref.set(self._submitone(cmd, encargsorres)) # pytype: disable=attribute-error
return next(batchable) return next(batchable)
setattr(plain, 'batchable', f) setattr(plain, 'batchable', f)
setattr(plain, '__name__', f.__name__) setattr(plain, '__name__', f.__name__)
return plain return plain
class future(object): class future(object):
self.requirecap(b'clonebundles', _(b'clone bundles')) self.requirecap(b'clonebundles', _(b'clone bundles'))
return self._call(b'clonebundles') return self._call(b'clonebundles')
@batchable @batchable
def lookup(self, key): def lookup(self, key):
self.requirecap(b'lookup', _(b'look up remote revision')) self.requirecap(b'lookup', _(b'look up remote revision'))
f = future() f = future()
yield {b'key': encoding.fromlocal(key)}, f yield {b'key': encoding.fromlocal(key)}, f
d = f.value d = f.value # pytype: disable=attribute-error
success, data = d[:-1].split(b" ", 1) success, data = d[:-1].split(b" ", 1)
if int(success): if int(success):
yield bin(data) yield bin(data)
else: else:
self._abort(error.RepoError(data)) self._abort(error.RepoError(data))
@batchable @batchable
def heads(self): def heads(self):
f = future() f = future()
yield {}, f yield {}, f
d = f.value d = f.value # pytype: disable=attribute-error
try: try:
yield wireprototypes.decodelist(d[:-1]) yield wireprototypes.decodelist(d[:-1])
except ValueError: except ValueError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
@batchable @batchable
def known(self, nodes): def known(self, nodes):
f = future() f = future()
yield {b'nodes': wireprototypes.encodelist(nodes)}, f yield {b'nodes': wireprototypes.encodelist(nodes)}, f
d = f.value d = f.value # pytype: disable=attribute-error
try: try:
yield [bool(int(b)) for b in pycompat.iterbytestr(d)] yield [bool(int(b)) for b in pycompat.iterbytestr(d)]
except ValueError: except ValueError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
@batchable @batchable
def branchmap(self): def branchmap(self):
f = future() f = future()
yield {}, f yield {}, f
d = f.value d = f.value # pytype: disable=attribute-error
try: try:
branchmap = {} branchmap = {}
for branchpart in d.splitlines(): for branchpart in d.splitlines():
branchname, branchheads = branchpart.split(b' ', 1) branchname, branchheads = branchpart.split(b' ', 1)
branchname = encoding.tolocal(urlreq.unquote(branchname)) branchname = encoding.tolocal(urlreq.unquote(branchname))
branchheads = wireprototypes.decodelist(branchheads) branchheads = wireprototypes.decodelist(branchheads)
branchmap[branchname] = branchheads branchmap[branchname] = branchheads
yield branchmap yield branchmap
except TypeError: except TypeError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
@batchable @batchable
def listkeys(self, namespace): def listkeys(self, namespace):
if not self.capable(b'pushkey'): if not self.capable(b'pushkey'):
yield {}, None yield {}, None
f = future() f = future()
self.ui.debug(b'preparing listkeys for "%s"\n' % namespace) self.ui.debug(b'preparing listkeys for "%s"\n' % namespace)
yield {b'namespace': encoding.fromlocal(namespace)}, f yield {b'namespace': encoding.fromlocal(namespace)}, f
d = f.value d = f.value # pytype: disable=attribute-error
self.ui.debug( self.ui.debug(
b'received listkey for "%s": %i bytes\n' % (namespace, len(d)) b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
) )
yield pushkeymod.decodekeys(d) yield pushkeymod.decodekeys(d)
@batchable @batchable
def pushkey(self, namespace, key, old, new): def pushkey(self, namespace, key, old, new):
if not self.capable(b'pushkey'): if not self.capable(b'pushkey'):
yield False, None yield False, None
f = future() f = future()
self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key)) self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key))
yield { yield {
b'namespace': encoding.fromlocal(namespace), b'namespace': encoding.fromlocal(namespace),
b'key': encoding.fromlocal(key), b'key': encoding.fromlocal(key),
b'old': encoding.fromlocal(old), b'old': encoding.fromlocal(old),
b'new': encoding.fromlocal(new), b'new': encoding.fromlocal(new),
}, f }, f
d = f.value d = f.value # pytype: disable=attribute-error
d, output = d.split(b'\n', 1) d, output = d.split(b'\n', 1)
try: try:
d = bool(int(d)) d = bool(int(d))
except ValueError: except ValueError:
raise error.ResponseError( raise error.ResponseError(
_(b'push failed (unexpected response):'), d _(b'push failed (unexpected response):'), d
) )
for l in output.splitlines(True): for l in output.splitlines(True):

mercurial/wireprotov1server.py

if exc.key is not None: if exc.key is not None:
part.addparam(b'key', exc.key, mandatory=False) part.addparam(b'key', exc.key, mandatory=False)
if exc.new is not None: if exc.new is not None:
part.addparam(b'new', exc.new, mandatory=False) part.addparam(b'new', exc.new, mandatory=False)
if exc.old is not None: if exc.old is not None:
part.addparam(b'old', exc.old, mandatory=False) part.addparam(b'old', exc.old, mandatory=False)
if exc.ret is not None: if exc.ret is not None:
part.addparam(b'ret', exc.ret, mandatory=False) part.addparam(b'ret', exc.ret, mandatory=False)
except error.BundleValueError as exc: except error.BundleValueError as exc: # Should be BundleUnknownFeatureError
errpart = bundler.newpart(b'error:unsupportedcontent') errpart = bundler.newpart(b'error:unsupportedcontent')
if exc.parttype is not None: # TODO: These are bugs!
errpart.addparam(b'parttype', exc.parttype) if exc.parttype is not None: # pytype: disable=attribute-error
if exc.params: errpart.addparam(b'parttype', exc.parttype) # pytype: disable=attribute-error
errpart.addparam(b'params', b'\0'.join(exc.params)) if exc.params: # pytype: disable=attribute-error
errpart.addparam(b'params', b'\0'.join(exc.params)) # pytype: disable=attribute-error
except error.Abort as exc: except error.Abort as exc:
manargs = [(b'message', stringutil.forcebytestr(exc))] manargs = [(b'message', stringutil.forcebytestr(exc))]
advargs = [] advargs = []
if exc.hint is not None: if exc.hint is not None:
advargs.append((b'hint', exc.hint)) advargs.append((b'hint', exc.hint))
bundler.addpart( bundler.addpart(
bundle2.bundlepart(b'error:abort', manargs, advargs) bundle2.bundlepart(b'error:abort', manargs, advargs)
) )
except error.PushRaced as exc: except error.PushRaced as exc:
bundler.newpart( bundler.newpart(
b'error:pushraced', b'error:pushraced',
[(b'message', stringutil.forcebytestr(exc))], [(b'message', stringutil.forcebytestr(exc))],
) )
return wireprototypes.streamreslegacy(gen=bundler.getchunks()) return wireprototypes.streamreslegacy(gen=bundler.getchunks())

mercurial/wireprotov2server.py

# At some point we'll want to use our own API instead of recycling the # At some point we'll want to use our own API instead of recycling the
# behavior of version 1 of the wire protocol... # behavior of version 1 of the wire protocol...
# TODO return reasonable responses - not responses that overload the # TODO return reasonable responses - not responses that overload the
# HTTP status line message for error reporting. # HTTP status line message for error reporting.
try: try:
checkperm(rctx, req, b'pull' if permission == b'ro' else b'push') checkperm(rctx, req, b'pull' if permission == b'ro' else b'push')
except hgwebcommon.ErrorResponse as e: except hgwebcommon.ErrorResponse as e:
res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e)) res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e)) # pytype: disable=wrong-arg-types
for k, v in e.headers: for k, v in e.headers:
res.headers[k] = v res.headers[k] = v
res.setbodybytes(b'permission denied') res.setbodybytes(b'permission denied')
return return
# We have a special endpoint to reflect the request back at the client. # We have a special endpoint to reflect the request back at the client.
if command == b'debugreflect': if command == b'debugreflect':
_processhttpv2reflectrequest(rctx.repo.ui, rctx.repo, req, res) _processhttpv2reflectrequest(rctx.repo.ui, rctx.repo, req, res)
if ( if (
command == b'rawstorefiledata' command == b'rawstorefiledata'
and not streamclone.allowservergeneration(repo) and not streamclone.allowservergeneration(repo)
): ):
continue continue
caps[b'commands'][command] = { caps[b'commands'][command] = {
b'args': args, b'args': args,
b'permissions': [entry.permission], b'permissions': [entry.permission], # pytype: disable=unsupported-operands
} }
if entry.extracapabilitiesfn: if entry.extracapabilitiesfn:
extracaps = entry.extracapabilitiesfn(repo, proto) extracaps = entry.extracapabilitiesfn(repo, proto)
caps[b'commands'][command].update(extracaps) caps[b'commands'][command].update(extracaps)
caps[b'rawrepoformats'] = sorted(repo.requirements & repo.supportedformats) caps[b'rawrepoformats'] = sorted(repo.requirements & repo.supportedformats)
b'protocol': target[b'protocol'], b'protocol': target[b'protocol'],
b'uris': target[b'uris'], b'uris': target[b'uris'],
} }
for key in (b'snirequired', b'tlsversions'): for key in (b'snirequired', b'tlsversions'):
if key in target: if key in target:
entry[key] = target[key] entry[key] = target[key]
caps[b'redirect'][b'targets'].append(entry) caps[b'redirect'][b'targets'].append(entry) # pytype: disable=attribute-error
return proto.addcapabilities(repo, caps) return proto.addcapabilities(repo, caps)
def getadvertisedredirecttargets(repo, proto): def getadvertisedredirecttargets(repo, proto):
"""Obtain a list of content redirect targets. """Obtain a list of content redirect targets.
Returns a list containing potential redirect targets that will be Returns a list containing potential redirect targets that will be

tests/test-check-pytype.t

#require pytype py3 slow #require pytype py3 slow
$ cd $RUNTESTDIR/.. $ cd $RUNTESTDIR/..
Many of the individual files that are excluded here confuse pytype Many of the individual files that are excluded here confuse pytype
because they do a mix of Python 2 and Python 3 things because they do a mix of Python 2 and Python 3 things
conditionally. There's no good way to help it out with that as far as conditionally. There's no good way to help it out with that as far as
I can tell, so let's just hide those files from it for now. We should I can tell, so let's just hide those files from it for now. We should
endeavor to empty this list out over time, as some of these are endeavor to empty this list out over time, as some of these are
probably hiding real problems. probably hiding real problems.
$ pytype -k mercurial \
$ pytype mercurial \
> -x mercurial/bundlerepo.py \
> -x mercurial/chgserver.py \ > -x mercurial/chgserver.py \
> -x mercurial/context.py \ > -x mercurial/context.py \
> -x mercurial/crecord.py \ > -x mercurial/crecord.py \
> -x mercurial/encoding.py \ > -x mercurial/encoding.py \
> -x mercurial/error.py \
> -x mercurial/exchange.py \
> -x mercurial/lsprof.py \ > -x mercurial/lsprof.py \
> -x mercurial/policy.py
> -x mercurial/pycompat.py \
> -x mercurial/urllibcompat.py \ > -x mercurial/urllibcompat.py \
> -x mercurial/i18n.py \ > -x mercurial/i18n.py \
> -x mercurial/sslutil.py \ > -x mercurial/sslutil.py \
> -x mercurial/scmwindows.py \
> -x mercurial/keepalive.py \ > -x mercurial/keepalive.py \
> -x mercurial/windows.py \
> -x mercurial/wireprotoframing.py \
> -x mercurial/utils/stringutil.py \ > -x mercurial/utils/stringutil.py \
> -x mercurial/hgweb/server.py \ > -x mercurial/hgweb/server.py \
> -x mercurial/hgweb/wsgicgi.py \ > -x mercurial/hgweb/wsgicgi.py \
> -x mercurial/minirst.py \ > -x mercurial/minirst.py \
> -x mercurial/interfaces \ > -x mercurial/interfaces \
> -x mercurial/cffi \ > -x mercurial/cffi \
> -x mercurial/pure \ > -x mercurial/pure \
> -x mercurial/thirdparty > -x mercurial/thirdparty \
> -x mercurial/httppeer.py \
> -x mercurial/repoview.py \
> -x mercurial/localrepo.py \
> -x mercurial/revlog.py \
> -x mercurial/merge.py \
> -x mercurial/testing/storage.py