ui.py
346 lines
| 12.5 KiB
| text/x-python
|
PythonLexer
/ mercurial / ui.py
|
r207 | # ui.py - user interface bits for mercurial | ||
# | ||||
|
r4635 | # Copyright 2005-2007 Matt Mackall <mpm@selenic.com> | ||
|
r207 | # | ||
|
r8225 | # This software may be used and distributed according to the terms of the | ||
# GNU General Public License version 2, incorporated herein by reference. | ||||
|
r207 | |||
|
r3891 | from i18n import _ | ||
|
r8312 | import errno, getpass, os, re, socket, sys, tempfile, traceback | ||
import config, util, error | ||||
|
r207 | |||
|
r8222 | _booleans = {'1': True, 'yes': True, 'true': True, 'on': True, | ||
'0': False, 'no': False, 'false': False, 'off': False} | ||||
|
r3344 | |||
|
r1559 | class ui(object): | ||
|
r8190 | def __init__(self, src=None): | ||
|
r8202 | self._buffers = [] | ||
|
r8205 | self.quiet = self.verbose = self.debugflag = self._traceback = False | ||
|
r8208 | self._reportuntrusted = True | ||
|
r8203 | self._ocfg = config.config() # overlay | ||
self._tcfg = config.config() # trusted | ||||
self._ucfg = config.config() # untrusted | ||||
|
r8478 | self._trustusers = set() | ||
self._trustgroups = set() | ||||
|
r8136 | |||
|
r8190 | if src: | ||
|
r8203 | self._tcfg = src._tcfg.copy() | ||
self._ucfg = src._ucfg.copy() | ||||
self._ocfg = src._ocfg.copy() | ||||
|
r8201 | self._trustusers = src._trustusers.copy() | ||
self._trustgroups = src._trustgroups.copy() | ||||
|
r8143 | self.fixconfig() | ||
else: | ||||
|
r3676 | # we always trust global config files | ||
|
r8142 | for f in util.rcpath(): | ||
|
r8200 | self.readconfig(f, trust=True) | ||
|
r8222 | |||
|
r8189 | def copy(self): | ||
|
r8220 | return self.__class__(self) | ||
|
r1839 | |||
|
r8141 | def _is_trusted(self, fp, f): | ||
|
r3677 | st = util.fstat(fp) | ||
if util.isowner(fp, st): | ||||
return True | ||||
|
r8141 | |||
|
r8201 | tusers, tgroups = self._trustusers, self._trustgroups | ||
|
r8141 | if '*' in tusers or '*' in tgroups: | ||
return True | ||||
user = util.username(st.st_uid) | ||||
group = util.groupname(st.st_gid) | ||||
if user in tusers or group in tgroups or user == util.username(): | ||||
return True | ||||
|
r8204 | if self._reportuntrusted: | ||
|
r8141 | self.warn(_('Not trusting file %s from untrusted ' | ||
'user %s, group %s\n') % (f, user, group)) | ||||
return False | ||||
|
r3551 | |||
|
r8200 | def readconfig(self, filename, root=None, trust=False, | ||
|
r8345 | sections=None, remap=None): | ||
|
r8142 | try: | ||
fp = open(filename) | ||||
except IOError: | ||||
if not sections: # ignore unless we were looking for something | ||||
return | ||||
raise | ||||
|
r8139 | |||
|
r8203 | cfg = config.config() | ||
|
r8200 | trusted = sections or trust or self._is_trusted(fp, filename) | ||
|
r3552 | |||
|
r8142 | try: | ||
|
r8345 | cfg.read(filename, fp, sections=sections, remap=remap) | ||
|
r8144 | except error.ConfigError, inst: | ||
|
r8142 | if trusted: | ||
|
r8144 | raise | ||
self.warn(_("Ignored: %s\n") % str(inst)) | ||||
|
r3552 | |||
|
r8142 | if trusted: | ||
|
r8203 | self._tcfg.update(cfg) | ||
self._tcfg.update(self._ocfg) | ||||
self._ucfg.update(cfg) | ||||
self._ucfg.update(self._ocfg) | ||||
|
r8139 | |||
|
r3347 | if root is None: | ||
root = os.path.expanduser('~') | ||||
self.fixconfig(root=root) | ||||
|
r3014 | |||
|
r8197 | def fixconfig(self, root=None): | ||
|
r3347 | # translate paths relative to root (or home) into absolute paths | ||
|
r8197 | root = root or os.getcwd() | ||
|
r8203 | for c in self._tcfg, self._ucfg, self._ocfg: | ||
|
r8197 | for n, p in c.items('paths'): | ||
if p and "://" not in p and not os.path.isabs(p): | ||||
c.set("paths", n, os.path.normpath(os.path.join(root, p))) | ||||
|
r3347 | |||
|
r8138 | # update ui options | ||
|
r8197 | self.debugflag = self.configbool('ui', 'debug') | ||
self.verbose = self.debugflag or self.configbool('ui', 'verbose') | ||||
self.quiet = not self.debugflag and self.configbool('ui', 'quiet') | ||||
if self.verbose and self.quiet: | ||||
self.quiet = self.verbose = False | ||||
|
r8204 | self._reportuntrusted = self.configbool("ui", "report_untrusted", True) | ||
|
r8205 | self._traceback = self.configbool('ui', 'traceback', False) | ||
|
r3350 | |||
|
r3551 | # update trust information | ||
|
r8478 | self._trustusers.update(self.configlist('trusted', 'users')) | ||
self._trustgroups.update(self.configlist('trusted', 'groups')) | ||||
|
r3551 | |||
|
r3344 | def setconfig(self, section, name, value): | ||
|
r8203 | for cfg in (self._ocfg, self._tcfg, self._ucfg): | ||
cfg.set(section, name, value) | ||||
|
r8197 | self.fixconfig() | ||
|
r960 | |||
|
r8199 | def _data(self, untrusted): | ||
|
r8203 | return untrusted and self._ucfg or self._tcfg | ||
|
r3552 | |||
|
r8182 | def configsource(self, section, name, untrusted=False): | ||
|
r8199 | return self._data(untrusted).source(section, name) or 'none' | ||
|
r8182 | |||
|
r8144 | def config(self, section, name, default=None, untrusted=False): | ||
|
r8199 | value = self._data(untrusted).get(section, name, default) | ||
|
r8204 | if self.debugflag and not untrusted and self._reportuntrusted: | ||
|
r8203 | uvalue = self._ucfg.get(section, name) | ||
|
r3552 | if uvalue is not None and uvalue != value: | ||
|
r8204 | self.debug(_("ignoring untrusted configuration option " | ||
|
r8222 | "%s.%s = %s\n") % (section, name, uvalue)) | ||
|
r3552 | return value | ||
|
r3341 | |||
|
r3552 | def configbool(self, section, name, default=False, untrusted=False): | ||
|
r8144 | v = self.config(section, name, None, untrusted) | ||
|
r8527 | if v is None: | ||
|
r8144 | return default | ||
if v.lower() not in _booleans: | ||||
raise error.ConfigError(_("%s.%s not a boolean ('%s')") | ||||
% (section, name, v)) | ||||
return _booleans[v.lower()] | ||||
|
r3552 | |||
def configlist(self, section, name, default=None, untrusted=False): | ||||
|
r2499 | """Return a list of comma/space separated strings""" | ||
|
r3552 | result = self.config(section, name, untrusted=untrusted) | ||
|
r2499 | if result is None: | ||
|
r2502 | result = default or [] | ||
if isinstance(result, basestring): | ||||
result = result.replace(",", " ").split() | ||||
return result | ||||
|
r2499 | |||
|
r4487 | def has_section(self, section, untrusted=False): | ||
|
r2343 | '''tell whether section exists in config.''' | ||
|
r8199 | return section in self._data(untrusted) | ||
|
r3552 | |||
def configitems(self, section, untrusted=False): | ||||
|
r8199 | items = self._data(untrusted).items(section) | ||
|
r8204 | if self.debugflag and not untrusted and self._reportuntrusted: | ||
|
r8222 | for k, v in self._ucfg.items(section): | ||
|
r8203 | if self._tcfg.get(section, k) != v: | ||
|
r8204 | self.debug(_("ignoring untrusted configuration option " | ||
|
r8144 | "%s.%s = %s\n") % (section, k, v)) | ||
return items | ||||
|
r285 | |||
|
r3552 | def walkconfig(self, untrusted=False): | ||
|
r8203 | cfg = self._data(untrusted) | ||
for section in cfg.sections(): | ||||
|
r3552 | for name, value in self.configitems(section, untrusted): | ||
|
r4085 | yield section, name, str(value).replace('\n', '\\n') | ||
|
r1028 | |||
|
r608 | def username(self): | ||
|
r1985 | """Return default username to be used in commits. | ||
Searched in this order: $HGUSER, [ui] section of hgrcs, $EMAIL | ||||
and stop searching if one of these is set. | ||||
|
r6862 | If not found and ui.askusername is True, ask the user, else use | ||
($LOGNAME or $USER or $LNAME or $USERNAME) + "@full.hostname". | ||||
|
r1985 | """ | ||
user = os.environ.get("HGUSER") | ||||
if user is None: | ||||
user = self.config("ui", "username") | ||||
if user is None: | ||||
user = os.environ.get("EMAIL") | ||||
|
r6862 | if user is None and self.configbool("ui", "askusername"): | ||
|
r7600 | user = self.prompt(_("enter a commit username:"), default=None) | ||
|
r4044 | if user is None: | ||
|
r3721 | try: | ||
user = '%s@%s' % (util.getuser(), socket.getfqdn()) | ||||
|
r4044 | self.warn(_("No username found, using '%s' instead\n") % user) | ||
|
r3721 | except KeyError: | ||
|
r4044 | pass | ||
if not user: | ||||
raise util.Abort(_("Please specify a username.")) | ||||
|
r6351 | if "\n" in user: | ||
|
r7470 | raise util.Abort(_("username %s contains a newline\n") % repr(user)) | ||
|
r1985 | return user | ||
|
r608 | |||
|
r1129 | def shortuser(self, user): | ||
"""Return a short representation of a user name or email address.""" | ||||
|
r1903 | if not self.verbose: user = util.shortuser(user) | ||
|
r1129 | return user | ||
|
r8196 | def _path(self, loc): | ||
p = self.config('paths', loc) | ||||
if p and '%%' in p: | ||||
ui.warn('(deprecated \'\%\%\' in path %s=%s from %s)\n' % | ||||
(loc, p, self.configsource('paths', loc))) | ||||
|
r8197 | p = p.replace('%%', '%') | ||
|
r8196 | return p | ||
|
r2494 | def expandpath(self, loc, default=None): | ||
|
r1892 | """Return repository location relative to cwd or from [paths]""" | ||
|
r4216 | if "://" in loc or os.path.isdir(os.path.join(loc, '.hg')): | ||
|
r1892 | return loc | ||
|
r8196 | path = self._path(loc) | ||
|
r2495 | if not path and default is not None: | ||
|
r8196 | path = self._path(default) | ||
|
r2498 | return path or loc | ||
|
r506 | |||
|
r3737 | def pushbuffer(self): | ||
|
r8202 | self._buffers.append([]) | ||
|
r3737 | |||
def popbuffer(self): | ||||
|
r8202 | return "".join(self._buffers.pop()) | ||
|
r3737 | |||
|
r207 | def write(self, *args): | ||
|
r8202 | if self._buffers: | ||
self._buffers[-1].extend([str(a) for a in args]) | ||||
|
r3737 | else: | ||
for a in args: | ||||
sys.stdout.write(str(a)) | ||||
|
r565 | |||
def write_err(self, *args): | ||||
|
r1989 | try: | ||
if not sys.stdout.closed: sys.stdout.flush() | ||||
for a in args: | ||||
sys.stderr.write(str(a)) | ||||
|
r4023 | # stderr may be buffered under win32 when redirected to files, | ||
# including stdout. | ||||
if not sys.stderr.closed: sys.stderr.flush() | ||||
|
r1989 | except IOError, inst: | ||
if inst.errno != errno.EPIPE: | ||||
raise | ||||
|
r565 | |||
|
r1837 | def flush(self): | ||
|
r2013 | try: sys.stdout.flush() | ||
except: pass | ||||
try: sys.stderr.flush() | ||||
except: pass | ||||
|
r1837 | |||
|
r8208 | def interactive(self): | ||
|
r8538 | i = self.configbool("ui", "interactive", None) | ||
if i is None: | ||||
return sys.stdin.isatty() | ||||
return i | ||||
|
r8208 | |||
|
r5337 | def _readline(self, prompt=''): | ||
|
r8208 | if sys.stdin.isatty(): | ||
|
r5036 | try: | ||
# magically add command line editing support, where | ||||
# available | ||||
import readline | ||||
# force demandimport to really load the module | ||||
readline.read_history_file | ||||
|
r7496 | # windows sometimes raises something other than ImportError | ||
except Exception: | ||||
|
r5036 | pass | ||
|
r5613 | line = raw_input(prompt) | ||
# When stdin is in binary mode on Windows, it can cause | ||||
# raw_input() to emit an extra trailing carriage return | ||||
if os.linesep == '\r\n' and line and line[-1] == '\r': | ||||
line = line[:-1] | ||||
return line | ||||
|
r5036 | |||
|
r8259 | def prompt(self, msg, choices=None, default="y"): | ||
"""Prompt user with msg, read response, and ensure it matches | ||||
one of the provided choices. choices is a sequence of acceptable | ||||
responses with the format: ('&None', 'E&xec', 'Sym&link') | ||||
No sequence implies no response checking. Responses are case | ||||
insensitive. If ui is not interactive, the default is returned. | ||||
|
r5751 | """ | ||
|
r8208 | if not self.interactive(): | ||
|
r7320 | self.note(msg, ' ', default, "\n") | ||
return default | ||||
|
r5671 | while True: | ||
try: | ||||
r = self._readline(msg + ' ') | ||||
|
r5709 | if not r: | ||
return default | ||||
|
r8259 | if not choices: | ||
|
r5671 | return r | ||
|
r8259 | resps = [s[s.index('&')+1].lower() for s in choices] | ||
if r.lower() in resps: | ||||
return r.lower() | ||||
|
r5671 | else: | ||
self.write(_("unrecognized response\n")) | ||||
except EOFError: | ||||
raise util.Abort(_('response expected')) | ||||
|
r5036 | |||
|
r2281 | def getpass(self, prompt=None, default=None): | ||
|
r8208 | if not self.interactive(): return default | ||
|
r7798 | try: | ||
return getpass.getpass(prompt or _('password: ')) | ||||
except EOFError: | ||||
raise util.Abort(_('response expected')) | ||||
|
r207 | def status(self, *msg): | ||
if not self.quiet: self.write(*msg) | ||||
|
r234 | def warn(self, *msg): | ||
|
r565 | self.write_err(*msg) | ||
|
r207 | def note(self, *msg): | ||
if self.verbose: self.write(*msg) | ||||
def debug(self, *msg): | ||||
if self.debugflag: self.write(*msg) | ||||
|
r1983 | def edit(self, text, user): | ||
|
r2206 | (fd, name) = tempfile.mkstemp(prefix="hg-editor-", suffix=".txt", | ||
text=True) | ||||
|
r1984 | try: | ||
f = os.fdopen(fd, "w") | ||||
f.write(text) | ||||
f.close() | ||||
|
r5660 | editor = self.geteditor() | ||
|
r207 | |||
|
r1984 | util.system("%s \"%s\"" % (editor, name), | ||
environ={'HGUSER': user}, | ||||
onerr=util.Abort, errprefix=_("edit failed")) | ||||
|
r608 | |||
|
r1984 | f = open(name) | ||
t = f.read() | ||||
f.close() | ||||
finally: | ||||
os.unlink(name) | ||||
|
r662 | |||
|
r207 | return t | ||
|
r2200 | |||
|
r8206 | def traceback(self): | ||
|
r2335 | '''print exception traceback if traceback printing enabled. | ||
only to call in exception handler. returns true if traceback | ||||
printed.''' | ||||
|
r8205 | if self._traceback: | ||
|
r2335 | traceback.print_exc() | ||
|
r8205 | return self._traceback | ||
|
r5660 | |||
def geteditor(self): | ||||
'''return editor to use''' | ||||
return (os.environ.get("HGEDITOR") or | ||||
self.config("ui", "editor") or | ||||
os.environ.get("VISUAL") or | ||||
os.environ.get("EDITOR", "vi")) | ||||