ui: refactor option setting

No more passing options as constructor keywords. Basic options are now
always stored in the overlay for simplicity and consistency.

mercurial/dispatch.py

@@ -1,424 +1,432
 # dispatch.py - command dispatching for mercurial
 #
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 from i18n import _
 import os, sys, atexit, signal, pdb, socket, errno, shlex, time
 import util, commands, hg, fancyopts, extensions, hook, error
 import cmdutil, encoding
 import ui as _ui
 
 def run():
     "run the command in sys.argv"
     sys.exit(dispatch(sys.argv[1:]))
 
 def dispatch(args):
     "run the command specified in args"
     try:
-        u = _ui.ui(traceback='--traceback' in args)
+        u = _ui.ui()
+        if '--traceback' in args:
+            u.setconfig('ui', 'traceback', 'on')
     except util.Abort, inst:
         sys.stderr.write(_("abort: %s\n") % inst)
         return -1
     return _runcatch(u, args)
 
 def _runcatch(ui, args):
     def catchterm(*args):
         raise error.SignalInterrupt
 
     for name in 'SIGBREAK', 'SIGHUP', 'SIGTERM':
         num = getattr(signal, name, None)
         if num: signal.signal(num, catchterm)
 
     try:
         try:
             # enter the debugger before command execution
             if '--debugger' in args:
                 pdb.set_trace()
             try:
                 return _dispatch(ui, args)
             finally:
                 ui.flush()
         except:
             # enter the debugger when we hit an exception
             if '--debugger' in args:
                 pdb.post_mortem(sys.exc_info()[2])
             ui.print_exc()
             raise
 
     # Global exception handling, alphabetically
     # Mercurial-specific first, followed by built-in and library exceptions
     except error.AmbiguousCommand, inst:
         ui.warn(_("hg: command '%s' is ambiguous:\n    %s\n") %
                 (inst.args[0], " ".join(inst.args[1])))
     except error.LockHeld, inst:
         if inst.errno == errno.ETIMEDOUT:
             reason = _('timed out waiting for lock held by %s') % inst.locker
         else:
             reason = _('lock held by %s') % inst.locker
         ui.warn(_("abort: %s: %s\n") % (inst.desc or inst.filename, reason))
     except error.LockUnavailable, inst:
         ui.warn(_("abort: could not lock %s: %s\n") %
                (inst.desc or inst.filename, inst.strerror))
     except error.ParseError, inst:
         if inst.args[0]:
             ui.warn(_("hg %s: %s\n") % (inst.args[0], inst.args[1]))
             commands.help_(ui, inst.args[0])
         else:
             ui.warn(_("hg: %s\n") % inst.args[1])
             commands.help_(ui, 'shortlist')
     except error.RepoError, inst:
         ui.warn(_("abort: %s!\n") % inst)
     except error.ResponseError, inst:
         ui.warn(_("abort: %s") % inst.args[0])
         if not isinstance(inst.args[1], basestring):
             ui.warn(" %r\n" % (inst.args[1],))
         elif not inst.args[1]:
             ui.warn(_(" empty string\n"))
         else:
             ui.warn("\n%r\n" % util.ellipsis(inst.args[1]))
     except error.RevlogError, inst:
         ui.warn(_("abort: %s!\n") % inst)
     except error.SignalInterrupt:
         ui.warn(_("killed!\n"))
     except error.UnknownCommand, inst:
         ui.warn(_("hg: unknown command '%s'\n") % inst.args[0])
         commands.help_(ui, 'shortlist')
     except util.Abort, inst:
         ui.warn(_("abort: %s\n") % inst)
     except ImportError, inst:
         m = str(inst).split()[-1]
         ui.warn(_("abort: could not import module %s!\n") % m)
         if m in "mpatch bdiff".split():
             ui.warn(_("(did you forget to compile extensions?)\n"))
         elif m in "zlib".split():
             ui.warn(_("(is your Python install correct?)\n"))
     except IOError, inst:
         if hasattr(inst, "code"):
             ui.warn(_("abort: %s\n") % inst)
         elif hasattr(inst, "reason"):
             try: # usually it is in the form (errno, strerror)
                 reason = inst.reason.args[1]
             except: # it might be anything, for example a string
                 reason = inst.reason
             ui.warn(_("abort: error: %s\n") % reason)
         elif hasattr(inst, "args") and inst.args[0] == errno.EPIPE:
             if ui.debugflag:
                 ui.warn(_("broken pipe\n"))
         elif getattr(inst, "strerror", None):
             if getattr(inst, "filename", None):
                 ui.warn(_("abort: %s: %s\n") % (inst.strerror, inst.filename))
             else:
                 ui.warn(_("abort: %s\n") % inst.strerror)
         else:
             raise
     except OSError, inst:
         if getattr(inst, "filename", None):
             ui.warn(_("abort: %s: %s\n") % (inst.strerror, inst.filename))
         else:
             ui.warn(_("abort: %s\n") % inst.strerror)
     except KeyboardInterrupt:
         try:
             ui.warn(_("interrupted!\n"))
         except IOError, inst:
             if inst.errno == errno.EPIPE:
                 if ui.debugflag:
                     ui.warn(_("\nbroken pipe\n"))
             else:
                 raise
     except MemoryError:
         ui.warn(_("abort: out of memory\n"))
     except SystemExit, inst:
         # Commands shouldn't sys.exit directly, but give a return code.
         # Just in case catch this and and pass exit code to caller.
         return inst.code
     except socket.error, inst:
         ui.warn(_("abort: %s\n") % inst.args[-1])
     except:
         ui.warn(_("** unknown exception encountered, details follow\n"))
         ui.warn(_("** report bug details to "
                  "http://www.selenic.com/mercurial/bts\n"))
         ui.warn(_("** or [e-mail]\n"))
         ui.warn(_("** Mercurial Distributed SCM (version %s)\n")
                % util.version())
         ui.warn(_("** Extensions loaded: %s\n")
                % ", ".join([x[0] for x in extensions.extensions()]))
         raise
 
     return -1
 
 def _findrepo(p):
     while not os.path.isdir(os.path.join(p, ".hg")):
         oldp, p = p, os.path.dirname(p)
         if p == oldp:
             return None
 
     return p
 
 def _parse(ui, args):
     options = {}
     cmdoptions = {}
 
     try:
         args = fancyopts.fancyopts(args, commands.globalopts, options)
     except fancyopts.getopt.GetoptError, inst:
         raise error.ParseError(None, inst)
 
     if args:
         cmd, args = args[0], args[1:]
         aliases, i = cmdutil.findcmd(cmd, commands.table,
                                      ui.config("ui", "strict"))
         cmd = aliases[0]
         defaults = ui.config("defaults", cmd)
         if defaults:
             args = shlex.split(defaults) + args
         c = list(i[1])
     else:
         cmd = None
         c = []
 
     # combine global options into local
     for o in commands.globalopts:
         c.append((o[0], o[1], options[o[1]], o[3]))
 
     try:
         args = fancyopts.fancyopts(args, c, cmdoptions, True)
     except fancyopts.getopt.GetoptError, inst:
         raise error.ParseError(cmd, inst)
 
     # separate global options back out
     for o in commands.globalopts:
         n = o[1]
         options[n] = cmdoptions[n]
         del cmdoptions[n]
 
     return (cmd, cmd and i[0] or None, args, options, cmdoptions)
 
 def _parseconfig(config):
     """parse the --config options from the command line"""
     parsed = []
     for cfg in config:
         try:
             name, value = cfg.split('=', 1)
             section, name = name.split('.', 1)
             if not section or not name:
                 raise IndexError
             parsed.append((section, name, value))
         except (IndexError, ValueError):
             raise util.Abort(_('malformed --config option: %s') % cfg)
     return parsed
 
 def _earlygetopt(aliases, args):
     """Return list of values for an option (or aliases).
 
     The values are listed in the order they appear in args.
     The options and values are removed from args.
     """
     try:
         argcount = args.index("--")
     except ValueError:
         argcount = len(args)
     shortopts = [opt for opt in aliases if len(opt) == 2]
     values = []
     pos = 0
     while pos < argcount:
         if args[pos] in aliases:
             if pos + 1 >= argcount:
                 # ignore and let getopt report an error if there is no value
                 break
             del args[pos]
             values.append(args.pop(pos))
             argcount -= 2
         elif args[pos][:2] in shortopts:
             # short option can have no following space, e.g. hg log -Rfoo
             values.append(args.pop(pos)[2:])
             argcount -= 1
         else:
             pos += 1
     return values
 
 def runcommand(lui, repo, cmd, fullargs, ui, options, d):
     # run pre-hook, and abort if it fails
     ret = hook.hook(lui, repo, "pre-%s" % cmd, False, args=" ".join(fullargs))
     if ret:
         return ret
     ret = _runcommand(ui, options, cmd, d)
     # run post-hook, passing command result
     hook.hook(lui, repo, "post-%s" % cmd, False, args=" ".join(fullargs),
               result = ret)
     return ret
 
 _loaded = {}
 def _dispatch(ui, args):
     # read --config before doing anything else
     # (e.g. to change trust settings for reading .hg/hgrc)
     config = _earlygetopt(['--config'], args)
     if config:
-        ui.updateopts(config=_parseconfig(config))
+        ui.updateopts(_parseconfig(config))
 
     # check for cwd
     cwd = _earlygetopt(['--cwd'], args)
     if cwd:
         os.chdir(cwd[-1])
 
     # read the local repository .hgrc into a local ui object
     path = _findrepo(os.getcwd()) or ""
     if not path:
         lui = ui
     if path:
         try:
             lui = _ui.ui(parentui=ui)
             lui.readconfig(os.path.join(path, ".hg", "hgrc"))
         except IOError:
             pass
 
     # now we can expand paths, even ones in .hg/hgrc
     rpath = _earlygetopt(["-R", "--repository", "--repo"], args)
     if rpath:
         path = lui.expandpath(rpath[-1])
         lui = _ui.ui(parentui=ui)
         lui.readconfig(os.path.join(path, ".hg", "hgrc"))
 
     extensions.loadall(lui)
     for name, module in extensions.extensions():
         if name in _loaded:
             continue
 
         # setup extensions
         # TODO this should be generalized to scheme, where extensions can
         #      redepend on other extensions.  then we should toposort them, and
         #      do initialization in correct order
         extsetup = getattr(module, 'extsetup', None)
         if extsetup:
             extsetup()
 
         cmdtable = getattr(module, 'cmdtable', {})
         overrides = [cmd for cmd in cmdtable if cmd in commands.table]
         if overrides:
             ui.warn(_("extension '%s' overrides commands: %s\n")
                     % (name, " ".join(overrides)))
         commands.table.update(cmdtable)
         _loaded[name] = 1
     # check for fallback encoding
     fallback = lui.config('ui', 'fallbackencoding')
     if fallback:
         encoding.fallbackencoding = fallback
 
     fullargs = args
     cmd, func, args, options, cmdoptions = _parse(lui, args)
 
     if options["config"]:
         raise util.Abort(_("Option --config may not be abbreviated!"))
     if options["cwd"]:
         raise util.Abort(_("Option --cwd may not be abbreviated!"))
     if options["repository"]:
         raise util.Abort(_(
             "Option -R has to be separated from other options (i.e. not -qR) "
             "and --repository may only be abbreviated as --repo!"))
 
     if options["encoding"]:
         encoding.encoding = options["encoding"]
     if options["encodingmode"]:
         encoding.encodingmode = options["encodingmode"]
     if options["time"]:
         def get_times():
             t = os.times()
             if t[4] == 0.0: # Windows leaves this as zero, so use time.clock()
                 t = (t[0], t[1], t[2], t[3], time.clock())
             return t
         s = get_times()
         def print_time():
             t = get_times()
             ui.warn(_("Time: real %.3f secs (user %.3f+%.3f sys %.3f+%.3f)\n") %
                 (t[4]-s[4], t[0]-s[0], t[2]-s[2], t[1]-s[1], t[3]-s[3]))
         atexit.register(print_time)
 
-    ui.updateopts(options["verbose"], options["debug"], options["quiet"],
-                 not options["noninteractive"], options["traceback"])
+    if options['verbose'] or options['debug'] or options['quiet']:
+        ui.setconfig('ui', 'verbose', str(bool(options['verbose'])))
+        ui.setconfig('ui', 'debug', str(bool(options['debug'])))
+        ui.setconfig('ui', 'quiet', str(bool(options['quiet'])))
+    if options['traceback']:
+        ui.setconfig('ui', 'traceback', 'on')
+    if options['noninteractive']:
+        ui.setconfig('ui', 'interactive', 'off')
 
     if options['help']:
         return commands.help_(ui, cmd, options['version'])
     elif options['version']:
         return commands.version_(ui)
     elif not cmd:
         return commands.help_(ui, 'shortlist')
 
     repo = None
     if cmd not in commands.norepo.split():
         try:
             repo = hg.repository(ui, path=path)
             ui = repo.ui
             if not repo.local():
                 raise util.Abort(_("repository '%s' is not local") % path)
             ui.setconfig("bundle", "mainreporoot", repo.root)
         except error.RepoError:
             if cmd not in commands.optionalrepo.split():
                 if args and not path: # try to infer -R from command args
                     repos = map(_findrepo, args)
                     guess = repos[0]
                     if guess and repos.count(guess) == len(repos):
                         return _dispatch(ui, ['--repository', guess] + fullargs)
                 if not path:
                     raise error.RepoError(_("There is no Mercurial repository"
                                       " here (.hg not found)"))
                 raise
         args.insert(0, repo)
     elif rpath:
         ui.warn("warning: --repository ignored\n")
 
     d = lambda: util.checksignature(func)(ui, *args, **cmdoptions)
     return runcommand(lui, repo, cmd, fullargs, ui, options, d)
 
 def _runcommand(ui, options, cmd, cmdfunc):
     def checkargs():
         try:
             return cmdfunc()
         except error.SignatureError:
             raise error.ParseError(cmd, _("invalid arguments"))
 
     if options['profile']:
         format = ui.config('profiling', 'format', default='text')
 
         if not format in ['text', 'kcachegrind']:
             ui.warn(_("unrecognized profiling format '%s'"
                         " - Ignored\n") % format)
             format = 'text'
 
         output = ui.config('profiling', 'output')
 
         if output:
             path = os.path.expanduser(output)
             path = ui.expandpath(path)
             ostream = open(path, 'wb')
         else:
             ostream = sys.stderr
 
         try:
             from mercurial import lsprof
         except ImportError:
             raise util.Abort(_(
                 'lsprof not available - install from '
                 'http://codespeak.net/svn/user/arigo/hack/misc/lsprof/'))
         p = lsprof.Profiler()
         p.enable(subcalls=True)
         try:
             return checkargs()
         finally:
             p.disable()
 
             if format == 'kcachegrind':
                 import lsprofcalltree
                 calltree = lsprofcalltree.KCacheGrind(p)
                 calltree.output(ostream)
             else:
                 # format == 'text'
                 stats = lsprof.Stats(p.getstats())
                 stats.sort()
                 stats.pprint(top=10, file=ostream, climit=5)
 
             if output:
                 ostream.close()
     else:
         return checkargs()

mercurial/hgweb/hgweb_mod.py

@@ -1,310 +1,312
 # hgweb/hgweb_mod.py - Web interface for a repository.
 #
 # Copyright 21 May 2005 - (c) 2005 Jake Edge <[e-mail]>
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 import os
 from mercurial import ui, hg, util, hook, error, encoding
 from mercurial import templater, templatefilters
 from common import get_mtime, ErrorResponse
 from common import HTTP_OK, HTTP_BAD_REQUEST, HTTP_NOT_FOUND, HTTP_SERVER_ERROR
 from common import HTTP_UNAUTHORIZED, HTTP_METHOD_NOT_ALLOWED
 from request import wsgirequest
 import webcommands, protocol, webutil
 
 perms = {
     'changegroup': 'pull',
     'changegroupsubset': 'pull',
     'unbundle': 'push',
     'stream_out': 'pull',
 }
 
 class hgweb(object):
     def __init__(self, repo, name=None):
         if isinstance(repo, str):
-            parentui = ui.ui(report_untrusted=False, interactive=False)
+            parentui = ui.ui()
+            parentui.setconfig('ui', 'report_untrusted', 'off')
+            parentui.setconfig('ui', 'interactive', 'off')
             self.repo = hg.repository(parentui, repo)
         else:
             self.repo = repo
 
         hook.redirect(True)
         self.mtime = -1
         self.reponame = name
         self.archives = 'zip', 'gz', 'bz2'
         self.stripecount = 1
         # a repo owner may set web.templates in .hg/hgrc to get any file
         # readable by the user running the CGI script
         self.templatepath = self.config('web', 'templates')
 
     # The CGI scripts are often run by a user different from the repo owner.
     # Trust the settings from the .hg/hgrc files by default.
     def config(self, section, name, default=None, untrusted=True):
         return self.repo.ui.config(section, name, default,
                                    untrusted=untrusted)
 
     def configbool(self, section, name, default=False, untrusted=True):
         return self.repo.ui.configbool(section, name, default,
                                        untrusted=untrusted)
 
     def configlist(self, section, name, default=None, untrusted=True):
         return self.repo.ui.configlist(section, name, default,
                                        untrusted=untrusted)
 
     def refresh(self):
         mtime = get_mtime(self.repo.root)
         if mtime != self.mtime:
             self.mtime = mtime
             self.repo = hg.repository(self.repo.ui, self.repo.root)
             self.maxchanges = int(self.config("web", "maxchanges", 10))
             self.stripecount = int(self.config("web", "stripes", 1))
             self.maxshortchanges = int(self.config("web", "maxshortchanges", 60))
             self.maxfiles = int(self.config("web", "maxfiles", 10))
             self.allowpull = self.configbool("web", "allowpull", True)
             self.encoding = self.config("web", "encoding", encoding.encoding)
 
     def run(self):
         if not os.environ.get('GATEWAY_INTERFACE', '').startswith("CGI/1."):
             raise RuntimeError("This function is only intended to be called while running as a CGI script.")
         import mercurial.hgweb.wsgicgi as wsgicgi
         wsgicgi.launch(self)
 
     def __call__(self, env, respond):
         req = wsgirequest(env, respond)
         return self.run_wsgi(req)
 
     def run_wsgi(self, req):
 
         self.refresh()
 
         # process this if it's a protocol request
         # protocol bits don't need to create any URLs
         # and the clients always use the old URL structure
 
         cmd = req.form.get('cmd', [''])[0]
         if cmd and cmd in protocol.__all__:
             try:
                 if cmd in perms:
                     try:
                         self.check_perm(req, perms[cmd])
                     except ErrorResponse, inst:
                         if cmd == 'unbundle':
                             req.drain()
                         raise
                 method = getattr(protocol, cmd)
                 return method(self.repo, req)
             except ErrorResponse, inst:
                 req.respond(inst, protocol.HGTYPE)
                 if not inst.message:
                     return []
                 return '0\n%s\n' % inst.message,
 
         # work with CGI variables to create coherent structure
         # use SCRIPT_NAME, PATH_INFO and QUERY_STRING as well as our REPO_NAME
 
         req.url = req.env['SCRIPT_NAME']
         if not req.url.endswith('/'):
             req.url += '/'
         if 'REPO_NAME' in req.env:
             req.url += req.env['REPO_NAME'] + '/'
 
         if 'PATH_INFO' in req.env:
             parts = req.env['PATH_INFO'].strip('/').split('/')
             repo_parts = req.env.get('REPO_NAME', '').split('/')
             if parts[:len(repo_parts)] == repo_parts:
                 parts = parts[len(repo_parts):]
             query = '/'.join(parts)
         else:
             query = req.env['QUERY_STRING'].split('&', 1)[0]
             query = query.split(';', 1)[0]
 
         # translate user-visible url structure to internal structure
 
         args = query.split('/', 2)
         if 'cmd' not in req.form and args and args[0]:
 
             cmd = args.pop(0)
             style = cmd.rfind('-')
             if style != -1:
                 req.form['style'] = [cmd[:style]]
                 cmd = cmd[style+1:]
 
             # avoid accepting e.g. style parameter as command
             if hasattr(webcommands, cmd):
                 req.form['cmd'] = [cmd]
             else:
                 cmd = ''
 
             if cmd == 'static':
                 req.form['file'] = ['/'.join(args)]
             else:
                 if args and args[0]:
                     node = args.pop(0)
                     req.form['node'] = [node]
                 if args:
                     req.form['file'] = args
 
             if cmd == 'archive':
                 fn = req.form['node'][0]
                 for type_, spec in self.archive_specs.iteritems():
                     ext = spec[2]
                     if fn.endswith(ext):
                         req.form['node'] = [fn[:-len(ext)]]
                         req.form['type'] = [type_]
 
         # process the web interface request
 
         try:
             tmpl = self.templater(req)
             ctype = tmpl('mimetype', encoding=self.encoding)
             ctype = templater.stringify(ctype)
 
             # check read permissions non-static content
             if cmd != 'static':
                 self.check_perm(req, None)
 
             if cmd == '':
                 req.form['cmd'] = [tmpl.cache['default']]
                 cmd = req.form['cmd'][0]
 
             if cmd not in webcommands.__all__:
                 msg = 'no such method: %s' % cmd
                 raise ErrorResponse(HTTP_BAD_REQUEST, msg)
             elif cmd == 'file' and 'raw' in req.form.get('style', []):
                 self.ctype = ctype
                 content = webcommands.rawfile(self, req, tmpl)
             else:
                 content = getattr(webcommands, cmd)(self, req, tmpl)
                 req.respond(HTTP_OK, ctype)
 
             return content
 
         except error.LookupError, err:
             req.respond(HTTP_NOT_FOUND, ctype)
             msg = str(err)
             if 'manifest' not in msg:
                 msg = 'revision not found: %s' % err.name
             return tmpl('error', error=msg)
         except (error.RepoError, error.RevlogError), inst:
             req.respond(HTTP_SERVER_ERROR, ctype)
             return tmpl('error', error=str(inst))
         except ErrorResponse, inst:
             req.respond(inst, ctype)
             return tmpl('error', error=inst.message)
 
     def templater(self, req):
 
         # determine scheme, port and server name
         # this is needed to create absolute urls
 
         proto = req.env.get('wsgi.url_scheme')
         if proto == 'https':
             proto = 'https'
             default_port = "443"
         else:
             proto = 'http'
             default_port = "80"
 
         port = req.env["SERVER_PORT"]
         port = port != default_port and (":" + port) or ""
         urlbase = '%s://%s%s' % (proto, req.env['SERVER_NAME'], port)
         staticurl = self.config("web", "staticurl") or req.url + 'static/'
         if not staticurl.endswith('/'):
             staticurl += '/'
 
         # some functions for the templater
 
         def header(**map):
             yield tmpl('header', encoding=self.encoding, **map)
 
         def footer(**map):
             yield tmpl("footer", **map)
 
         def motd(**map):
             yield self.config("web", "motd", "")
 
         # figure out which style to use
 
         vars = {}
         style = self.config("web", "style", "paper")
         if 'style' in req.form:
             style = req.form['style'][0]
             vars['style'] = style
 
         start = req.url[-1] == '?' and '&' or '?'
         sessionvars = webutil.sessionvars(vars, start)
         mapfile = templater.stylemap(style, self.templatepath)
 
         if not self.reponame:
             self.reponame = (self.config("web", "name")
                              or req.env.get('REPO_NAME')
                              or req.url.strip('/') or self.repo.root)
 
         # create the templater
 
         tmpl = templater.templater(mapfile, templatefilters.filters,
                                    defaults={"url": req.url,
                                              "staticurl": staticurl,
                                              "urlbase": urlbase,
                                              "repo": self.reponame,
                                              "header": header,
                                              "footer": footer,
                                              "motd": motd,
                                              "sessionvars": sessionvars
                                             })
         return tmpl
 
     def archivelist(self, nodeid):
         allowed = self.configlist("web", "allow_archive")
         for i, spec in self.archive_specs.iteritems():
             if i in allowed or self.configbool("web", "allow" + i):
                 yield {"type" : i, "extension" : spec[2], "node" : nodeid}
 
     archive_specs = {
         'bz2': ('application/x-tar', 'tbz2', '.tar.bz2', None),
         'gz': ('application/x-tar', 'tgz', '.tar.gz', None),
         'zip': ('application/zip', 'zip', '.zip', None),
         }
 
     def check_perm(self, req, op):
         '''Check permission for operation based on request data (including
         authentication info). Return if op allowed, else raise an ErrorResponse
         exception.'''
 
         user = req.env.get('REMOTE_USER')
 
         deny_read = self.configlist('web', 'deny_read')
         if deny_read and (not user or deny_read == ['*'] or user in deny_read):
             raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized')
 
         allow_read = self.configlist('web', 'allow_read')
         result = (not allow_read) or (allow_read == ['*'])
         if not (result or user in allow_read):
             raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized')
 
         if op == 'pull' and not self.allowpull:
             raise ErrorResponse(HTTP_UNAUTHORIZED, 'pull not authorized')
         elif op == 'pull' or op is None: # op is None for interface requests
             return
 
         # enforce that you can only push using POST requests
         if req.env['REQUEST_METHOD'] != 'POST':
             msg = 'push requires POST request'
             raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg)
 
         # require ssl by default for pushing, auth info cannot be sniffed
         # and replayed
         scheme = req.env.get('wsgi.url_scheme')
         if self.configbool('web', 'push_ssl', True) and scheme != 'https':
             raise ErrorResponse(HTTP_OK, 'ssl required')
 
         deny = self.configlist('web', 'deny_push')
         if deny and (not user or deny == ['*'] or user in deny):
             raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')
 
         allow = self.configlist('web', 'allow_push')
         result = allow and (allow == ['*'] or user in allow)
         if not result:
             raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')

mercurial/hgweb/hgwebdir_mod.py

@@ -1,327 +1,332
 # hgweb/hgwebdir_mod.py - Web interface for a directory of repositories.
 #
 # Copyright 21 May 2005 - (c) 2005 Jake Edge <[e-mail]>
 # Copyright 2005, 2006 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 import os
 from mercurial.i18n import _
 from mercurial import ui, hg, util, templater, templatefilters, error, encoding
 from common import ErrorResponse, get_mtime, staticfile, paritygen,\
                    get_contact, HTTP_OK, HTTP_NOT_FOUND, HTTP_SERVER_ERROR
 from hgweb_mod import hgweb
 from request import wsgirequest
 
 # This is a stopgap
 class hgwebdir(object):
     def __init__(self, config, parentui=None):
         def cleannames(items):
             return [(util.pconvert(name).strip('/'), path)
                     for name, path in items]
 
-        self.parentui = parentui or ui.ui(report_untrusted=False,
-                                          interactive = False)
+        if parentui:
+           self.parentui = parentui
+        else:
+            self.parentui = ui.ui()
+            self.parentui.setconfig('ui', 'report_untrusted', 'off')
+            self.parentui.setconfig('ui', 'interactive', 'off')
+
         self.motd = None
         self.style = 'paper'
         self.stripecount = None
         self.repos_sorted = ('name', False)
         self._baseurl = None
         if isinstance(config, (list, tuple)):
             self.repos = cleannames(config)
             self.repos_sorted = ('', False)
         elif isinstance(config, dict):
             self.repos = util.sort(cleannames(config.items()))
         else:
             if isinstance(config, util.configparser):
                 cp = config
             else:
                 cp = util.configparser()
                 cp.read(config)
             self.repos = []
             if cp.has_section('web'):
                 if cp.has_option('web', 'motd'):
                     self.motd = cp.get('web', 'motd')
                 if cp.has_option('web', 'style'):
                     self.style = cp.get('web', 'style')
                 if cp.has_option('web', 'stripes'):
                     self.stripecount = int(cp.get('web', 'stripes'))
                 if cp.has_option('web', 'baseurl'):
                     self._baseurl = cp.get('web', 'baseurl')
             if cp.has_section('paths'):
                 paths = cleannames(cp.items('paths'))
                 for prefix, root in paths:
                     roothead, roottail = os.path.split(root)
                     # "foo = /bar/*" makes every subrepo of /bar/ to be
                     # mounted as foo/subrepo
                     # and "foo = /bar/**" does even recurse inside the
                     # subdirectories, remember to use it without working dir.
                     try:
                         recurse = {'*': False, '**': True}[roottail]
                     except KeyError:
                         self.repos.append((prefix, root))
                         continue
                     roothead = os.path.normpath(roothead)
                     for path in util.walkrepos(roothead, followsym=True,
                                                recurse=recurse):
                         path = os.path.normpath(path)
                         name = util.pconvert(path[len(roothead):]).strip('/')
                         if prefix:
                             name = prefix + '/' + name
                         self.repos.append((name, path))
             if cp.has_section('collections'):
                 for prefix, root in cp.items('collections'):
                     for path in util.walkrepos(root, followsym=True):
                         repo = os.path.normpath(path)
                         name = repo
                         if name.startswith(prefix):
                             name = name[len(prefix):]
                         self.repos.append((name.lstrip(os.sep), repo))
             self.repos.sort()
 
     def run(self):
         if not os.environ.get('GATEWAY_INTERFACE', '').startswith("CGI/1."):
             raise RuntimeError("This function is only intended to be called while running as a CGI script.")
         import mercurial.hgweb.wsgicgi as wsgicgi
         wsgicgi.launch(self)
 
     def __call__(self, env, respond):
         req = wsgirequest(env, respond)
         return self.run_wsgi(req)
 
     def read_allowed(self, ui, req):
         """Check allow_read and deny_read config options of a repo's ui object
         to determine user permissions.  By default, with neither option set (or
         both empty), allow all users to read the repo.  There are two ways a
         user can be denied read access:  (1) deny_read is not empty, and the
         user is unauthenticated or deny_read contains user (or *), and (2)
         allow_read is not empty and the user is not in allow_read.  Return True
         if user is allowed to read the repo, else return False."""
 
         user = req.env.get('REMOTE_USER')
 
         deny_read = ui.configlist('web', 'deny_read', untrusted=True)
         if deny_read and (not user or deny_read == ['*'] or user in deny_read):
             return False
 
         allow_read = ui.configlist('web', 'allow_read', untrusted=True)
         # by default, allow reading if no allow_read option has been set
         if (not allow_read) or (allow_read == ['*']) or (user in allow_read):
             return True
 
         return False
 
     def run_wsgi(self, req):
 
         try:
             try:
 
                 virtual = req.env.get("PATH_INFO", "").strip('/')
                 tmpl = self.templater(req)
                 ctype = tmpl('mimetype', encoding=encoding.encoding)
                 ctype = templater.stringify(ctype)
 
                 # a static file
                 if virtual.startswith('static/') or 'static' in req.form:
                     if virtual.startswith('static/'):
                         fname = virtual[7:]
                     else:
                         fname = req.form['static'][0]
                     static = templater.templatepath('static')
                     return (staticfile(static, fname, req),)
 
                 # top-level index
                 elif not virtual:
                     req.respond(HTTP_OK, ctype)
                     return self.makeindex(req, tmpl)
 
                 # nested indexes and hgwebs
 
                 repos = dict(self.repos)
                 while virtual:
                     real = repos.get(virtual)
                     if real:
                         req.env['REPO_NAME'] = virtual
                         try:
                             repo = hg.repository(self.parentui, real)
                             return hgweb(repo).run_wsgi(req)
                         except IOError, inst:
                             msg = inst.strerror
                             raise ErrorResponse(HTTP_SERVER_ERROR, msg)
                         except error.RepoError, inst:
                             raise ErrorResponse(HTTP_SERVER_ERROR, str(inst))
 
                     # browse subdirectories
                     subdir = virtual + '/'
                     if [r for r in repos if r.startswith(subdir)]:
                         req.respond(HTTP_OK, ctype)
                         return self.makeindex(req, tmpl, subdir)
 
                     up = virtual.rfind('/')
                     if up < 0:
                         break
                     virtual = virtual[:up]
 
                 # prefixes not found
                 req.respond(HTTP_NOT_FOUND, ctype)
                 return tmpl("notfound", repo=virtual)
 
             except ErrorResponse, err:
                 req.respond(err, ctype)
                 return tmpl('error', error=err.message or '')
         finally:
             tmpl = None
 
     def makeindex(self, req, tmpl, subdir=""):
 
         def archivelist(ui, nodeid, url):
             allowed = ui.configlist("web", "allow_archive", untrusted=True)
             for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:
                 if i[0] in allowed or ui.configbool("web", "allow" + i[0],
                                                     untrusted=True):
                     yield {"type" : i[0], "extension": i[1],
                            "node": nodeid, "url": url}
 
         def entries(sortcolumn="", descending=False, subdir="", **map):
             def sessionvars(**map):
                 fields = []
                 if 'style' in req.form:
                     style = req.form['style'][0]
                     if style != get('web', 'style', ''):
                         fields.append(('style', style))
 
                 separator = url[-1] == '?' and ';' or '?'
                 for name, value in fields:
                     yield dict(name=name, value=value, separator=separator)
                     separator = ';'
 
             rows = []
             parity = paritygen(self.stripecount)
             for name, path in self.repos:
                 if not name.startswith(subdir):
                     continue
                 name = name[len(subdir):]
 
                 u = ui.ui(parentui=self.parentui)
                 try:
                     u.readconfig(os.path.join(path, '.hg', 'hgrc'))
                 except Exception, e:
                     u.warn(_('error reading %s/.hg/hgrc: %s\n') % (path, e))
                     continue
                 def get(section, name, default=None):
                     return u.config(section, name, default, untrusted=True)
 
                 if u.configbool("web", "hidden", untrusted=True):
                     continue
 
                 if not self.read_allowed(u, req):
                     continue
 
                 parts = [name]
                 if 'PATH_INFO' in req.env:
                     parts.insert(0, req.env['PATH_INFO'].rstrip('/'))
                 if req.env['SCRIPT_NAME']:
                     parts.insert(0, req.env['SCRIPT_NAME'])
                 url = ('/'.join(parts).replace("//", "/")) + '/'
 
                 # update time with local timezone
                 try:
                     d = (get_mtime(path), util.makedate()[1])
                 except OSError:
                     continue
 
                 contact = get_contact(get)
                 description = get("web", "description", "")
                 name = get("web", "name", name)
                 row = dict(contact=contact or "unknown",
                            contact_sort=contact.upper() or "unknown",
                            name=name,
                            name_sort=name,
                            url=url,
                            description=description or "unknown",
                            description_sort=description.upper() or "unknown",
                            lastchange=d,
                            lastchange_sort=d[1]-d[0],
                            sessionvars=sessionvars,
                            archives=archivelist(u, "tip", url))
                 if (not sortcolumn
                     or (sortcolumn, descending) == self.repos_sorted):
                     # fast path for unsorted output
                     row['parity'] = parity.next()
                     yield row
                 else:
                     rows.append((row["%s_sort" % sortcolumn], row))
             if rows:
                 rows.sort()
                 if descending:
                     rows.reverse()
                 for key, row in rows:
                     row['parity'] = parity.next()
                     yield row
 
         sortable = ["name", "description", "contact", "lastchange"]
         sortcolumn, descending = self.repos_sorted
         if 'sort' in req.form:
             sortcolumn = req.form['sort'][0]
             descending = sortcolumn.startswith('-')
             if descending:
                 sortcolumn = sortcolumn[1:]
             if sortcolumn not in sortable:
                 sortcolumn = ""
 
         sort = [("sort_%s" % column,
                  "%s%s" % ((not descending and column == sortcolumn)
                             and "-" or "", column))
                 for column in sortable]
 
         if self._baseurl is not None:
             req.env['SCRIPT_NAME'] = self._baseurl
 
         return tmpl("index", entries=entries, subdir=subdir,
                     sortcolumn=sortcolumn, descending=descending,
                     **dict(sort))
 
     def templater(self, req):
 
         def header(**map):
             yield tmpl('header', encoding=encoding.encoding, **map)
 
         def footer(**map):
             yield tmpl("footer", **map)
 
         def motd(**map):
             if self.motd is not None:
                 yield self.motd
             else:
                 yield config('web', 'motd', '')
 
         def config(section, name, default=None, untrusted=True):
             return self.parentui.config(section, name, default, untrusted)
 
         if self._baseurl is not None:
             req.env['SCRIPT_NAME'] = self._baseurl
 
         url = req.env.get('SCRIPT_NAME', '')
         if not url.endswith('/'):
             url += '/'
 
         staticurl = config('web', 'staticurl') or url + 'static/'
         if not staticurl.endswith('/'):
             staticurl += '/'
 
         style = self.style
         if style is None:
             style = config('web', 'style', '')
         if 'style' in req.form:
             style = req.form['style'][0]
         if self.stripecount is None:
             self.stripecount = int(config('web', 'stripes', 1))
         mapfile = templater.stylemap(style)
         tmpl = templater.templater(mapfile, templatefilters.filters,
                                    defaults={"header": header,
                                              "footer": footer,
                                              "motd": motd,
                                              "url": url,
                                              "staticurl": staticurl})
         return tmpl

mercurial/ui.py

@@ -1,478 +1,464
 # ui.py - user interface bits for mercurial
 #
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 from i18n import _
 import errno, getpass, os, re, socket, sys, tempfile
 import ConfigParser, traceback, util
 
 def dupconfig(orig):
     new = util.configparser(orig.defaults())
     updateconfig(orig, new)
     return new
 
 def updateconfig(source, dest, sections=None):
     if not sections:
         sections = source.sections()
     for section in sections:
         if not dest.has_section(section):
             dest.add_section(section)
         for name, value in source.items(section, raw=True):
             dest.set(section, name, value)
 
 class ui(object):
-    def __init__(self, verbose=False, debug=False, quiet=False,
-                 interactive=True, traceback=False, report_untrusted=True,
-                 parentui=None):
+    def __init__(self, parentui=None):
         self.buffers = []
+        self.quiet = self.verbose = self.debugflag = self.traceback = False
+        self.interactive = self.report_untrusted = True
+
         if parentui is None:
             # this is the parent of all ui children
             self.parentui = None
-            self.quiet = quiet
-            self.verbose = verbose
-            self.debugflag = debug
-            self.interactive = interactive
-            self.traceback = traceback
-            self.report_untrusted = report_untrusted
             self.trusted_users = {}
             self.trusted_groups = {}
             self.overlay = util.configparser()
             # if ucdata is not None, its keys must be a superset of cdata's
             self.cdata = util.configparser()
             self.ucdata = None
             # we always trust global config files
             self.readconfig(util.rcpath(), assumetrusted=True)
-            self.updateopts(verbose, debug, quiet, interactive)
         else:
             # parentui may point to an ui object which is already a child
             self.parentui = parentui.parentui or parentui
             self.trusted_users = parentui.trusted_users.copy()
             self.trusted_groups = parentui.trusted_groups.copy()
             self.cdata = dupconfig(self.parentui.cdata)
             self.overlay = dupconfig(self.parentui.overlay)
             if self.parentui.ucdata:
                 self.ucdata = dupconfig(self.parentui.ucdata)
             if self.parentui is not parentui:
                 self.overlay = util.configparser()
                 updateconfig(parentui.overlay, self.overlay)
             self.buffers = parentui.buffers
 
     def __getattr__(self, key):
         return getattr(self.parentui, key)
 
     _isatty = None
     def isatty(self):
         if ui._isatty is None:
+            try:
             ui._isatty = sys.stdin.isatty()
+            except AttributeError: # not a real file object
+                ui._isatty = False
         return ui._isatty
 
-    def updateopts(self, verbose=False, debug=False, quiet=False,
-                   interactive=True, traceback=False, config=[]):
+    def updateopts(self, config):
         for section, name, value in config:
             self.setconfig(section, name, value)
 
-        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)))
-
-        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')
 
         if self.debugflag:
             self.verbose = True
             self.quiet = False
         elif self.verbose and self.quiet:
             self.quiet = self.verbose = False
 
     def _is_trusted(self, fp, f, warn=True):
         st = util.fstat(fp)
         if util.isowner(fp, st):
             return True
         tusers = self.trusted_users
         tgroups = self.trusted_groups
         if not tusers:
             user = util.username()
             if user is not None:
                 self.trusted_users[user] = 1
                 self.fixconfig(section='trusted')
         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:
                 if warn and self.report_untrusted:
                     self.warn(_('Not trusting file %s from untrusted '
                                 'user %s, group %s\n') % (f, user, group))
                 return False
         return True
 
     def readconfig(self, fn, root=None, assumetrusted=False):
         if isinstance(fn, basestring):
             fn = [fn]
         for f in fn:
             try:
                 fp = open(f)
             except IOError:
                 continue
             cdata = self.cdata
             trusted = assumetrusted or 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()
 
             try:
                 cdata.readfp(fp, f)
             except ConfigParser.ParsingError, inst:
                 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)
         # override data from config files with data set with ui.setconfig
         updateconfig(self.overlay, self.cdata)
         if root is None:
             root = os.path.expanduser('~')
         self.fixconfig(root=root)
 
     def readsections(self, filename, *sections):
         """Read filename and add only the specified sections to the config data
 
         The settings are added to the trusted config data.
         """
         if not sections:
             return
 
         cdata = util.configparser()
         try:
             try:
                 fp = open(filename)
             except IOError, inst:
                 raise util.Abort(_("unable to open %s: %s") %
                                  (filename, getattr(inst, "strerror", inst)))
             try:
                 cdata.readfp(fp, filename)
             finally:
                 fp.close()
         except ConfigParser.ParsingError, inst:
             raise util.Abort(_("failed to parse %s\n%s") % (filename, inst))
 
         for section in sections:
             if not cdata.has_section(section):
                 cdata.add_section(section)
 
         updateconfig(cdata, self.cdata, sections)
         if self.ucdata:
             updateconfig(cdata, self.ucdata, sections)
 
     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 []
             for cdata in self.cdata, self.ucdata, self.overlay:
                 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.normpath(os.path.join(root, path)))
 
         # update verbosity/interactive/report_untrusted settings
         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':
                 interactive = self.configbool("ui", "interactive", None)
                 if interactive is None and self.interactive:
                     self.interactive = self.isatty()
                 else:
                     self.interactive = interactive
             if name is None or name == 'report_untrusted':
                 self.report_untrusted = (
                     self.configbool("ui", "report_untrusted", True))
+            self.traceback = self.configbool('ui', 'traceback', False)
 
         # update trust information
         if (section is None or section == 'trusted') and self.trusted_users:
             for user in self.configlist('trusted', 'users'):
                 self.trusted_users[user] = 1
             for group in self.configlist('trusted', 'groups'):
                 self.trusted_groups[group] = 1
 
     def setconfig(self, section, name, value):
         for cdata in (self.overlay, self.cdata, self.ucdata):
             if not cdata: continue
             if not cdata.has_section(section):
                 cdata.add_section(section)
             cdata.set(section, name, value)
         self.fixconfig(section, name, value)
 
     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):
             try:
                 func = getattr(cdata, funcname)
                 return func(section, name)
             except (ConfigParser.InterpolationError, ValueError), inst:
                 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)
         return default
 
     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
 
     def config(self, section, name, default=None, untrusted=False):
         return self._configcommon(section, name, default, 'get', untrusted)
 
     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):
         """Return a list of comma/space separated strings"""
         result = self.config(section, name, untrusted=untrusted)
         if result is None:
             result = default or []
         if isinstance(result, basestring):
             result = result.replace(",", " ").split()
         return result
 
     def has_section(self, section, untrusted=False):
         '''tell whether section exists in config.'''
         cdata = self._get_cdata(untrusted)
         return cdata.has_section(section)
 
     def _configitems(self, section, untrusted, abort):
         items = {}
         cdata = self._get_cdata(untrusted)
         if cdata.has_section(section):
             try:
                 items.update(dict(cdata.items(section)))
             except ConfigParser.InterpolationError, inst:
                 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)
             for k in util.sort(uitems):
                 if uitems[k] != items.get(k):
                     self.warn(_("Ignoring untrusted configuration option "
                                 "%s.%s = %s\n") % (section, k, uitems[k]))
         return util.sort(items.items())
 
     def walkconfig(self, untrusted=False):
         cdata = self._get_cdata(untrusted)
         sections = cdata.sections()
         sections.sort()
         for section in sections:
             for name, value in self.configitems(section, untrusted):
                 yield section, name, str(value).replace('\n', '\\n')
 
     def username(self):
         """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.
         If not found and ui.askusername is True, ask the user, else use
         ($LOGNAME or $USER or $LNAME or $USERNAME) + "@full.hostname".
         """
         user = os.environ.get("HGUSER")
         if user is None:
             user = self.config("ui", "username")
         if user is None:
             user = os.environ.get("EMAIL")
         if user is None and self.configbool("ui", "askusername"):
             user = self.prompt(_("enter a commit username:"), default=None)
         if user is None:
             try:
                 user = '%s@%s' % (util.getuser(), socket.getfqdn())
                 self.warn(_("No username found, using '%s' instead\n") % user)
             except KeyError:
                 pass
         if not user:
             raise util.Abort(_("Please specify a username."))
         if "\n" in user:
             raise util.Abort(_("username %s contains a newline\n") % repr(user))
         return user
 
     def shortuser(self, user):
         """Return a short representation of a user name or email address."""
         if not self.verbose: user = util.shortuser(user)
         return user
 
     def expandpath(self, loc, default=None):
         """Return repository location relative to cwd or from [paths]"""
         if "://" in loc or os.path.isdir(os.path.join(loc, '.hg')):
             return loc
 
         path = self.config("paths", loc)
         if not path and default is not None:
             path = self.config("paths", default)
         return path or loc
 
     def pushbuffer(self):
         self.buffers.append([])
 
     def popbuffer(self):
         return "".join(self.buffers.pop())
 
     def write(self, *args):
         if self.buffers:
             self.buffers[-1].extend([str(a) for a in args])
         else:
             for a in args:
                 sys.stdout.write(str(a))
 
     def write_err(self, *args):
         try:
             if not sys.stdout.closed: sys.stdout.flush()
             for a in args:
                 sys.stderr.write(str(a))
             # stderr may be buffered under win32 when redirected to files,
             # including stdout.
             if not sys.stderr.closed: sys.stderr.flush()
         except IOError, inst:
             if inst.errno != errno.EPIPE:
                 raise
 
     def flush(self):
         try: sys.stdout.flush()
         except: pass
         try: sys.stderr.flush()
         except: pass
 
     def _readline(self, prompt=''):
         if self.isatty():
             try:
                 # magically add command line editing support, where
                 # available
                 import readline
                 # force demandimport to really load the module
                 readline.read_history_file
                 # windows sometimes raises something other than ImportError
             except Exception:
                 pass
         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
 
     def prompt(self, msg, pat=None, default="y"):
         """Prompt user with msg, read response, and ensure it matches pat
 
         If not interactive -- the default is returned
         """
         if not self.interactive:
             self.note(msg, ' ', default, "\n")
             return default
         while True:
             try:
                 r = self._readline(msg + ' ')
                 if not r:
                     return default
                 if not pat or re.match(pat, r):
                     return r
                 else:
                     self.write(_("unrecognized response\n"))
             except EOFError:
                 raise util.Abort(_('response expected'))
 
     def getpass(self, prompt=None, default=None):
         if not self.interactive: return default
         try:
             return getpass.getpass(prompt or _('password: '))
         except EOFError:
             raise util.Abort(_('response expected'))
     def status(self, *msg):
         if not self.quiet: self.write(*msg)
     def warn(self, *msg):
         self.write_err(*msg)
     def note(self, *msg):
         if self.verbose: self.write(*msg)
     def debug(self, *msg):
         if self.debugflag: self.write(*msg)
     def edit(self, text, user):
         (fd, name) = tempfile.mkstemp(prefix="hg-editor-", suffix=".txt",
                                       text=True)
         try:
             f = os.fdopen(fd, "w")
             f.write(text)
             f.close()
 
             editor = self.geteditor()
 
             util.system("%s \"%s\"" % (editor, name),
                         environ={'HGUSER': user},
                         onerr=util.Abort, errprefix=_("edit failed"))
 
             f = open(name)
             t = f.read()
             f.close()
             t = re.sub("(?m)^HG:.*\n", "", t)
         finally:
             os.unlink(name)
 
         return t
 
     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
 
     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"))

tests/test-non-interactive-wsgi

@@ -1,71 +1,68
 #!/bin/sh
 # Tests if hgweb can run without touching sys.stdin, as is required
 # by the WSGI standard and strictly implemented by mod_wsgi.
 
 mkdir repo
 cd repo
 hg init
 echo foo > bar
 hg add bar
 hg commit -m "test" -d "0 0"
 hg tip
 
 cat > request.py <<EOF
 from mercurial import dispatch
 from mercurial.hgweb.hgweb_mod import hgweb
 from mercurial.ui import ui
 from mercurial import hg
 from StringIO import StringIO
 import os, sys
 
 class FileLike(object):
     def __init__(self, real):
         self.real = real
     def fileno(self):
         print >> sys.__stdout__, 'FILENO'
         return self.real.fileno()
     def read(self):
         print >> sys.__stdout__, 'READ'
         return self.real.read()
     def readline(self):
         print >> sys.__stdout__, 'READLINE'
         return self.real.readline()
-    def isatty(self):
-        print >> sys.__stdout__, 'ISATTY'
-        return False
 
 sys.stdin = FileLike(sys.stdin)
 errors = StringIO()
 input = StringIO()
 output = StringIO()
 
 def startrsp(headers, data):
 	print '---- HEADERS'
 	print headers
 	print '---- DATA'
 	print data
 	return output.write
 
 env = {
 	'wsgi.version': (1, 0),
 	'wsgi.url_scheme': 'http',
 	'wsgi.errors': errors,
 	'wsgi.input': input,
 	'wsgi.multithread': False,
 	'wsgi.multiprocess': False,
 	'wsgi.run_once': False,
 	'REQUEST_METHOD': 'GET',
 	'SCRIPT_NAME': '',
 	'PATH_INFO': '',
 	'QUERY_STRING': '',
 	'SERVER_NAME': '127.0.0.1',
 	'SERVER_PORT': os.environ['HGPORT'],
 	'SERVER_PROTOCOL': 'HTTP/1.0'
 }
 
 hgweb('.')(env, startrsp)
 print '---- ERRORS'
 print errors.getvalue()
 EOF
 
 python request.py

tests/test-trusted.py

@@ -1,211 +1,211
 # Since it's not easy to write a test that portably deals
 # with files from different users/groups, we cheat a bit by
 # monkey-patching some functions in the util module
 
 import os
 from mercurial import ui, util
 
 hgrc = os.environ['HGRCPATH']
 f = open(hgrc)
 basehgrc = f.read()
 f.close()
 
 def testui(user='foo', group='bar', tusers=(), tgroups=(),
            cuser='foo', cgroup='bar', debug=False, silent=False):
     # user, group => owners of the file
     # tusers, tgroups => trusted users/groups
     # cuser, cgroup => user/group of the current process
 
     # write a global hgrc with the list of trusted users/groups and
     # some setting so that we can be sure it was read
     f = open(hgrc, 'w')
     f.write(basehgrc)
     f.write('\n[paths]\n')
     f.write('global = /some/path\n\n')
 
     if tusers or tgroups:
         f.write('[trusted]\n')
         if tusers:
             f.write('users = %s\n' % ', '.join(tusers))
         if tgroups:
             f.write('groups = %s\n' % ', '.join(tgroups))
     f.close()
 
     # override the functions that give names to uids and gids
     def username(uid=None):
         if uid is None:
             return cuser
         return user
     util.username = username
 
     def groupname(gid=None):
         if gid is None:
             return 'bar'
         return group
     util.groupname = groupname
 
     def isowner(fp, st=None):
         return user == cuser
     util.isowner = isowner
 
     # try to read everything
     #print '# File belongs to user %s, group %s' % (user, group)
     #print '# trusted users = %s; trusted groups = %s' % (tusers, tgroups)
     kind = ('different', 'same')
     who = ('', 'user', 'group', 'user and the group')
     trusted = who[(user in tusers) + 2*(group in tgroups)]
     if trusted:
         trusted = ', but we trust the ' + trusted
     print '# %s user, %s group%s' % (kind[user == cuser], kind[group == cgroup],
                                      trusted)
 
     parentui = ui.ui()
-    parentui.updateopts(debug=debug)
+    parentui.setconfig('ui', 'debug', str(bool(debug)))
     u = ui.ui(parentui=parentui)
     u.readconfig('.hg/hgrc')
     if silent:
         return u
     print 'trusted'
     for name, path in u.configitems('paths'):
         print '   ', name, '=', path
     print 'untrusted'
     for name, path in u.configitems('paths', untrusted=True):
         print '.',
         u.config('paths', name) # warning with debug=True
         print '.',
         u.config('paths', name, untrusted=True) # no warnings
         print name, '=', path
     print
 
     return u
 
 os.mkdir('repo')
 os.chdir('repo')
 os.mkdir('.hg')
 f = open('.hg/hgrc', 'w')
 f.write('[paths]\n')
 f.write('local = /another/path\n\n')
 f.write('interpolated = %(global)s%(local)s\n\n')
 f.close()
 
 #print '# Everything is run by user foo, group bar\n'
 
 # same user, same group
 testui()
 # same user, different group
 testui(group='def')
 # different user, same group
 testui(user='abc')
 # ... but we trust the group
 testui(user='abc', tgroups=['bar'])
 # different user, different group
 testui(user='abc', group='def')
 # ... but we trust the user
 testui(user='abc', group='def', tusers=['abc'])
 # ... but we trust the group
 testui(user='abc', group='def', tgroups=['def'])
 # ... but we trust the user and the group
 testui(user='abc', group='def', tusers=['abc'], tgroups=['def'])
 # ... but we trust all users
 print '# we trust all users'
 testui(user='abc', group='def', tusers=['*'])
 # ... but we trust all groups
 print '# we trust all groups'
 testui(user='abc', group='def', tgroups=['*'])
 # ... but we trust the whole universe
 print '# we trust all users and groups'
 testui(user='abc', group='def', tusers=['*'], tgroups=['*'])
 # ... check that users and groups are in different namespaces
 print "# we don't get confused by users and groups with the same name"
 testui(user='abc', group='def', tusers=['def'], tgroups=['abc'])
 # ... lists of user names work
 print "# list of user names"
 testui(user='abc', group='def', tusers=['foo', 'xyz', 'abc', 'bleh'],
        tgroups=['bar', 'baz', 'qux'])
 # ... lists of group names work
 print "# list of group names"
 testui(user='abc', group='def', tusers=['foo', 'xyz', 'bleh'],
        tgroups=['bar', 'def', 'baz', 'qux'])
 
 print "# Can't figure out the name of the user running this process"
 testui(user='abc', group='def', cuser=None)
 
 print "# prints debug warnings"
 u = testui(user='abc', group='def', cuser='foo', debug=True)
 
 print "# ui.readsections"
 filename = 'foobar'
 f = open(filename, 'w')
 f.write('[foobar]\n')
 f.write('baz = quux\n')
 f.close()
 u.readsections(filename, 'foobar')
 print u.config('foobar', 'baz')
 
 print
 print "# read trusted, untrusted, new ui, trusted"
 u = ui.ui()
-u.updateopts(debug=True)
+u.setconfig('ui', 'debug', 'on')
 u.readconfig(filename)
 u2 = ui.ui(parentui=u)
 def username(uid=None):
     return 'foo'
 util.username = username
 u2.readconfig('.hg/hgrc')
 print 'trusted:'
 print u2.config('foobar', 'baz')
 print u2.config('paths', 'interpolated')
 print 'untrusted:'
 print u2.config('foobar', 'baz', untrusted=True)
 print u2.config('paths', 'interpolated', untrusted=True)
 
 print
 print "# error handling"
 
 def assertraises(f, exc=util.Abort):
     try:
         f()
     except exc, inst:
         print 'raised', inst.__class__.__name__
     else:
         print 'no exception?!'
 
 print "# file doesn't exist"
 os.unlink('.hg/hgrc')
 assert not os.path.exists('.hg/hgrc')
 testui(debug=True, silent=True)
 testui(user='abc', group='def', debug=True, silent=True)
 
 print
 print "# parse error"
 f = open('.hg/hgrc', 'w')
 f.write('foo = bar')
 f.close()
 testui(user='abc', group='def', silent=True)
 assertraises(lambda: testui(debug=True, silent=True))
 
 print
 print "# interpolation error"
 f = open('.hg/hgrc', 'w')
 f.write('[foo]\n')
 f.write('bar = %(')
 f.close()
 u = testui(debug=True, silent=True)
 print '# regular config:'
 print '  trusted',
 assertraises(lambda: u.config('foo', 'bar'))
 print 'untrusted',
 assertraises(lambda: u.config('foo', 'bar', untrusted=True))
 
 u = testui(user='abc', group='def', debug=True, silent=True)
 print '  trusted ',
 print u.config('foo', 'bar')
 print 'untrusted',
 assertraises(lambda: u.config('foo', 'bar', untrusted=True))
 
 print '# configitems:'
 print '  trusted ',
 print u.configitems('foo')
 print 'untrusted',
 assertraises(lambda: u.configitems('foo', untrusted=True))
 

tests/test-ui-verbosity

@@ -1,46 +1,49
 #!/usr/bin/env python
 
 import os
 from mercurial import ui
 
 hgrc = os.environ['HGRCPATH']
 f = open(hgrc)
 basehgrc = f.read()
 f.close()
 
 print '      hgrc settings    command line options      final result   '
 print '    quiet verbo debug   quiet verbo debug      quiet verbo debug'
 
 for i in xrange(64):
     hgrc_quiet   = bool(i & 1<<0)
     hgrc_verbose = bool(i & 1<<1)
     hgrc_debug   = bool(i & 1<<2)
     cmd_quiet    = bool(i & 1<<3)
     cmd_verbose  = bool(i & 1<<4)
     cmd_debug    = bool(i & 1<<5)
 
     f = open(hgrc, 'w')
     f.write(basehgrc)
     f.write('\n[ui]\n')
     if hgrc_quiet:
         f.write('quiet = True\n')
     if hgrc_verbose:
         f.write('verbose = True\n')
     if hgrc_debug:
         f.write('debug = True\n')
     f.close()
 
     u = ui.ui()
-    u.updateopts(quiet=cmd_quiet, verbose=cmd_verbose, debug=cmd_debug)
+    if cmd_quiet or cmd_debug or cmd_verbose:
+        u.setconfig('ui', 'quiet', str(bool(cmd_quiet)))
+        u.setconfig('ui', 'verbose', str(bool(cmd_verbose)))
+        u.setconfig('ui', 'debug', str(bool(cmd_debug)))
 
     check = ''
     if u.debugflag:
         if not u.verbose or u.quiet:
             check = ' *'
     elif u.verbose and u.quiet:
         check = ' +'
 
     print ('%2d  %5s %5s %5s   %5s %5s %5s  ->  %5s %5s %5s%s'
            % (i, hgrc_quiet, hgrc_verbose, hgrc_debug,
               cmd_quiet, cmd_verbose, cmd_debug,
               u.quiet, u.verbose, u.debugflag, check))