python类Response()的实例源码

def append_request_id(req, resp, resource, params):
    """Append request id which got from response
    header to resource.req_ids list.
    """
    def get_headers(resp):
        if hasattr(resp, 'headers'):
            return resp.headers
        if hasattr(resp, '_headers'):
            return resp._headers
        return None

    if(isinstance(resp, Response) or
       (get_headers(resp) is not None)):
        # Extract 'x-request-id' from headers if
        # response is a Response object.
        request_id = get_headers(resp).get('x-request-id')
    else:
        # If resp is of type string or None.
        request_id = resp

    if resource.req_ids is None:
        resource.req_ids = []

    if request_id not in resource.req_ids:
        resource.req_ids.append(request_id)

def test_response_unset_cookie(client):
    resp = falcon.Response()
    resp.unset_cookie('bad')
    resp.set_cookie('bad', 'cookie', max_age=300)
    resp.unset_cookie('bad')

    morsels = list(resp._cookies.values())
    len(morsels) == 1

    bad_cookie = morsels[0]
    bad_cookie['expires'] == -1

    output = bad_cookie.OutputString()
    assert 'bad=;' in output or 'bad="";' in output

    match = re.search('expires=([^;]+)', output)
    assert match

    expiration = http_date_to_dt(match.group(1), obs_date=True)
    assert expiration < datetime.utcnow()

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_process_request_user(self):
        ''' AuthMiddleware is expected to correctly identify the headers
            added to an authenticated request by keystonemiddleware in a
            PasteDeploy configuration
        '''

        req_env = TestAuthMiddleware.ks_user_env

        project_id = str(uuid.uuid4().hex)
        req_env['HTTP_X_PROJECT_ID'] = project_id
        user_id = str(uuid.uuid4().hex)
        req_env['HTTP_X_USER_ID'] = user_id
        token = str(uuid.uuid4().hex)
        req_env['HTTP_X_AUTH_TOKEN'] = token

        middleware = AuthMiddleware()
        request = DrydockRequest(req_env)
        response = falcon.Response()

        middleware.process_request(request, response)

        assert request.context.authenticated
        assert request.context.user_id == user_id

def test_process_request_user_noauth(self):
        ''' AuthMiddleware is expected to correctly identify the headers
            added to an unauthenticated (no token, bad token) request by
            keystonemiddleware in a PasteDeploy configuration
        '''

        req_env = TestAuthMiddleware.ks_user_env

        req_env['HTTP_X_IDENTITY_STATUS'] = 'Invalid'

        middleware = AuthMiddleware()
        request = DrydockRequest(req_env)
        response = falcon.Response()

        middleware.process_request(request, response)

        assert request.context.authenticated is False

def test_on_get(self):
        time.sleep(1)
        req = None
        resp = falcon.Response()
        resp.status = None
        resp.body = None
        self.elb_resource.on_get(req, resp)
        response_body = json.loads(resp.body)
        self.assertEquals(response_body['capacity'], 100)
        self.assertEquals(response_body['requests'], 1)
        self.assertGreaterEqual(response_body['uptime'], 1)
        self.assertEquals(resp.status, falcon.HTTP_200)

        self.load_balancer.check_if_model_to_workers_map_is_empty = MagicMock(return_value = True)

        resp_1 = falcon.Response()
        resp_1.status = None
        resp_1.body = None

        self.assertRaises(falcon.HTTPInternalServerError, lambda : self.elb_resource.on_get(req, resp_1))

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_valid_user(self):
        """
        Verify authenticate works with a proper JSON in Etcd, Authorization header, and a matching user.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            # Mock the return of the Etcd get result
            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()

            # Test the call
            req = falcon.Request(
                create_environ(headers={'Authorization': 'basic YTph'}))
            resp = falcon.Response()
            self.assertEquals(
                None,
                http_basic_auth.authenticate(req, resp))

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
        """
        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 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 on_delete(self, req, resp, name):
        """
        Handles the Deletion of a 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
        """
        resp.body = '{}'
        store_manager = cherrypy.engine.publish('get-store-manager')[0]
        try:
            store_manager.delete(Network.new(name=name))
            resp.status = falcon.HTTP_200
        except Exception as error:
            self.logger.warn('{}: {}'.format(type(error), error))
            resp.status = falcon.HTTP_404

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 _set_response_attributes_from_endpoint(response: falcon.Response, endpoint: Endpoint):
        response.status = getattr(falcon, f"HTTP_{endpoint.status}")
        response.body = endpoint.body
        if endpoint.content_type:
            response.content_type = endpoint.content_type
        for header_name, header_value in endpoint.headers.items():
            response.set_header(header_name, header_value)
        for cookie_name, cookie_value in endpoint.cookies.items():
            response.set_cookie(cookie_name, cookie_value)

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')