def test_call(self):
class FakeController(object):
def index(self, shirt, pants=None):
return shirt, pants
resource = wsgi.Resource(FakeController(), None, None)
def dispatch(obj, *args, **kwargs):
if isinstance(obj, wsgi.JSONRequestDeserializer):
return []
if isinstance(obj, wsgi.JSONResponseSerializer):
raise webob.exc.HTTPForbidden()
with mock.patch('glare.common.wsgi.Resource.dispatch',
side_effect=dispatch):
request = wsgi.Request.blank('/')
response = resource.__call__(request)
self.assertIsInstance(response, webob.exc.HTTPForbidden)
self.assertEqual(http.FORBIDDEN, response.status_code)
python类FORBIDDEN的实例源码
def catalog_1(self, *tokens):
"""Outputs the contents of the specified catalog file, using the
name in the request path, directly to the client."""
try:
name = tokens[0]
except IndexError:
raise cherrypy.HTTPError(http_client.FORBIDDEN,
_("Directory listing not allowed."))
try:
fpath = self.repo.catalog_1(name,
pub=self._get_req_pub())
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))
self.__set_response_expires("catalog", 86400, 86400)
return serve_file(fpath, "text/plain; charset=utf-8")
def _network_ping(self):
try:
# Ping the versions URL, rather than the default /
# so that we don't initialize the BUI code yet.
repourl = urljoin(self.url, "versions/0")
# Disable SSL peer verification, we just want to check
# if the depot is running.
urlopen(repourl,
context=ssl._create_unverified_context())
except HTTPError as e:
if e.code == http_client.FORBIDDEN:
return True
return False
except URLError:
return False
return True
def test_access_error(self):
self.request.return_value.status_code = http_client.FORBIDDEN
self.request.return_value.json.side_effect = ValueError('no json')
with self.assertRaisesRegex(exceptions.AccessError,
'unknown error') as cm:
self.conn._op('GET', 'http://foo.bar')
exc = cm.exception
self.assertEqual(http_client.FORBIDDEN, exc.status_code)
def raise_for_response(method, url, response):
"""Raise a correct error class, if needed."""
if response.status_code < http_client.BAD_REQUEST:
return
elif response.status_code == http_client.NOT_FOUND:
raise ResourceNotFoundError(method, url, response)
elif response.status_code == http_client.BAD_REQUEST:
raise BadRequestError(method, url, response)
elif response.status_code in (http_client.UNAUTHORIZED,
http_client.FORBIDDEN):
raise AccessError(method, url, response)
elif response.status_code >= http_client.INTERNAL_SERVER_ERROR:
raise ServerSideError(method, url, response)
else:
raise HTTPError(method, url, response)
def test_failure(self):
import mock
import requests
from six.moves import http_client
response = requests.Response()
response.status_code = http_client.FORBIDDEN
to_patch = 'ci_diff_helper._github._rate_limit_info'
with mock.patch(to_patch) as patched:
with self.assertRaises(requests.HTTPError):
self._call_function_under_test(response)
patched.assert_called_once_with(response)
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_resource_not_authorized(self):
class Controller(object):
def index(self, req):
raise exception.Forbidden()
req = webob.Request.blank('/tests')
app = fakes.TestRouter(Controller())
response = req.get_response(app)
self.assertEqual(response.status_int, http.FORBIDDEN)
def catalog_1(self, *tokens):
"""Outputs the contents of the specified catalog file, using the
name in the request path, directly to the client."""
try:
name = tokens[0]
except IndexError:
raise cherrypy.HTTPError(http_client.FORBIDDEN,
_("Directory listing not allowed."))
try:
fpath = self.repo.catalog_1(name,
pub=self._get_req_pub())
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))
# NASTY
# Stash catalog entry for later use.
# Toss out the list if it's larger than 1024 items.
if len(self.requested_catalogs) > 1024:
self.requested_catalogs = [fpath]
else:
self.requested_catalogs.append(fpath)
# NASTY
# Send the wrong catalog
if self.need_nasty_3():
cherrypy.log("NASTY catalog_1: wrong catalog file")
badpath = random.choice(self.requested_catalogs)
return serve_file(badpath, "text/plain; charset=utf-8")
# NASTY
return self.nasty_serve_file(fpath, "text/plain; charset=utf-8")
def _network_ping(self):
try:
urlopen(urljoin(self.url, "syspub/0"))
except HTTPError as e:
if e.code == http_client.FORBIDDEN:
return True
return False
except URLError:
return False
return True
def check_resp_status_and_retry(resp, image_id, url):
# Note(Jesse): This branch sorts errors into those that are permanent,
# those that are ephemeral, and those that are unexpected.
if resp.status in (httplib.BAD_REQUEST, # 400
httplib.UNAUTHORIZED, # 401
httplib.PAYMENT_REQUIRED, # 402
httplib.FORBIDDEN, # 403
httplib.METHOD_NOT_ALLOWED, # 405
httplib.NOT_ACCEPTABLE, # 406
httplib.PROXY_AUTHENTICATION_REQUIRED, # 407
httplib.CONFLICT, # 409
httplib.GONE, # 410
httplib.LENGTH_REQUIRED, # 411
httplib.PRECONDITION_FAILED, # 412
httplib.REQUEST_ENTITY_TOO_LARGE, # 413
httplib.REQUEST_URI_TOO_LONG, # 414
httplib.UNSUPPORTED_MEDIA_TYPE, # 415
httplib.REQUESTED_RANGE_NOT_SATISFIABLE, # 416
httplib.EXPECTATION_FAILED, # 417
httplib.UNPROCESSABLE_ENTITY, # 422
httplib.LOCKED, # 423
httplib.FAILED_DEPENDENCY, # 424
httplib.UPGRADE_REQUIRED, # 426
httplib.NOT_IMPLEMENTED, # 501
httplib.HTTP_VERSION_NOT_SUPPORTED, # 505
httplib.NOT_EXTENDED, # 510
):
raise PluginError("Got Permanent Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
# Nova service would process the exception
elif resp.status == httplib.NOT_FOUND: # 404
exc = XenAPI.Failure('ImageNotFound')
raise exc
# NOTE(nikhil): Only a sub-set of the 500 errors are retryable. We
# optimistically retry on 500 errors below.
elif resp.status in (httplib.REQUEST_TIMEOUT, # 408
httplib.INTERNAL_SERVER_ERROR, # 500
httplib.BAD_GATEWAY, # 502
httplib.SERVICE_UNAVAILABLE, # 503
httplib.GATEWAY_TIMEOUT, # 504
httplib.INSUFFICIENT_STORAGE, # 507
):
raise RetryableError("Got Ephemeral Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
else:
# Note(Jesse): Assume unexpected errors are retryable. If you are
# seeing this error message, the error should probably be added
# to either the ephemeral or permanent error list.
raise RetryableError("Got Unexpected Error response [%i] while "
"uploading image [%s] to glance [%s]"
% (resp.status, image_id, url))
def manifest_0(self, *tokens):
"""The request is an encoded pkg FMRI. If the version is
specified incompletely, we return an error, as the client is
expected to form correct requests based on its interpretation
of the catalog and its image policies."""
try:
pubs = self.repo.publishers
except Exception as e:
cherrypy.log("Request failed: {0}".format(e))
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
# A broken proxy (or client) has caused a fully-qualified FMRI
# to be split up.
comps = [t for t in tokens]
if not comps:
raise cherrypy.HTTPError(http_client.FORBIDDEN,
_("Directory listing not allowed."))
if len(comps) > 1 and comps[0] == "pkg:" and comps[1] in pubs:
# Only one slash here as another will be added below.
comps[0] += "/"
# Parse request into FMRI component and decode.
try:
# If more than one token (request path component) was
# specified, assume that the extra components are part
# of the fmri and have been split out because of bad
# proxy behaviour.
pfmri = "/".join(comps)
pfmri = fmri.PkgFmri(pfmri, None)
fpath = self.repo.manifest(pfmri,
pub=self._get_req_pub())
except (IndexError, fmri.FmriError) as e:
raise cherrypy.HTTPError(http_client.BAD_REQUEST, 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))
# Send manifest
self.__set_response_expires("manifest", 86400*365, 86400*365)
return serve_file(fpath, "text/plain; charset=utf-8")
def manifest_0(self, *tokens):
"""The request is an encoded pkg FMRI. If the version is
specified incompletely, we return an error, as the client is
expected to form correct requests based on its interpretation
of the catalog and its image policies."""
try:
pubs = self.repo.publishers
except Exception as e:
cherrypy.log("Request failed: {0}".format(e))
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
# A broken proxy (or client) has caused a fully-qualified FMRI
# to be split up.
comps = [t for t in tokens]
if not comps:
raise cherrypy.HTTPError(http_client.FORBIDDEN,
_("Directory listing not allowed."))
if len(comps) > 1 and comps[0] == "pkg:" and comps[1] in pubs:
# Only one slash here as another will be added below.
comps[0] += "/"
# Parse request into FMRI component and decode.
try:
# If more than one token (request path component) was
# specified, assume that the extra components are part
# of the fmri and have been split out because of bad
# proxy behaviour.
pfmri = "/".join(comps)
pfmri = fmri.PkgFmri(pfmri, None)
fpath = self.repo.manifest(pfmri,
pub=self._get_req_pub())
except (IndexError, fmri.FmriError) as e:
raise cherrypy.HTTPError(http_client.BAD_REQUEST, 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))
# NASTY
# Stash manifest entry for later use.
# Toss out the list if it's larger than 1024 items.
if len(self.requested_manifests) > 1024:
self.requested_manifests = [fpath]
else:
self.requested_manifests.append(fpath)
# NASTY
# Send the wrong manifest
if self.need_nasty_3():
cherrypy.log("NASTY manifest_0: serve wrong mfst")
badpath = random.choice(self.requested_manifests)
return serve_file(badpath, "text/plain; charset=utf-8")
# NASTY
# Call a misbehaving serve_file
return self.nasty_serve_file(fpath, "text/plain; charset=utf-8")
def test_disable_ops(self):
"""Verify that disable-ops works as expected."""
# For this disabled case, /catalog/1/ should return
# a NOT_FOUND error.
self.__dc.set_disable_ops(["catalog/1"])
self.__dc.set_port(self.next_free_port)
self.__dc.start()
durl = self.__dc.get_depot_url()
try:
urlopen("{0}/catalog/1/".format(durl))
except HTTPError as e:
self.assertEqual(e.code, http_client.NOT_FOUND)
self.__dc.stop()
# For this disabled case, all /catalog/ operations should return
# a NOT_FOUND error.
self.__dc.set_disable_ops(["catalog"])
self.__dc.set_port(self.next_free_port)
self.__dc.start()
durl = self.__dc.get_depot_url()
for ver in (0, 1):
try:
urlopen("{0}/catalog/{1:d}/".format(durl, ver))
except HTTPError as e:
self.assertEqual(e.code, http_client.NOT_FOUND)
self.__dc.stop()
# In the normal case, /catalog/1/ should return
# a FORBIDDEN error.
self.__dc.unset_disable_ops()
self.__dc.start()
durl = self.__dc.get_depot_url()
try:
urlopen("{0}/catalog/1/".format(durl))
except HTTPError as e:
self.assertEqual(e.code, http_client.FORBIDDEN)
self.__dc.stop()
# A bogus operation should prevent the depot from starting.
self.__dc.set_disable_ops(["no_such_op/0"])
self.__dc.start_expected_fail()
self.assertFalse(self.__dc.is_alive())
