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)
]
python类ErrorPage()的实例源码
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