Python twisted.web.resource 模块，ErrorPage() 实例源码

def test_render_503(self):
        type(self.mock_redir).active_node_ip_port = None
        expected_server = 'vault-redirector/%s/TwistedWeb/16.1.0' % _VERSION
        type(self.mock_request).uri = '/foo/bar'
        type(self.mock_request).path = '/foo/bar'
        self.mock_request.reset_mock()

        with patch('%s.logger' % pbm) as mock_logger:
            resp = self.cls.render(self.mock_request)
        assert self.mock_request.mock_calls == [
            call.setHeader('server', expected_server)
        ]
        assert isinstance(resp, resource.ErrorPage)
        assert resp.code == SERVICE_UNAVAILABLE
        assert resp.brief == 'No Active Node'
        assert resp.detail == 'No active Vault leader could be determined ' \
                              'from Consul API'
        assert mock_logger.mock_calls == [
            call.warning('RESPOND 503 for %s%s request for %s from %s:%s',
                         '', 'GET', '/foo/bar', '1.2.3.4', 12345)
        ]

def test_render_503_queued(self):
        type(self.mock_redir).active_node_ip_port = None
        expected_server = 'vault-redirector/%s/TwistedWeb/16.1.0' % _VERSION
        type(self.mock_request).uri = '/foo/bar'
        type(self.mock_request).path = '/foo/bar'
        type(self.mock_request).queued = 1
        self.mock_request.reset_mock()

        with patch('%s.logger' % pbm) as mock_logger:
            resp = self.cls.render(self.mock_request)
        assert self.mock_request.mock_calls == [
            call.setHeader('server', expected_server)
        ]
        assert isinstance(resp, resource.ErrorPage)
        assert resp.code == SERVICE_UNAVAILABLE
        assert resp.brief == 'No Active Node'
        assert resp.detail == 'No active Vault leader could be determined ' \
                              'from Consul API'
        assert mock_logger.mock_calls == [
            call.warning('RESPOND 503 for %s%s request for %s from %s:%s',
                         'QUEUED ', 'GET', '/foo/bar', '1.2.3.4', 12345)
        ]

def test_render_503_log_disabled(self):
        type(self.mock_redir).active_node_ip_port = None
        expected_server = 'vault-redirector/%s/TwistedWeb/16.1.0' % _VERSION
        type(self.mock_request).uri = '/foo/bar'
        type(self.mock_request).path = '/foo/bar'
        self.mock_request.reset_mock()
        type(self.mock_redir).log_enabled = False

        with patch('%s.logger' % pbm) as mock_logger:
            resp = self.cls.render(self.mock_request)
        assert self.mock_request.mock_calls == [
            call.setHeader('server', expected_server)
        ]
        assert isinstance(resp, resource.ErrorPage)
        assert resp.code == SERVICE_UNAVAILABLE
        assert resp.brief == 'No Active Node'
        assert resp.detail == 'No active Vault leader could be determined ' \
                              'from Consul API'
        assert mock_logger.mock_calls == []

def _authorizedResource(self, request):
        """
        Get the L{IResource} which the given request is authorized to receive.
        If the proper authorization headers are present, the resource will be
        requested from the portal.  If not, an anonymous login attempt will be
        made.
        """
        authheader = request.getHeader(b'authorization')
        if not authheader:
            return util.DeferredResource(self._login(Anonymous()))

        factory, respString = self._selectParseHeader(authheader)
        if factory is None:
            return UnauthorizedResource(self._credentialFactories)
        try:
            credentials = factory.decode(respString, request)
        except error.LoginFailed:
            return UnauthorizedResource(self._credentialFactories)
        except:
            log.err(None, "Unexpected failure from credentials factory")
            return ErrorPage(500, None, None)
        else:
            return util.DeferredResource(self._login(credentials))

def render(self, request):
        self.args = request.args
        self.command = self.getArg("command")

        if config.plugins.Webinterface.anti_hijack.value and len(request.args) > 0 and request.method == 'GET':
            request.setHeader("Allow", "POST")
            return resource.ErrorPage(http.NOT_ALLOWED, "Invalid method: GET!", "GET is not allowed here, please use POST").render(request)

        if self.command is not None:
            if self.command in IPKGResource.SIMPLECMDS:
                return self.execSimpleCmd(request)
            elif self.command in IPKGResource.PACKAGECMDS:
                return self.execPackageCmd(request)
            else:
                return self.doErrorPage(request, "Unknown command: "+ self.command)
        else:
            return self.doIndexPage(request)

def getChildWithDefault(self, name, request):
        """
        This should never be called; it's simply required to implement the
        :py:class:`twisted.web.resource.IResource` interface. Just returns
        a 404.

        See: :py:meth:`twisted.web.resource.IResource.getChildWithDefault`
        """
        return resource.ErrorPage(NOT_FOUND, "No Such Resource",
                                  "No Such Resource")

def test_getChildWithDefault(self):
        res = self.cls.getChildWithDefault(None, None)
        assert isinstance(res, resource.ErrorPage)
        assert res.code == NOT_FOUND
        assert res.brief == 'No Such Resource'
        assert res.detail == 'No Such Resource'

def getChild(self, name, txrequest):
        target = self.get_target()
        try:
            newtarget = getattr(target, name)
            return JsonRpcResource(self.crawler, newtarget)
        except AttributeError:
            return resource.ErrorPage(404, "No Such Resource", "No such child resource.")

def ResourceTemplate(path, registry):
    from quixote import ptl_compile

    glob = {'__file__': _coerceToFilesystemEncoding("", path),
            'resource': resource.ErrorPage(500, "Whoops! Internal Error",
                                           rpyNoResource),
            'registry': registry}

    with open(path) as f:  # Not closed by quixote as of 2.9.1
        e = ptl_compile.compile_template(f, path)
    code = compile(e, "<source>", "exec")
    eval(code, glob, glob)
    return glob['resource']

def _loginFailed(self, result):
        """
        Handle login failure by presenting either another challenge (for
        expected authentication/authorization-related failures) or a server
        error page (for anything else).
        """
        if result.check(error.Unauthorized, error.LoginFailed):
            return UnauthorizedResource(self._credentialFactories)
        else:
            log.err(
                result,
                "HTTPAuthSessionWrapper.getChildWithDefault encountered "
                "unexpected error")
            return ErrorPage(500, None, None)

def failed(self, failure):
        #XXX: Argh. FIXME.
        failure = str(failure)
        self.request.write(
            resource.ErrorPage(http.INTERNAL_SERVER_ERROR,
                               "Server Connection Lost",
                               "Connection to distributed server lost:" +
                               util._PRE(failure)).
            render(self.request))
        self.request.finish()
        log.msg(failure)

def processEnded(self, reason):
        if reason.value.exitCode != 0:
            log.msg("CGI %s exited with exit code %s" %
                    (self.request.uri, reason.value.exitCode))
        if self.errortext:
            log.msg("Errors from CGI %s: %s" % (self.request.uri, self.errortext))
        if self.handling_headers:
            log.msg("Premature end of headers in %s: %s" % (self.request.uri, self.headertext))
            self.request.write(
                resource.ErrorPage(http.INTERNAL_SERVER_ERROR,
                                   "CGI Script Error",
                                   "Premature end of script headers.").render(self.request))
        self.request.unregisterProducer()
        self.request.finish()

def send_error_page(request, status, brief, detail):
    error_page = resource.ErrorPage(status, brief, detail).render(request)
    request.write(error_page)
    request.finish()

def __init__(self, res):
        print "[HTTPRootResource}.__init__"
        resource.Resource.__init__(self)
        self.resource = res
        self.sessionInvalidResource = resource.ErrorPage(http.PRECONDITION_FAILED, "Precondition failed!", "sessionid is missing, invalid or expired!")
        self._sessions = {}

def __init__(self, res, realm):
        HTTPRootResource.__init__(self, res)
        self.realm = realm
        self.authorized = False
        self.unauthorizedResource = resource.ErrorPage(http.UNAUTHORIZED, "Access denied", "Authentication credentials invalid!")
        self._localNetworks = []

def render(self, request):
        path = self.path
        if os_path.isfile(path):
            lastComponent = path.split('/')[-1]

            # Set the Header according to what's requested
            if lastComponent in AppTextHeaderFiles:
                request.setHeader('Content-Type', 'application/text')
            elif lastComponent in TextHtmlHeaderFiles or (path.endswith(".html.xml") and lastComponent != "updates.html.xml"):
                request.setHeader('Content-Type', 'text/html; charset=UTF-8')
            elif lastComponent in TextJavascriptHeaderFiles:
                request.setHeader('Content-Type', 'text/javascript; charset=UTF-8')
            elif lastComponent not in NoExplicitHeaderFiles:
                request.setHeader('Content-Type', 'application/xhtml+xml; charset=UTF-8')
            # now go and write the Output
            # request.finish() is called inside webif.py (requestFinish() which is called via renderPage())
            webif.renderPage(request, path, self.session) # login?
            request.setResponseCode(http.OK)

        elif os_path.isdir(path) and self.addSlash is True:
            uri = "%s/" % (request.path)
            request.redirect(uri)
            return "";

        else:
            return resource.ErrorPage(http.NOT_FOUND, "Error 404 - Page not found", "The requested resource is not available").render(request);

        return server.NOT_DONE_YET