Python falcon 模块，HTTP_OK 实例源码

def on_put(self, req, resp, *args, **kwargs):
        status_code = falcon.HTTP_OK
        try:
            with self.session_scope(self.db_engine) as db_session:
                obj = self.get_object(req, resp, kwargs, for_update=True, db_session=db_session)

                data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
                data, errors = self.clean(data)
                if errors:
                    result = {'errors': errors}
                    status_code = falcon.HTTP_BAD_REQUEST
                else:
                    result = self.update(req, resp, data, obj, db_session)
        except (IntegrityError, ProgrammingError) as err:
            # Cases such as unallowed NULL value should have been checked before we got here (e.g. validate against
            # schema using falconjsonio) - therefore assume this is a UNIQUE constraint violation
            if isinstance(err, IntegrityError) or err.orig.args[1] == self.VIOLATION_FOREIGN_KEY:
                raise HTTPConflict('Conflict', 'Unique constraint violated')
            else:
                raise

        self.render_response(result, req, resp, status_code)

def on_put(self, req, resp, *args, **kwargs):
        """
        Updates a single record.
        This should set all missing fields to default values, but we're not going to be so strict.

        :param req: Falcon request
        :type req: falcon.request.Request

        :param resp: Falcon response
        :type resp: falcon.response.Response
        """
        obj = self.get_object(req, resp, kwargs, for_update=True)

        data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
        data, errors = self.clean(data)
        if errors:
            result = {'errors': errors}
            status_code = falcon.HTTP_BAD_REQUEST
        else:
            result = self.update(req, resp, data, obj)
            status_code = falcon.HTTP_OK
        self.render_response(result, req, resp, status_code)

def test_parse_form_as_params(client):

    class Resource:

        def on_post(self, req, resp, **kwargs):
            assert req.get_param('simple') == 'ok'
            assert req.get_param('afile').file.read() == b'filecontent'
            assert req.get_param('afile').filename == 'afile.txt'
            resp.body = 'parsed'
            resp.content_type = 'text/plain'

    application.add_route('/route', Resource())

    resp = client.post('/route', data={'simple': 'ok'},
                       files={'afile': ('filecontent', 'afile.txt')})
    assert resp.status == falcon.HTTP_OK
    assert resp.body == 'parsed'

def test_with_binary_file(client):
    here = os.path.dirname(os.path.realpath(__file__))
    filepath = os.path.join(here, 'image.jpg')
    image = open(filepath, 'rb')

    class Resource:

        def on_post(self, req, resp, **kwargs):
            resp.data = req.get_param('afile').file.read()
            resp.content_type = 'image/jpg'

    application.add_route('/route', Resource())

    resp = client.post('/route', data={'simple': 'ok'},
                       files={'afile': image})
    assert resp.status == falcon.HTTP_OK
    image.seek(0)
    assert resp.body == image.read()

def test_parse_non_ascii_filename_in_headers(client):

    class Resource:

        def on_post(self, req, resp, **kwargs):
            assert req.get_param('afile').file.read() == b'name,code\nnom,2\n'
            assert req.get_param('afile').filename == 'Na%C3%AFve%20file.txt'
            resp.body = 'parsed'
            resp.content_type = 'text/plain'

    application.add_route('/route', Resource())

    # Simulate browser sending non ascii filename.
    body = ('--boundary\r\nContent-Disposition: '
            'form-data; name="afile"; filename*=utf-8\'\'Na%C3%AFve%20file.txt'
            '\r\nContent-Type: text/csv\r\n\r\nname,code\nnom,2\n\r\n'
            '--boundary--\r\n')
    headers = {'Content-Type': 'multipart/form-data; boundary=boundary'}
    resp = client.post('/route', body=body, headers=headers)
    assert resp.status == falcon.HTTP_OK
    assert resp.body == 'parsed'

def test_list_images(client):
    doc = {
        'images': [
            {
                'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
            }
        ]
    }

    response = client.simulate_get('/images')
    result_doc = msgpack.unpackb(response.content, encoding='utf-8')

    assert result_doc == doc
    assert response.status == falcon.HTTP_OK


# With clever composition of fixtures, we can observe what happens with
# the mock injected into the image resource.

def test_should_report_healthy_if_kafka_healthy(self, kafka_check):
        kafka_check.healthcheck.return_value = healthcheck.CheckResult(True,
                                                                       'OK')
        self.resource._kafka_check = kafka_check

        ret = self.simulate_request(ENDPOINT,
                                    headers={
                                        'Content-Type': 'application/json'
                                    },
                                    decode='utf8',
                                    method='GET')
        self.assertEqual(falcon.HTTP_OK, self.srmock.status)

        ret = json.loads(ret)
        self.assertIn('kafka', ret)
        self.assertEqual('OK', ret.get('kafka'))

def render_response(result, req, resp, status=falcon.HTTP_OK):
        """
        :param result: Data to be returned in the response

        :param req: Falcon request
        :type req: falcon.request.Request

        :param resp: Falcon response
        :type resp: falcon.response.Response

        :param status: HTTP status code
        :type status: str
        """
        resp.body = result
        resp.status = status

def test_parse_multiple_values(client):

    class Resource:

        def on_post(self, req, resp, **kwargs):
            assert req.get_param_as_list('multi') == ['1', '2']
            resp.body = 'parsed'
            resp.content_type = 'text/plain'

    application.add_route('/route', Resource())

    resp = client.post('/route', data={'multi': ['1', '2']},
                       files={'afile': ('filecontent', 'afile.txt')})
    assert resp.status == falcon.HTTP_OK
    assert resp.body == 'parsed'

def test_get_folder_tree(self, client):
        '''Test we can serialize a directory tree structure
        '''
        # Get path within designated storage directory
        fid = id_from_path(settings['MODEL_DATA_DIR'])

        response = client.simulate_get('/files/tree/{}'.format(fid))
        assert response.status == falcon.HTTP_OK
        assert response.text == json.dumps(directory_tree_serializer(settings['MODEL_DATA_DIR']))

def on_get(self, req, resp, fid):
        ''' Return a JSON representation of the directory tree. The JSON response has the
        following attributes:

            - ``type``: file or folder
            - ``name``: The base name of the file/folder
            - ``path``: Absolute path to the object.
            - ``id``: URL-safe base64 encoding of the ``path``
            - ``children``: Only present if ``type`` is ``folder``. A list of all
                children of this folder, each having the same representation.

        Args:
            fid (str): the base64 encoded url safe ID for the path to the root folder

        Example::

            http localhost:8000/files/tree/{fid}

        '''
        try:
            self._data_store.validate_fid(fid)
            resp.status = falcon.HTTP_OK
            path = serializers.path_from_id(fid)
            resp.body = json.dumps(serializers.directory_tree_serializer(path))
        except PermissionError as error:
            resp.status = falcon.HTTP_BAD_REQUEST
            resp.body = str(error)

def on_get(self, req, resp, name):
        '''Retrieve a JSON representation of a single ``Model`` based on its'
        name.

        Example::

            http localhost:8000/models/{name}
        '''
        model = self.get_object(name)
        resp.status = falcon.HTTP_OK
        resp.body = model.to_json()