comparison mercurial/hgweb/hgweb_mod.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

    webcommands,
    webutil,
    wsgicgi,
)

def getstyle(req, configfn, templatepath):
    styles = (
        req.qsparams.get('style', None),
        configfn('web', 'style'),
        'paper',
    )
    return styles, templater.stylemap(styles, templatepath)

def makebreadcrumb(url, prefix=''):
    '''Return a 'URL breadcrumb' list

    A 'URL breadcrumb' is a list of URL-name pairs,

            break
        breadcrumb.append({'url': urlel, 'name': pathel})
        urlel = os.path.dirname(urlel)
    return templateutil.mappinglist(reversed(breadcrumb))

class requestcontext(object):
    """Holds state/context for an individual request.

    Servers can be multi-threaded. Holding state on the WSGI application
    is prone to race conditions. Instances of this class exist to hold
    mutable and race-free state for requests.
    """
    def __init__(self, app, repo, req, res):
        self.repo = repo
        self.reponame = app.reponame
        self.req = req
        self.res = res

        self.csp, self.nonce = cspvalues(self.repo.ui)

    # Trust the settings from the .hg/hgrc files by default.
    def config(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.config(section, name, default,
                                   untrusted=untrusted)

    def configbool(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configbool(section, name, default,
                                       untrusted=untrusted)

    def configint(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configint(section, name, default,
                                      untrusted=untrusted)

    def configlist(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configlist(section, name, default,
                                       untrusted=untrusted)

    def archivelist(self, nodeid):
        return webutil.archivelist(self.repo.ui, nodeid)

    def templater(self, req):
        # determine scheme, port and server name
        # this is needed to create absolute urls
        logourl = self.config('web', 'logourl')
        logoimg = self.config('web', 'logoimg')
        staticurl = (self.config('web', 'staticurl')
                     or req.apppath.rstrip('/') + '/static/')
        if not staticurl.endswith('/'):
            staticurl += '/'

        # figure out which style to use

        vars = {}
        styles, (style, mapfile) = getstyle(req, self.config,
                                            self.templatepath)
        if style == styles[0]:
            vars['style'] = style

        sessionvars = webutil.sessionvars(vars, '?')

        if not self.reponame:
            self.reponame = (self.config('web', 'name', '')
                             or req.reponame
                             or req.apppath
                             or self.repo.root)

        filters = {}
        templatefilter = registrar.templatefilter(filters)
        @templatefilter('websub', intype=bytes)
        def websubfilter(text):
            return templatefilters.websub(text, self.websubtable)

        # create the templater

            'pathdef': makebreadcrumb(req.apppath),
            'style': style,
            'nonce': self.nonce,
        }
        templatekeyword = registrar.templatekeyword(defaults)
        @templatekeyword('motd', requires=())
        def motd(context, mapping):
            yield self.config('web', 'motd')

        tres = formatter.templateresources(self.repo.ui, self.repo)
        tmpl = templater.templater.frommapfile(mapfile,
                                               filters=filters,
                                               defaults=defaults,
                                               resources=tres)
        return tmpl

    def sendtemplate(self, name, **kwargs):
        """Helper function to send a response generated from a template."""
        kwargs = pycompat.byteskwargs(kwargs)
        self.res.setbodygen(self.tmpl.generate(name, kwargs))
        return self.res.sendresponse()

class hgweb(object):
    """HTTP server for individual repositories.

    Instances of this class serve HTTP responses for a particular
    repository.

    Instances are typically used as WSGI applications.

    Some servers are multi-threaded. On these servers, there may
    be multiple active threads inside __call__.
    """
    def __init__(self, repo, name=None, baseui=None):
        if isinstance(repo, bytes):
            if baseui:
                u = baseui.copy()
            else:

        """Start a server from CGI environment.

        Modern servers should be using WSGI and should avoid this
        method, if possible.
        """
        if not encoding.environ.get('GATEWAY_INTERFACE',
                                    '').startswith("CGI/1."):
            raise RuntimeError("This function is only intended to be "
                               "called while running as a CGI script.")
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        """Run the WSGI application.

        # accordingly. But URL paths can conflict with subrepos and virtual
        # repos in hgwebdir. So until we have a workaround for this, only
        # expose the URLs if the feature is enabled.
        apienabled = rctx.repo.ui.configbool('experimental', 'web.apiserver')
        if apienabled and req.dispatchparts and req.dispatchparts[0] == b'api':
            wireprotoserver.handlewsgiapirequest(rctx, req, res,
                                                 self.check_perm)
            return res.sendresponse()

        handled = wireprotoserver.handlewsgirequest(
            rctx, req, res, self.check_perm)
        if handled:
            return res.sendresponse()

        # Old implementations of hgweb supported dispatching the request via
        # the initial query string parameter instead of using PATH_INFO.

        if 'cmd' not in req.qsparams and args and args[0]:
            cmd = args.pop(0)
            style = cmd.rfind('-')
            if style != -1:
                req.qsparams['style'] = cmd[:style]
                cmd = cmd[style + 1:]

            # avoid accepting e.g. style parameter as command
            if util.safehasattr(webcommands, cmd):
                req.qsparams['cmd'] = cmd

            if cmd == 'archive':
                fn = req.qsparams['node']
                for type_, spec in webutil.archivespecs.iteritems():
                    ext = spec[2]
                    if fn.endswith(ext):
                        req.qsparams['node'] = fn[:-len(ext)]
                        req.qsparams['type'] = type_
        else:
            cmd = req.qsparams.get('cmd', '')

        # process the web interface request

        try:
            rctx.tmpl = rctx.templater(req)
            ctype = rctx.tmpl.render('mimetype',
                                     {'encoding': encoding.encoding})

            # check read permissions non-static content
            if cmd != 'static':
                self.check_perm(rctx, req, None)

                res.headers['Content-Type'] = ctype
                return getattr(webcommands, cmd)(rctx)

        except (error.LookupError, error.RepoLookupError) as err:
            msg = pycompat.bytestr(err)
            if (util.safehasattr(err, 'name') and
                not isinstance(err,  error.ManifestLookupError)):
                msg = 'revision not found: %s' % err.name

            res.status = '404 Not Found'
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=msg)

    def check_perm(self, rctx, req, op):
        for permhook in permhooks:
            permhook(rctx, req, op)

def getwebview(repo):
    """The 'web.view' config controls changeset filter to hgweb. Possible
    values are ``served``, ``visible`` and ``all``. Default is ``served``.
    The ``served`` filter only shows changesets that can be pulled from the
    hgweb instance.  The``visible`` filter includes secret changesets but

43076:2372284d9457

    webcommands,
    webutil,
    wsgicgi,
)


def getstyle(req, configfn, templatepath):
    styles = (
        req.qsparams.get('style', None),
        configfn('web', 'style'),
        'paper',
    )
    return styles, templater.stylemap(styles, templatepath)


def makebreadcrumb(url, prefix=''):
    '''Return a 'URL breadcrumb' list

    A 'URL breadcrumb' is a list of URL-name pairs,

            break
        breadcrumb.append({'url': urlel, 'name': pathel})
        urlel = os.path.dirname(urlel)
    return templateutil.mappinglist(reversed(breadcrumb))


class requestcontext(object):
    """Holds state/context for an individual request.

    Servers can be multi-threaded. Holding state on the WSGI application
    is prone to race conditions. Instances of this class exist to hold
    mutable and race-free state for requests.
    """

    def __init__(self, app, repo, req, res):
        self.repo = repo
        self.reponame = app.reponame
        self.req = req
        self.res = res

        self.csp, self.nonce = cspvalues(self.repo.ui)

    # Trust the settings from the .hg/hgrc files by default.
    def config(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.config(section, name, default, untrusted=untrusted)

    def configbool(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configbool(
            section, name, default, untrusted=untrusted
        )

    def configint(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configint(
            section, name, default, untrusted=untrusted
        )

    def configlist(self, section, name, default=uimod._unset, untrusted=True):
        return self.repo.ui.configlist(
            section, name, default, untrusted=untrusted
        )

    def archivelist(self, nodeid):
        return webutil.archivelist(self.repo.ui, nodeid)

    def templater(self, req):
        # determine scheme, port and server name
        # this is needed to create absolute urls
        logourl = self.config('web', 'logourl')
        logoimg = self.config('web', 'logoimg')
        staticurl = (
            self.config('web', 'staticurl')
            or req.apppath.rstrip('/') + '/static/'
        )
        if not staticurl.endswith('/'):
            staticurl += '/'

        # figure out which style to use

        vars = {}
        styles, (style, mapfile) = getstyle(req, self.config, self.templatepath)
        if style == styles[0]:
            vars['style'] = style

        sessionvars = webutil.sessionvars(vars, '?')

        if not self.reponame:
            self.reponame = (
                self.config('web', 'name', '')
                or req.reponame
                or req.apppath
                or self.repo.root
            )

        filters = {}
        templatefilter = registrar.templatefilter(filters)

        @templatefilter('websub', intype=bytes)
        def websubfilter(text):
            return templatefilters.websub(text, self.websubtable)

        # create the templater

            'pathdef': makebreadcrumb(req.apppath),
            'style': style,
            'nonce': self.nonce,
        }
        templatekeyword = registrar.templatekeyword(defaults)

        @templatekeyword('motd', requires=())
        def motd(context, mapping):
            yield self.config('web', 'motd')

        tres = formatter.templateresources(self.repo.ui, self.repo)
        tmpl = templater.templater.frommapfile(
            mapfile, filters=filters, defaults=defaults, resources=tres
        )
        return tmpl

    def sendtemplate(self, name, **kwargs):
        """Helper function to send a response generated from a template."""
        kwargs = pycompat.byteskwargs(kwargs)
        self.res.setbodygen(self.tmpl.generate(name, kwargs))
        return self.res.sendresponse()


class hgweb(object):
    """HTTP server for individual repositories.

    Instances of this class serve HTTP responses for a particular
    repository.

    Instances are typically used as WSGI applications.

    Some servers are multi-threaded. On these servers, there may
    be multiple active threads inside __call__.
    """

    def __init__(self, repo, name=None, baseui=None):
        if isinstance(repo, bytes):
            if baseui:
                u = baseui.copy()
            else:

        """Start a server from CGI environment.

        Modern servers should be using WSGI and should avoid this
        method, if possible.
        """
        if not encoding.environ.get('GATEWAY_INTERFACE', '').startswith(
            "CGI/1."
        ):
            raise RuntimeError(
                "This function is only intended to be "
                "called while running as a CGI script."
            )
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        """Run the WSGI application.

        # accordingly. But URL paths can conflict with subrepos and virtual
        # repos in hgwebdir. So until we have a workaround for this, only
        # expose the URLs if the feature is enabled.
        apienabled = rctx.repo.ui.configbool('experimental', 'web.apiserver')
        if apienabled and req.dispatchparts and req.dispatchparts[0] == b'api':
            wireprotoserver.handlewsgiapirequest(
                rctx, req, res, self.check_perm
            )
            return res.sendresponse()

        handled = wireprotoserver.handlewsgirequest(
            rctx, req, res, self.check_perm
        )
        if handled:
            return res.sendresponse()

        # Old implementations of hgweb supported dispatching the request via
        # the initial query string parameter instead of using PATH_INFO.

        if 'cmd' not in req.qsparams and args and args[0]:
            cmd = args.pop(0)
            style = cmd.rfind('-')
            if style != -1:
                req.qsparams['style'] = cmd[:style]
                cmd = cmd[style + 1 :]

            # avoid accepting e.g. style parameter as command
            if util.safehasattr(webcommands, cmd):
                req.qsparams['cmd'] = cmd

            if cmd == 'archive':
                fn = req.qsparams['node']
                for type_, spec in webutil.archivespecs.iteritems():
                    ext = spec[2]
                    if fn.endswith(ext):
                        req.qsparams['node'] = fn[: -len(ext)]
                        req.qsparams['type'] = type_
        else:
            cmd = req.qsparams.get('cmd', '')

        # process the web interface request

        try:
            rctx.tmpl = rctx.templater(req)
            ctype = rctx.tmpl.render(
                'mimetype', {'encoding': encoding.encoding}
            )

            # check read permissions non-static content
            if cmd != 'static':
                self.check_perm(rctx, req, None)

                res.headers['Content-Type'] = ctype
                return getattr(webcommands, cmd)(rctx)

        except (error.LookupError, error.RepoLookupError) as err:
            msg = pycompat.bytestr(err)
            if util.safehasattr(err, 'name') and not isinstance(
                err, error.ManifestLookupError
            ):
                msg = 'revision not found: %s' % err.name

            res.status = '404 Not Found'
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=msg)

    def check_perm(self, rctx, req, op):
        for permhook in permhooks:
            permhook(rctx, req, op)


def getwebview(repo):
    """The 'web.view' config controls changeset filter to hgweb. Possible
    values are ``served``, ``visible`` and ``all``. Default is ``served``.
    The ``served`` filter only shows changesets that can be pulled from the
    hgweb instance.  The``visible`` filter includes secret changesets but