Python falcon 模块，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',
        })

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_custom_router_takes_req_positional_argument():
    def responder(req, resp):
        resp.body = 'OK'

    class CustomRouter(object):
        def find(self, uri, req):
            if uri == '/test' and isinstance(req, falcon.Request):
                return responder, {'GET': responder}, {}, None

    router = CustomRouter()
    app = falcon.API(router=router)
    client = testing.TestClient(app)
    response = client.simulate_request(path='/test')
    assert response.content == b'OK'

def test_custom_router_takes_req_keyword_argument():
    def responder(req, resp):
        resp.body = 'OK'

    class CustomRouter(object):
        def find(self, uri, req=None):
            if uri == '/test' and isinstance(req, falcon.Request):
                return responder, {'GET': responder}, {}, None

    router = CustomRouter()
    app = falcon.API(router=router)
    client = testing.TestClient(app)
    response = client.simulate_request(path='/test')
    assert response.content == b'OK'

def test_slots_request(self):
        env = testing.create_environ()
        req = Request(env)

        try:
            req.doesnt = 'exist'
        except AttributeError:
            pytest.fail('Unable to add additional variables dynamically')

def test_legacy_middleware_called_with_correct_args(self):
        global context
        app = falcon.API(middleware=[ExecutedFirstMiddleware()])
        app.add_route(TEST_ROUTE, MiddlewareClassResource())
        client = testing.TestClient(app)

        client.simulate_request(path=TEST_ROUTE)
        assert isinstance(context['req'], falcon.Request)
        assert isinstance(context['resp'], falcon.Response)
        assert isinstance(context['resource'], MiddlewareClassResource)

def test_error_composed_before_resp_middleware_called(self):
        mw = CaptureResponseMiddleware()
        app = falcon.API(middleware=mw)
        app.add_route('/', MiddlewareClassResource())
        client = testing.TestClient(app)

        response = client.simulate_request(path='/', method='POST')
        assert response.status == falcon.HTTP_403
        assert mw.resp.status == response.status

        composed_body = json.loads(mw.resp.body)
        assert composed_body['title'] == response.status

        assert not mw.req_succeeded

        # NOTE(kgriffs): Sanity-check the other params passed to
        # process_response()
        assert isinstance(mw.req, falcon.Request)
        assert isinstance(mw.resource, MiddlewareClassResource)

def test_request_cookie_parsing():
    # testing with a github-ish set of cookies
    headers = [
        (
            'Cookie',
            """
            logged_in=no;_gh_sess=eyJzZXXzaW9uX2lkIjoiN2;
            tz=Europe/Berlin; _ga=GA1.2.332347814.1422308165;
            _gat=1;
            _octo=GH1.1.201722077.1422308165
            """
        ),
    ]

    environ = testing.create_environ(headers=headers)
    req = falcon.Request(environ)

    assert req.cookies['logged_in'] == 'no'
    assert req.cookies['tz'] == 'Europe/Berlin'
    assert req.cookies['_octo'] == 'GH1.1.201722077.1422308165'

    assert 'logged_in' in req.cookies
    assert '_gh_sess' in req.cookies
    assert 'tz' in req.cookies
    assert '_ga' in req.cookies
    assert '_gat' in req.cookies
    assert '_octo' in req.cookies

def test_bounded_stream_property_empty_body(self):
        """Test that we can get a bounded stream outside of wsgiref."""
        environ = testing.create_environ()
        req = falcon.Request(environ)

        bounded_stream = req.bounded_stream

        # NOTE(kgriffs): Verify that we aren't creating a new object
        # each time the property is called. Also ensures branch
        # coverage of the property implementation.
        assert bounded_stream is req.bounded_stream
        data = bounded_stream.read()
        assert len(data) == 0

def test_request_repr(self):
        environ = testing.create_environ()
        req = falcon.Request(environ)
        _repr = '<%s: %s %r>' % (req.__class__.__name__, req.method, req.url)
        assert req.__repr__() == _repr

def test_content_header_missing(self, client):
        environ = testing.create_environ()
        req = falcon.Request(environ)
        for header in ('Content-Type', 'Content-Length'):
            assert req.get_header(header) is None

def test_headers_as_list(self, client):
        headers = [
            ('Client-ID', '692ba466-74bb-11e3-bf3f-7567c531c7ca'),
            ('Accept', 'audio/*; q=0.2, audio/basic')
        ]

        # Unit test
        environ = testing.create_environ(headers=headers)
        req = falcon.Request(environ)

        for name, value in headers:
            assert (name.upper(), value) in req.headers.items()

        # Functional test
        client.app.add_route('/', testing.SimpleTestResource(headers=headers))
        result = client.simulate_get()

        for name, value in headers:
            assert result.headers[name] == value

def test_missing_qs(self):
        env = testing.create_environ()
        if 'QUERY_STRING' in env:
            del env['QUERY_STRING']

        # Should not cause an exception when Request is instantiated
        Request(env)

def test_nonlatin_path(self, test_path):
        # NOTE(kgriffs): When a request comes in, web servers decode
        # the path.  The decoded path may contain UTF-8 characters,
        # but according to the WSGI spec, no strings can contain chars
        # outside ISO-8859-1. Therefore, to reconcile the URI
        # encoding standard that allows UTF-8 with the WSGI spec
        # that does not, WSGI servers tunnel the string via
        # ISO-8859-1. falcon.testing.create_environ() mimics this
        # behavior, e.g.:
        #
        #   tunnelled_path = path.encode('utf-8').decode('iso-8859-1')
        #
        # falcon.Request does the following to reverse the process:
        #
        #   path = tunnelled_path.encode('iso-8859-1').decode('utf-8', 'replace')
        #

        req = Request(testing.create_environ(
            host='com',
            path=test_path,
            headers=self.headers))

        assert req.path == falcon.uri.decode(test_path)

def test_uri_https(self):
        # =======================================================
        # Default port, implicit
        # =======================================================
        req = Request(testing.create_environ(
            path='/hello', scheme='https'))
        uri = ('https://' + testing.DEFAULT_HOST + '/hello')

        assert req.uri == uri

        # =======================================================
        # Default port, explicit
        # =======================================================
        req = Request(testing.create_environ(
            path='/hello', scheme='https', port=443))
        uri = ('https://' + testing.DEFAULT_HOST + '/hello')

        assert req.uri == uri

        # =======================================================
        # Non-default port
        # =======================================================
        req = Request(testing.create_environ(
            path='/hello', scheme='https', port=22))
        uri = ('https://' + testing.DEFAULT_HOST + ':22/hello')

        assert req.uri == uri

def test_relative_uri(self):
        assert self.req.relative_uri == self.app + self.relative_uri
        assert self.req_noqs.relative_uri == self.app + self.path

        req_noapp = Request(testing.create_environ(
            path='/hello',
            query_string=self.qs,
            headers=self.headers))

        assert req_noapp.relative_uri == self.relative_uri

        req_noapp = Request(testing.create_environ(
            path='/hello/',
            query_string=self.qs,
            headers=self.headers))

        # NOTE(kgriffs): Call twice to check caching works
        assert req_noapp.relative_uri == self.relative_uri
        assert req_noapp.relative_uri == self.relative_uri

        options = RequestOptions()
        options.strip_url_path_trailing_slash = False
        req_noapp = Request(testing.create_environ(
            path='/hello/',
            query_string=self.qs,
            headers=self.headers),
            options=options)

        assert req_noapp.relative_uri == '/hello/' + '?' + self.qs

def test_client_accepts_props(self):
        headers = {'Accept': 'application/xml'}
        req = Request(testing.create_environ(headers=headers))
        assert req.client_accepts_xml
        assert not req.client_accepts_json
        assert not req.client_accepts_msgpack

        headers = {'Accept': 'application/*'}
        req = Request(testing.create_environ(headers=headers))
        assert req.client_accepts_xml
        assert req.client_accepts_json
        assert req.client_accepts_msgpack

        headers = {'Accept': 'application/json'}
        req = Request(testing.create_environ(headers=headers))
        assert not req.client_accepts_xml
        assert req.client_accepts_json
        assert not req.client_accepts_msgpack

        headers = {'Accept': 'application/x-msgpack'}
        req = Request(testing.create_environ(headers=headers))
        assert not req.client_accepts_xml
        assert not req.client_accepts_json
        assert req.client_accepts_msgpack

        headers = {'Accept': 'application/msgpack'}
        req = Request(testing.create_environ(headers=headers))
        assert not req.client_accepts_xml
        assert not req.client_accepts_json
        assert req.client_accepts_msgpack

        headers = {
            'Accept': 'application/json,application/xml,application/x-msgpack'
        }
        req = Request(testing.create_environ(headers=headers))
        assert req.client_accepts_xml
        assert req.client_accepts_json
        assert req.client_accepts_msgpack

def test_client_prefers(self):
        headers = {'Accept': 'application/xml'}
        req = Request(testing.create_environ(headers=headers))
        preferred_type = req.client_prefers(['application/xml'])
        assert preferred_type == 'application/xml'

        headers = {'Accept': '*/*'}
        preferred_type = req.client_prefers(('application/xml',
                                             'application/json'))

        # NOTE(kgriffs): If client doesn't care, "prefer" the first one
        assert preferred_type == 'application/xml'

        headers = {'Accept': 'text/*; q=0.1, application/xhtml+xml; q=0.5'}
        req = Request(testing.create_environ(headers=headers))
        preferred_type = req.client_prefers(['application/xhtml+xml'])
        assert preferred_type == 'application/xhtml+xml'

        headers = {'Accept': '3p12845j;;;asfd;'}
        req = Request(testing.create_environ(headers=headers))
        preferred_type = req.client_prefers(['application/xhtml+xml'])
        assert preferred_type is None

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

        headers = {'Range': 'bytes=10-20'}
        req = Request(testing.create_environ(headers=headers))
        assert req.range == (10, 20)

        headers = {'Range': 'bytes=-10240'}
        req = Request(testing.create_environ(headers=headers))
        assert req.range == (-10240, -1)

        headers = {'Range': 'bytes=0-2'}
        req = Request(testing.create_environ(headers=headers))
        assert req.range == (0, 2)

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

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

def test_missing_attribute_header(self):
        req = Request(testing.create_environ())
        assert req.range is None

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