mercurial-scm/hg-stable: mercurial/context.py comparison

comparison mercurial/context.py @ 43077:687b865b95ad

formatting: byteify all mercurial/ and hgext/ string literals Done with python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py') black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**') # skip-blame mass-reformatting only Differential Revision: https://phab.mercurial-scm.org/D6972

author	Augie Fackler <augie@google.com>
date	Sun, 06 Oct 2019 09:48:39 -0400
parents	2372284d9457
children	eef9a2d67051

comparison

equal deleted inserted replaced

-:2372284d9457
+:687b865b95ad
 def matchfileset(self, expr, badfn=None):
 return fileset.match(self, expr, badfn=badfn)
 def obsolete(self):
 """True if the changeset is obsolete"""
-return self.rev() in obsmod.getrevs(self._repo, 'obsolete')
+return self.rev() in obsmod.getrevs(self._repo, b'obsolete')
 def extinct(self):
 """True if the changeset is extinct"""
-return self.rev() in obsmod.getrevs(self._repo, 'extinct')
+return self.rev() in obsmod.getrevs(self._repo, b'extinct')
 def orphan(self):
 """True if the changeset is not obsolete, but its ancestor is"""
-return self.rev() in obsmod.getrevs(self._repo, 'orphan')
+return self.rev() in obsmod.getrevs(self._repo, b'orphan')
 def phasedivergent(self):
 """True if the changeset tries to be a successor of a public changeset
 Only non-public and non-obsolete changesets may be phase-divergent.
 """
-return self.rev() in obsmod.getrevs(self._repo, 'phasedivergent')
+return self.rev() in obsmod.getrevs(self._repo, b'phasedivergent')
 def contentdivergent(self):
 """Is a successor of a changeset with multiple possible successor sets
 Only non-public and non-obsolete changesets may be content-divergent.
 """
-return self.rev() in obsmod.getrevs(self._repo, 'contentdivergent')
+return self.rev() in obsmod.getrevs(self._repo, b'contentdivergent')
 def isunstable(self):
 """True if the changeset is either orphan, phase-divergent or
 content-divergent"""
 return self.orphan() or self.phasedivergent() or self.contentdivergent()
 - phase-divergent,
 - content-divergent.
 """
 instabilities = []
 if self.orphan():
-instabilities.append('orphan')
+instabilities.append(b'orphan')
 if self.phasedivergent():
-instabilities.append('phase-divergent')
+instabilities.append(b'phase-divergent')
 if self.contentdivergent():
-instabilities.append('content-divergent')
+instabilities.append(b'content-divergent')
 return instabilities
 def parents(self):
 """return contexts for each parent changeset"""
 return self._parents
 if r'_manifest' in self.__dict__:
 try:
 return self._manifest[path], self._manifest.flags(path)
 except KeyError:
 raise error.ManifestLookupError(
-self._node, path, _('not found in manifest')
+self._node, path, _(b'not found in manifest')
 )
 if r'_manifestdelta' in self.__dict__ or path in self.files():
 if path in self._manifestdelta:
 return (
 self._manifestdelta[path],
 mfl = self._repo.manifestlog
 try:
 node, flag = mfl[self._changeset.manifest].find(path)
 except KeyError:
 raise error.ManifestLookupError(
-self._node, path, _('not found in manifest')
+self._node, path, _(b'not found in manifest')
 )
 return node, flag
 def filenode(self, path):
 def flags(self, path):
 try:
 return self._fileinfo(path)[1]
 except error.LookupError:
-return ''
+return b''
 @propertycache
 def _copies(self):
 return copies.computechangesetcopies(self)
 def match(
 self,
 pats=None,
 include=None,
 exclude=None,
-default='glob',
+default=b'glob',
 listsubrepos=False,
 badfn=None,
 ):
 r = self._repo
 return matchmod.match(
 clean=listclean,
 unknown=listunknown,
 listsubrepos=True,
 )
 for rfiles, sfiles in zip(r, s):
-rfiles.extend("%s/%s" % (subpath, f) for f in sfiles)
+rfiles.extend(b"%s/%s" % (subpath, f) for f in sfiles)
 for l in r:
 l.sort()
 return r
 modified.difference_update(self.filesadded())
 modified.difference_update(self.filesremoved())
 return sorted(modified)
 def filesadded(self):
-source = self._repo.ui.config('experimental', 'copies.read-from')
+source = self._repo.ui.config(b'experimental', b'copies.read-from')
 filesadded = self._changeset.filesadded
-if source == 'changeset-only':
+if source == b'changeset-only':
 if filesadded is None:
 filesadded = []
-elif source == 'compatibility':
+elif source == b'compatibility':
 if filesadded is None:
 filesadded = scmutil.computechangesetfilesadded(self)
 else:
 filesadded = scmutil.computechangesetfilesadded(self)
 return filesadded
 def filesremoved(self):
-source = self._repo.ui.config('experimental', 'copies.read-from')
+source = self._repo.ui.config(b'experimental', b'copies.read-from')
 filesremoved = self._changeset.filesremoved
-if source == 'changeset-only':
+if source == b'changeset-only':
 if filesremoved is None:
 filesremoved = []
-elif source == 'compatibility':
+elif source == b'compatibility':
 if filesremoved is None:
 filesremoved = scmutil.computechangesetfilesremoved(self)
 else:
 filesremoved = scmutil.computechangesetfilesremoved(self)
 return filesremoved
 @propertycache
 def _copies(self):
-source = self._repo.ui.config('experimental', 'copies.read-from')
+source = self._repo.ui.config(b'experimental', b'copies.read-from')
 p1copies = self._changeset.p1copies
 p2copies = self._changeset.p2copies
 # If config says to get copy metadata only from changeset, then return
 # that, defaulting to {} if there was no copy metadata.
 # In compatibility mode, we return copy data from the changeset if
 # it was recorded there, and otherwise we fall back to getting it from
 # the filelogs (below).
-if source == 'changeset-only':
+if source == b'changeset-only':
 if p1copies is None:
 p1copies = {}
 if p2copies is None:
 p2copies = {}
-elif source == 'compatibility':
+elif source == b'compatibility':
 if p1copies is None:
 # we are in compatiblity mode and there is not data in the
 # changeset), we get the copy metadata from the filelogs.
 p1copies, p2copies = super(changectx, self)._copies
 else:
 def description(self):
 return self._changeset.description
 def branch(self):
-return encoding.tolocal(self._changeset.extra.get("branch"))
+return encoding.tolocal(self._changeset.extra.get(b"branch"))
 def closesbranch(self):
-return 'close' in self._changeset.extra
+return b'close' in self._changeset.extra
 def extra(self):
 """Return a dict of extra information."""
 return self._changeset.extra
 def phase(self):
 return self._repo._phasecache.phase(self._repo, self._rev)
 def hidden(self):
-return self._rev in repoview.filterrevs(self._repo, 'visible')
+return self._rev in repoview.filterrevs(self._repo, b'visible')
 def isinmemory(self):
 return False
 def children(self):
 anc = nullid
 elif len(cahs) == 1:
 anc = cahs[0]
 else:
 # experimental config: merge.preferancestor
-for r in self._repo.ui.configlist('merge', 'preferancestor'):
+for r in self._repo.ui.configlist(b'merge', b'preferancestor'):
 try:
 ctx = scmutil.revsymbol(self._repo, r)
 except error.RepoLookupError:
 continue
 anc = ctx.node()
 else:
 anc = self._repo.changelog.ancestor(self._node, n2)
 if warn:
 self._repo.ui.status(
 (
-_("note: using %s as ancestor of %s and %s\n")
+_(b"note: using %s as ancestor of %s and %s\n")
 % (short(anc), short(self._node), short(n2))
 )
-+ ''.join(
++ b''.join(
 _(
-"      alternatively, use --config "
+b"      alternatively, use --config "
-"merge.preferancestor=%s\n"
+b"merge.preferancestor=%s\n"
 )
 % short(n)
 for n in sorted(cahs)
 if n != anc
 )
 # Wrap match.bad method to have message with nodeid
 def bad(fn, msg):
 # The manifest doesn't know about subrepos, so don't complain about
 # paths into valid subrepos.
-if any(fn == s or fn.startswith(s + '/') for s in self.substate):
+if any(fn == s or fn.startswith(s + b'/') for s in self.substate):
 return
-match.bad(fn, _('no such file in rev %s') % self)
+match.bad(fn, _(b'no such file in rev %s') % self)
 m = matchmod.badmatch(self._repo.narrowmatch(match), bad)
 return self._manifest.walk(m)
 def matches(self, match):
 __bool__ = __nonzero__
 def __bytes__(self):
 try:
-return "%s@%s" % (self.path(), self._changectx)
+return b"%s@%s" % (self.path(), self._changectx)
 except error.LookupError:
-return "%s@???" % self.path()
+return b"%s@???" % self.path()
 __str__ = encoding.strmethod(__bytes__)
 def __repr__(self):
 return r"<%s %s>" % (type(self).__name__, str(self))
 return stringutil.binary(self.data())
 except IOError:
 return False
 def isexec(self):
-return 'x' in self.flags()
+return b'x' in self.flags()
 def islink(self):
-return 'l' in self.flags()
+return b'l' in self.flags()
 def isabsent(self):
 """whether this filectx represents a file not in self._changectx
 This is mainly for merge code to detect change/delete conflicts. This is
 if fctx._customcmp:
 return fctx.cmp(self)
 if self._filenode is None:
 raise error.ProgrammingError(
-'filectx.cmp() must be reimplemented if not backed by revlog'
+b'filectx.cmp() must be reimplemented if not backed by revlog'
 )
 if fctx._filenode is None:
 if self._repo._encodefilterpats:
 # can't rely on size() because wdir content may be decoded
 self._repo = repo
 self._path = path
 assert (
 changeid is not None or fileid is not None or changectx is not None
-), "bad args: changeid=%r, fileid=%r, changectx=%r" % (
+), b"bad args: changeid=%r, fileid=%r, changectx=%r" % (
 changeid,
 fileid,
 changectx,
 )
 def data(self):
 try:
 return self._filelog.read(self._filenode)
 except error.CensoredNodeError:
-if self._repo.ui.config("censor", "policy") == "ignore":
+if self._repo.ui.config(b"censor", b"policy") == b"ignore":
-return ""
+return b""
 raise error.Abort(
-_("censored node: %s") % short(self._filenode),
+_(b"censored node: %s") % short(self._filenode),
-hint=_("set censor.policy to ignore errors"),
+hint=_(b"set censor.policy to ignore errors"),
 )
 def size(self):
 return self._filelog.size(self._filerev)
 wants the ability to commit, e.g. workingctx or memctx."""
 def __init__(
 self,
 repo,
-text="",
+text=b"",
 user=None,
 date=None,
 extra=None,
 changes=None,
 branch=None,
 self._extra = {}
 if extra:
 self._extra = extra.copy()
 if branch is not None:
-self._extra['branch'] = encoding.fromlocal(branch)
+self._extra[b'branch'] = encoding.fromlocal(branch)
-if not self._extra.get('branch'):
+if not self._extra.get(b'branch'):
-self._extra['branch'] = 'default'
+self._extra[b'branch'] = b'default'
 def __bytes__(self):
-return bytes(self._parents[0]) + "+"
+return bytes(self._parents[0]) + b"+"
 __str__ = encoding.strmethod(__bytes__)
 def __nonzero__(self):
 return True
 return self._repo.ui.username()
 @propertycache
 def _date(self):
 ui = self._repo.ui
-date = ui.configdate('devel', 'default-date')
+date = ui.configdate(b'devel', b'default-date')
 if date is None:
 date = dateutil.makedate()
 return date
 def subrev(self, subpath):
 filesmodified = modified
 filesadded = added
 filesremoved = removed
 def branch(self):
-return encoding.tolocal(self._extra['branch'])
+return encoding.tolocal(self._extra[b'branch'])
 def closesbranch(self):
-return 'close' in self._extra
+return b'close' in self._extra
 def extra(self):
 return self._extra
 def isinmemory(self):
 changes - a list of file lists as returned by localrepo.status()
 or None to use the repository status.
 """
 def __init__(
-self, repo, text="", user=None, date=None, extra=None, changes=None
+self, repo, text=b"", user=None, date=None, extra=None, changes=None
 ):
 branch = None
-if not extra or 'branch' not in extra:
+if not extra or b'branch' not in extra:
 try:
 branch = repo.dirstate.branch()
 except UnicodeDecodeError:
-raise error.Abort(_('branch name not in UTF-8!'))
+raise error.Abort(_(b'branch name not in UTF-8!'))
 super(workingctx, self).__init__(
 repo, text, user, date, extra, changes, branch=branch
 )
 def __iter__(self):
 d = self._repo.dirstate
 for f in d:
-if d[f] != 'r':
+if d[f] != b'r':
 yield f
 def __contains__(self, key):
-return self._repo.dirstate[key] not in "?r"
+return self._repo.dirstate[key] not in b"?r"
 def hex(self):
 return wdirhex
 @propertycache
 return fl1
 if fl1 == fla:
 return fl2
 if fl2 == fla:
 return fl1
-return ''  # punt for conflicts
+return b''  # punt for conflicts
 return func
 @propertycache
 def _flagfunc(self):
 def flags(self, path):
 if r'_manifest' in self.__dict__:
 try:
 return self._manifest.flags(path)
 except KeyError:
-return ''
+return b''
 try:
 return self._flagfunc(path)
 except OSError:
-return ''
+return b''
 def filectx(self, path, filelog=None):
 """get a file context from the working directory"""
 return workingfilectx(
 self._repo, path, workingctx=self, filelog=filelog
 )
 def dirty(self, missing=False, merge=True, branch=True):
-"check whether a working directory is modified"
+b"check whether a working directory is modified"
 # check subrepos first
 for s in sorted(self.substate):
 if self.sub(s).dirty(missing=missing):
 return True
 # check current working dir
 or self.added()
 or self.removed()
 or (missing and self.deleted())
 )
-def add(self, list, prefix=""):
+def add(self, list, prefix=b""):
 with self._repo.wlock():
 ui, ds = self._repo.ui, self._repo.dirstate
 uipath = lambda f: ds.pathto(pathutil.join(prefix, f))
 rejected = []
 lstat = self._repo.wvfs.lstat
 # Windows, since it contains the drive letter and colon.
 scmutil.checkportable(ui, os.path.join(prefix, f))
 try:
 st = lstat(f)
 except OSError:
-ui.warn(_("%s does not exist!\n") % uipath(f))
+ui.warn(_(b"%s does not exist!\n") % uipath(f))
 rejected.append(f)
 continue
-limit = ui.configbytes('ui', 'large-file-limit')
+limit = ui.configbytes(b'ui', b'large-file-limit')
 if limit != 0 and st.st_size > limit:
 ui.warn(
 _(
-"%s: up to %d MB of RAM may be required "
+b"%s: up to %d MB of RAM may be required "
-"to manage this file\n"
+b"to manage this file\n"
-"(use 'hg revert %s' to cancel the "
+b"(use 'hg revert %s' to cancel the "
-"pending addition)\n"
+b"pending addition)\n"
 )
 % (f, 3 * st.st_size // 1000000, uipath(f))
 )
 if not (stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode)):
 ui.warn(
 _(
-"%s not added: only files and symlinks "
+b"%s not added: only files and symlinks "
-"supported currently\n"
+b"supported currently\n"
 )
 % uipath(f)
 )
 rejected.append(f)
-elif ds[f] in 'amn':
+elif ds[f] in b'amn':
-ui.warn(_("%s already tracked!\n") % uipath(f))
+ui.warn(_(b"%s already tracked!\n") % uipath(f))
-elif ds[f] == 'r':
+elif ds[f] == b'r':
 ds.normallookup(f)
 else:
 ds.add(f)
 return rejected
-def forget(self, files, prefix=""):
+def forget(self, files, prefix=b""):
 with self._repo.wlock():
 ds = self._repo.dirstate
 uipath = lambda f: ds.pathto(pathutil.join(prefix, f))
 rejected = []
 for f in files:
 if f not in ds:
-self._repo.ui.warn(_("%s not tracked!\n") % uipath(f))
+self._repo.ui.warn(_(b"%s not tracked!\n") % uipath(f))
 rejected.append(f)
-elif ds[f] != 'a':
+elif ds[f] != b'a':
 ds.remove(f)
 else:
 ds.drop(f)
 return rejected
 st = self._repo.wvfs.lstat(dest)
 except OSError as err:
 if err.errno != errno.ENOENT:
 raise
 self._repo.ui.warn(
-_("%s does not exist!\n") % self._repo.dirstate.pathto(dest)
+_(b"%s does not exist!\n") % self._repo.dirstate.pathto(dest)
 )
 return
 if not (stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode)):
 self._repo.ui.warn(
-_("copy failed: %s is not a file or a " "symbolic link\n")
+_(b"copy failed: %s is not a file or a " b"symbolic link\n")
 % self._repo.dirstate.pathto(dest)
 )
 else:
 with self._repo.wlock():
 ds = self._repo.dirstate
-if ds[dest] in '?':
+if ds[dest] in b'?':
 ds.add(dest)
-elif ds[dest] in 'r':
+elif ds[dest] in b'r':
 ds.normallookup(dest)
 ds.copy(source, dest)
 def match(
 self,
 pats=None,
 include=None,
 exclude=None,
-default='glob',
+default=b'glob',
 listsubrepos=False,
 badfn=None,
 ):
 r = self._repo
 # via user error or dereferencing by NFS or Samba servers,
 # so we filter out any placeholders that don't look like a
 # symlink
 sane = []
 for f in files:
-if self.flags(f) == 'l':
+if self.flags(f) == b'l':
 d = self[f].data()
 if (
-d == ''
+d == b''
 or len(d) >= 1024
-or '\n' in d
+or b'\n' in d
 or stringutil.binary(d)
 ):
 self._repo.ui.debug(
-'ignoring suspect symlink placeholder' ' "%s"\n' % f
+b'ignoring suspect symlink placeholder' b' "%s"\n' % f
 )
 continue
 sane.append(f)
 return sane
 # in this case, writing changes out breaks
 # consistency, because .hg/dirstate was
 # already changed simultaneously after last
 # caching (see also issue5584 for detail)
 self._repo.ui.debug(
-'skip updating dirstate: ' 'identity mismatch\n'
+b'skip updating dirstate: ' b'identity mismatch\n'
 )
 except error.LockError:
 pass
 finally:
 # Even if the wlock couldn't be grabbed, clear out the list.
 self._repo.clearpostdsstatus()
 def _dirstatestatus(self, match, ignored=False, clean=False, unknown=False):
 '''Gets the status from the dirstate -- internal use only.'''
 subrepos = []
-if '.hgsub' in self:
+if b'.hgsub' in self:
 subrepos = sorted(self.substate)
 cmp, s = self._repo.dirstate.status(
 match, subrepos, ignored=ignored, clean=clean, unknown=unknown
 )
 s = self._dirstatestatus(match, listignored, listclean, listunknown)
 # Filter out symlinks that, in the case of FAT32 and NTFS filesystems,
 # might have accidentally ended up with the entire contents of the file
 # they are supposed to be linking to.
 s.modified[:] = self._filtersuspectsymlink(s.modified)
-if other != self._repo['.']:
+if other != self._repo[b'.']:
 s = super(workingctx, self)._buildstatus(
 other, s, match, listignored, listclean, listunknown
 )
 return s
 comparing against the parent changeset.
 If we aren't comparing against the working directory's parent, then we
 just use the default match object sent to us.
 """
-if other != self._repo['.']:
+if other != self._repo[b'.']:
 def bad(f, msg):
 # 'f' may be a directory pattern from 'match.files()',
 # so 'f not in ctx1' is not enough
 if f not in other and not other.hasdir(f):
 self._repo.ui.warn(
-'%s: %s\n' % (self._repo.dirstate.pathto(f), msg)
+b'%s: %s\n' % (self._repo.dirstate.pathto(f), msg)
 )
 match.bad = bad
 return match
 )
 def matches(self, match):
 match = self._repo.narrowmatch(match)
 ds = self._repo.dirstate
-return sorted(f for f in ds.matches(match) if ds[f] != 'r')
+return sorted(f for f in ds.matches(match) if ds[f] != b'r')
 def markcommitted(self, node):
 with self._repo.dirstate.parentchange():
 for f in self.modified() + self.added():
 self._repo.dirstate.normal(f)
 # invert comparison to reuse the same code path
 return fctx.cmp(self)
 def remove(self, ignoremissing=False):
 """wraps unlink for a repo's working directory"""
-rmdir = self._repo.ui.configbool('experimental', 'removeemptydirs')
+rmdir = self._repo.ui.configbool(b'experimental', b'removeemptydirs')
 self._repo.wvfs.unlinkpath(
 self._path, ignoremissing=ignoremissing, rmdir=rmdir
 )
 def write(self, data, flags, backgroundclose=False, **kwargs):
 ``write()`` can be called successfully.
 """
 wvfs = self._repo.wvfs
 f = self._path
 wvfs.audit(f)
-if self._repo.ui.configbool('experimental', 'merge.checkpathconflicts'):
+if self._repo.ui.configbool(
+b'experimental', b'merge.checkpathconflicts'
+):
 # remove files under the directory as they should already be
 # warned and backed up
 if wvfs.isdir(f) and not wvfs.islink(f):
 wvfs.rmtree(f, forcibly=True)
 for p in reversed(list(util.finddirs(f))):
 self._wrappedctx = wrappedctx
 self._parents = [wrappedctx]
 # Drop old manifest cache as it is now out of date.
 # This is necessary when, e.g., rebasing several nodes with one
 # ``overlayworkingctx`` (e.g. with --collapse).
-util.clearcachedproperty(self, '_manifest')
+util.clearcachedproperty(self, b'_manifest')
 def data(self, path):
 if self.isdirty(path):
-if self._cache[path]['exists']:
+if self._cache[path][b'exists']:
-if self._cache[path]['data'] is not None:
+if self._cache[path][b'data'] is not None:
-return self._cache[path]['data']
+return self._cache[path][b'data']
 else:
 # Must fallback here, too, because we only set flags.
 return self._wrappedctx[path].data()
 else:
 raise error.ProgrammingError(
-"No such file or directory: %s" % path
+b"No such file or directory: %s" % path
 )
 else:
 return self._wrappedctx[path].data()
 @propertycache
 return man
 @propertycache
 def _flagfunc(self):
 def f(path):
-return self._cache[path]['flags']
+return self._cache[path][b'flags']
 return f
 def files(self):
 return sorted(self.added() + self.modified() + self.removed())
 def modified(self):
 return [
 f
 for f in self._cache.keys()
-if self._cache[f]['exists'] and self._existsinparent(f)
+if self._cache[f][b'exists'] and self._existsinparent(f)
 ]
 def added(self):
 return [
 f
 for f in self._cache.keys()
-if self._cache[f]['exists'] and not self._existsinparent(f)
+if self._cache[f][b'exists'] and not self._existsinparent(f)
 ]
 def removed(self):
 return [
 f
 for f in self._cache.keys()
-if not self._cache[f]['exists'] and self._existsinparent(f)
+if not self._cache[f][b'exists'] and self._existsinparent(f)
 ]
 def p1copies(self):
 copies = self._repo._wrappedctx.p1copies().copy()
 narrowmatch = self._repo.narrowmatch()
 for f in self._cache.keys():
 if not narrowmatch(f):
 continue
 copies.pop(f, None)  # delete if it exists
-source = self._cache[f]['copied']
+source = self._cache[f][b'copied']
 if source:
 copies[f] = source
 return copies
 def p2copies(self):
 narrowmatch = self._repo.narrowmatch()
 for f in self._cache.keys():
 if not narrowmatch(f):
 continue
 copies.pop(f, None)  # delete if it exists
-source = self._cache[f]['copied']
+source = self._cache[f][b'copied']
 if source:
 copies[f] = source
 return copies
 def isinmemory(self):
 return True
 def filedate(self, path):
 if self.isdirty(path):
-return self._cache[path]['date']
+return self._cache[path][b'date']
 else:
 return self._wrappedctx[path].date()
 def markcopied(self, path, origin):
 self._markdirty(
 copied=origin,
 )
 def copydata(self, path):
 if self.isdirty(path):
-return self._cache[path]['copied']
+return self._cache[path][b'copied']
 else:
 return None
 def flags(self, path):
 if self.isdirty(path):
-if self._cache[path]['exists']:
+if self._cache[path][b'exists']:
-return self._cache[path]['flags']
+return self._cache[path][b'flags']
 else:
 raise error.ProgrammingError(
-"No such file or directory: %s" % self._path
+b"No such file or directory: %s" % self._path
 )
 else:
 return self._wrappedctx[path].flags()
 def __contains__(self, key):
 if key in self._cache:
-return self._cache[key]['exists']
+return self._cache[key][b'exists']
 return key in self.p1()
 def _existsinparent(self, path):
 try:
 # ``commitctx` raises a ``ManifestLookupError`` if a path does not
 """
 def fail(path, component):
 # p1() is the base and we're receiving "writes" for p2()'s
 # files.
-if 'l' in self.p1()[component].flags():
+if b'l' in self.p1()[component].flags():
 raise error.Abort(
-"error: %s conflicts with symlink %s "
+b"error: %s conflicts with symlink %s "
-"in %d." % (path, component, self.p1().rev())
+b"in %d." % (path, component, self.p1().rev())
 )
 else:
 raise error.Abort(
-"error: '%s' conflicts with file '%s' in "
+b"error: '%s' conflicts with file '%s' in "
-"%d." % (path, component, self.p1().rev())
+b"%d." % (path, component, self.p1().rev())
 )
 # Test that each new directory to be created to write this path from p2
 # is not a file in p1.
-components = path.split('/')
+components = path.split(b'/')
 for i in pycompat.xrange(len(components)):
-component = "/".join(components[0:i])
+component = b"/".join(components[0:i])
 if component in self:
 fail(path, component)
 # Test the other direction -- that this path from p2 isn't a directory
 # in p1 (test that p1 doesn't have any paths matching `path/*`).
 # omit the files which are deleted in current IMM wctx
 mfiles = [m for m in mfiles if m in self]
 if not mfiles:
 return
 raise error.Abort(
-"error: file '%s' cannot be written because "
+b"error: file '%s' cannot be written because "
-" '%s/' is a directory in %s (containing %d "
+b" '%s/' is a directory in %s (containing %d "
-"entries: %s)"
+b"entries: %s)"
-% (path, path, self.p1(), len(mfiles), ', '.join(mfiles))
+% (path, path, self.p1(), len(mfiles), b', '.join(mfiles))
 )
-def write(self, path, data, flags='', **kwargs):
+def write(self, path, data, flags=b'', **kwargs):
 if data is None:
-raise error.ProgrammingError("data must be non-None")
+raise error.ProgrammingError(b"data must be non-None")
 self._auditconflicts(path)
 self._markdirty(
 path, exists=True, data=data, date=dateutil.makedate(), flags=flags
 )
 def setflags(self, path, l, x):
-flag = ''
+flag = b''
 if l:
-flag = 'l'
+flag = b'l'
 elif x:
-flag = 'x'
+flag = b'x'
 self._markdirty(path, exists=True, date=dateutil.makedate(), flags=flag)
 def remove(self, path):
 self._markdirty(path, exists=False)
 """
 if self.isdirty(path):
 # If this path exists and is a symlink, "follow" it by calling
 # exists on the destination path.
 if (
-self._cache[path]['exists']
+self._cache[path][b'exists']
-and 'l' in self._cache[path]['flags']
+and b'l' in self._cache[path][b'flags']
 ):
-return self.exists(self._cache[path]['data'].strip())
+return self.exists(self._cache[path][b'data'].strip())
 else:
-return self._cache[path]['exists']
+return self._cache[path][b'exists']
 return self._existsinparent(path)
 def lexists(self, path):
 """lexists returns True if the path exists"""
 if self.isdirty(path):
-return self._cache[path]['exists']
+return self._cache[path][b'exists']
 return self._existsinparent(path)
 def size(self, path):
 if self.isdirty(path):
-if self._cache[path]['exists']:
+if self._cache[path][b'exists']:
-return len(self._cache[path]['data'])
+return len(self._cache[path][b'data'])
 else:
 raise error.ProgrammingError(
-"No such file or directory: %s" % self._path
+b"No such file or directory: %s" % self._path
 )
 return self._wrappedctx[path].size()
 def tomemctx(
 self,
 parents = (self._repo[parents[0]], self._repo[parents[1]])
 files = self.files()
 def getfile(repo, memctx, path):
-if self._cache[path]['exists']:
+if self._cache[path][b'exists']:
 return memfilectx(
 repo,
 memctx,
 path,
-self._cache[path]['data'],
+self._cache[path][b'data'],
-'l' in self._cache[path]['flags'],
+b'l' in self._cache[path][b'flags'],
-'x' in self._cache[path]['flags'],
+b'x' in self._cache[path][b'flags'],
-self._cache[path]['copied'],
+self._cache[path][b'copied'],
 )
 else:
 # Returning None, but including the path in `files`, is
 # necessary for memctx to register a deletion.
 return None
 for path in self._cache.keys():
 cache = self._cache[path]
 try:
 underlying = self._wrappedctx[path]
 if (
-underlying.data() == cache['data']
+underlying.data() == cache[b'data']
-and underlying.flags() == cache['flags']
+and underlying.flags() == cache[b'flags']
 ):
 keys.append(path)
 except error.ManifestLookupError:
 # Path not in the underlying manifest (created).
 continue
 for path in keys:
 del self._cache[path]
 return keys
 def _markdirty(
-self, path, exists, data=None, date=None, flags='', copied=None
+self, path, exists, data=None, date=None, flags=b'', copied=None
 ):
 # data not provided, let's see if we already have some; if not, let's
 # grab it from our underlying context, so that we always have data if
 # the file is marked as existing.
 if exists and data is None:
 oldentry = self._cache.get(path) or {}
-data = oldentry.get('data')
+data = oldentry.get(b'data')
 if data is None:
 data = self._wrappedctx[path].data()
 self._cache[path] = {
-'exists': exists,
+b'exists': exists,
-'data': data,
+b'data': data,
-'date': date,
+b'date': date,
-'flags': flags,
+b'flags': flags,
-'copied': copied,
+b'copied': copied,
 }
 def filectx(self, path, filelog=None):
 return overlayworkingfilectx(
 self._repo, path, parent=self, filelog=filelog
 This hides changes in the working directory, if they aren't
 committed in this context.
 """
 def __init__(
-self, repo, changes, text="", user=None, date=None, extra=None
+self, repo, changes, text=b"", user=None, date=None, extra=None
 ):
 super(workingcommitctx, self).__init__(
 repo, text, user, date, extra, changes
 )
 copied is the source file path if current file was copied in the
 revision being committed, or None."""
 super(memfilectx, self).__init__(repo, path, None, changectx)
 self._data = data
 if islink:
-self._flags = 'l'
+self._flags = b'l'
 elif isexec:
-self._flags = 'x'
+self._flags = b'x'
 else:
-self._flags = ''
+self._flags = b''
 self._copysource = copysource
 def copysource(self):
 return self._copysource
 self._path = path
 def cmp(self, fctx):
 # filecmp follows symlinks whereas `cmp` should not, so skip the fast
 # path if either side is a symlink.
-symlinks = 'l' in self.flags() or 'l' in fctx.flags()
+symlinks = b'l' in self.flags() or b'l' in fctx.flags()
 if not symlinks and isinstance(fctx, workingfilectx) and self._repo:
 # Add a fast-path for merge if both sides are disk-backed.
 # Note that filecmp uses the opposite return values (True if same)
 # from our cmp functions (True if different).
 return not filecmp.cmp(self.path(), self._repo.wjoin(fctx.path()))
 def path(self):
 return self._path
 def flags(self):
-return ''
+return b''
 def data(self):
 return util.readfile(self._path)
 def decodeddata(self):
-with open(self._path, "rb") as f:
+with open(self._path, b"rb") as f:
 return f.read()
 def remove(self):
 util.unlink(self._path)
 def write(self, data, flags, **kwargs):
 assert not flags
-with open(self._path, "wb") as f:
+with open(self._path, b"wb") as f:
 f.write(data)

Mercurial > public > mercurial-scm > hg-stable

comparison mercurial/context.py @ 43077:687b865b95ad