mercurial: comparison mercurial/ui.py

equal deleted inserted replaced

-:98e7be1ed6c5
+:0449fb7729d7
 import subprocess
 import sys
 import traceback
 from typing import (
+Any,
+Callable,
 Dict,
 List,
+NoReturn,
 Optional,
 Tuple,
+Type,
+TypeVar,
 Union,
 cast,
 )
 from .i18n import _
 resourceutil,
 stringutil,
 urlutil,
 )
+_ConfigItems = Dict[Tuple[bytes, bytes], object]  # {(section, name) : value}
 # The **opts args of the various write() methods can be basically anything, but
 # there's no way to express it as "anything but str".  So type it to be the
 # handful of known types that are used.
 _MsgOpts = Union[bytes, bool, List["_PromptChoice"]]
 _PromptChoice = Tuple[bytes, bytes]
+_Tui = TypeVar('_Tui', bound="ui")
 urlreq = util.urlreq
 # for use with str.translate(None, _keepalnum), to keep just alphanumerics
-_keepalnum = b''.join(
+_keepalnum: bytes = b''.join(
 c for c in map(pycompat.bytechr, range(256)) if not c.isalnum()
 )
 # The config knobs that will be altered (if unset) by ui.tweakdefaults.
-tweakrc = b"""
+tweakrc: bytes = b"""
 [ui]
 # The rollback command is dangerous. As a rule, don't use it.
 rollback = False
 # Make `hg status` report copy information
 statuscopies = yes
 git = 1
 showfunc = 1
 word-diff = 1
 """
-samplehgrcs = {
+samplehgrcs: Dict[bytes, bytes] = {
 b'user': b"""# example user config (see 'hg help config' for more info)
 [ui]
 # name and email, e.g.
 # username = Jane Doe <jdoe@example.com>
 username =
 class httppasswordmgrdbproxy:
 """Delays loading urllib2 until it's needed."""
-def __init__(self):
+def __init__(self) -> None:
 self._mgr = None
 def _get_mgr(self):
 if self._mgr is None:
 self._mgr = urlreq.httppasswordmgrwithdefaultrealm()
 return _maybebytesurl(
 mgr.find_user_password(_maybestrurl(realm), _maybestrurl(uri))
 )
-def _catchterm(*args):
+def _catchterm(*args) -> NoReturn:
 raise error.SignalInterrupt
 # unique object used to detect no default value has been provided when
 # retrieving configuration value.
 _unset = object()
 # _reqexithandlers: callbacks run at the end of a request
-_reqexithandlers = []
+_reqexithandlers: List = []
 class ui:
-def __init__(self, src=None):
+def __init__(self, src: Optional["ui"] = None) -> None:
 """Create a fresh new ui object if no src given
 Use uimod.ui.load() to create a ui which knows global and user configs.
 In most cases, you should use ui.copy() to create a copy of an existing
 ui object.
 self._exportableenviron = {}
 for k in allowed:
 if k in self.environ:
 self._exportableenviron[k] = self.environ[k]
-def _new_source(self):
+def _new_source(self) -> None:
 self._ocfg.new_source()
 self._tcfg.new_source()
 self._ucfg.new_source()
 @classmethod
-def load(cls):
+def load(cls: Type[_Tui]) -> _Tui:
 """Create a ui and load global and user configs"""
 u = cls()
 # we always trust global config files and environment variables
 for t, f in rcutil.rccomponents():
 if t == b'path':
 raise error.ProgrammingError(b'unknown rctype: %s' % t)
 u._maybetweakdefaults()
 u._new_source()  # anything after that is a different level
 return u
-def _maybetweakdefaults(self):
+def _maybetweakdefaults(self) -> None:
 if not self.configbool(b'ui', b'tweakdefaults'):
 return
 if self._tweaked or self.plain(b'tweakdefaults'):
 return
 for section in tmpcfg:
 for name, value in tmpcfg.items(section):
 if not self.hasconfig(section, name):
 self.setconfig(section, name, value, b"<tweakdefaults>")
-def copy(self):
+def copy(self: _Tui) -> _Tui:
 return self.__class__(self)
-def resetstate(self):
+def resetstate(self) -> None:
 """Clear internal state that shouldn't persist across commands"""
 if self._progbar:
 self._progbar.resetstate()  # reset last-print time of progress bar
 self.httppasswordmgrdb = httppasswordmgrdbproxy()
 @contextlib.contextmanager
-def timeblockedsection(self, key):
+def timeblockedsection(self, key: bytes):
 # this is open-coded below - search for timeblockedsection to find them
 starttime = util.timer()
 try:
 yield
 finally:
 self._uninterruptible = True
 yield
 finally:
 self._uninterruptible = False
-def formatter(self, topic, opts):
+def formatter(self, topic: bytes, opts):
 return formatter.formatter(self, self, topic, opts)
-def _trusted(self, fp, f):
+def _trusted(self, fp, f: bytes) -> bool:
 st = util.fstat(fp)
 if util.isowner(st):
 return True
 tusers, tgroups = self._trustusers, self._trustgroups
 )
 return False
 def read_resource_config(
 self, name, root=None, trust=False, sections=None, remap=None
-):
+) -> None:
 try:
 fp = resourceutil.open_resource(name[0], name[1])
 except IOError:
 if not sections:  # ignore unless we were looking for something
 return
 b'resource:%s.%s' % name, fp, root, trust, sections, remap
 )
 def readconfig(
 self, filename, root=None, trust=False, sections=None, remap=None
-):
+) -> None:
 try:
 fp = open(filename, 'rb')
 except IOError:
 if not sections:  # ignore unless we were looking for something
 return
 self._readconfig(filename, fp, root, trust, sections, remap)
 def _readconfig(
 self, filename, fp, root=None, trust=False, sections=None, remap=None
-):
+) -> None:
 with fp:
 cfg = config.config()
 trusted = sections or trust or self._trusted(fp, filename)
 try:
 _(b'ignored %s: %s\n') % (inst.location, inst.message)
 )
 self._applyconfig(cfg, trusted, root)
-def applyconfig(self, configitems, source=b"", root=None):
+def applyconfig(
+self, configitems: _ConfigItems, source=b"", root=None
+) -> None:
 """Add configitems from a non-file source.  Unlike with ``setconfig()``,
 they can be overridden by subsequent config file reads.  The items are
 in the same format as ``configoverride()``, namely a dict of the
 following structures: {(section, name) : value}
 for (section, name), value in configitems.items():
 cfg.set(section, name, value, source)
 self._applyconfig(cfg, True, root)
-def _applyconfig(self, cfg, trusted, root):
+def _applyconfig(self, cfg, trusted, root) -> None:
 if self.plain():
 for k in (
 b'debug',
 b'fallbackencoding',
 b'quiet',
 if root is None:
 root = os.path.expanduser(b'~')
 self.fixconfig(root=root)
-def fixconfig(self, root=None, section=None):
+def fixconfig(self, root=None, section=None) -> None:
 if section in (None, b'paths'):
 # expand vars and ~
 # translate paths relative to root (or home) into absolute paths
 root = root or encoding.getcwd()
 for c in self._tcfg, self._ucfg, self._ocfg:
 self._ocfg.backup(section, item),
 self._tcfg.backup(section, item),
 self._ucfg.backup(section, item),
 )
-def restoreconfig(self, data):
+def restoreconfig(self, data) -> None:
 self._ocfg.restore(data[0])
 self._tcfg.restore(data[1])
 self._ucfg.restore(data[2])
-def setconfig(self, section, name, value, source=b''):
+def setconfig(self, section, name, value, source=b'') -> None:
 for cfg in (self._ocfg, self._tcfg, self._ucfg):
 cfg.set(section, name, value, source)
 self.fixconfig(section=section)
 self._maybetweakdefaults()
 cfg = self._data(untrusted)
 for section in cfg.sections():
 for name, value in self.configitems(section, untrusted):
 yield section, name, value
-def plain(self, feature=None):
+def plain(self, feature: Optional[bytes] = None) -> bool:
 """is plain mode active?
 Plain mode means that all configuration variables which affect
 the behavior and output of Mercurial should be
 ignored. Additionally, the output should be stable,
 raise error.Abort(
 _(b"username %r contains a newline\n") % pycompat.bytestr(user)
 )
 return user
-def shortuser(self, user):
+def shortuser(self, user: bytes) -> bytes:
 """Return a short representation of a user name or email address."""
 if not self.verbose:
 user = stringutil.shortuser(user)
 return user
 def fmsg(self, f):
 self._fmsg = f
 self._fmsgout, self._fmsgerr = _selectmsgdests(self)
 @contextlib.contextmanager
-def silent(self, error=False, subproc=False, labeled=False):
+def silent(
+self, error: bool = False, subproc: bool = False, labeled: bool = False
+):
 self.pushbuffer(error=error, subproc=subproc, labeled=labeled)
 try:
 yield
 finally:
 self.popbuffer()
-def pushbuffer(self, error=False, subproc=False, labeled=False):
+def pushbuffer(
+self, error: bool = False, subproc: bool = False, labeled: bool = False
+) -> None:
 """install a buffer to capture standard output of the ui object
 If error is True, the error output will be captured too.
 If subproc is True, output from subprocesses (typically hooks) will be
 """
 self._buffers.append([])
 self._bufferstates.append((error, subproc, labeled))
 self._bufferapplylabels = labeled
-def popbuffer(self):
+def popbuffer(self) -> bytes:
 '''pop the last buffer and return the buffered output'''
 self._bufferstates.pop()
 if self._bufferstates:
 self._bufferapplylabels = self._bufferstates[-1][2]
 else:
 self._bufferapplylabels = None
 return b"".join(self._buffers.pop())
-def _isbuffered(self, dest):
+def _isbuffered(self, dest) -> bool:
 if dest is self._fout:
 return bool(self._buffers)
 if dest is self._ferr:
 return bool(self._bufferstates and self._bufferstates[-1][0])
 return False
-def canwritewithoutlabels(self):
+def canwritewithoutlabels(self) -> bool:
 '''check if write skips the label'''
 if self._buffers and not self._bufferapplylabels:
 return True
 return self._colormode is None
-def canbatchlabeledwrites(self):
+def canbatchlabeledwrites(self) -> bool:
 '''check if write calls with labels are batchable'''
 # Windows color printing is special, see ``write``.
 return self._colormode != b'win32'
 def write(self, *args: bytes, **opts: _MsgOpts) -> None:
 finally:
 self._blockedtimes[b'stdio_blocked'] += (
 util.timer() - starttime
 ) * 1000
-def _isatty(self, fh):
+def _isatty(self, fh) -> bool:
 if self.configbool(b'ui', b'nontty'):
 return False
 return procutil.isatty(fh)
 def protectfinout(self):
 try:
 yield fin, fout
 finally:
 self.restorefinout(fin, fout)
-def disablepager(self):
+def disablepager(self) -> None:
 self._disablepager = True
-def pager(self, command):
+def pager(self, command: bytes) -> None:
 """Start a pager for subsequent command output.
 Commands which produce a long stream of output should call
 this function to activate the user's preferred pagination
 mechanism (which may be no pager). Calling this function
 # If the pager can't be spawned in dispatch when --pager=on is
 # given, don't try again when the command runs, to avoid a duplicate
 # warning about a missing pager command.
 self.disablepager()
-def _runpager(self, command, env=None):
+def _runpager(self, command: bytes, env=None) -> bool:
 """Actually start the pager and set up file descriptors.
 This is separate in part so that extensions (like chg) can
 override how a pager is invoked.
 """
 Handlers do not stay registered across request boundaries."""
 self._exithandlers.append((func, args, kwargs))
 return func
-def interface(self, feature):
+def interface(self, feature: bytes) -> bytes:
 """what interface to use for interactive console features?
 The interface is controlled by the value of `ui.interface` but also by
 the value of feature-specific configuration. For example:
 # Default interface for all the features
 defaultinterface = b"text"
 i = self.config(b"ui", b"interface")
 if i in alldefaults:
-defaultinterface = i
+defaultinterface = cast(bytes, i)  # cast to help pytype
-choseninterface = defaultinterface
+choseninterface: bytes = defaultinterface
 f = self.config(b"ui", b"interface.%s" % feature)
 if f in availableinterfaces:
-choseninterface = f
+choseninterface = cast(bytes, f)  # cast to help pytype
 if i is not None and defaultinterface != i:
 if f is not None:
 self.warn(_(b"invalid value for ui.interface: %s\n") % (i,))
 else:
 # usually those are non-interactive
 return self._isatty(self._fin)
 return i
-def termwidth(self):
+def termwidth(self) -> int:
 """how wide is the terminal in columns?"""
 if b'COLUMNS' in encoding.environ:
 try:
 return int(encoding.environ[b'COLUMNS'])
 except ValueError:
 warnnoi18n = warn
 writenoi18n = write
 def edit(
 self,
-text,
+text: bytes,
-user,
+user: bytes,
-extra=None,
+extra: Optional[Dict[bytes, Any]] = None,  # TODO: value type of bytes?
 editform=None,
 pending=None,
-repopath=None,
+repopath: Optional[bytes] = None,
-action=None,
+action: Optional[bytes] = None,
-):
+) -> bytes:
 if action is None:
 self.develwarn(
 b'action is None but will soon be a required '
 b'parameter to ui.edit()'
 )
 return t
 def system(
 self,
-cmd,
+cmd: bytes,
 environ=None,
-cwd=None,
+cwd: Optional[bytes] = None,
-onerr=None,
+onerr: Optional[Callable[[bytes], Exception]] = None,
-errprefix=None,
+errprefix: Optional[bytes] = None,
-blockedtag=None,
+blockedtag: Optional[bytes] = None,
-):
+) -> int:
 """execute shell command with appropriate output stream. command
 output will be redirected if fout is not stdout.
 if command fails and onerr is None, return status, else raise onerr
 object as exception.
 if errprefix:
 errmsg = b'%s: %s' % (errprefix, errmsg)
 raise onerr(errmsg)
 return rc
-def _runsystem(self, cmd, environ, cwd, out):
+def _runsystem(self, cmd: bytes, environ, cwd: Optional[bytes], out) -> int:
 """actually execute the given shell command (can be overridden by
 extensions like chg)"""
 return procutil.system(cmd, environ=environ, cwd=cwd, out=out)
-def traceback(self, exc=None, force=False):
+def traceback(self, exc=None, force: bool = False):
 """print exception traceback if traceback printing enabled or forced.
 only to call in exception handler. returns true if traceback
 printed."""
 if self.tracebackflag or force:
 if exc is None:
 def getlogger(self, name):
 """Returns a logger of the given name; or None if not registered"""
 return self._loggers.get(name)
-def setlogger(self, name, logger):
+def setlogger(self, name, logger) -> None:
 """Install logger which can be identified later by the given name
 More than one loggers can be registered. Use extension or module
 name to uniquely identify the logger instance.
 """
 self._loggers[name] = logger
-def log(self, event, msgfmt, *msgargs, **opts):
+def log(self, event, msgfmt, *msgargs, **opts) -> None:
 """hook for logging facility extensions
 event should be a readily-identifiable subsystem, which will
 allow filtering.
 hgweb.
 """
 return self._exportableenviron
 @contextlib.contextmanager
-def configoverride(self, overrides, source=b""):
+def configoverride(self, overrides: _ConfigItems, source: bytes = b""):
 """Context manager for temporary config overrides
 `overrides` must be a dict of the following structure:
 {(section, name) : value}"""
 backups = {}
 try:
 # just restoring ui.quiet config to the previous value is not enough
 # as it does not update ui.quiet class member
 if (b'ui', b'quiet') in overrides:
 self.fixconfig(section=b'ui')
-def estimatememory(self):
+def estimatememory(self) -> Optional[int]:
 """Provide an estimate for the available system memory in Bytes.
 This can be overriden via ui.available-memory. It returns None, if
 no estimate can be computed.
 """
 def haveprogbar() -> bool:
 return _progresssingleton is not None
-def _selectmsgdests(ui):
+def _selectmsgdests(ui: ui):
 name = ui.config(b'ui', b'message-output')
 if name == b'channel':
 if ui.fmsg:
 return ui.fmsg, ui.fmsg
 else:

changeset 49797	0449fb7729d7
parent 49795	f1e820cda2f5
child 49798	a51328ba33ca