python类Request()的实例源码

def test_netloc_from_env(self, protocol):
        port = 9000
        host = 'example.org'
        env = testing.create_environ(
            protocol=protocol,
            host=host,
            port=port,
            app=self.app,
            path='/hello',
            query_string=self.qs,
            headers=self.headers)

        req = Request(env)

        assert req.port == port
        assert req.netloc == '{0}:{1}'.format(host, port)

def test_range_unit(self):
        headers = {'Range': 'bytes=10-'}
        req = Request(testing.create_environ(headers=headers))
        assert req.range == (10, -1)
        assert req.range_unit == 'bytes'

        headers = {'Range': 'items=10-'}
        req = Request(testing.create_environ(headers=headers))
        assert req.range == (10, -1)
        assert req.range_unit == 'items'

        headers = {'Range': ''}
        req = Request(testing.create_environ(headers=headers))
        with pytest.raises(falcon.HTTPInvalidHeader):
            req.range_unit

        req = Request(testing.create_environ())
        assert req.range_unit is None

def on_get(self, req, resp, **kwargs):
        """GET responder.

        Captures `req`, `resp`, and `kwargs`. Also sets up a sample response.

        Args:
            req: Falcon ``Request`` instance.
            resp: Falcon ``Response`` instance.
            kwargs: URI template *name=value* pairs, if any, along with
                any extra args injected by middleware.

        """

        # Don't try this at home - classes aren't recreated
        # for every request
        self.req, self.resp, self.kwargs = req, resp, kwargs

        self.called = True
        resp.status = falcon.HTTP_200
        resp.body = self.sample_body
        resp.set_headers(self.resp_headers)

def on_get(self, req, resp, **kwargs):
        """GET responder.

        Captures `req`, `resp`, and `kwargs`. Also sets up a sample response.

        Args:
            req: Falcon ``Request`` instance.
            resp: Falcon ``Response`` instance.
            kwargs: URI template *name=value* pairs, if any, along with
                any extra args injected by middleware.

        """

        # Don't try this at home - classes aren't recreated
        # for every request
        self.req, self.resp, self.kwargs = req, resp, kwargs

        self.called = True
        resp.status = falcon.HTTP_200
        resp.body = self.sample_body
        resp.set_headers(self.resp_headers)

def test_authenticate_with_invalid_password(self):
        """
        Verify authenticate denies with a proper JSON file, Authorization header, and the wrong password.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            return_value = mock.MagicMock(etcd.EtcdResult)
            with open(self.user_config, 'r') as users_file:
                return_value.value = users_file.read()

            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            manager.get.return_value = return_value

            # Reload with the data from the mock'd Etcd
            http_basic_auth = httpbasicauth.HTTPBasicAuth()

            req = falcon.Request(
                create_environ(headers={'Authorization': 'basic YTpiCg=='}))
            resp = falcon.Response()
            self.assertRaises(
                falcon.HTTPForbidden,
                http_basic_auth.authenticate,
                req, resp)

def test_valid_certs(self):
        """
        Verify authenticate succeeds when cn matches, fails when it doesn't
        """
        self.expect_forbidden(data=self.cert, cn="other-cn")

        auth = httpauthclientcert.HTTPClientCertAuth(cn="system:master-proxy")
        req = falcon.Request(create_environ())
        req.env[SSL_CLIENT_VERIFY] = self.cert
        resp = falcon.Response()
        self.assertEqual(None, auth.authenticate(req, resp))

        # With no cn any is valid
        auth = httpauthclientcert.HTTPClientCertAuth()
        req = falcon.Request(create_environ())
        req.env[SSL_CLIENT_VERIFY] = self.cert
        resp = falcon.Response()
        self.assertEqual(None, auth.authenticate(req, resp))

def authenticate(self, req, resp):
        """
        Implements the authentication logic.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :raises: falcon.HTTPForbidden
        """
        cert = req.env.get(SSL_CLIENT_VERIFY, {})
        if cert:
            for obj in cert.get('subject', ()):
                for key, value in obj:
                    if key == 'commonName' and \
                            (not self.cn or value == self.cn):
                        return

        # Forbid by default
        raise falcon.HTTPForbidden('Forbidden', 'Forbidden')

def authenticate(self, req, resp):
        """
        Implements the authentication logic.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :raises: falcon.HTTPForbidden
        """
        user, passwd = self._decode_basic_auth(req)
        if user is not None and passwd is not None:
            if user in self._data.keys():
                self.logger.debug('User {0} found in datastore.'.format(user))
                if self.check_authentication(user, passwd):
                    return  # Authentication is good

        # Forbid by default
        raise falcon.HTTPForbidden('Forbidden', 'Forbidden')

def _decode_bearer_auth(self, req):
        """
        Decodes basic auth from the header.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :returns: token or None if empty.
        :rtype: str
        """
        self.logger.debug('header: {}'.format(req.auth))
        if req.auth is not None:
            if req.auth.lower().startswith('bearer '):
                decoded = req.auth[7:]
                self.logger.debug('Token given: {0}'.format(decoded))
                return decoded
            else:
                self.logger.debug(
                    'Did not find bearer in the Authorization '
                    'header from {0}.'.format(req.remote_addr))
        # Default meaning no user or password
        self.logger.debug('Authentication for {0} failed.'.format(
            req.remote_addr))
        return None

def on_delete(self, req, resp, name):
        """
        Handles the deletion of a Cluster.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :param name: The name of the Cluster being deleted.
        :type name: str
        """
        resp.body = '{}'

        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            store_manager.delete(Cluster.new(name=name))
            resp.status = falcon.HTTP_200
            self.logger.info(
                'Deleted cluster {0} per request.'.format(name))
        except:
            self.logger.info(
                'Deleting for non-existent cluster {0} requested.'.format(
                    name))
            resp.status = falcon.HTTP_404

def on_get(self, req, resp, name):
        """
        Handles GET requests for Cluster hosts.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :param name: The name of the Cluster being requested.
        :type name: str
        """
        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            cluster = store_manager.get(Cluster.new(name=name))
        except:
            resp.status = falcon.HTTP_404
            return

        resp.body = json.dumps(cluster.hostset)
        resp.status = falcon.HTTP_200

def on_put(self, req, resp, name, address):
        """
        Handles PUT requests for individual hosts in a Cluster.
        This adds a single host to the cluster, idempotently.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :param name: The name of the Cluster being requested.
        :type name: str
        :param address: The address of the Host being requested.
        :type address: str
        """
        try:
            util.etcd_cluster_add_host(name, address)
            resp.status = falcon.HTTP_200
        except KeyError:
            resp.status = falcon.HTTP_404

def on_delete(self, req, resp, name, address):
        """
        Handles DELETE requests for individual hosts in a Cluster.
        This removes a single host from the cluster, idempotently.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :param name: The name of the Cluster being requested.
        :type name: str
        :param address: The address of the Host being requested.
        :type address: str
        """
        try:
            util.etcd_cluster_remove_host(name, address)
            resp.status = falcon.HTTP_200
        except KeyError:
            resp.status = falcon.HTTP_404

def on_get(self, req, resp):
        """
        Handles GET requests for Hosts.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        """

        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            hosts = store_manager.list(Hosts(hosts=[]))
            if len(hosts.hosts) == 0:
                raise Exception()
            resp.status = falcon.HTTP_200
            req.context['model'] = hosts
        except Exception:
            # This was originally a "no content" but I think a 404 makes
            # more sense if there are no hosts
            self.logger.warn(
                'Store does not have any hosts. Returning [] and 404.')
            resp.status = falcon.HTTP_404
            req.context['model'] = None
            return

def on_get(self, req, resp):
        """
        Handles GET requests for Networks.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        """

        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            networks = store_manager.list(Networks(networks=[]))
            if len(networks.networks) == 0:
                raise Exception()
            resp.status = falcon.HTTP_200
            resp.body = json.dumps([
                network.name for network in networks.networks])
        except Exception:
            self.logger.warn(
                'Store does not have any networks. Returning [] and 404.')
            resp.status = falcon.HTTP_404
            req.context['model'] = None
            return

def on_get(self, req, resp, name):
        """
        Handles retrieval of an existing Network.

        :param req: Request instance that will be passed through.
        :type req: falcon.Request
        :param resp: Response instance that will be passed through.
        :type resp: falcon.Response
        :param name: The friendly name of the network.
        :type address: str
        """
        try:
            store_manager = cherrypy.engine.publish('get-store-manager')[0]
            network = store_manager.get(Network.new(name=name))
            resp.status = falcon.HTTP_200
            req.context['model'] = network
        except:
            resp.status = falcon.HTTP_404
            return

def _handle_all(self, request: falcon.Request, response: falcon.Response):
        route = (request.method.lower(), self.base_uri + request.path.rstrip("/"))
        endpoint = self._endpoints.get(route, None)
        if endpoint:
            self._set_response_attributes_from_endpoint(response, endpoint)
            endpoint.called()
        else:
            error_endpoint = Endpoint(request.method, self.base_uri + request.path).once()
            error_endpoint.called()
            self._set_response_attributes_from_endpoint(response, error_endpoint)

        self._update_statistics(request, route)

def _update_statistics(self, request: falcon.Request, route: Tuple[str, str]):
        self._statistics.setdefault(route, Statistic(route[0], route[1]))
        statistic = self._statistics.get(route)
        statistic.requests.append(RequestedParams(cookies=request.cookies,
                                                  body=request.bounded_stream.read(),
                                                  content_type=request.content_type,
                                                  files=self._get_files(request),
                                                  headers=request.headers,
                                                  query_params=request.params))

def _get_files(request: falcon.Request) -> Optional[Dict[str, bytes]]:
        files = {
            param_name: param_value.file.read()
            for param_name, param_value in request.params.items()
            if hasattr(param_value, "file")
        }

        return files if files else None

def process_request(self, req, resp):
        """Process the request before routing it.

        Args:
            req: Request object that will eventually be
                routed to an on_* responder method.
            resp: Response object that will be routed to
                the on_* responder.
        """
        pass

def process_resource(self, req, resp, resource, params):
        """Process the request after routing.

        Note:
            This method is only called when the request matches
            a route to a resource.

        Args:
            req: Request object that will be passed to the
                routed responder.
            resp: Response object that will be passed to the
                responder.
            resource: Resource object to which the request was
                routed.
            params: A dict-like object representing any additional
                params derived from the route's URI template fields,
                that will be passed to the resource's responder
                method as keyword arguments.
        """
        pass

def process_response(self, req, resp, resource):
        """Post-processing of the response (after routing).

        Args:
            req: Request object.
            resp: Response object.
            resource: Resource object to which the request was
                routed. May be None if no route was found
                for the request.
        """
        pass

def test_update_withtout_pk(self):
        """
        test how update function will handle when we are not going to pass pk value
        """
        resource = SingleResource(objects_class=None)
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {}
        }
        resp = Response()

        with self.assertRaises(Exception):
            resource.on_patch(req, resp)

def test_update_get_object(self):
        """
        Test `get_object` func
        """
        resource = SingleResource(objects_class=None)
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {'pk': 1}
        }
        resp = Response()

        resource.objects_class = FakeObjectList()
        obj = resource.get_object(req=req, resp=resp, path_params={})
        self.assertEqual(obj.pk, 1)

def test_update_when_no_expected_params_is_set(self):
        """
        Test if update func will not update param if it's not defined in expected params
        """
        resource = SingleResource(objects_class=None)
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {'pk': 1, 'name': 'TestNewName'}
        }

        def clean(data):
            return {}, {}

        resource.clean = clean

        resp = Response()

        resource.objects_class = FakeObjectList()
        resource.serialize = fake_serialize
        resource.on_patch(req, resp)
        self.assertEqual(
            {'pk': 1, 'name': 'OldName', '_saved': True},
            resp.body
        )

def test_update_params(self):
        """
        Test if param is updated and returned
        """
        resource = SingleResource(objects_class=None)
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {'pk': 1, 'name': 'TestNewName'},
        }
        resp = Response()

        resource.objects_class = FakeObjectList()
        resource.serialize = fake_serialize
        resource.on_patch(req, resp)
        self.assertEqual(
            {'pk': 1, 'name': 'TestNewName', '_saved': True},
            resp.body
        )

def test_render_response_status_200(self):
        """
        need to check if status of the response is set for 200 and
        """
        env = create_environ(path='/')
        req = Request(env)
        resp = Response()
        result = None
        BaseResource.render_response(
            result=result, req=req, resp=resp
        )
        self.assertTrue(resp.status, 200)

def test_on_get(self):
        """
        need to check if status of the response is set for 200 and
        """
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {}
        }
        req.params[mongoengine.CollectionResource.PARAM_TOTAL_COUNT] = '1'
        resp = Response()
        resource = mongoengine.CollectionResource(objects_class=Mock(return_value=[1, 2, 3]), max_limit=2)
        resource.get_object_list = Mock(return_value=[1, 2])
        resource.get_total_objects = Mock(return_value={'total_count': 3})
        resource.on_get(req=req, resp=resp)
        self.assertEqual(resp.body, {'results': [1, 2], 'total': 3, 'returned': 2})
        self.assertEqual(resp.get_header('x-api-total'), '3')
        self.assertEqual(resp.get_header('x-api-returned'), '2')

def test_on_head(self):
        """
        need to check if status of the response is set for 200 and
        """
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {}
        }
        req.params[mongoengine.CollectionResource.PARAM_TOTAL_COUNT] = '1'
        resp = Response()
        resource = mongoengine.CollectionResource(objects_class=Mock(return_value=[1, 2, 3]), max_limit=2)
        resource.get_object_list = Mock(return_value=[1, 2])
        resource.get_total_objects = Mock(return_value={'total_count': 3})
        resource.on_head(req=req, resp=resp)
        self.assertIsNot(resp.body, [1, 2, 3])
        self.assertEqual(resp.get_header('x-api-total'), '3')
        self.assertEqual(resp.get_header('x-api-returned'), '2')

def test_on_options(self):
        """
        need to check if status of the response is set for 200 and
        """
        env = create_environ(path='/')
        req = Request(env)
        req.context = {
            'doc': {}
        }
        resp = Response()
        resource = mongoengine.CollectionResource(objects_class=FakeObjectList, max_limit=2)
        resource.on_options(req=req, resp=resp)
        self.assertEqual(resp.get_header('Allow'), 'GET, HEAD, OPTIONS, POST, PUT')
        self.assertEqual(resp.body, {
            'name': 'FakeObjectList',
            'description': 'Extend list to match interface of List resource',
        })