Python six.moves.http_client 模块,FORBIDDEN 实例源码

我们从Python开源项目中,提取了以下14个代码示例,用于说明如何使用six.moves.http_client.FORBIDDEN

项目:glare    作者:openstack    | 项目源码 | 文件源码
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)
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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")
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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
项目:sushy    作者:openstack    | 项目源码 | 文件源码
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)
项目:sushy    作者:openstack    | 项目源码 | 文件源码
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)
项目:ci-diff-helper    作者:dhermes    | 项目源码 | 文件源码
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)
项目:masakari    作者:openstack    | 项目源码 | 文件源码
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)
项目:masakari    作者:openstack    | 项目源码 | 文件源码
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)
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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")
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
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))
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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")
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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")
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
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())