我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用falcon.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 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')
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', })
def test_clean_check_error_raising(self): """ Check if clean function returns errors dict when `clean_param_name` raise `ParamException` """ resource = CollectionResource(objects_class=None) env = create_environ(path='/') req = Request(env) req.context = { 'doc': { 'id': 1, 'name': 'Opentopic' } } Response() def clean_name_test(self): raise ParamException('Test Error Message') resource.clean_name = clean_name_test data, errors = resource.clean(req.context['doc']) self.assertEqual(data, { 'id': 1, }) self.assertEqual(errors, {'name': ['Test Error Message']})
def test_on_put_success_result(self): """ Test if we will receive correct response """ resource = CollectionResource(objects_class=FakeObjectClass) env = create_environ(path='/') req = Request(env) req.context = { 'doc': { 'id': 1, 'name': 'Opentopic' } } resp = Response() resource.on_put(req=req, resp=resp) self.assertEqual( resp.body, {'id': 1, 'name': 'Opentopic'} )
def test_slots_response(self): resp = Response() try: resp.doesnt = 'exist' except AttributeError: pytest.fail('Unable to add additional variables dynamically')
def test_response_set_content_type_set(): resp = falcon.Response() resp._set_media_type(MEDIA_TEXT) assert resp._headers['content-type'] == MEDIA_TEXT
def test_response_set_content_type_not_set(): resp = falcon.Response() assert 'content-type' not in resp._headers
def test_append_body(self): text = 'Hello beautiful world! ' resp = falcon.Response() resp.body = '' for token in text.split(): resp.body += token resp.body += ' ' assert resp.body == text
def test_response_repr(self): resp = falcon.Response() _repr = '<%s: %s>' % (resp.__class__.__name__, resp.status) assert resp.__repr__() == _repr
def client(): app = falcon.API() app.add_route('/', CookieResource()) app.add_route('/test-convert', CookieResourceMaxAgeFloatString()) return testing.TestClient(app) # ===================================================================== # Response # =====================================================================
def test_cookies_setable(client): resp = falcon.Response() assert resp._cookies is None resp.set_cookie('foo', 'wrong-cookie', max_age=301) resp.set_cookie('foo', 'bar', max_age=300) morsel = resp._cookies['foo'] assert isinstance(morsel, Morsel) assert morsel.key == 'foo' assert morsel.value == 'bar' assert morsel['max-age'] == 300
def test_unicode_inside_ascii_range(): resp = falcon.Response() # should be ok resp.set_cookie('non_unicode_ascii_name_1', 'ascii_value') resp.set_cookie(u'unicode_ascii_name_1', 'ascii_value') resp.set_cookie('non_unicode_ascii_name_2', u'unicode_ascii_value') resp.set_cookie(u'unicode_ascii_name_2', u'unicode_ascii_value')
def test_non_ascii_name(name): resp = falcon.Response() with pytest.raises(KeyError): resp.set_cookie(name, 'ok_value')
def test_default_response_context(self): resp = Response() assert isinstance(resp.context, dict)
def test_custom_response_context(self): class MyCustomContextType(object): pass class MyCustomResponse(Response): context_type = MyCustomContextType resp = MyCustomResponse() assert isinstance(resp.context, MyCustomContextType)
def test_custom_response_context_failure(self): class MyCustomResponse(Response): context_type = False with pytest.raises(TypeError): MyCustomResponse()
def test_custom_response_context_factory(self): def create_context(resp): return {'resp': resp} class MyCustomResponse(Response): context_type = create_context resp = MyCustomResponse() assert isinstance(resp.context, dict) assert resp.context['resp'] is resp
def test_get_versions(mocker): api = VersionsResource() # Configure mocked request and response req = mocker.MagicMock(spec=falcon.Request) resp = mocker.MagicMock(spec=falcon.Response) api.on_get(req, resp) expected = api.to_json({'v1.0': {'path': '/api/v1.0', 'status': 'stable'}}) assert resp.body == expected assert resp.status == falcon.HTTP_200
def test_get_health(mocker): api = HealthResource() # Configure mocked request and response req = mocker.MagicMock(spec=falcon.Request) resp = mocker.MagicMock(spec=falcon.Response) api.on_get(req, resp) assert resp.status == falcon.HTTP_204
def process_request(self, req, resp): # type: (Request, Response) -> None """Ensure requests accept JSON or specify JSON as content type :param req: the passed request object :param resp: the passed repsonse object :raises HttpNotAcceptable: if the request does not specify "application/json" responses as acceptable :raises HttpUnsupportedContentType: if a request of a type specified by "required_methods" does not specify a content-type of "application/json" """ log.debug('JSONEnforcer.process_request(%s, %s)', req, resp) if not req.client_accepts_json: raise HTTPNotAcceptable( description=( 'This server only supports responses encoded as JSON. ' 'Please update your "Accept" header to include ' '"application/json".' ) ) if req.method in JSON_CONTENT_REQUIRED_METHODS: if (req.content_type is None or 'application/json' not in req.content_type): raise HTTPUnsupportedMediaType( description=( '%s requests must have "application/json" in their ' '"Content-Type" header.' % req.method ) )
