python类ErrorPage()的实例源码

test_redirector.py 文件源码 项目:vault-redirector-twisted 作者: manheim 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
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)
        ]
test_redirector.py 文件源码 项目:vault-redirector-twisted 作者: manheim 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
        ]
test_redirector.py 文件源码 项目:vault-redirector-twisted 作者: manheim 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
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 == []
wrapper.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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))
IPKG.py 文件源码 项目:enigma2-plugins 作者: opendreambox 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
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)
redirector.py 文件源码 项目:vault-redirector-twisted 作者: manheim 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
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")
test_redirector.py 文件源码 项目:vault-redirector-twisted 作者: manheim 项目源码 文件源码 阅读 30 收藏 0 点赞 0 评论 0
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'
webservice.py 文件源码 项目:scrappy 作者: DormyMo 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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.")
script.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
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']
wrapper.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
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)
distrib.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
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)
twcgi.py 文件源码 项目:zenchmarks 作者: squeaky-pl 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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()
utils.py 文件源码 项目:txasgiresource 作者: JohnDoee 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def send_error_page(request, status, brief, detail):
    error_page = resource.ErrorPage(status, brief, detail).render(request)
    request.write(error_page)
    request.finish()
plugin.py 文件源码 项目:enigma2-plugins 作者: opendreambox 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
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 = {}
plugin.py 文件源码 项目:enigma2-plugins 作者: opendreambox 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
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 = []
Screenpage.py 文件源码 项目:enigma2-plugins 作者: opendreambox 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
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


问题


面经


文章

微信
公众号

扫码关注公众号