我们从Python开源项目中,提取了以下14个代码示例,用于说明如何使用six.moves.http_client.FORBIDDEN。
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)
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())