ui.py
452 lines
| 16.6 KiB
| text/x-python
|
PythonLexer
/ mercurial / ui.py
|
r207 | # ui.py - user interface bits for mercurial | ||
# | ||||
|
r2859 | # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> | ||
|
r207 | # | ||
# This software may be used and distributed according to the terms | ||||
# of the GNU General Public License, incorporated herein by reference. | ||||
|
r1400 | from i18n import gettext as _ | ||
|
r613 | from demandload import * | ||
|
r2889 | demandload(globals(), "errno getpass os re socket sys tempfile") | ||
|
r3385 | demandload(globals(), "ConfigParser traceback util") | ||
|
r207 | |||
|
r3344 | def dupconfig(orig): | ||
|
r3425 | new = util.configparser(orig.defaults()) | ||
|
r3344 | updateconfig(orig, new) | ||
return new | ||||
|
r3433 | def updateconfig(source, dest, sections=None): | ||
if not sections: | ||||
sections = source.sections() | ||||
for section in sections: | ||||
|
r3344 | if not dest.has_section(section): | ||
dest.add_section(section) | ||||
for name, value in source.items(section, raw=True): | ||||
dest.set(section, name, value) | ||||
|
r1559 | class ui(object): | ||
|
r207 | def __init__(self, verbose=False, debug=False, quiet=False, | ||
|
r3557 | interactive=True, traceback=False, report_untrusted=True, | ||
parentui=None): | ||||
|
r3344 | self.overlay = None | ||
|
r3737 | self.buffers = [] | ||
|
r1839 | if parentui is None: | ||
|
r1874 | # this is the parent of all ui children | ||
self.parentui = None | ||||
|
r3014 | self.readhooks = [] | ||
|
r3350 | self.quiet = quiet | ||
self.verbose = verbose | ||||
self.debugflag = debug | ||||
self.interactive = interactive | ||||
self.traceback = traceback | ||||
|
r3557 | self.report_untrusted = report_untrusted | ||
|
r3551 | self.trusted_users = {} | ||
self.trusted_groups = {} | ||||
|
r3552 | # if ucdata is not None, its keys must be a superset of cdata's | ||
|
r3425 | self.cdata = util.configparser() | ||
|
r3552 | self.ucdata = None | ||
|
r3676 | # we always trust global config files | ||
self.check_trusted = False | ||||
|
r1951 | self.readconfig(util.rcpath()) | ||
|
r3676 | self.check_trusted = True | ||
|
r1839 | self.updateopts(verbose, debug, quiet, interactive) | ||
|
r1866 | else: | ||
|
r1874 | # parentui may point to an ui object which is already a child | ||
self.parentui = parentui.parentui or parentui | ||||
|
r3338 | self.readhooks = self.parentui.readhooks[:] | ||
|
r3551 | self.trusted_users = parentui.trusted_users.copy() | ||
self.trusted_groups = parentui.trusted_groups.copy() | ||||
|
r3344 | self.cdata = dupconfig(self.parentui.cdata) | ||
|
r3552 | if self.parentui.ucdata: | ||
self.ucdata = dupconfig(self.parentui.ucdata) | ||||
|
r3344 | if self.parentui.overlay: | ||
self.overlay = dupconfig(self.parentui.overlay) | ||||
|
r1839 | |||
def __getattr__(self, key): | ||||
return getattr(self.parentui, key) | ||||
|
r1071 | |||
def updateopts(self, verbose=False, debug=False, quiet=False, | ||||
|
r2293 | interactive=True, traceback=False, config=[]): | ||
|
r3346 | for section, name, value in config: | ||
self.setconfig(section, name, value) | ||||
|
r285 | |||
|
r3350 | if quiet or verbose or debug: | ||
self.setconfig('ui', 'quiet', str(bool(quiet))) | ||||
self.setconfig('ui', 'verbose', str(bool(verbose))) | ||||
self.setconfig('ui', 'debug', str(bool(debug))) | ||||
self.verbosity_constraints() | ||||
if not interactive: | ||||
self.setconfig('ui', 'interactive', 'False') | ||||
self.interactive = False | ||||
self.traceback = self.traceback or traceback | ||||
def verbosity_constraints(self): | ||||
self.quiet = self.configbool('ui', 'quiet') | ||||
self.verbose = self.configbool('ui', 'verbose') | ||||
self.debugflag = self.configbool('ui', 'debug') | ||||
|
r3349 | if self.debugflag: | ||
self.verbose = True | ||||
self.quiet = False | ||||
elif self.verbose and self.quiet: | ||||
|
r3350 | self.quiet = self.verbose = False | ||
|
r3349 | |||
|
r3551 | def _is_trusted(self, fp, f, warn=True): | ||
|
r3676 | if not self.check_trusted: | ||
return True | ||||
|
r3677 | st = util.fstat(fp) | ||
if util.isowner(fp, st): | ||||
return True | ||||
|
r3551 | tusers = self.trusted_users | ||
tgroups = self.trusted_groups | ||||
|
r3678 | if not tusers: | ||
user = util.username() | ||||
if user is not None: | ||||
self.trusted_users[user] = 1 | ||||
self.fixconfig(section='trusted') | ||||
|
r3551 | if (tusers or tgroups) and '*' not in tusers and '*' not in tgroups: | ||
user = util.username(st.st_uid) | ||||
group = util.groupname(st.st_gid) | ||||
if user not in tusers and group not in tgroups: | ||||
|
r3557 | if warn and self.report_untrusted: | ||
|
r3552 | self.warn(_('Not trusting file %s from untrusted ' | ||
|
r3551 | 'user %s, group %s\n') % (f, user, group)) | ||
return False | ||||
return True | ||||
|
r1893 | def readconfig(self, fn, root=None): | ||
|
r1483 | if isinstance(fn, basestring): | ||
fn = [fn] | ||||
for f in fn: | ||||
try: | ||||
|
r3551 | fp = open(f) | ||
except IOError: | ||||
continue | ||||
|
r3552 | cdata = self.cdata | ||
trusted = self._is_trusted(fp, f) | ||||
if not trusted: | ||||
if self.ucdata is None: | ||||
self.ucdata = dupconfig(self.cdata) | ||||
cdata = self.ucdata | ||||
elif self.ucdata is not None: | ||||
# use a separate configparser, so that we don't accidentally | ||||
# override ucdata settings later on. | ||||
cdata = util.configparser() | ||||
|
r3551 | try: | ||
|
r3552 | cdata.readfp(fp, f) | ||
|
r1483 | except ConfigParser.ParsingError, inst: | ||
|
r3552 | msg = _("Failed to parse %s\n%s") % (f, inst) | ||
if trusted: | ||||
raise util.Abort(msg) | ||||
self.warn(_("Ignored: %s\n") % msg) | ||||
if trusted: | ||||
if cdata != self.cdata: | ||||
updateconfig(cdata, self.cdata) | ||||
if self.ucdata is not None: | ||||
updateconfig(cdata, self.ucdata) | ||||
|
r3344 | # override data from config files with data set with ui.setconfig | ||
if self.overlay: | ||||
updateconfig(self.overlay, self.cdata) | ||||
|
r3347 | if root is None: | ||
root = os.path.expanduser('~') | ||||
self.fixconfig(root=root) | ||||
|
r2944 | for hook in self.readhooks: | ||
hook(self) | ||||
|
r337 | |||
|
r3014 | def addreadhook(self, hook): | ||
self.readhooks.append(hook) | ||||
|
r3433 | def readsections(self, filename, *sections): | ||
|
r3552 | """Read filename and add only the specified sections to the config data | ||
The settings are added to the trusted config data. | ||||
""" | ||||
|
r3433 | if not sections: | ||
return | ||||
cdata = util.configparser() | ||||
try: | ||||
cdata.read(filename) | ||||
except ConfigParser.ParsingError, inst: | ||||
|
r3474 | raise util.Abort(_("failed to parse %s\n%s") % (filename, | ||
inst)) | ||||
|
r3433 | |||
for section in sections: | ||||
if not cdata.has_section(section): | ||||
cdata.add_section(section) | ||||
updateconfig(cdata, self.cdata, sections) | ||||
|
r3552 | if self.ucdata: | ||
updateconfig(cdata, self.ucdata, sections) | ||||
|
r3433 | |||
|
r3347 | def fixconfig(self, section=None, name=None, value=None, root=None): | ||
# translate paths relative to root (or home) into absolute paths | ||||
if section is None or section == 'paths': | ||||
if root is None: | ||||
root = os.getcwd() | ||||
items = section and [(name, value)] or [] | ||||
|
r3552 | for cdata in self.cdata, self.ucdata, self.overlay: | ||
|
r3347 | if not cdata: continue | ||
if not items and cdata.has_section('paths'): | ||||
pathsitems = cdata.items('paths') | ||||
else: | ||||
pathsitems = items | ||||
for n, path in pathsitems: | ||||
if path and "://" not in path and not os.path.isabs(path): | ||||
cdata.set("paths", n, os.path.join(root, path)) | ||||
|
r3350 | # update quiet/verbose/debug and interactive status | ||
if section is None or section == 'ui': | ||||
if name is None or name in ('quiet', 'verbose', 'debug'): | ||||
self.verbosity_constraints() | ||||
if name is None or name == 'interactive': | ||||
self.interactive = self.configbool("ui", "interactive", True) | ||||
|
r3551 | # update trust information | ||
|
r3678 | if (section is None or section == 'trusted') and self.trusted_users: | ||
for user in self.configlist('trusted', 'users'): | ||||
|
r3551 | self.trusted_users[user] = 1 | ||
|
r3678 | for group in self.configlist('trusted', 'groups'): | ||
self.trusted_groups[group] = 1 | ||||
|
r3551 | |||
|
r3344 | def setconfig(self, section, name, value): | ||
if not self.overlay: | ||||
|
r3425 | self.overlay = util.configparser() | ||
|
r3552 | for cdata in (self.overlay, self.cdata, self.ucdata): | ||
if not cdata: continue | ||||
|
r3344 | if not cdata.has_section(section): | ||
cdata.add_section(section) | ||||
cdata.set(section, name, value) | ||||
|
r3347 | self.fixconfig(section, name, value) | ||
|
r960 | |||
|
r3552 | def _get_cdata(self, untrusted): | ||
if untrusted and self.ucdata: | ||||
return self.ucdata | ||||
return self.cdata | ||||
def _config(self, section, name, default, funcname, untrusted, abort): | ||||
cdata = self._get_cdata(untrusted) | ||||
if cdata.has_option(section, name): | ||||
|
r1876 | try: | ||
|
r3552 | func = getattr(cdata, funcname) | ||
|
r3341 | return func(section, name) | ||
|
r1876 | except ConfigParser.InterpolationError, inst: | ||
|
r3552 | msg = _("Error in configuration section [%s] " | ||
"parameter '%s':\n%s") % (section, name, inst) | ||||
if abort: | ||||
raise util.Abort(msg) | ||||
self.warn(_("Ignored: %s\n") % msg) | ||||
|
r3343 | return default | ||
|
r3341 | |||
|
r3552 | def _configcommon(self, section, name, default, funcname, untrusted): | ||
value = self._config(section, name, default, funcname, | ||||
untrusted, abort=True) | ||||
if self.debugflag and not untrusted and self.ucdata: | ||||
uvalue = self._config(section, name, None, funcname, | ||||
untrusted=True, abort=False) | ||||
if uvalue is not None and uvalue != value: | ||||
self.warn(_("Ignoring untrusted configuration option " | ||||
"%s.%s = %s\n") % (section, name, uvalue)) | ||||
return value | ||||
|
r3341 | |||
|
r3552 | def config(self, section, name, default=None, untrusted=False): | ||
return self._configcommon(section, name, default, 'get', untrusted) | ||||
|
r285 | |||
|
r3552 | def configbool(self, section, name, default=False, untrusted=False): | ||
return self._configcommon(section, name, default, 'getboolean', | ||||
untrusted) | ||||
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 | |||
|
r3552 | def has_config(self, section, untrusted=False): | ||
|
r2343 | '''tell whether section exists in config.''' | ||
|
r3552 | cdata = self._get_cdata(untrusted) | ||
return cdata.has_section(section) | ||||
|
r2343 | |||
|
r3552 | def _configitems(self, section, untrusted, abort): | ||
|
r1839 | items = {} | ||
|
r3552 | cdata = self._get_cdata(untrusted) | ||
if cdata.has_section(section): | ||||
|
r1876 | try: | ||
|
r3552 | items.update(dict(cdata.items(section))) | ||
|
r1876 | except ConfigParser.InterpolationError, inst: | ||
|
r3552 | msg = _("Error in configuration section [%s]:\n" | ||
"%s") % (section, inst) | ||||
if abort: | ||||
raise util.Abort(msg) | ||||
self.warn(_("Ignored: %s\n") % msg) | ||||
return items | ||||
def configitems(self, section, untrusted=False): | ||||
items = self._configitems(section, untrusted=untrusted, abort=True) | ||||
if self.debugflag and not untrusted and self.ucdata: | ||||
uitems = self._configitems(section, untrusted=True, abort=False) | ||||
keys = uitems.keys() | ||||
keys.sort() | ||||
for k in keys: | ||||
if uitems[k] != items.get(k): | ||||
self.warn(_("Ignoring untrusted configuration option " | ||||
"%s.%s = %s\n") % (section, k, uitems[k])) | ||||
|
r1839 | x = items.items() | ||
x.sort() | ||||
return x | ||||
|
r285 | |||
|
r3552 | def walkconfig(self, untrusted=False): | ||
cdata = self._get_cdata(untrusted) | ||||
sections = cdata.sections() | ||||
|
r3342 | sections.sort() | ||
for section in sections: | ||||
|
r3552 | for name, value in self.configitems(section, untrusted): | ||
|
r3342 | yield section, name, value.replace('\n', '\\n') | ||
|
r1028 | |||
|
r1071 | def extensions(self): | ||
|
r2403 | result = self.configitems("extensions") | ||
for i, (key, value) in enumerate(result): | ||||
if value: | ||||
result[i] = (key, os.path.expanduser(value)) | ||||
return result | ||||
|
r1071 | |||
|
r2003 | def hgignorefiles(self): | ||
|
r2403 | result = [] | ||
for key, value in self.configitems("ui"): | ||||
if key == 'ignore' or key.startswith('ignore.'): | ||||
result.append(os.path.expanduser(value)) | ||||
return result | ||||
|
r2003 | |||
|
r2072 | def configrevlog(self): | ||
|
r2403 | result = {} | ||
for key, value in self.configitems("revlog"): | ||||
result[key.lower()] = value | ||||
return result | ||||
|
r2388 | |||
|
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. | ||||
|
r3721 | If not found, 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") | ||||
|
r3489 | if not user: | ||
|
r3721 | try: | ||
user = '%s@%s' % (util.getuser(), socket.getfqdn()) | ||||
except KeyError: | ||||
raise util.Abort(_("Please specify a username.")) | ||||
self.warn(_("No username found, using '%s' instead\n" % 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 | ||
|
r2494 | def expandpath(self, loc, default=None): | ||
|
r1892 | """Return repository location relative to cwd or from [paths]""" | ||
|
r2624 | if "://" in loc or os.path.isdir(loc): | ||
|
r1892 | return loc | ||
|
r2495 | path = self.config("paths", loc) | ||
if not path and default is not None: | ||||
path = self.config("paths", default) | ||||
|
r2498 | return path or loc | ||
|
r506 | |||
|
r3737 | def pushbuffer(self): | ||
self.buffers.append([]) | ||||
def popbuffer(self): | ||||
return "".join(self.buffers.pop()) | ||||
|
r207 | def write(self, *args): | ||
|
r3737 | if self.buffers: | ||
self.buffers[-1].extend([str(a) for a in args]) | ||||
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)) | ||||
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 | |||
|
r207 | def readline(self): | ||
return sys.stdin.readline()[:-1] | ||||
|
r2281 | def prompt(self, msg, pat=None, default="y"): | ||
|
r207 | if not self.interactive: return default | ||
while 1: | ||||
self.write(msg, " ") | ||||
r = self.readline() | ||||
|
r2281 | if not pat or re.match(pat, r): | ||
|
r207 | return r | ||
else: | ||||
|
r1402 | self.write(_("unrecognized response\n")) | ||
|
r2281 | def getpass(self, prompt=None, default=None): | ||
if not self.interactive: return default | ||||
return getpass.getpass(prompt or _('password: ')) | ||||
|
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() | ||||
editor = (os.environ.get("HGEDITOR") or | ||||
self.config("ui", "editor") or | ||||
os.environ.get("EDITOR", "vi")) | ||||
|
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() | ||||
t = re.sub("(?m)^HG:.*\n", "", t) | ||||
finally: | ||||
os.unlink(name) | ||||
|
r662 | |||
|
r207 | return t | ||
|
r2200 | |||
|
r2335 | def print_exc(self): | ||
'''print exception traceback if traceback printing enabled. | ||||
only to call in exception handler. returns true if traceback | ||||
printed.''' | ||||
if self.traceback: | ||||
traceback.print_exc() | ||||
return self.traceback | ||||