def test_extensions_expected_error(self):
@extensions.expected_errors(http.NOT_FOUND)
def fake_func():
raise webob.exc.HTTPNotFound()
self.assertRaises(webob.exc.HTTPNotFound, fake_func)
python类NOT_FOUND的实例源码
def test_extensions_expected_error_from_list(self):
@extensions.expected_errors((http.NOT_FOUND, http.FORBIDDEN))
def fake_func():
raise webob.exc.HTTPNotFound()
self.assertRaises(webob.exc.HTTPNotFound, fake_func)
def test_extensions_unexpected_error(self):
@extensions.expected_errors(http.NOT_FOUND)
def fake_func():
raise webob.exc.HTTPConflict()
self.assertRaises(webob.exc.HTTPInternalServerError, fake_func)
def test_extensions_unexpected_policy_not_authorized_error(self):
@extensions.expected_errors(http.NOT_FOUND)
def fake_func():
raise exception.PolicyNotAuthorized(action="foo")
self.assertRaises(exception.PolicyNotAuthorized, fake_func)
def test_get_version_1_versions_invalid(self, mock_get_client):
req = webob.Request.blank('/v1/versions/1234/foo')
req.accept = "application/json"
res = req.get_response(self.wsgi_app)
self.assertEqual(http.NOT_FOUND, res.status_int)
def test_override_default_code(self):
robj = wsgi.ResponseObject({}, code=http.NOT_FOUND)
self.assertEqual(robj.code, http.NOT_FOUND)
def test_override_modified_code(self):
robj = wsgi.ResponseObject({}, code=http.NOT_FOUND)
robj._default_code = http.ACCEPTED
self.assertEqual(robj.code, http.NOT_FOUND)
def test_get_failed_not_found(self, mock_novaclient):
mock_novaclient.return_value.servers.get.side_effect = (
nova_exception.NotFound(http.NOT_FOUND, '404'))
self.assertRaises(exception.NotFound,
self.api.get_server, self.ctx, uuidsentinel.fake_server)
def file_2(self, *tokens):
"""Outputs the contents of the file, named by the SHA hash
name in the request path, directly to the client."""
method = cherrypy.request.method
if method == "HEAD":
try:
fhash = tokens[0]
except IndexError:
fhash = None
try:
fpath = self.repo.file(fhash,
pub=self._get_req_pub())
except srepo.RepositoryFileNotFoundError as e:
raise cherrypy.HTTPError(http_client.NOT_FOUND,
str(e))
except srepo.RepositoryError as e:
# Treat any remaining repository error as a 404,
# but log the error and include the real failure
# information.
cherrypy.log("Request failed: {0}".format(
str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND,
str(e))
csize, chashes = misc.compute_compressed_attrs(fhash,
file_path=fpath)
response = cherrypy.response
for i, attr in enumerate(chashes):
response.headers["X-Ipkg-Attr-{0}".format(i)] = \
"{0}={1}".format(attr, chashes[attr])
# set expiration of response to one day
self.__set_response_expires("file", 86400, 86400)
return serve_file(fpath, "application/data")
return self.file_1(*tokens)
def index_0(self, *tokens):
"""Provides an administrative interface for search indexing.
Returns no output if successful; otherwise the response body
will contain the failure details.
"""
try:
cmd = tokens[0]
except IndexError:
cmd = ""
# These commands cause the operation requested to be queued
# for later execution. This does mean that if the operation
# fails, the client won't know about it, but this is necessary
# since these are long running operations (are likely to exceed
# connection timeout limits).
try:
if cmd == "refresh":
# Update search indexes.
self.__bgtask.put(self.repo.refresh_index,
pub=self._get_req_pub())
else:
err = "Unknown index subcommand: {0}".format(
cmd)
cherrypy.log(err)
raise cherrypy.HTTPError(http_client.NOT_FOUND, err)
except queue.Full:
raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE,
"Another operation is already in progress; try "
"again later.")
def publisher_1(self, *tokens):
"""Returns a pkg(7) information datastream based on the
the request's publisher or all if not specified."""
prefix = self._get_req_pub()
pubs = []
if not prefix:
pubs = self.repo.get_publishers()
else:
try:
pub = self.repo.get_publisher(prefix)
pubs.append(pub)
except Exception as e:
# If the Publisher object creation fails, return
# a not found error to the client so it will
# treat it as an unsupported operation.
cherrypy.log("Request failed: {0}".format(
str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND,
str(e))
buf = cStringIO()
try:
p5i.write(buf, pubs)
except Exception as e:
# Treat any remaining error as a 404, but log it and
# include the real failure information.
cherrypy.log("Request failed: {0}".format(str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e))
buf.seek(0)
self.__set_response_expires("publisher", 86400*365, 86400*365)
return buf.getvalue()
def status_0(self, *tokens):
"""Return a JSON formatted dictionary containing statistics
information for the repository being served."""
self.__set_response_expires("versions", 5*60, 5*60)
dump_struct = self.repo.get_status()
try:
out = json.dumps(dump_struct, ensure_ascii=False,
indent=2, sort_keys=True)
except Exception as e:
raise cherrypy.HTTPError(http_client.NOT_FOUND, _("Unable "
"to generate statistics."))
return misc.force_bytes(out + "\n")
def search_1(self, *args, **params):
# Raise assorted errors; if not, call superclass search_1.
if self.need_nasty():
errs = [http_client.NOT_FOUND, http_client.BAD_REQUEST,
http_client.SERVICE_UNAVAILABLE]
code = random.choice(errs)
cherrypy.log("NASTY search_1: HTTP {0:d}".format(code))
raise cherrypy.HTTPError(code)
return DepotHTTP.search_1(self, *args, **params)
def feed(depot, request, response, pub):
if depot.repo.mirror:
raise cherrypy.HTTPError(http_client.NOT_FOUND,
"Operation not supported in current server mode.")
if not depot.repo.get_catalog(pub).updates:
raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE,
"No update history; unable to generate feed.")
return pkg.server.feed.handle(depot, request, response, pub)
def __handle_error(path, error):
# All errors are treated as a 404 since reverse proxies such as Apache
# don't handle 500 errors in a desirable way. For any error but a 404,
# an error is logged.
if error != http_client.NOT_FOUND:
cherrypy.log("Error encountered while processing "
"template: {0}\n".format(path), traceback=True)
raise cherrypy.NotFound()
def default_error_page(status=http_client.NOT_FOUND, message="oops",
traceback=None, version=None):
"""This function is registered as the default error page
for CherryPy errors. This sets the response headers to
be uncacheable, and then returns a HTTP response."""
response = cherrypy.response
for key in ('Cache-Control', 'Pragma'):
if key in response.headers:
del response.headers[key]
# Server errors are interesting, so let's log them. In the case
# of an internal server error, we send a 404 to the client. but
# log the full details in the server log.
if (status == http_client.INTERNAL_SERVER_ERROR or
status.startswith("500 ")):
# Convert the error to a 404 to obscure implementation
# from the client, but log the original error to the
# server logs.
error = cherrypy._cperror._HTTPErrorTemplate % \
{"status": http_client.NOT_FOUND,
"message": http_client.responses[http_client.NOT_FOUND],
"traceback": "",
"version": cherrypy.__version__}
print("Path that raised exception was {0}".format(
cherrypy.request.path_info))
print(message)
return error
else:
error = cherrypy._cperror._HTTPErrorTemplate % \
{"status": http_client.NOT_FOUND, "message": message,
"traceback": "", "version": cherrypy.__version__}
return error